# -*- 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 app.core.config import settings from app.services.service import Service from app.utils.date import get_time_str, TIME_FMT 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 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 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) -> pd.DataFrame: url = self._base_url.join('duoduo-service/transfer/environment/target') 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: custom_target_df = pd.DataFrame(raw_info.get('data')) custom_target_df.set_index('time', inplace=True) except KeyError: custom_target_df = pd.DataFrame() return custom_target_df async def get_current_temperature_target(self) -> object: current_targets = await self.get_custom_target() if len(current_targets) > 0: 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 = 0.0, 0.0 else: current_lower_target, current_upper_target = np.NAN, np.NAN return (current_lower_target + current_upper_target) / 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 EquipmentInfoService(Service): def __init__(self, client: AsyncClient, project_id: str, server_settings=settings): super(EquipmentInfoService, 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_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