123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307 |
- from typing import List
- import arrow
- import numpy as np
- 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, ACATAHSupplyAirTempSetRequest
- from app.schemas.equipment import VAVBox
- from app.services.platform import DataPlatformService, InfoCode
- from app.services.transfer import Duoduo, Season
- 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,
- season: Season,
- ):
- 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
- self.season = season
- 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.current + 1.5
- 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.0
- 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.0
- 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,
- box.supply_air_flow_upper_limit,
- box.supply_air_flow_lower_limit,
- box.supply_air_flow_set,
- box.valve_opening,
- self.season,
- )
- cold += temp if temp < 0 else 0
- total += abs(temp)
- try:
- cold_ratio = abs(cold / total)
- except ZeroDivisionError:
- cold_ratio = np.NAN
- logger.debug(f"cold ratio: {cold_ratio}")
- return cold_ratio
- def get_normal_ratio(self):
- normal = 0
- for box in self.vav_boxes_list:
- if (
- abs(box.virtual_realtime_temperature - box.virtual_target_temperature)
- <= 1
- ):
- normal += 1
- try:
- ratio = normal / len(self.vav_boxes_list)
- except ZeroDivisionError:
- ratio = np.NAN
- return ratio
- def build(self) -> float:
- if not self.is_off_to_on:
- normal_ratio = self.get_normal_ratio()
- if normal_ratio < 0.9:
- cold_ratio = self.get_cold_ratio()
- temperature = self.calculate_by_cold_vav(cold_ratio)
- else:
- temperature = self.current
- else:
- if self.season == Season.heating:
- temperature = 27.0
- elif self.season == Season.cooling:
- temperature = 20.0
- else:
- temperature = 25.0
- if self.season == Season.heating:
- temperature = max(20.0, min(30.0, temperature))
- else:
- temperature = max(18.0, min(25.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)
- duoduo = Duoduo(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
- )
- season = await duoduo.get_season()
- # 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,
- season,
- )
- 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)
- logger.debug(f"next supply air temperature set: {device_id} - {new}")
- return new
- @logger.catch()
- def build_acatah_supply_air_temperature_set(
- params: ACATAHSupplyAirTempSetRequest,
- ) -> float:
- try:
- vav_list = list()
- for raw_vav in params.vav_list:
- vav = VAVBox(**raw_vav.dict())
- vav.virtual_target_temperature = raw_vav.virtual_temperature_target
- vav_list.append(vav)
- if params.chill_water_valve_opening_set_list[-1] == 0.0:
- thermal_mode = ThermalMode.heating
- else:
- thermal_mode = ThermalMode.cooling
- is_off_to_on = False
- if params.equip_switch_set_list[-1] == 1.0:
- for item in params.equip_switch_set_list[::-1]:
- if item == 0.0:
- is_off_to_on = True
- break
- is_thermal_mode_switched = False
- if len(set([item for item in params.hot_water_valve_opening_set_list])) > 1:
- is_thermal_mode_switched = True
- if len(set([item for item in params.chill_water_valve_opening_set_list])) > 1:
- is_thermal_mode_switched = True
- controller = ACATAHSupplyAirTemperatureController(
- vav_list,
- params.supply_air_temperature,
- params.return_air_temperature,
- thermal_mode,
- is_off_to_on,
- is_thermal_mode_switched,
- Season(params.season),
- )
- supply_air_temperature_set = controller.build()
- except (KeyError, IndexError):
- controller = ACATAHSupplyAirTemperatureDefaultController(params.is_clear_day)
- supply_air_temperature_set = controller.build()
- return supply_air_temperature_set
|