from typing import List import arrow from httpx import AsyncClient from loguru import logger from app.controllers.equipment.ahu.thermal_mode import count_vav_box_weight, fetch_status_params from app.models.domain.devices import ThermalMode from app.schemas.equipment import VAVBox from app.services.platform import DataPlatformService, InfoCode from app.services.weather import WeatherService from app.utils.date import get_time_str, TIME_FMT class ACATAHSupplyAirTemperatureController: """ Supply air temperature setting logic version 2 by WuXu. """ def __init__( self, vav_boxes_list: List[VAVBox], current: float, return_air: float, thermal_mode: ThermalMode, is_off_to_on: bool, is_thermal_mode_switched: bool ): super(ACATAHSupplyAirTemperatureController, self).__init__() self.vav_boxes_list = vav_boxes_list self.current = current self.return_air = return_air self.thermal_mode = thermal_mode self.is_off_to_on = is_off_to_on self.is_thermal_mode_switched = is_thermal_mode_switched def calculate_by_cold_vav(self, cold_ratio: float) -> float: if self.thermal_mode == ThermalMode.cooling: if cold_ratio < 0.3: new = self.current - 1.0 elif cold_ratio < 0.45: new = self.current - 0.5 elif cold_ratio <= 0.55: new = self.current elif cold_ratio <= 0.7: new = self.current + 1.0 elif cold_ratio <= 1.0: new = self.return_air else: new = self.current elif self.thermal_mode == ThermalMode.heating: if cold_ratio < 0.3: new = self.return_air elif cold_ratio < 0.45: new = self.current - 1 elif cold_ratio <= 0.55: new = self.current elif cold_ratio <= 0.7: new = self.current + 0.5 elif cold_ratio <= 1.0: new = self.current + 1 else: new = self.current else: new = self.current return new def get_cold_ratio(self): cold, total = 0, 0 for box in self.vav_boxes_list: temp = count_vav_box_weight(box.virtual_realtime_temperature, box.virtual_target_temperature) cold += temp if temp < 0 else 0 total += abs(temp) return abs(cold / total) def build(self) -> float: if not self.is_off_to_on: cold_ratio = self.get_cold_ratio() temperature = self.calculate_by_cold_vav(cold_ratio) else: temperature = 25.0 temperature = max(20.0, min(30.0, temperature)) return temperature class ACATAHSupplyAirTemperatureDefaultController: """ Determine supply air temperature when missing data. """ def __init__(self, is_clear_day: bool): super(ACATAHSupplyAirTemperatureDefaultController, self).__init__() self.is_clear_day = is_clear_day def build(self) -> float: now = get_time_str() now_time_str = arrow.get(now, TIME_FMT).time().strftime('%H%M%S') if '080000' <= now_time_str < '100000': is_morning = True else: is_morning = False if is_morning: temperature = 27.0 else: if self.is_clear_day: temperature = 23.0 else: temperature = 25.0 return temperature async def get_planned(project_id: str, device_id: str) -> float: vav_boxes_params = await fetch_status_params(project_id, device_id) vav_boxes_lit = vav_boxes_params['vav_boxes_list'] async with AsyncClient() as client: platform = DataPlatformService(client, project_id) current_supply_air_temperature = await platform.get_realtime_supply_air_temperature(device_id) return_air_temperature = await platform.query_realtime_return_air_temperature(device_id) hot_water_valve_opening_set_duration = await platform.get_duration( InfoCode.hot_water_valve_opening_set, device_id, 15 * 60 ) chill_water_valve_opening_set_duration = await platform.get_duration( InfoCode.chill_water_valve_opening_set, device_id, 15 * 60 ) on_off_set_duration = await platform.get_duration(InfoCode.equip_switch_set, device_id, 20 * 60) # if hot_water_valve_opening_set_duration[-1]['value'] == 0.0: # thermal_mode = ThermalMode.cooling if chill_water_valve_opening_set_duration[-1]['value'] == 0.0: thermal_mode = ThermalMode.heating else: thermal_mode = ThermalMode.cooling is_off_to_on = False if on_off_set_duration[-1]['value'] == 1.0: for item in on_off_set_duration[::-1]: if item['value'] == 0.0: is_off_to_on = True break logger.debug(f'{device_id} was off to on: {is_off_to_on}') is_thermal_mode_switched = False if len(set([item['value'] for item in hot_water_valve_opening_set_duration])) > 1: is_thermal_mode_switched = True if len(set([item['value'] for item in chill_water_valve_opening_set_duration])) > 1: is_thermal_mode_switched = True controller = ACATAHSupplyAirTemperatureController( vav_boxes_lit, current_supply_air_temperature, return_air_temperature, thermal_mode, is_off_to_on, is_thermal_mode_switched ) next_supply_air_temperature_set = controller.build() return next_supply_air_temperature_set async def get_default(project_id: str) -> float: async with AsyncClient() as client: weather_service = WeatherService(client) realtime_weather = await weather_service.get_realtime_weather(project_id) if realtime_weather.get('text') == '晴': is_clear_day = True else: is_clear_day = False controller = ACATAHSupplyAirTemperatureDefaultController(is_clear_day) next_supply_air_temperature_set = controller.build() return next_supply_air_temperature_set @logger.catch() async def get_next_supply_air_temperature_set(project_id: str, device_id: str) -> float: try: new = await get_planned(project_id, device_id) except KeyError and IndexError: new = await get_default(project_id) return new