# -*- coding: utf-8 -*- from typing import Dict import numpy as np from httpx import AsyncClient from loguru import logger from app.controllers.equipment.controller import EquipmentController from app.schemas.equipment import AirValveSpeed, FCU from app.schemas.space import Space from app.services.platform import DataPlatformService, InfoCode from app.services.transfer import SpaceInfoService, Duoduo, Season from app.utils.math import round_half_up class FCUController(EquipmentController): def __init__(self, equipment: FCU, season: Season): super(FCUController, self).__init__() self._equipment = equipment self.season = season async def get_temperature_target(self) -> float: target = self._equipment.space.temperature_target if not np.isnan(target): setting_target = target else: setting_target = 25.0 self._equipment.setting_temperature = setting_target return setting_target async def get_mode(self) -> int: temperature_target = self._equipment.space.temperature_target if self.season == Season.heating: if temperature_target >= self._equipment.space.realtime_temperature: mode = 2 else: mode = 1 elif self.season == Season.cooling: mode = 1 else: mode = 0 self._equipment.work_mode = mode return mode async def get_air_valve_speed(self): temperature_target = self._equipment.space.temperature_target if temperature_target > 0: if self.season == Season.heating: diff = abs(self._equipment.space.realtime_temperature - temperature_target) if diff >= 1.5: self._equipment.air_valve_speed = AirValveSpeed.high elif diff >= 1.0: self._equipment.air_valve_speed = AirValveSpeed.medium elif diff >= 0.5: self._equipment.air_valve_speed = AirValveSpeed.low else: self._equipment.air_valve_speed = AirValveSpeed.off else: diff = self._equipment.space.realtime_temperature - temperature_target if diff >= 1.0: self._equipment.air_valve_speed = AirValveSpeed.high elif diff >= 0.5: self._equipment.air_valve_speed = AirValveSpeed.medium elif diff >= 0.0: self._equipment.air_valve_speed = AirValveSpeed.low else: self._equipment.air_valve_speed = AirValveSpeed.off else: self._equipment.air_valve_speed = AirValveSpeed.off async def get_running_status(self): if self._equipment.air_valve_speed == AirValveSpeed.off: self._equipment.running_status = False else: self._equipment.running_status = True async def run(self): # await self.get_temperature_target() await self.get_air_valve_speed() await self.get_running_status() await self.get_mode() def get_results(self): return self._equipment class FCUControllerV2(EquipmentController): """Writen by Wenyan""" def __init__(self, equipment: FCU, season: Season): super(FCUControllerV2, self).__init__() self.equipment = equipment self.season = season def get_mode(self) -> int: if not np.isnan(self.equipment.water_in_temperature): if self.equipment.water_in_temperature > self.equipment.space.temperature_target: mode = 2 else: mode = 1 elif not np.isnan(self.equipment.supply_air_temperature): if self.equipment.supply_air_temperature > self.equipment.space.temperature_target: mode = 2 else: mode = 1 else: if self.season == Season.heating: mode = 2 elif self.season == Season.cooling: mode = 1 else: mode = 0 self.equipment.work_mode = mode return mode def build_air_valve_speed(self): target = self.equipment.space.temperature_target if target > 0: if self.equipment.space.realtime_temperature > 0: mode = self.get_mode() if mode == 2: diff = target - self.equipment.space.realtime_temperature elif mode == 1: diff = self.equipment.space.realtime_temperature - target else: diff = 0 if diff < 0: speed = AirValveSpeed.off elif diff < 0.5: speed = AirValveSpeed.low elif diff < 1.5: speed = AirValveSpeed.medium else: speed = AirValveSpeed.high self.equipment.air_valve_speed = speed else: self.equipment.air_valve_speed = AirValveSpeed.medium else: self.equipment.air_valve_speed = AirValveSpeed.off def build_running_status(self): if self.equipment.air_valve_speed == AirValveSpeed.off: self.equipment.running_status = False else: self.equipment.running_status = True def build_temperature_set(self): if np.isnan(self.equipment.space.temperature_target): self.equipment.setting_temperature = 0.0 else: self.equipment.setting_temperature = self.equipment.space.temperature_target async def run(self): self.build_air_valve_speed() self.build_running_status() self.get_mode() self.build_temperature_set() def get_results(self): return self.equipment @logger.catch() async def get_fcu_control_result(project_id: str, equipment_id: str) -> Dict: async with AsyncClient() as client: duo_duo = Duoduo(client, project_id) platform = DataPlatformService(client, project_id) spaces = await duo_duo.get_space_by_equipment(equipment_id) if project_id == 'Pj1101150003': supply_air_temperature = await platform.get_realtime_data(InfoCode.supply_temperature, equipment_id) water_in_temperature = await platform.get_realtime_data(InfoCode.water_in_temperature, equipment_id) else: supply_air_temperature = np.NAN water_in_temperature = np.NAN for sp in spaces: transfer = SpaceInfoService(client, project_id, sp.get('id')) current_target = await transfer.get_current_temperature_target() realtime_temperature = await platform.get_realtime_temperature(sp.get('id')) season = await duo_duo.get_season() temp_space_params = { 'id': sp.get('id'), 'temperature_target': current_target, 'realtime_temperature': realtime_temperature } space = Space(**temp_space_params) break fcu_temp_params = { 'id': equipment_id, 'space': space, 'supply_air_temperature': supply_air_temperature, 'water_in_temperature': water_in_temperature } fcu = FCU(**fcu_temp_params) if project_id == 'Pj1101050030': fcu_controller = FCUController(fcu, season) else: fcu_controller = FCUControllerV2(fcu, season) await fcu_controller.run() regulated_fcu = fcu_controller.get_results() output = { 'onOff': 1 if regulated_fcu.running_status else 0, 'mode': regulated_fcu.work_mode, 'speed': regulated_fcu.air_valve_speed.value, 'temperature': float(round_half_up(regulated_fcu.setting_temperature, 1)), 'water': 1 if regulated_fcu.running_status else 0 } return output