env_regulation_api/app/controllers/diagnosis/thermal_comfort.py

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