thermal_mode.py 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. from typing import Dict, List
  2. from httpx import AsyncClient
  3. from loguru import logger
  4. from app.models.domain.devices import ThermalMode
  5. from app.schemas.equipment import VAVBox
  6. from app.services.platform import DataPlatformService
  7. from app.services.transfer import Duoduo
  8. def count_vav_box_weight(realtime: float, target: float) -> float:
  9. diff = abs(realtime - target)
  10. if diff > 3:
  11. weight = 4
  12. elif diff > 2:
  13. weight = 3
  14. elif diff > 1:
  15. weight = 2
  16. elif diff > 0:
  17. weight = 1
  18. else:
  19. weight = 0
  20. sign = 1 if realtime - target > 0 else -1
  21. return weight * sign
  22. class ACATAHThermalModeController:
  23. """
  24. Decide whether to use cooling or heating mode according to space condition controlled by VAV Box.
  25. Writen by WuXu
  26. """
  27. def __init__(self, vav_boxes_list: List[VAVBox]):
  28. super(ACATAHThermalModeController, self).__init__()
  29. self.vav_boxes_list = vav_boxes_list
  30. def build(self) -> str:
  31. weight = 0.0
  32. for box in self.vav_boxes_list:
  33. weight += count_vav_box_weight(box.virtual_realtime_temperature, box.virtual_target_temperature)
  34. if weight > 0:
  35. mode = 'cooling'
  36. elif weight < 0:
  37. mode = 'heating'
  38. else:
  39. mode = 'hold'
  40. return mode
  41. async def fetch_status_params(project_id: str, device_id: str) -> Dict:
  42. async with AsyncClient() as client:
  43. platform = DataPlatformService(client, project_id)
  44. duoduo = Duoduo(client, project_id)
  45. relations = await platform.query_relations(from_id=device_id, graph_id='GtControlEquipNetwork001')
  46. vav_id_list = [item.get('to_id') for item in relations]
  47. vav_boxes_list = []
  48. for vav_id in vav_id_list:
  49. virtual_realtime_temperature = await duoduo.query_device_virtual_data(
  50. vav_id,
  51. 'VirtualRealtimeTemperature'
  52. )
  53. virtual_temperature_target = await duoduo.query_device_virtual_data(vav_id, 'TargetTemperatureSet')
  54. vav_params = {
  55. 'id': vav_id,
  56. 'virtual_realtime_temperature': virtual_realtime_temperature,
  57. 'virtual_target_temperature': virtual_temperature_target
  58. }
  59. vav = VAVBox(**vav_params)
  60. vav_boxes_list.append(vav)
  61. return {'vav_boxes_list': vav_boxes_list}
  62. async def get_thermal_mode(project_id: str, device_id: str) -> ThermalMode:
  63. prams = await fetch_status_params(project_id, device_id)
  64. controller = ACATAHThermalModeController(prams.get('vav_boxes_list'))
  65. mode = controller.build()
  66. return ThermalMode(mode)