123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157 |
- 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:
- if 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
|