123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300 |
- # -*- coding: utf-8 -*-
- from enum import Enum
- from typing import Dict, List, Optional, 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_flow_set = 'SupplyAirFlowSet'
- supply_air_temperature = 'SupplyAirTemp'
- supply_air_temperature_set = 'SupplyAirTempSet'
- fan_speed = 'FanGear'
- fan_speed_set = 'FanGearSet'
- fan_freq = 'FanFreq'
- fan_freq_set = 'FanFreqSet'
- supply_static_press = 'SupplyStaticPress'
- supply_static_press_set = 'SupplyStaticPressSet'
- running_status = 'RunStatus'
- cloud_status = 'InCloudStatus'
- equip_switch_set = 'EquipSwitchSet'
- return_air_temperature = 'ReturnAirTemp'
- chill_water_valve_opening_set = 'ChillWaterValveOpeningSet'
- hot_water_valve_opening_set = 'HotWaterValveOpeningSet'
- water_valve_switch_set = 'WaterValveSwitchSet'
- in_cloud_set = 'InCloudSet'
- work_mode_set = 'WorkModeSet'
- supply_temperature = 'SupplyTemp'
- water_out_temperature = 'WaterOutTemp'
- water_in_temperature = 'WaterInTemp'
- valve_opening = 'ValveOpening'
- 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)
- 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 (IndexError, KeyError, TypeError):
- value = np.NAN
- return value
- async def get_duration(self, code: InfoCode, object_id: str, duration: int) -> List[Dict]:
- url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
- params = self._common_parameters()
- start_time = get_time_str(duration, 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)
- try:
- content = raw_info.get('Content')
- latest_time = content[-1].get('receivetime')
- if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT):
- result = []
- logger.info(f'delayed data - {object_id}: ({latest_time})')
- else:
- result = [
- {
- 'timestamp': item['receivetime'],
- 'value': item['data']
- }
- for item in content
- ]
- except (KeyError, TypeError, IndexError):
- result = []
- return result
- async def get_past_data(self, code: InfoCode, object_id: str, interval: int) -> float:
- """
- Query past data from data platform.
- :param code: Info code
- :param object_id:
- :param interval: time interval(seconds) from now to past
- :return: a past value
- """
- url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
- params = self._common_parameters()
- start_time = get_time_str(60 * 60 + interval, flag='ago')
- end_time = get_time_str(interval, flag='ago')
- payload = {
- 'criteria': {
- 'id': object_id,
- 'code': code.value,
- 'receivetime': {
- '$gte': start_time,
- '$lte': end_time,
- }
- }
- }
- raw_info = await self._post(url, params, payload)
- 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(end_time, TIME_FMT):
- logger.info(f'delayed data - {object_id}: ({latest_time}, {latest_data})')
- value = round_half_up(latest_data, 2)
- except (KeyError, IndexError, TypeError):
- value = np.NAN
- except TypeError:
- value = -1.0
- return value
- async def query_relations(
- self,
- from_id: Optional[str] = None,
- graph_id: Optional[str] = None,
- relation_type: Optional[str] = None
- ) -> List[Dict]:
- url = self._base_url.join('data-platform-3/relation/query')
- params = self._common_parameters()
- criteria = dict()
- if from_id:
- criteria.update({'from_id': from_id})
- if graph_id:
- criteria.update({'graph_id': graph_id})
- if relation_type:
- criteria.update({'relation_type': relation_type})
- payload = {
- 'criteria': criteria
- }
- raw_info = await self._post(url, params, payload)
- return raw_info.get('Content')
- async def get_realtime_temperature(self, space_id: str) -> float:
- return await self.get_realtime_data(InfoCode.temperature, space_id)
- async def get_past_temperature(self, space_id: str, interval: int) -> float:
- return await self.get_past_data(InfoCode.temperature, space_id, interval)
- 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_realtime_supply_air_temperature_set(self, equipment_id: str) -> float:
- return await self.get_realtime_data(InfoCode.supply_air_temperature_set, 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, IndexError, TypeError) 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
- async def get_schedule(self, equipment_id: str) -> Tuple[str, str]:
- on_time = await self.get_static_info('ctm-OnTime', equipment_id)
- off_time = await self.get_static_info('ctm-OffTime', equipment_id)
- if not on_time:
- on_time = '080000'
- if not off_time:
- off_time = '190000'
- return on_time, off_time
- async def get_realtime_fan_freq_set(self, equipment_id: str) -> float:
- return await self.get_realtime_data(InfoCode.fan_freq_set, equipment_id)
- async def get_realtime_supply_static_press(self, system_id: str) -> float:
- return await self.get_realtime_data(InfoCode.supply_static_press, system_id)
- async def get_realtime_supply_static_press_set(self, system_id: str) -> float:
- return await self.get_realtime_data(InfoCode.supply_static_press_set, system_id)
- async def get_realtime_running_status(self, equipment_id: str) -> float:
- return await self.get_realtime_data(InfoCode.running_status, equipment_id)
- async def get_cloud_status(self, equipment_id: str) -> float:
- return await self.get_realtime_data(InfoCode.cloud_status, equipment_id)
- async def query_realtime_return_air_temperature(self, device_id: str) -> float:
- return await self.get_realtime_data(InfoCode.return_air_temperature, device_id)
- async def set_code_value(self, object_id: str, code: InfoCode, value: float):
- url = self._base_url.join('data-platform-3/parameter/setting')
- params = self._common_parameters()
- payload = {
- 'id': object_id,
- 'code': code.value,
- 'value': value
- }
- await self._post(url, params, payload)
- async def get_items_by_category(self, code) -> List:
- url = self._base_url.join('data-platform-3/object/subset_query')
- params = self._common_parameters()
- payload = {
- 'customInfo': True,
- 'criteria': {
- 'type': [code]
- }
- }
- raw_info = await self._post(url, params, payload)
- items = raw_info.get('Content')
- results = items if items else []
- return results
|