123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486 |
- # -*- 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
|