# -*- coding: utf-8 -*- from typing import List, Optional import numpy as np from httpx import AsyncClient from loguru import logger from app.models.domain.devices import ACATAHFreqSetRequest from app.schemas.equipment import AHU from app.schemas.system import ACAT from app.services.platform import DataPlatformService, InfoCode from app.services.transfer import Duoduo class AHUController: def __init__(self, equipment: Optional[AHU] = None, system: Optional[ACAT] = None): super(AHUController, self).__init__() self._equipment = equipment self._system = system async def build_freq_set( self, supply_air_temperature_set_duration: List, hot_rate: float ) -> float: extent = 5 if ( np.isnan(self._equipment.fan_freq_set) or np.isnan(self._system.supply_static_press) or np.isnan(self._system.supply_static_press_set) ): temp_freq_set = 80.0 else: pre_fan_freq_set = self._equipment.fan_freq_set if ( self._system.supply_static_press < self._system.supply_static_press_set - extent ): temp_freq_set = pre_fan_freq_set + 2 elif ( self._system.supply_static_press > self._system.supply_static_press_set + extent ): temp_freq_set = pre_fan_freq_set - 2 else: temp_freq_set = pre_fan_freq_set temperature_value_list = np.array(supply_air_temperature_set_duration) freq_upper_limit = 90.0 if ( temperature_value_list.size > 0 and np.all(temperature_value_list == temperature_value_list[0]) and temperature_value_list[0] <= 18.0 and hot_rate >= 0.5 ): freq_upper_limit = 100.0 freq_set = min(temp_freq_set, freq_upper_limit) self._equipment.fan_freq_set = freq_set return freq_set @logger.catch() async def get_freq_controlled(project_id: str, equipment_id: str) -> None: async with AsyncClient() as client: platform = DataPlatformService(client, project_id) review_service = Duoduo(client, project_id) fan_freq_set = await platform.get_realtime_fan_freq_set(equipment_id) # system = await equip_service.get_system_by_equipment(equipment_id) system = ["Sy1101050030eab54ad66b084a1cb1b919950b263280"] press = await platform.get_realtime_supply_static_press(system[0]) press_set = await platform.get_realtime_supply_static_press_set(system[0]) count = await review_service.get_fill_count() try: hot_rate = count.get("hotNum") / ( count.get("hotNum") + count.get("coldNum") + count.get("normalNum") ) except (ZeroDivisionError, KeyError): hot_rate = 0.0 raw_supply_air_temperature_set = await platform.get_duration( InfoCode.supply_air_temperature_set, equipment_id, 30 * 60 ) supply_air_temperature_set = [ item["value"] for item in raw_supply_air_temperature_set ] equip_params = {"id": equipment_id, "fan_freq_set": fan_freq_set} ahu = AHU(**equip_params) system_params = { "id": system[0], "supply_static_press": press, "supply_static_press_set": press_set, } acat_system = ACAT(**system_params) ahu_controller = AHUController(ahu, acat_system) new_freq_set = await ahu_controller.build_freq_set( supply_air_temperature_set, hot_rate ) logger.debug(f"{equipment_id} freq set: {new_freq_set}") async with AsyncClient() as client: platform = DataPlatformService(client, project_id) await platform.set_code_value(equipment_id, InfoCode.fan_freq_set, new_freq_set) @logger.catch() async def build_acatah_freq_set(params: ACATAHFreqSetRequest) -> float: acat_system = ACAT( supply_static_press=params.system_supply_static_press, supply_static_press_set=params.system_supply_static_press_set, ) ahu = AHU(fan_freq_set=params.current_freq_set) ahu_controller = AHUController(ahu, acat_system) new_freq_set = await ahu_controller.build_freq_set( params.supply_air_temperature_set_list, params.spaces_hot_rate ) return new_freq_set