123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108 |
- from typing import Tuple
- from httpx import AsyncClient
- from loguru import logger
- from app.models.domain.devices import ACATFUSupplyAirTempSetRequest
- from app.services.platform import DataPlatformService, InfoCode
- from app.services.transfer import Duoduo
- class ACATFUSupplyAirTemperatureController:
- """
- Supply air temperature setting logic version 1 by Wenyan.
- """
- def __init__(self, current_supply_air_temp: float, hot_rate: float, cold_rate: float):
- self.current_air_temp = current_supply_air_temp
- self.hot_rate = hot_rate
- self.cold_rate = cold_rate
- def get_next_set(self, is_just_booted: bool) -> float:
- try:
- cold_hot_ratio = self.cold_rate / self.hot_rate
- except ZeroDivisionError:
- cold_hot_ratio = 99
- if is_just_booted:
- if cold_hot_ratio < 0.5:
- next_temperature_set = 8.0
- elif cold_hot_ratio < 0.9:
- next_temperature_set = 11.0
- elif cold_hot_ratio < 1.1:
- next_temperature_set = 14.0
- elif cold_hot_ratio < 1.5:
- next_temperature_set = 17.0
- elif cold_hot_ratio >= 1.5:
- next_temperature_set = 20.0
- else: # If cold hot ratio is nan.
- next_temperature_set = 20.0
- else:
- if cold_hot_ratio < 0.5:
- diff = -3
- elif cold_hot_ratio < 0.9:
- diff = -2
- elif cold_hot_ratio < 1.1:
- diff = 0
- elif cold_hot_ratio < 1.5:
- diff = 2
- elif cold_hot_ratio >= 1.5:
- diff = 3
- else: # If cold hot ratio is nan.
- diff = 0
- next_temperature_set = self.current_air_temp + diff
- next_temperature_set = max(8.0, min(20.0, next_temperature_set))
- return next_temperature_set
- async def fetch_params(project_id: str, device_id: str) -> Tuple[float, float, float, bool]:
- 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)
- running_status_duration = await platform.get_duration(InfoCode.running_status, device_id, 15 * 60)
- hot_rate, cold_rate = await duoduo.query_fill_rate_by_device(device_id)
- is_just_booted = False
- if running_status_duration[-1]['value'] == 1.0:
- for item in running_status_duration[::-1]:
- if item['value'] == 0.0:
- is_just_booted = True
- break
- return current_supply_air_temperature, hot_rate, cold_rate, is_just_booted
- @logger.catch()
- async def get_next_acatfu_supply_air_temperature_set(project_id: str, device_id: str) -> float:
- current_supply_air_temperature, hot_rate, cold_rate, is_just_booted = await fetch_params(project_id, device_id)
- controller = ACATFUSupplyAirTemperatureController(
- current_supply_air_temperature,
- hot_rate,
- cold_rate
- )
- next_temperature_set = controller.get_next_set(is_just_booted)
- return next_temperature_set
- @logger.catch()
- def build_acatfu_supply_air_temperature(params: ACATFUSupplyAirTempSetRequest) -> float:
- is_just_booted = False
- if params.running_status_list[-1] == 1.0:
- for item in params.running_status_list[::-1]:
- if item == 0.0:
- is_just_booted = True
- break
- controller = ACATFUSupplyAirTemperatureController(
- params.supply_air_temperature,
- params.hot_ratio,
- params.cold_ratio
- )
- supply_air_temperature_set = controller.get_next_set(is_just_booted)
- return supply_air_temperature_set
|