# -*- 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.models.domain.devices import ( ACATFC2InstructionsRequest, ACATFC4InstructionsRequest, ) from app.schemas.equipment import AirValveSpeed, FCU from app.schemas.instructions import ACATFCInstructions 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 self.equipment.water_in_temperature > 0: if ( self.equipment.water_in_temperature > self.equipment.space.temperature_target ): mode = 2 else: mode = 1 elif self.equipment.supply_air_temperature > 0: 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 @logger.catch() async def build_acatfc2_instructions( params: ACATFC2InstructionsRequest, ) -> ACATFCInstructions: space = Space( temperature_target=params.space_temperature_target, realtime_temperature=params.space_realtime_temperature, ) fcu = FCU(space=space, supply_air_temperature=-1.0, water_in_temperature=-1.0) controller = FCUControllerV2(fcu, Season(params.season)) await controller.run() regulated_fcu = controller.get_results() instructions = ACATFCInstructions( switch_set=1 if regulated_fcu.running_status else 0, speed_set=regulated_fcu.air_valve_speed.value, temperature_set=float(round_half_up(regulated_fcu.setting_temperature, 1)), mode_set=regulated_fcu.work_mode, water_valve_switch_set=1 if regulated_fcu.running_status else 0, ) return instructions @logger.catch() async def build_acatfc4_instructions( params: ACATFC4InstructionsRequest, ) -> ACATFCInstructions: space = Space( temperature_target=params.space_temperature_target, realtime_temperature=params.space_realtime_temperature, ) fcu = FCU(space=space) controller = FCUController(fcu, Season(params.season)) await controller.run() regulated_fcu = controller.get_results() instructions = ACATFCInstructions( switch_set=1 if regulated_fcu.running_status else 0, speed_set=regulated_fcu.air_valve_speed.value, temperature_set=float(round_half_up(regulated_fcu.setting_temperature, 1)), mode_set=regulated_fcu.work_mode, water_valve_switch_set=1 if regulated_fcu.running_status else 0, ) return instructions