# -*- coding: utf-8 -*- from typing import Dict, Tuple 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: mode = 2 if ( temperature_target > 0 and self._equipment.space.realtime_temperature > 0 ): if temperature_target < self._equipment.space.realtime_temperature: 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 class ATFC2Controller: """Two-wipe ATFC logic""" def __init__(self, space_target: float, space_realtime: float, season: Season): self.space_target = space_target self.space_realtime = space_realtime self.season = season def build_speed_set(self) -> str: if self.space_realtime > 0.0 and self.space_target > 0.0: if self.season == Season.cooling: diff = self.space_realtime - self.space_target elif self.season == Season.heating: diff = self.space_target - self.space_realtime else: diff = 0.0 if diff <= 0.0: speed = "off" elif diff < 0.5: speed = "low" elif diff < 1.5: speed = "medium" else: speed = "high" else: speed = "hold" if self.space_target < 0.0: speed = "off" return speed def build_switch_set(self, new_speed_set: str) -> str: if new_speed_set == "off": switch_set = "off" else: switch_set = "on" return switch_set def build_water_valve_set(self, new_switch_set: str) -> str: return new_switch_set def run(self) -> dict: speed = self.build_speed_set() switch_set = self.build_switch_set(speed) water_switch_set = self.build_water_valve_set(switch_set) res = { "speed_set": speed, "switch_set": switch_set, "water_valve_set": water_switch_set, } return res class ATFC4Controller(ATFC2Controller): """Four-wipe ATFC logic""" def __init__(self, space_target: float, space_realtime: float, season: Season): super().__init__(space_target, space_realtime, season) def build_speed_set(self) -> str: speed = super().build_speed_set() if self.season == Season.heating: if self.space_target > 0.0 and self.space_realtime > 0.0: diff = abs(self.space_realtime - self.space_target) if diff < 0.2: speed = "off" elif diff < 0.8: speed = "low" elif diff < 1.5: speed = "medium" else: speed = "high" return speed def build_water_valve_set(self, new_switch_set: str) -> Tuple[str, str]: if new_switch_set == "off": chill_water_valve_set = "off" hot_water_valve_set = "off" else: if self.season == Season.heating: if self.space_realtime > 0.0: if self.space_realtime > self.space_target: chill_water_valve_set = "on" hot_water_valve_set = "off" else: chill_water_valve_set = "off" hot_water_valve_set = "on" else: chill_water_valve_set = "hold" hot_water_valve_set = "hold" else: chill_water_valve_set = "on" hot_water_valve_set = "off" return chill_water_valve_set, hot_water_valve_set def run(self) -> dict: speed = self.build_speed_set() switch_set = self.build_switch_set(speed) chill_water_valve_set, hot_water_valve_set = self.build_water_valve_set( switch_set ) res = { "speed_set": speed, "switch_set": switch_set, "chill_water_valve_set": chill_water_valve_set, "hot_water_valve_set": hot_water_valve_set, } return res @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 @logger.catch() async def build_acatfc2_instructions_v2(params: ACATFC2InstructionsRequest) -> dict: space_realtime = ( params.space_realtime_temperature if params.space_realtime_temperature else -1 ) controller = ATFC2Controller( params.space_temperature_target, space_realtime, Season(params.season) ) instructions = controller.run() return instructions @logger.catch() async def build_acatfc4_instructions_v2(params: ACATFC4InstructionsRequest) -> dict: space_realtime = ( params.space_realtime_temperature if params.space_realtime_temperature else -1 ) controller = ATFC4Controller( params.space_temperature_target, space_realtime, Season(params.season) ) instructions = controller.run() return instructions