# -*- coding: utf-8 -*- from enum import Enum from typing import Dict, Tuple import arrow import numpy as np 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 InfoCode(str, Enum): temperature = 'Tdb' co2 = 'CO2' hcho = 'HCHO' pm2d5 = 'PM2d5' humidity = 'RH' supply_air_flow = 'SupplyAirFlow' supply_air_temperature = 'SupplyAirTemp' fan_speed = 'FanGear' class DataPlatformService(Service): def __init__( self, client: AsyncClient, project_id: str, server_settings=settings ): super(DataPlatformService, self).__init__(client) self._project_id = project_id self._base_url = URL(server_settings.PLATFORM_HOST) self._now_time = get_time_str() self._secret = server_settings.PLATFORM_SECRET def _common_parameters(self) -> Dict: return {'projectId': self._project_id, 'secret': self._secret} async def get_realtime_data(self, code: InfoCode, object_id: str) -> float: url = self._base_url.join('data-platform-3/hisdata/query_by_obj') params = self._common_parameters() start_time = get_time_str(60 * 60, flag='ago') payload = { 'criteria': { 'id': object_id, 'code': code.value, 'receivetime': { '$gte': start_time, '$lte': self._now_time, } } } raw_info = await self._post(url, params, payload) # value = np.NAN # if raw_info.get('Content'): try: latest_data = raw_info.get('Content')[-1].get('data') latest_time = raw_info.get('Content')[-1].get('receivetime') if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT): logger.info(f'delayed data - {object_id}: ({latest_time}, {latest_data})') value = round_half_up(latest_data, 2) except KeyError: value = np.NAN return value async def get_realtime_temperature(self, space_id: str) -> float: return await self.get_realtime_data(InfoCode.temperature, space_id) async def get_realtime_co2(self, space_id: str) -> float: return await self.get_realtime_data(InfoCode.co2, space_id) async def get_realtime_hcho(self, space_id: str) -> float: return await self.get_realtime_data(InfoCode.hcho, space_id) async def get_realtime_pm2d5(self, space_id: str) -> float: return await self.get_realtime_data(InfoCode.pm2d5, space_id) async def get_realtime_humidity(self, space_id: str) -> float: return await self.get_realtime_data(InfoCode.humidity, space_id) async def get_realtime_supply_air_flow(self, equipment_id: str) -> float: return await self.get_realtime_data(InfoCode.supply_air_flow, equipment_id) async def get_realtime_supply_air_temperature(self, equipment_id: str) -> float: return await self.get_realtime_data(InfoCode.supply_air_temperature, equipment_id) async def get_fan_speed(self, equipment_id: str) -> float: return await self.get_realtime_data(InfoCode.fan_speed, equipment_id) async def get_static_info(self, code: str, object_id: str): url = self._base_url.join('data-platform-3/object/batch_query') params = self._common_parameters() payload = { 'customInfo': True, 'criterias': [ { 'id': object_id } ] } raw_info = await self._post(url, params, payload) try: info = raw_info['Content'][0]['infos'][code] except KeyError as e: logger.error(f'id: {object_id}, details: {e}') info = None return info async def get_air_flow_limit(self, equipment_id: str) -> Tuple[float, float]: lower = await self.get_static_info('MinAirFlow', equipment_id) upper = await self.get_static_info('MaxAirFlow', equipment_id) if not lower: lower = 150.0 if not upper: upper = 2000.0 return lower, upper