123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225 |
- 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
- 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 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:
- 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(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)
- 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
|