platform.py 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  1. # -*- coding: utf-8 -*-
  2. from enum import Enum
  3. from typing import Dict, List, Tuple
  4. import arrow
  5. import numpy as np
  6. from httpx import AsyncClient, URL
  7. from loguru import logger
  8. from app.core.config import settings
  9. from app.services.service import Service
  10. from app.utils.date import get_time_str, TIME_FMT
  11. from app.utils.math import round_half_up
  12. class InfoCode(str, Enum):
  13. temperature = 'Tdb'
  14. co2 = 'CO2'
  15. hcho = 'HCHO'
  16. pm2d5 = 'PM2d5'
  17. humidity = 'RH'
  18. supply_air_flow = 'SupplyAirFlow'
  19. supply_air_temperature = 'SupplyAirTemp'
  20. supply_air_temperature_set = 'SupplyAirTempSet'
  21. fan_speed = 'FanGear'
  22. fan_freq = 'FanFreq'
  23. fan_freq_set = 'FanFreqSet'
  24. supply_static_press = 'SupplyStaticPress'
  25. supply_static_press_set = 'SupplyStaticPressSet'
  26. class DataPlatformService(Service):
  27. def __init__(
  28. self,
  29. client: AsyncClient,
  30. project_id: str,
  31. server_settings=settings
  32. ):
  33. super(DataPlatformService, self).__init__(client)
  34. self._project_id = project_id
  35. self._base_url = URL(server_settings.PLATFORM_HOST)
  36. self._now_time = get_time_str()
  37. self._secret = server_settings.PLATFORM_SECRET
  38. def _common_parameters(self) -> Dict:
  39. return {'projectId': self._project_id, 'secret': self._secret}
  40. async def get_realtime_data(self, code: InfoCode, object_id: str) -> float:
  41. url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
  42. params = self._common_parameters()
  43. start_time = get_time_str(60 * 60, flag='ago')
  44. payload = {
  45. 'criteria': {
  46. 'id': object_id,
  47. 'code': code.value,
  48. 'receivetime': {
  49. '$gte': start_time,
  50. '$lte': self._now_time,
  51. }
  52. }
  53. }
  54. raw_info = await self._post(url, params, payload)
  55. try:
  56. latest_data = raw_info.get('Content')[-1].get('data')
  57. latest_time = raw_info.get('Content')[-1].get('receivetime')
  58. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT):
  59. logger.info(f'delayed data - {object_id}: ({latest_time}, {latest_data})')
  60. value = round_half_up(latest_data, 2)
  61. except KeyError and IndexError:
  62. value = np.NAN
  63. except TypeError:
  64. value = -1.0
  65. return value
  66. async def get_duration(self, code: InfoCode, object_id: str, duration: int) -> List[Dict]:
  67. url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
  68. params = self._common_parameters()
  69. start_time = get_time_str(duration, flag='ago')
  70. payload = {
  71. 'criteria': {
  72. 'id': object_id,
  73. 'code': code.value,
  74. 'receivetime': {
  75. '$gte': start_time,
  76. '$lte': self._now_time,
  77. }
  78. }
  79. }
  80. raw_info = await self._post(url, params, payload)
  81. try:
  82. content = raw_info.get('Content')
  83. latest_time = content[-1].get('receivetime')
  84. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT):
  85. result = []
  86. logger.info(f'delayed data - {object_id}: ({latest_time})')
  87. else:
  88. result = [
  89. {
  90. 'timestamp': item['receivetime'],
  91. 'value': item['data']
  92. }
  93. for item in content
  94. ]
  95. except KeyError:
  96. result = []
  97. return result
  98. async def get_past_data(self, code: InfoCode, object_id: str, interval: int) -> float:
  99. """
  100. Query past data from data platform.
  101. :param code: Info code
  102. :param object_id:
  103. :param interval: time interval(seconds) from now to past
  104. :return: a past value
  105. """
  106. url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
  107. params = self._common_parameters()
  108. start_time = get_time_str(60 * 60 + interval, flag='ago')
  109. end_time = get_time_str(interval, flag='ago')
  110. payload = {
  111. 'criteria': {
  112. 'id': object_id,
  113. 'code': code.value,
  114. 'receivetime': {
  115. '$gte': start_time,
  116. '$lte': end_time,
  117. }
  118. }
  119. }
  120. raw_info = await self._post(url, params, payload)
  121. try:
  122. latest_data = raw_info.get('Content')[-1].get('data')
  123. latest_time = raw_info.get('Content')[-1].get('receivetime')
  124. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(end_time, TIME_FMT):
  125. logger.info(f'delayed data - {object_id}: ({latest_time}, {latest_data})')
  126. value = round_half_up(latest_data, 2)
  127. except KeyError and IndexError:
  128. value = np.NAN
  129. except TypeError:
  130. value = -1.0
  131. return value
  132. async def get_realtime_temperature(self, space_id: str) -> float:
  133. return await self.get_realtime_data(InfoCode.temperature, space_id)
  134. async def get_past_temperature(self, space_id: str, interval: int) -> float:
  135. return await self.get_past_data(InfoCode.temperature, space_id, interval)
  136. async def get_realtime_co2(self, space_id: str) -> float:
  137. return await self.get_realtime_data(InfoCode.co2, space_id)
  138. async def get_realtime_hcho(self, space_id: str) -> float:
  139. return await self.get_realtime_data(InfoCode.hcho, space_id)
  140. async def get_realtime_pm2d5(self, space_id: str) -> float:
  141. return await self.get_realtime_data(InfoCode.pm2d5, space_id)
  142. async def get_realtime_humidity(self, space_id: str) -> float:
  143. return await self.get_realtime_data(InfoCode.humidity, space_id)
  144. async def get_realtime_supply_air_flow(self, equipment_id: str) -> float:
  145. return await self.get_realtime_data(InfoCode.supply_air_flow, equipment_id)
  146. async def get_realtime_supply_air_temperature(self, equipment_id: str) -> float:
  147. return await self.get_realtime_data(InfoCode.supply_air_temperature, equipment_id)
  148. async def get_realtime_supply_air_temperature_set(self, equipment_id: str) -> float:
  149. return await self.get_realtime_data(InfoCode.supply_air_temperature_set, equipment_id)
  150. async def get_fan_speed(self, equipment_id: str) -> float:
  151. return await self.get_realtime_data(InfoCode.fan_speed, equipment_id)
  152. async def get_static_info(self, code: str, object_id: str):
  153. url = self._base_url.join('data-platform-3/object/batch_query')
  154. params = self._common_parameters()
  155. payload = {
  156. 'customInfo': True,
  157. 'criterias': [
  158. {
  159. 'id': object_id
  160. }
  161. ]
  162. }
  163. raw_info = await self._post(url, params, payload)
  164. try:
  165. info = raw_info['Content'][0]['infos'][code]
  166. except KeyError as e:
  167. logger.error(f'id: {object_id}, details: {e}')
  168. info = None
  169. return info
  170. async def get_air_flow_limit(self, equipment_id: str) -> Tuple[float, float]:
  171. lower = await self.get_static_info('MinAirFlow', equipment_id)
  172. upper = await self.get_static_info('MaxAirFlow', equipment_id)
  173. if not lower:
  174. lower = 150.0
  175. if not upper:
  176. upper = 2000.0
  177. return lower, upper
  178. async def get_realtime_fan_freq_set(self, equipment_id: str) -> float:
  179. return await self.get_realtime_data(InfoCode.fan_freq_set, equipment_id)
  180. async def get_realtime_supply_static_press(self, system_id: str) -> float:
  181. return await self.get_realtime_data(InfoCode.supply_static_press, system_id)
  182. async def get_realtime_supply_static_press_set(self, system_id: str) -> float:
  183. return await self.get_realtime_data(InfoCode.supply_static_press_set, system_id)
  184. async def set_code_value(self, object_id: str, code: InfoCode, value: float):
  185. url = self._base_url.join('data-platform-3/parameter/setting')
  186. params = self._common_parameters()
  187. payload = {
  188. 'id': object_id,
  189. 'code': code.value,
  190. 'value': value
  191. }
  192. await self._post(url, params, payload)