from abc import ABC, abstractmethod from typing import Optional import numpy as np from loguru import logger from app.models.domain.diagnosis import ThermalComfortDiagnosisRequest from app.schemas.diagnosis import FaultCategory from app.schemas.equipment import AHU, FCU, VAVBox, VRF from app.services.transfer import Season class DiagnotorBase(ABC): def __init__(self): self._result = FaultCategory() def controller_check( self, setting_value: float, feedback_value: float, is_categorical: bool = True, tolerance_pct: float = 0.0 ) -> None: if is_categorical: if setting_value != feedback_value: self._result.controller_err = True else: if not setting_value * (1 - tolerance_pct) <= feedback_value <= setting_value * (1 + tolerance_pct): self._result.controller_err = True class ThermalComfortDiagnotor(DiagnotorBase): def __init__(self): super(ThermalComfortDiagnotor, self).__init__() class FCUDiagnotor(DiagnotorBase): def __init__(self, fcu: FCU, ahu: AHU, duration: float, target_temp: float, realtime_temp: float, season: Season): super(FCUDiagnotor, self).__init__() self._fcu = fcu self._ahu = ahu self._duration = duration self._target_temp = target_temp self._realtime_temp = realtime_temp self._season = season def low_in_heating(self) -> None: if self._fcu.air_valve_speed == 'off': if self._fcu.speed_limit == 'off': self._result.over_constrained = True else: self._result.control_logic_err = True else: if self._fcu.air_valve_speed == self._fcu.recommended_speed: self.controller_check(self._fcu.air_valve_speed_set, self._fcu.air_valve_speed) if not self._result.controller_err: if self._fcu.air_valve_speed_set == 'high': if self._duration > 120: if self._fcu.supply_air_temperature < self._target_temp + 1: self._result.insufficient_heating = True else: self._result.undefined_err = True else: self._result.no_problems_found = True else: self._result.control_logic_err = True else: if self._fcu.speed_limit == self._fcu.air_valve_speed: self._result.over_constrained = True else: self._result.obj_info_err = True def low_in_cooling(self) -> None: if self._fcu.recommended_speed == 'off': if self._fcu.air_valve_speed == self._fcu.recommended_speed: self.controller_check(self._fcu.air_valve_speed_set, self._fcu.air_valve_speed) if not self._result.controller_err: if self._duration > 120: self._result.undefined_err = True else: self._result.no_problems_found = True else: self._result.obj_info_err = True else: self._result.control_logic_err = True def high_in_heating(self) -> None: self.low_in_cooling() def high_in_cooling(self) -> None: self.low_in_heating() if self._result.insufficient_heating: self._result.insufficient_heating = False self._result.insufficient_cooling = True def run(self) -> None: if self._season == Season.heating or self._fcu.supply_air_temperature > self._target_temp: if self._realtime_temp < self._target_temp - 1: self.low_in_heating() elif self._realtime_temp > self._target_temp + 1: self.high_in_heating() elif self._season == Season.cooling or self._fcu.supply_air_temperature < self._target_temp: if self._realtime_temp < self._target_temp - 1: self.low_in_cooling() elif self._realtime_temp > self._target_temp + 1: self.high_in_cooling() class VAVDiagnotor(DiagnotorBase): def __init__(self, vav: VAVBox, ahu: AHU, known_err: FaultCategory, duration: float, season: Season): super(VAVDiagnotor, self).__init__() self._vav = vav self._ahu = ahu self._known_err = known_err self._duration = duration self._season = season def low_in_heating(self) -> None: self.controller_check( self._vav.supply_air_flow_set, self._vav.supply_air_flow, is_categorical=False, tolerance_pct=0.1 ) if self._vav.recommended_supply_air_flow >= self._vav.supply_air_flow_upper_limit: if self._vav.recommended_supply_air_flow == self._vav.supply_air_flow_set: if not self._result.controller_err: if self._duration > 120.0: if self._vav.supply_air_temperature < self._vav.virtual_target_temperature + 1 \ or self._ahu.supply_air_temperature < self._vav.virtual_target_temperature + 2: self._result.insufficient_heating = True else: if self._vav.supply_air_flow_upper_limit < 1200.0: self._result.unreasonable_vav_flow_limit = True else: if not self._known_err.sensor_err: self._result.undefined_err = True else: self._result.no_problems_found = True else: self._result.obj_info_err = True else: if self._vav.recommended_supply_air_flow != self._vav.supply_air_flow_set: self._result.obj_info_err = True def low_in_cooling(self) -> None: if self._vav.recommended_supply_air_flow == self._vav.supply_air_flow_set: self.controller_check( self._vav.supply_air_flow_set, self._vav.supply_air_flow, is_categorical=False, tolerance_pct=0.1 ) if self._vav.supply_air_flow_set <= self._vav.supply_air_flow_lower_limit: if not self._result.controller_err: if self._duration >= 120.0: if self._vav.supply_air_flow_lower_limit > 400: self._result.unreasonable_vav_flow_limit = True else: if not self._known_err.sensor_err: if self._vav.supply_air_temperature < 18 or self._ahu.supply_air_temperature < 18: self._result.excessive_cooling = True else: self._result.obj_info_err = True else: self._result.no_problems_found = True else: self._result.no_problems_found = True else: self._result.obj_info_err = True def high_in_heating(self) -> None: self.low_in_cooling() if self._result.excessive_cooling: self._result.excessive_cooling = False self._result.excessive_heating = True def high_in_cooling(self) -> None: self.low_in_heating() if self._vav.recommended_supply_air_flow >= self._vav.supply_air_flow_upper_limit: if self._vav.recommended_supply_air_flow == self._vav.supply_air_flow_set: if not self._result.controller_err: if self._duration > 120.0: if self._vav.supply_air_temperature > self._vav.virtual_target_temperature + 1 \ or self._ahu.supply_air_temperature > self._vav.virtual_target_temperature + 2: self._result.insufficient_cooling = True def run(self) -> None: if self._season == Season.heating or self._vav.supply_air_temperature > self._vav.virtual_target_temperature: if self._vav.virtual_realtime_temperature < self._vav.virtual_target_temperature - 1: self.low_in_heating() elif self._vav.virtual_realtime_temperature > self._vav.virtual_target_temperature + 1: self.high_in_heating() elif self._season == Season.cooling or self._vav.supply_air_temperature < self._vav.virtual_target_temperature: if self._vav.virtual_realtime_temperature < self._vav.virtual_target_temperature - 1: self.low_in_cooling() elif self._vav.virtual_realtime_temperature > self._vav.virtual_target_temperature + 1: self.high_in_cooling() class VRFDiagnotor(DiagnotorBase): def __init__(self, vrf: VRF): super(VRFDiagnotor, self).__init__() self._vrf = vrf def get_diagnosis_result(params: ThermalComfortDiagnosisRequest) -> FaultCategory: if params.fcu_list: for fcu in params.fcu_list: fcu_diagnotor = FCUDiagnotor( fcu, params.ahu, params.duration_minutes, params.target_temp, params.realtime_temp, params.season )