# -*- coding: utf-8 -*- from app.api.errors.iot import MissingIOTDataError 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.transfer import Season from app.utils.math import round_half_up from loguru import logger 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 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 and self._equipment.space.realtime_temperature: 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: 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): try: await self.get_air_valve_speed() await self.get_running_status() await self.get_mode() await self.get_temperature_target() except TypeError: raise MissingIOTDataError 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 and self.equipment.space.temperature_target: if self.equipment.water_in_temperature > self.equipment.space.temperature_target: mode = 2 else: mode = 1 elif self.equipment.supply_air_temperature and self.equipment.space.temperature_target: 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: if self.equipment.space.realtime_temperature: mode = self.get_mode() logger.debug(mode) if mode == 2: diff = target - self.equipment.space.realtime_temperature elif mode == 1: diff = self.equipment.space.realtime_temperature - target else: diff = 0 logger.debug(diff) 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 self.equipment.space.temperature_target: self.equipment.setting_temperature = self.equipment.space.temperature_target else: self.equipment.setting_temperature = 0.0 async def run(self): try: self.build_air_valve_speed() self.build_running_status() self.get_mode() self.build_temperature_set() except TypeError: raise MissingIOTDataError 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.feedback != FeedbackValue.null: speed = Speed.hold if not self.space_target: 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() else: if not self.space_target: speed = Speed.off else: if self.space_realtime: 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.3 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 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: try: speed = self.build_speed_set() switch_set = self.build_switch_set(speed) water_switch_set = self.build_water_valve_set(switch_set) except TypeError: raise MissingIOTDataError 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.feedback != FeedbackValue.null: if not self.space_target: 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() else: speed = Speed.hold else: if not self.space_target: speed = Speed.off else: if self.space_realtime: 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: speed = Speed.hold 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 and self.space_target: 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: try: 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) except TypeError: raise MissingIOTDataError 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 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) 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 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 async def build_acatfc2_instructions_v2(params: ACATFC2InstructionsRequest) -> dict: controller = ATFC2Controller( params.space_temperature_target, params.space_realtime_temperature, Season(params.season), params.feedback, params.running_status, params.speed, ) instructions = controller.run() return instructions async def build_acatfc4_instructions_v2(params: ACATFC4InstructionsRequest) -> dict: controller = ATFC4Controller( params.space_temperature_target, params.space_realtime_temperature, Season(params.season), params.feedback, params.running_status, params.speed, ) instructions = controller.run() return instructions