Reliability Prototype

h2integrate/converters/natural_gas/natural_gas_cc_ct.py

@@ -2,6 +2,7 @@
 from attrs import field, define, validators
 
 from h2integrate.core.utilities import BaseConfig, merge_shared_inputs
+from h2integrate.reliability.models import WeibullReliabilityModel
 from h2integrate.core.model_baseclasses import (
     CostModelBaseClass,
     CostModelBaseConfig,
@@ -66,6 +67,7 @@ def initialize(self):
         self.commodity = "electricity"
         self.commodity_rate_units = "MW"
         self.commodity_amount_units = "MW*h"
+        self.reliability_model = None
 
     def setup(self):
         super().setup()
@@ -74,6 +76,12 @@ def setup(self):
             merge_shared_inputs(self.options["tech_config"]["model_inputs"], "performance"),
             additional_cls_name=self.__class__.__name__,
         )
+        if use_reliability := "reliability" in self.options["tech_config"]["model_inputs"]:
+            self.reliability_model = WeibullReliabilityModel.from_dict(
+                merge_shared_inputs(self.options["tech_config"]["model_inputs"], "reliability"),
+                additional_cls_name=self.__class__.__name__,
+            )
+        self.use_reliability = use_reliability
 
         # Add natural gas consumed output
         self.add_output(
@@ -153,6 +161,8 @@ def compute(self, inputs, outputs):
             inputs["electricity_command_value"],
         )
         natural_gas_demand = electricity_command_value * heat_rate_mmbtu_per_mwh
+        if self.use_reliability:
+            natural_gas_demand * self.reliability_model.availability
 
         # available feedstock, saturated at maximum system feedstock consumption
         natural_gas_available = np.where(

h2integrate/reliability/models.py

@@ -0,0 +1,349 @@
+from abc import ABC, abstractmethod
+
+import numpy as np
+from attrs import field, define, validators
+from numpy.typing import ArrayLike
+
+from h2integrate.core.utilities import BaseConfig
+
+
+# generated from np.random.SeedSequence().entropy
+rng = np.random.default_rng(279299947538423226929715083173412195503)
+
+
+N_TIMESTEPS = 8760
+
+
+def create_failure_model(config: dict):
+    """Retrieves and initializes a matching reliability model."""
+    name = config["failure_model"]
+    fail_config = config["failure_parameters"]
+    match name:
+        case "WeibullReliability":
+            return WeibullReliability.from_dict(fail_config)
+        case "FixedIntervalReliability":
+            return FixedIntervalReliability.from_dict(fail_config)
+        case _:
+            raise NotImplementedError(f"{name} is not a valid model name")
+
+
+def create_maintenance_model(config: dict):
+    """Retrieves and initializes a matching reliability model."""
+    name = config["maintenance_model"]
+    maintenance_config = config["maintenance_parameters"]
+    match name:
+        case "WeibullReliability":
+            return WeibullReliability.from_dict(maintenance_config)
+        case "FixedIntervalReliability":
+            return FixedIntervalReliability.from_dict(maintenance_config)
+        case _:
+            raise NotImplementedError(f"{name} is not a valid model name")
+
+
+def generate_downtime_model(config: dict | int):
+    if not isinstance(config, dict):
+        return FixedDowntime(hours=config)
+    name = config["model"]
+    parameters = {k: v for k, v in config.items() if k != "model"}
+    match name:
+        case "LogNormalDowntime":
+            return LogNormalDowntime.from_dict(parameters)
+        case "FixedDowntime":
+            return FixedDowntime.from_dict(parameters)
+        case _:
+            raise NotImplementedError(f"{name} is not a valid model name")
+
+
+def update_dimensions(n_components: int, *args: np.ndarray):
+    """Update the dimensionality of a series of arrays and the value of :py:attr:`n_components`
+    to match. If the size of an array passed :py:attr:`args` is 1 and :py:attr:`n_components`
+    is greater than 1, all arrays passed to :py:attr:`args` will be broadcast to an array shaped
+    (:py:attr:`n_components`, 1). If the arrays are already larger than 1, then
+    :py:attr:`n_components will be updated to the size of the arrays.
+
+    Args:
+        n_components (int): Number of components in the model
+        args (np.ndarray): NumPy array of attribute values used to create a model.
+
+    Returns:
+        n_components: The value as passed or updated to match the size of arrays in :py:attr:`args`.
+        args: The arrays as passed or the arrays reshaped to shape (:py:attr:`n_components`, 1).
+    """
+    if args[0].size == 1 and n_components > 1:
+        shape = (n_components, 1)
+        for arg in args:
+            arg = np.broadcast_to(arg, shape)
+    n_components = args[0].size
+    return n_components, *args
+
+
+def float_array_converter(val: int | float | ArrayLike):
+    return np.array(val).astype(float).reshape(-1, 1)
+
+
+def int_array_converter(val: int | float | ArrayLike):
+    return np.array(val).astype(int).reshape(-1, 1)
+
+
+# NOTE: yes, I see the irony in creating a custom converter after removing all the attrs duplicates
+# NOTE: yes, I'll also be removing the above comment, but I'm sure someone will appreciate it
+def array_ge(val):
+    """Validates that all values of an array are greater than or equal to :py:attr:`val`."""
+
+    def validator(instance, attribute, value):
+        if any(value < val):
+            raise ValueError(f"'{attribute.name}' must have all values of at least {val}.")
+
+    return validator
+
+
+def array_gt(val):
+    """Validates that all values of an array are greater than or equal to :py:attr:`val`."""
+
+    def validator(instance, attribute, value):
+        if any(value <= val):
+            raise ValueError(f"'{attribute.name}' must have all values greater than {val}.")
+
+    return validator
+
+
+def match_shape(other):
+    """Validates the shape of an array matches the shape of :py:attr:`other`."""
+
+    def validator(instance, attribute, value):
+        other_arr = getattr(instance, other, None)
+        if other_arr is None:
+            raise ValueError(f"'{other}' does not exist.")
+        if other_arr.shape != value.shape:
+            msg = (
+                f"Shape of '{attribute.name}' {value.shape} does not match"
+                f" '{other}' {other_arr.shape}."
+            )
+            raise ValueError(msg)
+
+    return validator
+
+
+@define(kw_only=True)
+class BaseDowntime(ABC, BaseConfig):
+    @abstractmethod
+    def sample_downtime(self) -> np.ndarray: ...
+
+
+@define(kw_only=True)
+class BaseReliability(ABC, BaseConfig):
+    """Base reliability class responsible for common definitions and functionality.
+
+    Args:
+        burn_in (float): Number of years into the simulation to use as the starting point of the
+            the simulation's availability record. Defaults to 0.
+        n_components (int): Number of identical components to sample to avoid defining an array of
+            mean and sigma values when they are the same. After initialization this value changes
+            to align with the number of components being simulated, regardless of identicality.
+            Defaults to 1.
+        downtime (dict): Configuration for a downtime model.
+
+    Attributes:
+        time_to_failures (np.ndarray): Array of the hours to the next failure for each modeled
+            component. Generated by each subclass' :py:meth:`sample_events`.
+        downtime_per_event (np.ndarray): Number of hours of downtime corresponding to each downtime
+            event in :py:attr:`time_to_failures`.
+        availability (np.ndarray): Operational ratio of each modeled component for every time step
+            of the simulation.
+        system_availability (np.ndarray): Minimum operational ratio of all components for every time
+            time step of the simulation.
+    """
+
+    burn_in: float = field(default=0, converter=float, validator=validators.ge(0))
+    n_components: int = field(default=1, validator=(validators.instance_of(int), validators.ge(1)))
+    downtime: dict | BaseDowntime = field(converter=generate_downtime_model)
+
+    time_to_failures: np.ndarray = field(init=False, validator=validators.instance_of(np.ndarray))
+    downtime_per_event: np.ndarray = field(init=False, validator=validators.instance_of(np.ndarray))
+    availability: np.ndarray = field(init=False, validator=validators.instance_of(np.ndarray))
+    system_availability: np.ndarray = field(
+        init=False, validator=validators.instance_of(np.ndarray)
+    )
+
+    def sample_events(self) -> np.ndarray:
+        raise NotImplementedError("Failed to successfully subclass, please implement me.")
+
+    def create_downtime_events(self):
+        """Creates the ``time_to_failure`` and ``downtime_per_event`` arrays."""
+        self.time_to_failures = self.sample_events()
+        self.downtime_per_event = self.downtime.sample_downtime()
+
+    def calculate_availability(self):
+        """Determine the timing and duration of outages for a single year of simulation time."""
+        burn_in_time = np.ceil(self.burn_in * N_TIMESTEPS).astype(int)
+        simulation_end = burn_in_time + N_TIMESTEPS
+
+        accumulated = np.zeros((self.n_components, 1), dtype=int)
+        availability = np.ones((self.n_components, simulation_end), dtype=float)
+
+        while any(accumulated < simulation_end):
+            if not self.time_to_failures.size == 0:
+                self.create_downtime_events()
+            event = self.time_to_failures[:, 0].reshape(-1, 1)
+            self.time_to_failures = self.time_to_failures[:, 1:]
+            duration = self.downtime_per_event[:, 0].reshape(-1, 1)
+            self.downtime_per_event = self.downtime_per_event[:, 1:]
+
+            if all(event + accumulated) > simulation_end:
+                break
+
+            start = accumulated + event
+            end = start + duration
+            for i, (s, e) in enumerate(zip(start.flatten(), end.flatten())):
+                if s < simulation_end:
+                    e = min(simulation_end, e)
+                    availability[i, s:e] = 0
+            accumulated = end
+
+        self.availability = availability[:, burn_in_time:simulation_end]
+        self.system_availability = np.min(self.availability, axis=0)
+
+
+@define(kw_only=True)
+class FixedDowntime(BaseDowntime):
+    """Basic log-normal downtime model for generating the length of downtime for a given event.
+
+    Args:
+        hours (int | array-like): Length of downtime per event, in hours. Must be at least 1 hour.
+        n_components (int): Number of identical components to sample to avoid defining an array of
+            :py:attr:`hours` values when they are the same. Defaults to 1.
+    """
+
+    hours: int | ArrayLike = field(
+        converter=int_array_converter,
+        validator=(validators.instance_of(np.ndarray), array_ge(1)),
+    )
+    n_components: int = field(default=1, validator=(validators.instance_of(int), validators.ge(1)))
+
+    def __attrs_post_init__(self):
+        self.n_components, self.hours = update_dimensions(self.n_components, self.hours)
+
+    def sample_downtime(self):
+        """Return an array of 100 :py:attr:`hours`."""
+        return np.ones((1, 100), dtype=int) * self.hours
+
+
+@define(kw_only=True)
+class LogNormalDowntime(BaseDowntime):
+    """Basic log-normal downtime model for generating the length of downtime for a given event.
+
+    Args:
+        mean (float | array-like): Average length of downtime per event, in hours.
+        sigma (float | array-like): Standard deviation of the distribution(s), in hours.
+    """
+
+    mean: float = field(
+        converter=float_array_converter,
+        validator=(validators.instance_of(np.ndarray), array_ge(0)),
+    )
+    sigma: float = field(
+        converter=float_array_converter,
+        validator=(validators.instance_of(np.ndarray), match_shape("mean"), array_ge(0)),
+    )
+    n_components: int = field(default=1, validator=(validators.instance_of(int), validators.ge(1)))
+
+    def __attrs_post_init__(self):
+        self.n_components, self.mean, self.sigma = update_dimensions(
+            self.n_components, self.mean, self.sigma
+        )
+
+    def sample_downtime(self) -> np.ndarray:
+        """Return an array of 100 samples of each lognormal distribution."""
+        return np.ceil(rng.lognormal(self.mean, self.sigma, size=(self.mean.shape[0], 100))).astype(
+            int
+        )
+
+
+@define(kw_only=True)
+class WeibullReliability(BaseReliability):
+    r"""Basic reliability model for operating/not operating statuses.
+
+    Assumes a full operational shutdown with zero ramping of production for an hourly, 1 year
+    simulation.
+
+    Args:
+        scale (float): Also referred to as :math:`\lambda` or :math:`\alpha`. Determines
+            the scale of distribution, and is equivalent to the mean time
+            between failure in years (MTBF), or 1 / annual failure rate.
+        shape (float): Also referred to as ``k`` or :math:`\beta`. A value less than 1
+            corresponds to a decreasing hazard rate over time (break-in period failures);
+            a value greater than 1 corresponds to an increasing hazard rate over time (
+            aging/wear-out failures); and a value of 1 corresponds to a constant hazard
+            rate over time (exponential distribution).
+        downtime (float): Average amount of downtime per failure.
+
+    Attributes:
+        rng (np.random._generator.Generator): NumPy random generator object.
+
+    TODO:
+        - how to pass n_timesteps through from plant?
+        - burn-in
+    """
+
+    scale: float = field(
+        converter=float_array_converter,
+        validator=validators.instance_of(np.ndarray),
+    )
+    shape: float = field(
+        converter=float_array_converter,
+        validator=(validators.instance_of(np.ndarray), match_shape("scale")),
+    )
+
+    def __attrs_post_init__(self):
+        self.n_components, self.scale, self.shape = update_dimensions(
+            self.n_components, self.scale, self.shape
+        )
+
+        self.create_downtime_events()
+        self.calculate_availability()
+
+    def sample_events(self):
+        """Samples 100 events for each simulated component, rounding up to the nearest timestep."""
+        return np.ceil(
+            rng.weibull(self.shape, size=(self.shape.size, 100)) * self.scale * N_TIMESTEPS
+        ).astype(int)
+
+
+@define(kw_only=True)
+class FixedIntervalReliability(BaseReliability):
+    """Basic fixed interval downtime reliability model.
+
+    Args:
+        frequency (int | float | array-like): The annual frequency of events, e.g., 4 is equivalent
+            to a quarterly downtime event and 0.25 is equivalent to an every 4 years downtime event.
+            For all events the timing of the first event will be sampled within the first year or
+            interval period to offset events from being based on January 1st in an 8760.
+        downtime (int | float | dict): Either fixed length of each downtime, in hours, or a
+            configuration dictionary for a downtime length model.
+    """
+
+    frequency: float = field(
+        converter=float_array_converter,
+        validator=(validators.instance_of(np.ndarray), array_gt(0)),
+    )
+
+    def __attrs_post_init__(self):
+        self.n_components, self.frequency = update_dimensions(self.n_components, self.frequency)
+
+        self.create_downtime_events()
+        self.calculate_availability()
+
+    def sample_events(self):
+        """Creates the time to next failure array for each event's modality with the first event
+        occurring randomly either in the first year or first failure interval, whichever
+        comes first.
+
+        Returns:
+            time_to_failures (np.ndarray): An array of the next 100 events' time to next failure.
+        """
+        interval = np.ceil(8760 / (1 / self.frequency)).astype(int)
+        first_occurrence = rng.integers(0, np.where(interval > 8760, 8760, interval))
+        time_to_failures = np.hstack(
+            (first_occurrence, np.broadcast_to(interval, (interval.size, 99)))
+        )
+        return time_to_failures