# -*- coding: utf-8 -*- from typing import Dict, List, Optional import numpy as np from httpx import AsyncClient from loguru import logger from app.controllers.events import ahu_supply_air_temp_set_dict from app.schemas.equipment import AHU from app.schemas.system import ACAT from app.services.platform import DataPlatformService, InfoCode from app.services.transfer import ReviewService, EquipmentInfoService from app.services.weather import WeatherService class AHUController: def __init__(self, equipment: Optional[AHU] = None, system: Optional[ACAT] = None): super(AHUController, self).__init__() self._equipment = equipment self._system = system self._supply_air_temperature_command_dict = ahu_supply_air_temp_set_dict.get('dataframe') 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([item['value'] for item in supply_air_temperature_set_duration]) 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_set = 100.0 else: freq_set = min(temp_freq_set, 90.0) return freq_set async def build_supply_air_temperature_set(self, outdoor_temperature: float) -> float: command_df = self._supply_air_temperature_command_dict temp_command = command_df.loc[command_df['outdoor_temperature'] == outdoor_temperature]['temperature_set'] if len(temp_command) > 0: command = float(temp_command) else: command = 20.0 return command @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 = ReviewService(client, project_id) # equip_service = EquipmentInfoService(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]) hot_rate = await review_service.get_fill_rate() supply_air_temperature_set = await platform.get_duration( InfoCode.supply_air_temperature_set, equipment_id, 30 * 60 ) 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'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 get_supply_air_temperature_controlled(project_id: str, equipment_id: str) -> None: async with AsyncClient() as client: weather_service = WeatherService(client) realtime_weather = await weather_service.get_realtime_weather(project_id) outdoor_temperature = realtime_weather.get('temperature') ahu_controller = AHUController() new_supply_air_temperature_set = await ahu_controller.build_supply_air_temperature_set(outdoor_temperature) logger.debug(f'supply air temperature set: {new_supply_air_temperature_set}') async with AsyncClient() as client: platform = DataPlatformService(client, project_id) await platform.set_code_value(equipment_id, InfoCode.supply_air_temperature_set, new_supply_air_temperature_set)