platform.py 2.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485
  1. # -*- coding: utf-8 -*-
  2. from enum import Enum
  3. from typing import Dict
  4. import arrow
  5. import numpy as np
  6. from httpx import AsyncClient, URL
  7. from loguru import logger
  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. from app.core.config import PlatformSettings
  12. platform_settings = PlatformSettings()
  13. class InfoCode(str, Enum):
  14. temperature = 'Tdb'
  15. co2 = 'CO2'
  16. hcho = 'HCHO'
  17. pm2d5 = 'PM2d5'
  18. humidity = 'RH'
  19. class DataPlatformService(Service):
  20. def __init__(
  21. self,
  22. client: AsyncClient,
  23. project_id: str,
  24. space_id: str,
  25. settings: PlatformSettings = platform_settings
  26. ):
  27. super(DataPlatformService, self).__init__(client)
  28. self._project_id = project_id
  29. self._space_id = space_id
  30. self._base_url = URL(settings.platform_host)
  31. self._now_time = get_time_str()
  32. self._secret = settings.platform_secret
  33. def _common_parameters(self) -> Dict:
  34. return {'projectId': self._project_id, 'secret': self._secret}
  35. async def get_realtime_data(self, code: InfoCode) -> float:
  36. url = self._base_url.join('data-platform-3/hisdata/query_by_obj')
  37. params = self._common_parameters()
  38. start_time = get_time_str(60 * 60, flag='ago')
  39. payload = {
  40. 'criteria': {
  41. 'id': self._space_id,
  42. 'code': code.value,
  43. 'receivetime': {
  44. '$gte': start_time,
  45. '$lte': self._now_time,
  46. }
  47. }
  48. }
  49. raw_info = await self._post(url, params, payload)
  50. value = np.NAN
  51. if raw_info.get('Content'):
  52. latest_data = raw_info.get('Content')[-1].get('data')
  53. latest_time = raw_info.get('Content')[-1].get('receivetime')
  54. if arrow.get(latest_time, TIME_FMT).shift(minutes=15) < arrow.get(self._now_time, TIME_FMT):
  55. logger.info(f'delayed data - {self._space_id}: ({latest_time}, {latest_data})')
  56. value = round_half_up(latest_data, 2)
  57. return value
  58. async def get_realtime_temperature(self) -> float:
  59. return await self.get_realtime_data(InfoCode.temperature)
  60. async def get_realtime_co2(self) -> float:
  61. return await self.get_realtime_data(InfoCode.co2)
  62. async def get_realtime_hcho(self) -> float:
  63. return await self.get_realtime_data(InfoCode.hcho)
  64. async def get_realtime_pm2d5(self) -> float:
  65. return await self.get_realtime_data(InfoCode.pm2d5)
  66. async def get_realtime_humidity(self) -> float:
  67. return await self.get_realtime_data(InfoCode.humidity)