from abc import ABC 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: pass