AHU fault equations lack how to handle air side energy recovery units (ERV)

Think about creating a fault code 16 for an AHU to handle ERV. Maybe something like:

Assume 70 to 80% heat transfer between the exhaust and outdoor air streams. The mixed air temperature to the AHU is the return air temperature multiplied by the return air percent and the ERV leaving outdoor air temperature multiplied by the percent leaving outdoor air temperature.

Fault Condition 16: ERV Ineffective Fault

Objective:

To detect if the ERV is not performing efficiently, which could be indicated by the mixed air temperature not aligning with the expected temperatures based on the return air, outdoor air, and ERV heat exchange efficiency.

Fault Logic:

The mixed air temperature (MAT) should be calculated as a weighted average of the return air temperature (RAT) and the ERV-leaving outdoor air temperature (ERV_OAT_LEAVING).
On the exhaust side, the temperature difference between the ERV exhaust entering air temperature (ERV_EAT_ENTER) and the ERV exhaust leaving air temperature (ERV_EAT_LEAVING) should be within the expected range given the ERV’s heat transfer efficiency.

import pandas as pd
import numpy as np
from open_fdd.air_handling_unit.faults.fault_condition import (
    FaultCondition,
    MissingColumnError,
)
import sys

class FaultConditionSixteen(FaultCondition):
    """Class provides the definitions for Fault Condition 16.
    ERV Ineffective Process.
    """

    def __init__(self, dict_):
        super().__init__()

        # Threshold parameters
        self.erv_efficiency_min = dict_.get("ERV_EFFICIENCY_MIN", 0.7)  # 70% min expected efficiency
        self.erv_efficiency_max = dict_.get("ERV_EFFICIENCY_MAX", 0.8)  # 80% max expected efficiency

        # Validate that efficiency parameters are floats
        for param, value in [
            ("erv_efficiency_min", self.erv_efficiency_min),
            ("erv_efficiency_max", self.erv_efficiency_max),
        ]:
            if not isinstance(value, float):
                raise InvalidParameterError(
                    f"The parameter '{param}' should be a float, but got {type(value).__name__}."
                )

        # Other attributes
        self.erv_oat_enter_col = dict_.get("ERV_OAT_ENTER_COL", None)
        self.erv_oat_leaving_col = dict_.get("ERV_OAT_LEAVING_COL", None)
        self.erv_eat_enter_col = dict_.get("ERV_EAT_ENTER_COL", None)
        self.erv_eat_leaving_col = dict_.get("ERV_EAT_LEAVING_COL", None)
        self.mat_col = dict_.get("MAT_COL", None)
        self.rat_col = dict_.get("RAT_COL", None)
        self.supply_vfd_speed_col = dict_.get("SUPPLY_VFD_SPEED_COL", None)
        self.troubleshoot_mode = dict_.get("TROUBLESHOOT_MODE", False)
        self.rolling_window_size = dict_.get("ROLLING_WINDOW_SIZE", None)

        self.equation_string = (
            "fc16_flag = 1 if ERV process is ineffective based on temperatures and expected efficiency "
            "for N consecutive values else 0 \n"
        )
        self.description_string = (
            "Fault Condition 16: ERV Ineffective Fault \n"
        )
        self.required_column_description = (
            "Required inputs are the ERV outside air entering temperature, ERV outside air leaving temperature, "
            "ERV exhaust entering temperature, ERV exhaust leaving temperature, mixed air temperature, return air temperature, "
            "and supply fan VFD speed \n"
        )
        self.error_string = "One or more required columns are missing or None \n"

        self.set_attributes(dict_)

        # Set required columns specific to this fault condition
        self.required_columns = [
            self.erv_oat_enter_col,
            self.erv_oat_leaving_col,
            self.erv_eat_enter_col,
            self.erv_eat_leaving_col,
            self.mat_col,
            self.rat_col,
            self.supply_vfd_speed_col,
        ]

        # Check if any of the required columns are None
        if any(col is None for col in self.required_columns):
            raise MissingColumnError(
                f"{self.error_string}"
                f"{self.equation_string}"
                f"{self.description_string}"
                f"{self.required_column_description}"
                f"{self.required_columns}"
            )

        # Ensure all required columns are strings
        self.required_columns = [str(col) for col in self.required_columns]

        self.mapped_columns = (
            f"Your config dictionary is mapped as: {', '.join(self.required_columns)}"
        )

    def get_required_columns(self) -> str:
        """Returns a string representation of the required columns."""
        return (
            f"{self.equation_string}"
            f"{self.description_string}"
            f"{self.required_column_description}"
            f"{self.mapped_columns}"
        )

    def apply(self, df: pd.DataFrame) -> pd.DataFrame:
        try:
            # Ensure all required columns are present
            self.check_required_columns(df)

            if self.troubleshoot_mode:
                self.troubleshoot_cols(df)

            # Calculate expected ERV efficiency on the outdoor air stream
            df["erv_efficiency_oa"] = (
                (df[self.erv_oat_leaving_col] - df[self.erv_oat_enter_col]) / 
                (df[self.rat_col] - df[self.erv_oat_enter_col])
            )

            # Calculate expected ERV efficiency on the exhaust air stream
            df["erv_efficiency_ea"] = (
                (df[self.erv_eat_enter_col] - df[self.erv_eat_leaving_col]) / 
                (df[self.erv_eat_enter_col] - df[self.oat_col])
            )

            # Determine if efficiency is within acceptable range
            df["erv_efficiency_check"] = (
                (df["erv_efficiency_oa"] < self.erv_efficiency_min) | 
                (df["erv_efficiency_oa"] > self.erv_efficiency_max) |
                (df["erv_efficiency_ea"] < self.erv_efficiency_min) | 
                (df["erv_efficiency_ea"] > self.erv_efficiency_max)
            )

            # Check if mixed air temperature is within the expected range
            df["mat_expected"] = (
                df[self.rat_col] * (1 - df["erv_efficiency_oa"]) + 
                df[self.erv_oat_leaving_col] * df["erv_efficiency_oa"]
            )

            df["mat_check"] = abs(df[self.mat_col] - df["mat_expected"]) > self.mix_degf_err_thres

            # Combined fault condition check
            df["combined_check"] = df["erv_efficiency_check"] & df["mat_check"] & (
                df[self.supply_vfd_speed_col] > 0.01
            )

            # Rolling sum to count consecutive trues
            rolling_sum = df["combined_check"].rolling(window=self.rolling_window_size).sum()
            df["fc16_flag"] = (rolling_sum >= self.rolling_window_size).astype(int)

            if self.troubleshoot_mode:
                print("Troubleshoot mode enabled - not removing helper columns")
                sys.stdout.flush()

            # Optionally remove temporary columns
            df.drop(
                columns=["erv_efficiency_oa", "erv_efficiency_ea", "erv_efficiency_check", "mat_expected", "mat_check", "combined_check"],
                inplace=True,
            )

            return df

        except MissingColumnError as e:
            print(f"Error: {e.message}")
            sys.stdout.flush()
            raise e
        except InvalidParameterError as e:
            print(f"Error: {e.message}")
            sys.stdout.flush()
            raise e

bbartling / open-fdd

AHU fault equations lack how to handle air side energy recovery units (ERV) #22

Fault Condition 16: ERV Ineffective Fault