# -*- 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, Speed, ) from app.models.domain.feedback import FeedbackValue 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 SpeedController: def __init__(self, current_speed: Speed, running_status: bool) -> None: if running_status: self.current_speed = current_speed else: self.current_speed = Speed.off self.next_speed = current_speed def turn_up(self) -> Speed: if self.current_speed == Speed.off: self.next_speed = Speed.low elif self.current_speed == Speed.low: self.next_speed = Speed.medium elif self.current_speed == Speed.medium: self.next_speed = Speed.high return self.next_speed def turn_down(self) -> Speed: if self.current_speed == Speed.high: self.next_speed = Speed.medium elif self.current_speed == Speed.medium: self.next_speed = Speed.low elif self.current_speed == Speed.low: self.next_speed = Speed.off return self.next_speed def turn_on(self) -> Speed: self.next_speed = Speed.medium return self.next_speed def turn_off(self) -> Speed: self.next_speed = Speed.off return self.next_speed def get_next_speed(self) -> Speed: return self.next_speed class ATFC2Controller: """Two-wipe ATFC logic""" def __init__( self, space_target: float, space_realtime: float, season: Season, feedback: FeedbackValue, running_status: bool, speed: Speed, ): self.space_target = space_target self.space_realtime = space_realtime self.season = season self.feedback = feedback self.running_status = running_status self.speed = speed def build_speed_set(self) -> Speed: speed_controller = SpeedController(self.speed, self.running_status) 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.2: speed = Speed.off elif diff <= 0.2: speed = Speed.hold elif diff < 0.5: speed = Speed.low elif diff < 1.5: speed = Speed.medium else: speed = Speed.high else: speed = Speed.hold if self.space_target < 0.0: speed = Speed.off else: if self.feedback == FeedbackValue.switch_on: speed = Speed.medium elif ( self.feedback == FeedbackValue.so_hot or self.feedback == FeedbackValue.a_little_hot ): if self.season == Season.cooling: speed = speed_controller.turn_up() elif self.season == Season.heating: speed = speed_controller.turn_down() elif ( self.feedback == FeedbackValue.so_cold or self.feedback == FeedbackValue.a_little_cold ): if self.season == Season.cooling: speed = speed_controller.turn_down() elif self.season == Season.heating: speed = speed_controller.turn_up() return speed @staticmethod def build_switch_set(new_speed_set: Speed) -> str: if new_speed_set == Speed.off: switch_set = "off" elif new_speed_set == Speed.hold: switch_set = "hold" 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, feedback: FeedbackValue, running_status: bool, speed: Speed, ): super().__init__( space_target, space_realtime, season, feedback, running_status, speed ) def build_speed_set(self) -> Speed: speed = super().build_speed_set() if self.season == Season.heating: speed_controller = SpeedController(self.speed, self.running_status) 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 = Speed.hold elif diff < 0.5: speed = Speed.low elif diff < 1.5: speed = Speed.medium else: speed = Speed.high else: if self.space_target < 0.0: speed = Speed.off else: if self.feedback == FeedbackValue.switch_on: speed = Speed.medium elif ( self.feedback == FeedbackValue.so_hot or self.feedback == FeedbackValue.a_little_hot ): speed = speed_controller.turn_up() elif ( self.feedback == FeedbackValue.so_cold or self.feedback == FeedbackValue.a_little_cold ): speed = speed_controller.turn_up() 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 >= 0.2: chill_water_valve_set = "on" hot_water_valve_set = "off" elif self.space_target - self.space_realtime >= 0.2: 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 = "hold" hot_water_valve_set = "hold" if self.feedback == FeedbackValue.switch_on: chill_water_valve_set = "off" hot_water_valve_set = "on" elif ( self.feedback == FeedbackValue.so_hot or self.feedback == FeedbackValue.a_little_hot ): chill_water_valve_set = "on" hot_water_valve_set = "off" elif ( self.feedback == FeedbackValue.so_cold or self.feedback == FeedbackValue.a_little_cold ): chill_water_valve_set = "off" hot_water_valve_set = "on" 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), params.feedback, params.running_status, params.speed, ) 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), params.feedback, params.running_status, params.speed, ) instructions = controller.run() return instructions