Strategy Pattern with a Perception Layer

[project]
name = "Thermal Runaway Predictor - ML 1.2.2"
version = "0.1.0"
description = "ML thermal runaway predictor"
authors = [{ name = "John Doe", email = "[email protected]" }]
dependencies = [
    "composabl-core",
    "scikit-learn==1.2.2"
]

[composabl]
type = "perceptor"
entrypoint = "thermal_runaway_predictor.perceptor:ThermalRunawayPredict"

# Include additional data files
[tool.setuptools.packages.find]
where = ["thermal_runaway_predictor"]

[tool.setuptools.package-data]
"*" = ["*.json", "*.pkl"]

from composabl_core import PerceptorImpl

#######
import os
import pickle

# Determine the directory where the current script is located
path = os.path.dirname(os.path.realpath(__file__))

class ThermalRunawayPredict(PerceptorImpl):
    def __init__(self, *args, **kwargs):
        """
        Initialize the ThermalRunawayPredict perceptor with default values and load the machine learning model.
        
        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.
        """
        # Initialize the prediction output variable
        self.y = 0
        
        # Initialize a flag to indicate thermal runaway status
        self.thermal_run = 0
        
        # Load the pre-trained machine learning model from a pickle file
        # The model is expected to be located in the 'ml_models' directory relative to the script's path
        model_path = os.path.join(path, "ml_models", "ml_predict_temperature_122.pkl")
        try:
            with open(model_path, 'rb') as model_file:
                self.ml_model = pickle.load(model_file)
        except FileNotFoundError:
            print(f"Machine learning model not found at {model_path}. Please ensure the model file exists.")
            self.ml_model = None
        except Exception as e:
            print(f"An error occurred while loading the ML model: {e}")
            self.ml_model = None
        
        # Initialize a list to store historical ML predictions if needed
        self.ML_list = []
        
        # Initialize the last recorded 'Tc' value to compute its change (ΔTc)
        self.last_Tc = 0

    async def compute(self, obs_spec, obs):
        """
        Compute the thermal runaway prediction based on current sensor observations.
        
        Args:
            obs_spec: Observation specification (not used in this implementation).
            obs: Current sensor observations. Can be a list or a dictionary.
        
        Returns:
            dict: A dictionary containing the thermal runaway prediction.
                  Example: {"thermal_runaway_predict": 1}
        """
        # Ensure that 'obs' is a dictionary. If not, convert it using predefined sensor keys.
        if not isinstance(obs, dict):
            # Define the expected sensor keys
            obs_keys = ['T', 'Tc', 'Ca', 'Cref', 'Tref', 'Conc_Error', 'Eps_Yield', 'Cb_Prod']
            # Convert the list to a dictionary by zipping it with the sensor keys
            obs = dict(zip(obs_keys, obs))
            print("Converted 'obs' to dictionary format using predefined sensor keys.")
        
        # Calculate the change in 'Tc' (ΔTc) since the last observation
        if self.last_Tc == 0:
            # If this is the first observation, assume an initial ΔTc of 5
            self.ΔTc = 5
        else:
            # Compute ΔTc as the difference between current 'Tc' and the last recorded 'Tc'
            try:
                current_Tc = float(obs['Tc'])
                self.ΔTc = current_Tc - self.last_Tc
            except (KeyError, ValueError, TypeError) as e:
                # Handle cases where 'Tc' is missing or cannot be converted to float
                print(f"Error accessing or converting 'Tc': {e}")
                self.ΔTc = 0  # Default to 0 if there's an error
        
        # Initialize the prediction output
        y = 0
        
        # Check if the current temperature 'T' exceeds or equals 340
        try:
            current_T = float(obs['T'])
        except (KeyError, ValueError, TypeError) as e:
            print(f"Error accessing or converting 'T': {e}")
            current_T = 0  # Default to 0 if there's an error
        
        if current_T >= 340:
            # Prepare the feature vector for the ML model
            try:
                Ca = float(obs['Ca'])
                Cref = float(obs['Cref'])
            except (KeyError, ValueError, TypeError) as e:
                print(f"Error accessing or converting 'Ca' or 'Cref': {e}")
                Ca = 0
                Cref = 0
            
            # Feature vector: [Ca, T, Tc, ΔTc]
            X = [[Ca, current_T, self.ΔTc]]
            
            # If the ML model was loaded successfully, make a prediction
            if self.ml_model:
                try:
                    # Predict the probability of thermal runaway
                    y_proba = self.ml_model.predict_proba(X)
                    
                    # Get the predicted class label (e.g., 0 or 1)
                    y = self.ml_model.predict(X)[0]
                    
                    # Optionally, use the probability to adjust prediction confidence
                    # For example, set y=1 only if the probability of class 1 is >= 0.3
                    if y_proba[0][1] >= 0.3:
                        y = 1
                    else:
                        y = 0
                except Exception as e:
                    print(f"Error during ML model prediction: {e}")
                    y = 0
            else:
                print("ML model is not loaded. Cannot make predictions.")
                y = 0
        
        # Update the last recorded 'Tc' with the current value for the next computation
        try:
            self.last_Tc = float(obs['Tc'])
        except (KeyError, ValueError, TypeError) as e:
            print(f"Error accessing or converting 'Tc' for updating last_Tc: {e}")
            self.last_Tc = self.last_Tc  # Keep the previous value if there's an error
        
        # Optionally, store the prediction in ML_list for historical tracking
        self.ML_list.append(y)
        
        # Update the prediction output variable
        self.y = y
        
        # Return the prediction as a dictionary
        return {"thermal_runaway_predict": y}

    def filtered_sensor_space(self, obs):
        """
        Define which sensors are relevant for this perceptor.
        
        Args:
            obs: Current sensor observations (not used in this implementation).
        
        Returns:
            list: Names of the sensors to be used.
        """
        # Specify the sensors that this perceptor will use
        return ['T', 'Tc', 'Ca', 'Cref', 'Tref', 'Conc_Error', 'Eps_Yield', 'Cb_Prod']