PrognosticsModel#
- class progpy.PrognosticsModel(**kwargs)#
A general time-variant state space model of system degradation behavior.
The PrognosticsModel class is a wrapper around a mathematical model of a system as represented by a state, output, input, event_state and threshold equation.
A Model also has a parameters structure, which contains fields for various model parameters.
- Keyword Arguments
process_noise (Optional, float or dict[str, float]) – Process noise (applied at dx/next_state). Can be number (e.g., .2) applied to every state, a dictionary of values for each state (e.g., {‘x1’: 0.2, ‘x2’: 0.3}), or a function (x) -> x
process_noise_dist (Optional, str) – distribution for process noise (e.g., normal, uniform, triangular)
measurement_noise (Optional, float or dict[str, float]) – Measurement noise (applied in output eqn). Can be number (e.g., .2) applied to every output, a dictionary of values for each output (e.g., {‘z1’: 0.2, ‘z2’: 0.3}), or a function (z) -> z
measurement_noise_dist (Optional, str) – distribution for measurement noise (e.g., normal, uniform, triangular)
integration_method (Optional, str or OdeSolver) – Integration method used by next_state in continuous models, e.g. ‘rk4’ or ‘euler’ (default: ‘euler’). Could also be a SciPy integrator (e.g., scipy.integrate.RK45). If the model is discrete, this parameter will raise an exception.
Additional parameters specific to the model
- Raises
Example
>>> m = PrognosticsModel(process_noise=3.2) Traceback (most recent call last): ... TypeError: Must have `states` attribute >>> m2 = PrognosticsModel(integration_method='rk4') Traceback (most recent call last): ... TypeError: Must have `states` attribute >>> m3 = PrognosticsModel(integration_method=sp.integrate.RK45) Traceback (most recent call last): ... NameError: name 'sp' is not defined
- is_vectorized#
True if the model is vectorized, False otherwise. Default is False
- Type
bool, optional
- parameters#
Parameters for the specific model object. This is created automatically from the default_parameters and kwargs
- state_limits#
Limits on the state variables format {‘state_name’: (lower_limit, upper_limit)}
- OutputContainer#
Class for output container - used for representing output
- Type
DictLikeMatrixWrapper
- apply_limits(x)#
Apply state bound limits. Any state outside of limits will be set to the closest limit.
- Parameters
x (StateContainer or dict) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Returns
x – Bounded state, with keys defined by model.states e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Return type
StateContainer or dict
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> x = m.apply_limits(x) # Returns bounded state
- apply_measurement_noise(z)#
Apply measurement noise to the measurement
- Parameters
z (OutputContainer) –
output, with keys defined by model.outputs
e.g., z = m.OutputContainer({‘abc’: 332.1, ‘def’: 221.003}) given outputs = [‘abc’, ‘def’]
- Returns
z – output, with applied noise, with keys defined by model.outputs
e.g., z = m.OutputContainer({‘abc’: 332.2, ‘def’: 221.043}) given outputs = [‘abc’, ‘def’]
- Return type
OutputContainer
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() >>> z = m.OutputContainer({'z1': 2.2}) >>> z = m.apply_measurement_noise(z)
Note
Configured using parameters measurement_noise and measurement_noise_dist
- apply_process_noise(x, dt: float = 1)#
Apply process noise to the state
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
dt (float, optional) – Time step (e.g., dt = 0.1)
- Returns
x – state, with applied noise, with keys defined by model.states e.g., x = m.StateContainer({‘abc’: 332.2, ‘def’: 221.043}) given states = [‘abc’, ‘def’]
- Return type
StateContainer
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> x = m.apply_process_noise(x) # {'tb': 292.04728504765023, 'qb': 7856.398056966799, 'qcp': 0.1736435913459429, 'qcs': 0.053705829558158086}
Note
Configured using parameters process_noise and process_noise_dist
- calc_error(times: List[float], inputs: List[progpy.utils.containers.DictLikeMatrixWrapper], outputs: List[progpy.utils.containers.DictLikeMatrixWrapper], _loc=None, **kwargs) float#
Calculate Error between simulated and observed data using selected Error Calculation Method
- Parameters
- Keyword Arguments
method (str, optional) –
Error method to use when calculating error. Supported methods include:
MSE (Mean Squared Error) - DEFAULT
RMSE (Root Mean Squared Error)
MAX_E (Maximum Error)
MAE (Mean Absolute Error)
MAPE (Mean Absolute Percentage Error)
DTW (Dynamic Time Warping)
x0 (StateContainer, optional) – Initial state
dt (float, optional) – Maximum time step in simulation. Time step used in simulation is lower of dt and time between samples. Defaults to time between samples.
stability_tol (double, optional) –
Configurable parameter. Configurable cutoff value, between 0 and 1, that determines the fraction of the data points for which the model must be stable. In some cases, a prognostics model will become unstable under certain conditions, after which point the model can no longer represent behavior. stability_tol represents the fraction of the provided argument times that are required to be met in simulation, before the model goes unstable in order to produce a valid estimate of mean squared error.
If the model goes unstable before stability_tol is met, a ValueError is raised. Else, model goes unstable after stability_tol is met, the mean squared error calculated from data up to the instability is returned.
aggr_method (func, optional) – When multiple runs are provided, users can state how to aggregate the results of the errors. Defaults to taking the mean.
- Returns
error
- Return type
See also
calc_error.MSE()calc_error.RMSE()calc_error.MAX_E()calc_error.MAPE()calc_error.MAE():func:’calc_error.DTW’
- dx(x, u)#
Calculate the first derivative of state x at a specific time t, given state and input
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
u (InputContainer) –
Inputs, with keys defined by model.inputs
e.g., u = m.InputContainer({‘i’:3.2}) given inputs = [‘i’]
- Returns
dx – First derivative of state, with keys defined by model.states
e.g., dx = m.StateContainer({‘abc’: 3.1, ‘def’: -2.003}) given states = [‘abc’, ‘def’]
- Return type
StateContainer
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'i': 2.0}) >>> z = m.OutputContainer({'v': 3.2, 't': 295}) >>> x = m.initialize(u, z) # Initialize first state >>> dx = m.dx(x, u) # Returns first derivative of state given input u
See also
Note
A model should overwrite either next_state or dx. Override dx for continuous models, and next_state for discrete, where the behavior cannot be described by the first derivative
- estimate_params(runs: List[tuple] = None, keys: List[str] = None, times: List[float] = None, inputs: List[progpy.utils.containers.DictLikeMatrixWrapper] = None, outputs: List[progpy.utils.containers.DictLikeMatrixWrapper] = None, method: str = 'nelder-mead', **kwargs) None#
Estimate the model parameters given data. Overrides model parameters
- Keyword Arguments
inputs (list[InputContainer]) – Array of input containers where input[x] corresponds to time[x]
outputs (list[OutputContainer]) – Array of output containers where output[x] corresponds to time[x]
method (str, optional) – Optimization method- see scipy.optimize.minimize for options
tol (int, optional) – Tolerance for termination. Depending on the provided minimization method, specifying tolerance sets solver-specific options to tol
error_method (str, optional) – Method to use in calculating error. See calc_error for options
bounds (tuple or dict, optional) – Bounds for optimization in format ((lower1, upper1), (lower2, upper2), …) or {key1: (lower1, upper1), key2: (lower2, upper2), …}
options (dict, optional) – Options passed to optimizer. see scipy.optimize.minimize for options
runs (list[tuple], depreciated) – data from all runs, where runs[0] is the data from run 0. Each run consists of a tuple of arrays of times, input dicts, and output dicts. Use inputs, outputs, states, times, etc. instead
- Returns
Scipy minimize Optimization Result from estimating parameters. See scipy’s scipy.optimize.OptimizeResult documentation for details.
- Return type
OptimizeResult
See: examples.param_est
- event_state(x) dict#
Calculate event states (i.e., measures of progress towards event (0-1, where 0 means event has occurred))
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Returns
event_state – Event States, with keys defined by prognostics_model.events.
e.g., event_state = {‘EOL’:0.32} given events = [‘EOL’]
- Return type
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> event_state = m.event_state(x) # {'EOD': 1.0}
Note
If not overridden, will return 0.0 if threshold_met returns True, otherwise 1.0. If neither threshold_met or event_state is overridden, will return an empty dictionary (i.e., no events)
See also
- classmethod from_json(data: str)#
Create a new prognostics model from a previously generated model that was serialized as a JSON object
- Parameters
data (str) – JSON serialized parameters necessary to build a model See to_json method
- Returns
Model generated from serialized parameters
- Return type
See also
Note
This serialization only works for models that include all parameters necessary to generate the model in model.parameters.
- generate_surrogate(load_functions: List[collections.abc.Callable], method: str = 'dmd', **kwargs)#
Generate a surrogate model to approximate the higher-fidelity model
- Parameters
load_functions (List[abc.Callable]) – Each index is a callable loading function of (t, x = None) -> z used to predict future loading (output) at a given time (t) and state (x)
method (str, optional) – list[ indicating surrogate modeling method to be used
- Keyword Arguments
dt (float or abc.Callable, optional) – Same as in simulate_to_threshold; for DMD, this value is the time step of the training data
save_freq (float, optional) – Same as in simulate_to_threshold; for DMD, this value is the time step with which the surrogate model is generated
state_keys (List[str], optional) – List of state keys to be included in the surrogate model generation. keys must be a subset of those defined in the PrognosticsModel
input_keys (List[str], optional) – List of input keys to be included in the surrogate model generation. keys must be a subset of those defined in the PrognosticsModel
output_keys (List[str], optional) – List of output keys to be included in the surrogate model generation. keys must be a subset of those defined in the PrognosticsModel
event_keys (List[str], optional) – List of event_state keys to be included in the surrogate model generation. keys must be a subset of those defined in the PrognosticsModel
... (optional) – Keyword arguments from simulate_to_threshold (except save_pts)
- Returns
SurrogateModel() – Instance of SurrogateModel class
- Return type
class
Example
See examples/generate_surrogate
- initialize(u=None, z=None)#
Calculate initial state given inputs and outputs. If not defined for a model, it will return parameters[‘x0’]
- Parameters
u (InputContainer) –
Inputs, with keys defined by model.inputs
e.g., u = m.InputContainer({‘i’:3.2}) given inputs = [‘i’]
z (OutputContainer) –
Outputs, with keys defined by model.outputs
e.g., z = m.OutputContainer({‘t’:12.4, ‘v’:3.3}) given outputs = [‘t’, ‘v’]
- Returns
x – First state, with keys defined by model.states
e.g., x = StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Return type
StateContainer
Example
: >>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace above with specific model being simulated ^ >>> u = m.InputContainer({‘u1’: 3.2}) >>> z = m.OutputContainer({‘z1’: 2.2}) >>> x = m.initialize(u, z) # Initialize first state
- property is_continuous: bool#
returns: is_continuous – True if model is continuous, False if discrete :rtype: bool
- property is_direct: bool#
New in version 1.5.0.
If the model is a “direct model” - i.e., a model that directly estimates time of event from system state, rather than using state transition. This is useful for data-driven models that map from sensor data to time of event, and for physics-based models where state transition differential equations can be solved.
- Returns
if the model is a direct model
- Return type
- property is_discrete: bool#
returns: is_discrete – True if model is discrete, False if continuous :rtype: bool
- property is_state_transition_model: bool#
New in version 1.5.0.
If the model is a “state transition model” - i.e., a model that uses state transition differential equations to propagate state forward.
- Returns
if the model is a state transition model
- Return type
- next_state(x, u, dt: float)#
State transition equation: Calculate next state
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
u (InputContainer) –
Inputs, with keys defined by model.inputs
e.g., u = m.InputContainer({‘i’:3.2}) given inputs = [‘i’]
dt (float) –
Timestep size in seconds (≥ 0)
e.g., dt = 0.1
- Returns
x – Next state, with keys defined by model.states e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Return type
StateContainer
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> x = m.next_state(x, u, 0.1) # Returns state at 3.1 seconds given input u
???{‘tb’: 292.1, ‘qb’: nan, ‘qcp’: nan, ‘qcs’: nan}
See also
Note
A model should overwrite either next_state or dx. Override dx for continuous models, and next_state for discrete, where the behavior cannot be described by the first derivative
- output(x)#
Calculate output given state
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Returns
z – Outputs, with keys defined by model.outputs.
e.g., z = m.OutputContainer({‘t’:12.4, ‘v’:3.3}) given outputs = [‘t’, ‘v’]
- Return type
OutputContainer
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> z = m.output(x) {'t': 292.1, 'v': 4.182999999010731}
- performance_metrics(x) dict#
Calculate performance metrics where
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Returns
pm – Performance Metrics, with keys defined by model.performance_metric_keys.
e.g., pm = {‘tMax’:33, ‘iMax’:19} given performance_metric_keys = [‘tMax’, ‘iMax’]
- Return type
Example
>>> from progpy.models import BatteryElectroChemEOD >>> m = BatteryElectroChemEOD() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> pm = m.performance_metrics(x) # {'max_i': array([8.83810109])}
- simulate_to(time: float, future_loading_eqn: collections.abc.Callable = <function PrognosticsModel.<lambda>>, first_output=None, **kwargs) collections.namedtuple#
Simulate prognostics model for a given number of seconds
- Parameters
time (float) –
Time to which the model will be simulated in seconds (≥ 0.0)
e.g., time = 200
future_loading_eqn (abc.Callable) – Function of (t) -> z used to predict future loading (output) at a given time (t)
first_output (OutputContainer, optional) – First measured output, needed to initialize state for some classes. Can be omitted for classes that don’t use this
- Returns
times (list[float]) – Times for each simulated point
inputs (SimResult) – Future input (from future_loading_eqn) for each time in times
states (SimResult) – Estimated states for each time in times
outputs (SimResult) – Estimated outputs for each time in times
event_states (SimResult) – Estimated event state (e.g., SOH), between 1-0 where 0 is event occurrence, for each time in times
- Raises
ProgModelInputException –
Note: – See simulate_to_threshold for supported keyword arguments
See also
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> def future_load_eqn(t, x = None): ... if t < 5.0: # Load is 2.0 for first 5 seconds ... return m.InputContainer({'i': 2.0}) ... else: ... return m.InputContainer({'i': 2.2}) >>> first_output = m.OutputContainer({'v': 3.2, 't': 295}) >>> (results) = m.simulate_to(200, future_load_eqn, first_output)
- simulate_to_threshold(future_loading_eqn: collections.abc.Callable = None, first_output=None, threshold_keys: list = None, **kwargs) collections.namedtuple#
Simulate prognostics model until any or specified threshold(s) have been met
- Parameters
future_loading_eqn (abc.Callable) – Function of (t) -> z used to predict future loading (output) at a given time (t)
- Keyword Arguments
t0 (float, optional) – Starting time for simulation in seconds (default: 0.0)
dt (float, tuple, str, or function, optional) –
float: constant time step (s), e.g. dt = 0.1
function (t, x) -> dt
tuple: (mode, dt), where modes could be constant or auto. If auto, dt is maximum step size
str: mode - ‘auto’ or ‘constant’
integration_method (str, optional) – Integration method, e.g. ‘rk4’ or ‘euler’ (default: ‘euler’)
save_freq (float, optional) – Frequency at which output is saved (s), e.g., save_freq = 10. A save_freq of 0 will save every step.
save_pts (list[float], optional) – Additional ordered list of custom times where output is saved (s), e.g., save_pts= [50, 75]
eval_pts (list[float], optional) – Additional ordered list of custom times where simulation is guarenteed to be evaluated (though results are not saved, as with save_pts) when dt is auto (s), e.g., eval_pts= [50, 75]
horizon (float, optional) – maximum time that the model will be simulated forward (s), e.g., horizon = 1000
first_output (OutputContainer, optional) – First measured output, needed to initialize state for some classes. Can be omitted for classes that don’t use this
threshold_keys (abc.Sequence[str] or str, optional) – Keys for events that will trigger the end of simulation. If blank, simulation will occur if any event will be met ()
x (StateContainer, optional) – initial state, e.g., x= m.StateContainer({‘x1’: 10, ‘x2’: -5.3})
thresholds_met_eqn (abc.Callable, optional) – custom equation to indicate logic for when to stop sim f(thresholds_met) -> bool
print (bool, optional) –
toggle intermediate printing, e.g., print = True
e.g., m.simulate_to_threshold(eqn, z, dt=0.1, save_pts=[1, 2])
progress (bool, optional) – toggle progress bar printing, e.g., progress = True
- Returns
times (list[float]) – Times for each simulated point
inputs (SimResult) – Future input (from future_loading_eqn) for each time in times
states (SimResult) – Estimated states for each time in times
outputs (SimResult) – Estimated outputs for each time in times
event_states (SimResult) – Estimated event state (e.g., SOH), between 1-0 where 0 is event occurrence, for each time in times
- Raises
See also
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> def future_load_eqn(t, x = None): ... if t< 5.0: # Load is 3.0 for first 5 seconds ... return m.InputContainer({'i': 2.0}) ... else: ... return m.InputContainer({'i': 2.2}) >>> first_output = m.OutputContainer({'v': 3.2, 't': 295}) >>> (results) = m.simulate_to_threshold(future_load_eqn, first_output)
Note
configuration of the model is set through model.parameters.
- state_at_event(x, future_loading_eqn=<function PrognosticsModel.<lambda>>, **kwargs)#
New in version 1.5.0.
Calculate the state at the time that each event occurs (i.e., the event threshold is met). state_at_event can be implemented by a direct model. For a state transition model, this returns the state at which threshold_met returns true for each event.
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
future_loading_eqn (abc.Callable, optional) – Function of (t) -> z used to predict future loading (output) at a given time (t). Defaults to no outputs
- Returns
state at each events occurrence, with keys defined by model.events
e.g., state_at_event = {‘impact’: {‘x1’: 10, ‘x2’: 11}, ‘falling’: {‘x1’: 15, ‘x2’: 20}} given events = [‘impact’, ‘falling’] and states = [‘x1’, ‘x2’]
- Return type
Note
Also supports arguments from
simulate_to_threshold()See also
threshold_met
- threshold_met(x) dict#
For each event threshold, calculate if it has been met
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
- Returns
If each threshold has been met (bool), with keys defined by prognostics_model.events
e.g., thresholds_met = {‘EOL’: False} given events = [‘EOL’]
- Return type
thresholds_met (dict)
Example
>>> from progpy.models import BatteryCircuit >>> m = BatteryCircuit() # Replace with specific model being simulated >>> u = m.InputContainer({'u1': 3.2}) >>> z = m.OutputContainer({'z1': 2.2}) >>> x = m.initialize(u, z) # Initialize first state >>> threshold_met = m.threshold_met(x) # {'EOD': False}
Note
If not overridden, will return True if event_state is <= 0, otherwise False. If neither threshold_met or event_state is overridden, will return an empty dictionary (i.e., no events)
See also
event_state
- time_of_event(x, future_loading_eqn=<function PrognosticsModel.<lambda>>, **kwargs) dict#
New in version 1.5.0.
Calculate the time at which each event occurs (i.e., the event threshold is met). time_of_event must be implemented by any direct model. For a state transition model, this returns the time at which threshold_met returns true for each event. A model that implements this is called a “direct model”.
- Parameters
x (StateContainer) –
state, with keys defined by model.states
e.g., x = m.StateContainer({‘abc’: 332.1, ‘def’: 221.003}) given states = [‘abc’, ‘def’]
future_loading_eqn (abc.Callable, optional) – Function of (t) -> z used to predict future loading (output) at a given time (t). Defaults to no outputs
- Returns
- time_of_event (dict)
time of each event, with keys defined by model.events
e.g., time_of_event = {‘impact’: 8.2, ‘falling’: 4.077} given events = [‘impact’, ‘falling’]
Note
Also supports arguments from
simulate_to_threshold()See also
threshold_met