123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301 |
- # -*- coding: utf-8 -*-
- from enum import Enum
- from typing import Dict, List
- import arrow
- import numpy as np
- import pandas as pd
- from httpx import AsyncClient, URL
- from loguru import logger
- from app.core.config import settings
- from app.services.service import Service
- from app.utils.date import get_time_str, TIME_FMT
- from app.utils.math import round_half_up
- class Season(str, Enum):
- cooling = 'Cooling'
- heating = 'Warm'
- transition = 'Transition'
- class SpaceInfoService(Service):
- def __init__(
- self,
- client: AsyncClient,
- project_id: str,
- space_id: str,
- server_settings=settings
- ) -> None:
- super(SpaceInfoService, self).__init__(client)
- self._project_id = project_id
- self._space_id = space_id
- self._base_url = URL(server_settings.TRANSFER_HOST)
- self._now_time = get_time_str()
- def _common_parameters(self) -> Dict:
- return {'projectId': self._project_id, 'spaceId': self._space_id}
- async def is_customized(self) -> bool:
- url = self._base_url.join('duoduo-service/custom-service/custom/timetarget')
- time_str = arrow.get(arrow.get(self._now_time, TIME_FMT).shift(minutes=15).timestamp
- // 900 * 900).strftime('%Y%m%d%H%M%S')
- params = {
- 'projectId': self._project_id,
- 'objectId': self._space_id,
- 'timepoint': time_str,
- }
- raw_info = await self._get(url, params)
- flag = False
- if raw_info.get('data'):
- flag = True
- return flag
- async def is_temporary(self) -> bool:
- url = self._base_url.join('duoduo-service/transfer/environment/temp/target')
- params = self._common_parameters()
- params.update({'time': self._now_time})
- raw_info = await self._get(url, params)
- flag = False
- if raw_info.get('flag') == 1:
- flag = True
- return flag
- async def get_feedback(self, wechat_time: str) -> Dict:
- url = self._base_url.join('duoduo-service/transfer/environment/feedbackCount')
- params = self._common_parameters()
- params.update({'time': wechat_time})
- raw_info = await self._get(url, params)
- meaning_dict = {
- 'Id1': 'a little cold',
- 'Id2': 'so cold',
- 'Id3': 'a little hot',
- 'Id4': 'so hot',
- 'Id5': 'noisy or blowy',
- 'Id6': 'so stuffy',
- 'Id7': 'more sunshine',
- 'Id8': 'less sunshine',
- 'Id9': 'send a repairman',
- 'Id10': 'switch off',
- 'Id11': 'nice',
- 'Id12': 'switch on',
- }
- feedback_dic = {meaning_dict.get(k): v for k, v in raw_info.items() if k != 'result'}
- return feedback_dic
- async def get_custom_target(self) -> Dict[str, pd.DataFrame]:
- url = self._base_url.join('duoduo-service/transfer/environment/normalAndPreDayTarget')
- params = self._common_parameters()
- params.update({'date': arrow.get(self._now_time, TIME_FMT).date().strftime('%Y%m%d')})
- raw_info = await self._get(url, params)
- try:
- pre_target_df = pd.DataFrame(raw_info.get('preTargets'))
- pre_target_df.set_index('time', inplace=True)
- except (KeyError, TypeError):
- pre_target_df = pd.DataFrame()
- try:
- normal_target_df = pd.DataFrame(raw_info.get('normalTargets'))
- normal_target_df.set_index('time', inplace=True)
- except (KeyError, TypeError):
- normal_target_df = pd.DataFrame()
- return {
- 'pre_targets': pre_target_df,
- 'normal_targets': normal_target_df
- }
- async def get_current_temperature_target(self) -> float:
- targets = await self.get_custom_target()
- if len(targets.get('pre_targets')) > 0:
- current_targets = targets.get('pre_targets').append(targets.get('normal_targets'))
- else:
- current_targets = targets.get('normal_targets')
- temp = arrow.get(self._now_time, TIME_FMT).shift(minutes=15).timestamp // (15 * 60) * (15 * 60)
- next_quarter_minutes = arrow.get(temp).time().strftime('%H%M%S')
- try:
- current_lower_target = current_targets['temperatureMin'].loc[next_quarter_minutes]
- current_upper_target = current_targets['temperatureMax'].loc[next_quarter_minutes]
- except KeyError:
- current_lower_target, current_upper_target = np.NAN, np.NAN
- return round_half_up((current_lower_target + current_upper_target) / 2, 2)
- async def env_database_set(self, form: str, value: float) -> None:
- url = self._base_url.join('duoduo-service/transfer/environment/hispoint/set')
- params = self._common_parameters()
- time_str = arrow.get(arrow.get(self._now_time, TIME_FMT).timestamp // 900 * 900).strftime('%Y%m%d%H%M%S')
- params.update({'time': time_str, 'type': form, 'value': value})
- await self._get(url, params)
- async def env_database_get(self) -> Dict[str, pd.DataFrame]:
- url = self._base_url.join('duoduo-service/transfer/environment/hispoint/get')
- params = self._common_parameters()
- params.update({'date': arrow.get(self._now_time, TIME_FMT).date().strftime('%Y%m%d')})
- raw_info = await self._get(url, params)
- result = {}
- if raw_info.get('result') == 'success':
- for k, v in raw_info.items():
- if k != 'result':
- if len(v) > 0:
- temp = {}
- data = np.array(v)
- temp.update({'timestamp': data[:, 0]})
- temp.update({'value': data[:, 1].astype(np.float)})
- result.update({k: pd.DataFrame(temp)})
- else:
- result.update({k: pd.DataFrame()})
- return result
- async def set_custom_target(self, form: str, target_value: Dict[str, List[float]], flag: str = '1') -> None:
- url = self._base_url.join('duoduo-service/transfer/environment/target/setting')
- params = {
- 'projectId': self._project_id,
- 'spaceId': self._space_id,
- 'timepoint': self._now_time,
- 'type': form,
- 'flag': flag
- }
- await self._post(url, params=params, payload=target_value)
- async def set_temporary_custom(self) -> None:
- url = self._base_url.join('duoduo-service/transfer/environment/setServiceFlag')
- params = self._common_parameters()
- params.update({'time': self._now_time})
- await self._get(url, params)
- async def get_equipment(self) -> List[dict]:
- url = self._base_url.join('duoduo-service/object-service/object/equipment/findForServe')
- params = self._common_parameters()
- raw_info = await self._post(url, params)
- result = []
- for eq in raw_info.get('data'):
- result.append({'id': eq.get('id'), 'category': eq.get('equipmentCategory')})
- return result
- class Duoduo(Service):
- def __init__(self, client: AsyncClient, project_id: str, server_settings=settings):
- super(Duoduo, self).__init__(client)
- self._project_id = project_id
- self._base_url = URL(server_settings.TRANSFER_HOST)
- self._now_time = get_time_str()
- async def get_season(self) -> Season:
- url = self._base_url.join('duoduo-service/transfer/environment/getSeasonType')
- params = {
- 'projectId': self._project_id,
- 'date': self._now_time,
- }
- raw_info = await self._get(url, params)
- return Season(raw_info.get('data'))
- async def get_fill_count(self) -> Dict:
- url = self._base_url.join('duoduo-service/review-service/space/report/quarter/query')
- payload = {
- 'criteria': {
- 'projectId': self._project_id,
- 'date': arrow.get(self._now_time, TIME_FMT).date().strftime('%Y%m%d')
- },
- 'orders': [
- {
- 'column': 'time',
- 'asc': False
- }
- ],
- 'page': 1,
- 'size': 1
- }
- raw_info = await self._post(url, payload=payload)
- try:
- result = raw_info.get('content')[-1]
- except (IndexError, TypeError):
- result = {}
- return result
- async def get_space_by_equipment(self, equipment_id: str) -> List[dict]:
- url = self._base_url.join('duoduo-service/object-service/object/space/findForServe')
- params = {
- 'projectId': self._project_id,
- 'objectId': equipment_id
- }
- raw_info = await self._post(url, params)
- result = []
- for sp in raw_info.get('data'):
- if sp.get('isControlled'):
- result.append({'id': sp.get('id')})
- return result
- async def get_system_by_equipment(self, equipment_id: str) -> List:
- url = self._base_url.join('duoduo-service/object-service/object/system/findForCompose')
- params = {
- 'projectId': self._project_id,
- 'equipmentId': equipment_id
- }
- raw_info = await self._post(url, params)
- system_list = []
- for sy in raw_info.get('data'):
- system_list.append({'id': sy.get('id')})
- return system_list
- async def get_day_type(self) -> Dict:
- url = self._base_url.join('duoduo-service/custom-service/custom/getDateInfo')
- params = {
- 'projectId': self._project_id,
- 'date': arrow.get(self._now_time, TIME_FMT).date().strftime('%Y%m%d')
- }
- raw_info = await self._get(url, params)
- result = {
- 'day_type': raw_info.get('dayType'),
- 'season': raw_info.get('seasonType')
- }
- return result
- async def query_device_virtual_data(self, device_id: str, info_code: str) -> float:
- url = self._base_url.join('duoduo-service/review-service/equipment/order/query')
- payload = {
- 'criteria': {
- 'projectId': self._project_id,
- 'objectId': device_id,
- 'date': arrow.get(self._now_time, TIME_FMT).date().strftime('%Y%m%d'),
- 'funcId': info_code
- }
- }
- raw_info = await self._post(url, payload=payload)
- try:
- latest_data = raw_info.get('data')[-1].get('value')
- latest_time = raw_info.get('data')[-1].get('realTime')
- if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT):
- value = np.NAN
- else:
- value = latest_data
- except (KeyError, TypeError, IndexError):
- value = np.NAN
- return value
|