LinearModel#

class progpy.LinearModel(**kwargs)#

Bases: progpy.prognostics_model.PrognosticsModel, abc.ABC

A linear prognostics model. Used when behavior can be described using a simple linear time-series model defined by the following equations: .. math:

\dfrac{dx}{dt} = Ax + Bu + E

z = Cx + D

es = Fx + G

where x is state, u is input, z is output and es is event state

Linear Models must inherit from this class and define the following properties:

  • A: 2-d np.array[float], dimensions: n_states x n_states

  • B: 2-d np.array[float], optional (zeros by default), dimensions: n_states x n_inputs

  • C: 2-d np.array[float], dimensions: n_outputs x n_states

  • D: 1-d np.array[float], optional (zeros by default), dimensions: n_outputs x 1

  • E: 1-d np.array[float], optional (zeros by default), dimensions: n_states x 1

  • F: 2-d np.array[float], dimensions: n_es x n_states

  • G: 1-d np.array[float], optional (zeros by default), dimensions: n_es x 1

  • inputs: list[str] - input keys

  • states: list[str] - state keys

  • outputs: list[str] - output keys

  • events: list[str] - event keys

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

  • times (list[float]) – array of times for each sample

  • inputs (list[InputContainer]) – array of input dictionaries where input[x] corresponds to time[x]

  • outputs (list[OutputContainer]) – array of output dictionaries where output[x] corresponds to time[x]

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

float

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

next_state

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

  • keys (list[str]) – Parameter keys to optimize

  • times (list[float]) – Array of times for each sample

  • 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)#

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

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
>>> 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

threshold_met

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

PrognosticsModel

See also

to_json

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

bool

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

bool

matrixCheck() → None#

Public class method for checking matrices dimensions across all properties of the model.

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

dx

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

dict

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

simulate_to_threshold

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

ValueError –

See also

simulate_to

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

state_at_event (dict[str, StateContainer])

Note

Also supports arguments from simulate_to_threshold()

See also

threshold_met

threshold_met(x)#

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

to_json() → str#

Serialize parameters as JSON objects

Returns

Serialized PrognosticsModel parameters as string

Return type

str

See also

from_json

Note

This method only serializes the values of the prognostics model parameters (model.parameters)