from typing import Dict, List from httpx import AsyncClient from loguru import logger from app.models.domain.devices import ThermalMode from app.models.domain.devices import ACATAHThermalModeSetRequest 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, valve_opening: 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 or valve_opening > 90.0: if season == Season.cooling: if diff > 0: flag = True if season == Season.heating: if diff < 0: flag = True if flag: if abs(diff) < 1: weight = 0.0 else: weight = diff 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, box.valve_opening, 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 ) valve_opening = await platform.get_realtime_data( InfoCode.valve_opening, 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, "valve_opening": valve_opening, } 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) @logger.catch() def build_acatah_thermal_mode_set(params: ACATAHThermalModeSetRequest) -> str: vav_list = list() for raw_vav in params.vav_list: vav = VAVBox( virtual_realtime_temperature=raw_vav.virtual_realtime_temperature, virtual_target_temperature=raw_vav.virtual_temperature_target, supply_air_flow_lower_limit=raw_vav.supply_air_flow_lower_limit, supply_air_flow_upper_limit=raw_vav.supply_air_flow_upper_limit, supply_air_flow_set=raw_vav.supply_air_flow_set, valve_opening=raw_vav.valve_opening, ) vav_list.append(vav) controller = ACATAHThermalModeController(vav_list, Season(params.season)) mode = controller.build() return mode