123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116 |
- 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, InfoCode
- from app.services.transfer import Duoduo, Season
- def count_vav_box_weight(
- realtime: float,
- target: float,
- upper_limit_flow: float,
- lower_limit_flow: float,
- current_flow_set: float,
- season: Season
- ) -> float:
- diff = realtime - target
- flag = False
- if current_flow_set < lower_limit_flow * 1.1:
- if season == Season.cooling:
- if diff < 0:
- flag = True
- if season == Season.heating:
- if diff > 0:
- flag = True
- elif current_flow_set > upper_limit_flow * 0.9:
- if season == Season.cooling:
- if diff > 0:
- flag = True
- if season == Season.heating:
- if diff < 0:
- flag = True
- if flag:
- weight = round(diff, 0)
- weight = max(-4.0, min(4.0, weight))
- else:
- weight = 0
- return weight
- 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], season: Season):
- super(ACATAHThermalModeController, self).__init__()
- self.vav_boxes_list = vav_boxes_list
- self.season = season
- 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,
- box.supply_air_flow_upper_limit,
- box.supply_air_flow_lower_limit,
- box.supply_air_flow_set,
- self.season
- )
- 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)
- season = duoduo.get_season()
- 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')
- lower_limit_flow, upper_limit_flow = await platform.get_air_flow_limit(vav_id)
- current_flow_set = await platform.get_realtime_data(InfoCode.supply_air_flow_set, vav_id)
- vav_params = {
- 'id': vav_id,
- 'virtual_realtime_temperature': virtual_realtime_temperature,
- 'virtual_target_temperature': virtual_temperature_target,
- 'supply_air_flow_lower_limit': lower_limit_flow,
- 'supply_air_flow_upper_limit': upper_limit_flow,
- 'supply_air_flow_set': current_flow_set
- }
- vav = VAVBox(**vav_params)
- vav_boxes_list.append(vav)
- return {'vav_boxes_list': vav_boxes_list, 'season': season}
- @logger.catch()
- 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'), prams.get('season'))
- mode = controller.build()
- return ThermalMode(mode)
|