123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384 |
- from typing import Dict, List
- from httpx import AsyncClient
- from loguru import logger
- from app.models.domain.devices import ThermalMode
- from app.schemas.equipment import VAVBox
- from app.services.platform import DataPlatformService
- from app.services.transfer import Duoduo
- def count_vav_box_weight(realtime: float, target: float) -> float:
- diff = abs(realtime - target)
- if diff > 3:
- weight = 4
- elif diff > 2:
- weight = 3
- elif diff > 1:
- weight = 2
- elif diff > 0:
- weight = 1
- else:
- weight = 0
- sign = 1 if realtime - target > 0 else -1
- return weight * sign
- class ACATAHThermalModeController:
- """
- Decide whether to use cooling or heating mode according to space condition controlled by VAV Box.
- Writen by WuXu
- """
- def __init__(self, vav_boxes_list: List[VAVBox]):
- super(ACATAHThermalModeController, self).__init__()
- self.vav_boxes_list = vav_boxes_list
- def build(self) -> str:
- weight = 0.0
- for box in self.vav_boxes_list:
- weight += count_vav_box_weight(box.virtual_realtime_temperature, box.virtual_target_temperature)
- if weight > 0:
- mode = 'cooling'
- elif weight < 0:
- mode = 'heating'
- else:
- mode = 'hold'
- return mode
- async def fetch_status_params(project_id: str, device_id: str) -> Dict:
- async with AsyncClient() as client:
- platform = DataPlatformService(client, project_id)
- duoduo = Duoduo(client, project_id)
- relations = await platform.query_relations(from_id=device_id, graph_id='GtControlEquipNetwork001')
- vav_id_list = [item.get('to_id') for item in relations]
- vav_boxes_list = []
- for vav_id in vav_id_list:
- virtual_realtime_temperature = await duoduo.query_device_virtual_data(
- vav_id,
- 'VirtualRealtimeTemperature'
- )
- virtual_temperature_target = await duoduo.query_device_virtual_data(vav_id, 'TargetTemperatureSet')
- vav_params = {
- 'id': vav_id,
- 'virtual_realtime_temperature': virtual_realtime_temperature,
- 'virtual_target_temperature': virtual_temperature_target
- }
- vav = VAVBox(**vav_params)
- vav_boxes_list.append(vav)
- return {'vav_boxes_list': vav_boxes_list}
- async def get_thermal_mode(project_id: str, device_id: str) -> ThermalMode:
- prams = await fetch_status_params(project_id, device_id)
- controller = ACATAHThermalModeController(prams.get('vav_boxes_list'))
- mode = controller.build()
- return ThermalMode(mode)
|