platform.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290
  1. # -*- coding: utf-8 -*-
  2. from enum import Enum
  3. import arrow
  4. import numpy as np
  5. from httpx import AsyncClient, URL
  6. from loguru import logger
  7. from app.core.config import settings
  8. from app.services.service import Service
  9. from app.utils.date import get_time_str, TIME_FMT
  10. from app.utils.math import round_half_up
  11. class InfoCode(str, Enum):
  12. temperature = "Tdb"
  13. co2 = "CO2"
  14. hcho = "HCHO"
  15. pm2d5 = "PM2d5"
  16. humidity = "RH"
  17. supply_air_flow = "SupplyAirFlow"
  18. supply_air_flow_set = "SupplyAirFlowSet"
  19. supply_air_temperature = "SupplyAirTemp"
  20. supply_air_temperature_set = "SupplyAirTempSet"
  21. fan_speed = "FanGear"
  22. fan_speed_set = "FanGearSet"
  23. fan_freq = "FanFreq"
  24. fan_freq_set = "FanFreqSet"
  25. supply_static_press = "SupplyStaticPress"
  26. supply_static_press_set = "SupplyStaticPressSet"
  27. running_status = "RunStatus"
  28. cloud_status = "InCloudStatus"
  29. equip_switch_set = "EquipSwitchSet"
  30. return_air_temperature = "ReturnAirTemp"
  31. chill_water_valve_opening_set = "ChillWaterValveOpeningSet"
  32. hot_water_valve_opening_set = "HotWaterValveOpeningSet"
  33. water_valve_switch_set = "WaterValveSwitchSet"
  34. in_cloud_set = "InCloudSet"
  35. work_mode_set = "WorkModeSet"
  36. supply_temperature = "SupplyTemp"
  37. water_out_temperature = "WaterOutTemp"
  38. water_in_temperature = "WaterInTemp"
  39. valve_opening = "ValveOpening"
  40. class DataPlatformService(Service):
  41. def __init__(self, client: AsyncClient, project_id: str, server_settings=settings):
  42. super(DataPlatformService, self).__init__(client)
  43. self._project_id = project_id
  44. self._base_url = URL(server_settings.PLATFORM_HOST)
  45. self._now_time = get_time_str()
  46. self._secret = server_settings.PLATFORM_SECRET
  47. def _common_parameters(self) -> dict:
  48. return {"projectId": self._project_id, "secret": self._secret}
  49. async def get_realtime_data(self, code: InfoCode, object_id: str) -> float:
  50. url = self._base_url.join("data-platform-3/hisdata/query_by_obj")
  51. params = self._common_parameters()
  52. start_time = get_time_str(60 * 60, flag="ago")
  53. payload = {
  54. "criteria": {
  55. "id": object_id,
  56. "code": code.value,
  57. "receivetime": {
  58. "$gte": start_time,
  59. "$lte": self._now_time,
  60. },
  61. }
  62. }
  63. raw_info = await self._post(url, params, payload)
  64. try:
  65. latest_data = raw_info.get("Content")[-1].get("data")
  66. latest_time = raw_info.get("Content")[-1].get("receivetime")
  67. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(
  68. self._now_time, TIME_FMT
  69. ):
  70. logger.info(
  71. f"delayed data - {object_id}: ({latest_time}, {latest_data})"
  72. )
  73. value = round_half_up(latest_data, 2)
  74. except (IndexError, KeyError, TypeError):
  75. value = np.NAN
  76. return value
  77. async def get_duration(
  78. self, code: InfoCode, object_id: str, duration: int
  79. ) -> list[dict]:
  80. url = self._base_url.join("data-platform-3/hisdata/query_by_obj")
  81. params = self._common_parameters()
  82. start_time = get_time_str(duration, flag="ago")
  83. payload = {
  84. "criteria": {
  85. "id": object_id,
  86. "code": code.value,
  87. "receivetime": {
  88. "$gte": start_time,
  89. "$lte": self._now_time,
  90. },
  91. }
  92. }
  93. raw_info = await self._post(url, params, payload)
  94. try:
  95. content = raw_info.get("Content")
  96. latest_time = content[-1].get("receivetime")
  97. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(
  98. self._now_time, TIME_FMT
  99. ):
  100. result = []
  101. logger.info(f"delayed data - {object_id}: ({latest_time})")
  102. else:
  103. result = [
  104. {"timestamp": item["receivetime"], "value": item["data"]}
  105. for item in content
  106. ]
  107. except (KeyError, TypeError, IndexError):
  108. result = []
  109. return result
  110. async def get_past_data(
  111. self, code: InfoCode, object_id: str, interval: int
  112. ) -> float:
  113. """
  114. Query past data from data platform.
  115. :param code: Info code
  116. :param object_id:
  117. :param interval: time interval(seconds) from now to past
  118. :return: a past value
  119. """
  120. url = self._base_url.join("data-platform-3/hisdata/query_by_obj")
  121. params = self._common_parameters()
  122. start_time = get_time_str(60 * 60 + interval, flag="ago")
  123. end_time = get_time_str(interval, flag="ago")
  124. payload = {
  125. "criteria": {
  126. "id": object_id,
  127. "code": code.value,
  128. "receivetime": {
  129. "$gte": start_time,
  130. "$lte": end_time,
  131. },
  132. }
  133. }
  134. raw_info = await self._post(url, params, payload)
  135. try:
  136. latest_data = raw_info.get("Content")[-1].get("data")
  137. latest_time = raw_info.get("Content")[-1].get("receivetime")
  138. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(
  139. end_time, TIME_FMT
  140. ):
  141. logger.info(
  142. f"delayed data - {object_id}: ({latest_time}, {latest_data})"
  143. )
  144. value = round_half_up(latest_data, 2)
  145. except (KeyError, IndexError, TypeError):
  146. value = np.NAN
  147. except TypeError:
  148. value = -1.0
  149. return value
  150. async def query_relations(
  151. self,
  152. from_id: str | None,
  153. graph_id: str | None,
  154. relation_type: str | None,
  155. ) -> list[dict]:
  156. url = self._base_url.join("data-platform-3/relation/query")
  157. params = self._common_parameters()
  158. criteria = dict()
  159. if from_id:
  160. criteria.update({"from_id": from_id})
  161. if graph_id:
  162. criteria.update({"graph_id": graph_id})
  163. if relation_type:
  164. criteria.update({"relation_type": relation_type})
  165. payload = {"criteria": criteria}
  166. raw_info = await self._post(url, params, payload)
  167. return raw_info.get("Content")
  168. async def get_realtime_temperature(self, space_id: str) -> float:
  169. return await self.get_realtime_data(InfoCode.temperature, space_id)
  170. async def get_past_temperature(self, space_id: str, interval: int) -> float:
  171. return await self.get_past_data(InfoCode.temperature, space_id, interval)
  172. async def get_realtime_co2(self, space_id: str) -> float:
  173. return await self.get_realtime_data(InfoCode.co2, space_id)
  174. async def get_realtime_hcho(self, space_id: str) -> float:
  175. return await self.get_realtime_data(InfoCode.hcho, space_id)
  176. async def get_realtime_pm2d5(self, space_id: str) -> float:
  177. return await self.get_realtime_data(InfoCode.pm2d5, space_id)
  178. async def get_realtime_humidity(self, space_id: str) -> float:
  179. return await self.get_realtime_data(InfoCode.humidity, space_id)
  180. async def get_realtime_supply_air_flow(self, equipment_id: str) -> float:
  181. return await self.get_realtime_data(InfoCode.supply_air_flow, equipment_id)
  182. async def get_realtime_supply_air_temperature(self, equipment_id: str) -> float:
  183. return await self.get_realtime_data(
  184. InfoCode.supply_air_temperature, equipment_id
  185. )
  186. async def get_realtime_supply_air_temperature_set(self, equipment_id: str) -> float:
  187. return await self.get_realtime_data(
  188. InfoCode.supply_air_temperature_set, equipment_id
  189. )
  190. async def get_fan_speed(self, equipment_id: str) -> float:
  191. return await self.get_realtime_data(InfoCode.fan_speed, equipment_id)
  192. async def get_static_info(self, code: str, object_id: str):
  193. url = self._base_url.join("data-platform-3/object/batch_query")
  194. params = self._common_parameters()
  195. payload = {"customInfo": True, "criterias": [{"id": object_id}]}
  196. raw_info = await self._post(url, params, payload)
  197. try:
  198. info = raw_info["Content"][0]["infos"][code]
  199. except (KeyError, IndexError, TypeError) as e:
  200. logger.error(f"id: {object_id}, details: {e}")
  201. info = None
  202. return info
  203. async def get_air_flow_limit(self, equipment_id: str) -> tuple[float, float]:
  204. lower = await self.get_static_info("MinAirFlow", equipment_id)
  205. upper = await self.get_static_info("MaxAirFlow", equipment_id)
  206. if lower is None:
  207. lower = 150.0
  208. if upper is None:
  209. upper = 2000.0
  210. return lower, upper
  211. async def get_schedule(self, equipment_id: str) -> tuple[str, str]:
  212. on_time = await self.get_static_info("ctm-OnTime", equipment_id)
  213. off_time = await self.get_static_info("ctm-OffTime", equipment_id)
  214. if not on_time:
  215. on_time = "080000"
  216. if not off_time:
  217. off_time = "190000"
  218. return on_time, off_time
  219. async def get_realtime_fan_freq_set(self, equipment_id: str) -> float:
  220. return await self.get_realtime_data(InfoCode.fan_freq_set, equipment_id)
  221. async def get_realtime_supply_static_press(self, system_id: str) -> float:
  222. return await self.get_realtime_data(InfoCode.supply_static_press, system_id)
  223. async def get_realtime_supply_static_press_set(self, system_id: str) -> float:
  224. return await self.get_realtime_data(InfoCode.supply_static_press_set, system_id)
  225. async def get_realtime_running_status(self, equipment_id: str) -> float:
  226. return await self.get_realtime_data(InfoCode.running_status, equipment_id)
  227. async def get_cloud_status(self, equipment_id: str) -> float:
  228. return await self.get_realtime_data(InfoCode.cloud_status, equipment_id)
  229. async def query_realtime_return_air_temperature(self, device_id: str) -> float:
  230. return await self.get_realtime_data(InfoCode.return_air_temperature, device_id)
  231. async def set_code_value(self, object_id: str, code: InfoCode, value: float):
  232. url = self._base_url.join("data-platform-3/parameter/setting")
  233. params = self._common_parameters()
  234. payload = {"id": object_id, "code": code.value, "value": value}
  235. await self._post(url, params, payload)
  236. async def get_items_by_category(self, code) -> list:
  237. url = self._base_url.join("data-platform-3/object/subset_query")
  238. params = self._common_parameters()
  239. payload = {"customInfo": True, "criteria": {"type": [code]}}
  240. raw_info = await self._post(url, params, payload)
  241. items = raw_info.get("Content")
  242. results = items if items else []
  243. return results