123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687 |
- from typing import List
- import arrow
- from httpx import AsyncClient
- from loguru import logger
- 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, cold_space_count: float, hot_space_count: float):
- self.current_air_temp = current_supply_air_temp
- self.cold_space_count = cold_space_count
- self.hot_space_count = hot_space_count
- def get_next_set(self, is_just_booted: bool) -> float:
- try:
- cold_hot_ratio = self.cold_space_count / self.hot_space_count
- 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
- return next_temperature_set
- async def fetch_params(project_id: str, device_id: str):
- 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)
- count = await duoduo.get_fill_count()
- 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, count.get('coldNum'), count.get('hotNum'), 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, cold_count, hot_count, is_just_booted = await fetch_params(project_id, device_id)
- controller = ACATFUSupplyAirTemperatureController(
- current_supply_air_temperature,
- cold_count,
- hot_count
- )
- next_temperature_set = controller.get_next_set(is_just_booted)
- return next_temperature_set
|