Fault Injection¶

simweave.faults injects physically-meaningful parameter faults into continuous dynamic systems to produce labelled time-series data for predictive-maintenance (PdM) model training.

A faulted simulation produces a :class:~simweave.faults.FaultDataset whose columns include:

Column	Description
features	State / sensor observations (optionally noise-corrupted)
health_index	Float in [0, 1] — 1 = healthy, 0 = fully failed
rul	Remaining useful life in simulation time units
is_failed	Boolean, True from failure time onward
failure_mode	String label identifying which failure mode is active

These labels support three common PdM tasks simultaneously:

Task	Target column
Binary fault detection	`is_failed`
RUL regression (LSTM, TCN)	`rul`
Multi-class mode identification	`failure_mode`

Concepts¶

FaultProfile¶

A :class:~simweave.faults.FaultProfile defines when and how fast a fault evolves. It maps a simulation time t to a health index in [0, 1]:

health index
    1 ─────────────┐
                   │  degradation window
                   └──────────── 0
    onset_time      failure_time

Three built-in degradation shapes are available:

Shape	Description	Typical use case
`"linear"`	Constant degradation rate	Uniform surface wear, corrosion
`"exponential"`	Slow start, rapid end (convex)	Fatigue, crack propagation
`"abrupt"`	Healthy then instant failure	Electrical short, fracture
callable	Custom curve from progress ∈ [0,1]	Any empirically-fitted profile

All profiles are monotonically decreasing by design. Stochastic variation is modelled at the sensor level (see noise_std in :meth:~simweave.faults.FaultDataset.from_result) rather than at the profile level, keeping the degradation trajectory deterministic and reproducible across seeds.

ParameterFault¶

A :class:~simweave.faults.ParameterFault maps a FaultProfile to a named attribute on a :class:~simweave.continuous.solver.DynamicSystem:

ParameterFault(
    param="R_th",        # attribute on the DynamicSystem
    profile=profile,
    max_delta=3.0,       # 3× increase at full failure (relative=True)
    relative=True,
)

At each ODE step the injector computes:

fault_fraction = 1 − health_index(t)        # 0 = healthy, 1 = failed
perturbed = nominal × (1 + max_delta × fault_fraction)   # relative mode

Common parameter–system pairings:

System	Parameter	Physical meaning
`ThermalRC`	`R_th`	Insulation loss / thermal resistance increase
`ThermalRC`	`C_th`	Phase-change material degradation
`MassSpringDamper`	`stiffness`	Spring fatigue / crack
`MassSpringDamper`	`damping`	Damper seal wear
`SeriesRLC`	`R`	Conductor corrosion
`SeriesRLC`	`C`	Capacitor electrolyte degradation
`QuarterCarModel`	`k_s`	Suspension spring rate change

FaultInjector¶

:class:~simweave.faults.FaultInjector wraps any :class:~simweave.continuous.solver.DynamicSystem and satisfies the :class:~simweave.continuous.solver.SupportsDynamics protocol, so it plugs directly into :func:~simweave.continuous.solver.simulate or a :class:~simweave.continuous.solver.ContinuousProcess without any changes to the solver.

Multiple simultaneous faults are fully supported — attach several ParameterFault objects to one injector. The system-level health index is the minimum across all faults, and the active failure mode is labelled from the most-degraded fault. This enables multi-class classification datasets covering several independent failure modes.

FaultDataset¶

:class:~simweave.faults.FaultDataset assembles all labels into a numpy-backed dataclass. Labels are computed analytically from the fault profiles at each time step, so no separate recorder is needed for standalone :func:~simweave.continuous.solver.simulate runs.

Key methods:

Method	Description
`from_result(result, injector)`	Build from a `SimulationResult` (standalone path)
`from_recorder(recorder, result)`	Build from a `FaultRecorder` (hybrid-env path)
`concat([ds1, ds2, ...])`	Stack multiple runs into one training corpus
`train_test_split(test_frac)`	Sequential or shuffled split
`to_dataframe()`	Export to `pandas.DataFrame` (requires `pip install pandas`)

Quick start¶

Standalone simulate path (most common)¶

import numpy as np
import simweave as sw
from simweave.faults import FaultProfile, ParameterFault, FaultInjector, FaultDataset

# 1. Describe the degradation
profile = FaultProfile(
    onset_time=200,          # seconds — degradation begins
    failure_time=800,        # seconds — fully failed
    mode="insulation_loss",
    shape="exponential",     # slow start, rapid end
)

# 2. Map to a system parameter
fault = ParameterFault(
    param="R_th",            # thermal resistance on ThermalRC
    profile=profile,
    max_delta=3.0,           # rises to 4× nominal at full failure
    relative=True,
)

# 3. Wrap the system
system = sw.ThermalRC(thermal_resistance=1.0, thermal_capacitance=800.0)
injector = FaultInjector(system=system, faults=[fault])

# 4. Simulate (same API as always)
def heat_input(t):
    return 50.0   # constant 50 W

result = sw.simulate(injector, t_span=(0, 1000), dt=0.5, inputs=heat_input)

# 5. Build labelled dataset with sensor noise
rng = np.random.default_rng(42)
ds = FaultDataset.from_result(result, injector, noise_std=0.5, rng=rng)

print(ds)
# FaultDataset(n=2001, features=['temperature'], modes=['healthy', 'insulation_loss'], failed=401)

# 6. Export to pandas
df = ds.to_dataframe()
print(df.head())

# 7. Split for training
train, test = ds.train_test_split(test_frac=0.2)
print(f"Train: {len(train)}  Test: {len(test)}")

Multiple failure modes¶

from simweave.faults import FaultProfile, ParameterFault, FaultInjector, FaultDataset
import simweave as sw

system = sw.ThermalRC(thermal_resistance=1.0, thermal_capacitance=800.0)

faults = [
    ParameterFault(
        param="R_th",
        profile=FaultProfile(onset_time=300, failure_time=700,
                             mode="insulation_loss", shape="linear"),
        max_delta=3.0,
    ),
    ParameterFault(
        param="C_th",
        profile=FaultProfile(onset_time=500, failure_time=900,
                             mode="capacitance_loss", shape="exponential"),
        max_delta=0.6,    # C_th drops to 40% of nominal
        relative=True,
    ),
]

injector = FaultInjector(system=system, faults=faults)
result = sw.simulate(injector, t_span=(0, 1000), dt=0.5)
ds = FaultDataset.from_result(result, injector, noise_std=0.3)

print(set(ds.failure_mode))
# {'healthy', 'insulation_loss', 'capacitance_loss'}

Assembling a training corpus¶

Combine healthy and faulted runs from the same system into a single dataset:

import numpy as np
from simweave.faults import FaultDataset, FaultInjector, FaultProfile, ParameterFault
from simweave.continuous.solver import simulate
from simweave.continuous.systems import ThermalRC

def run(onset, failure, shape, seed):
    sys = ThermalRC(thermal_resistance=1.0, thermal_capacitance=800.0)
    if onset is None:
        # Healthy run
        inj = FaultInjector(sys, faults=[])
    else:
        profile = FaultProfile(onset_time=onset, failure_time=failure,
                               mode="fault", shape=shape)
        fault = ParameterFault(param="R_th", profile=profile, max_delta=3.0)
        inj = FaultInjector(sys, faults=[fault])
    result = simulate(inj, t_span=(0, 1000), dt=1.0)
    return FaultDataset.from_result(result, inj, noise_std=0.5,
                                    rng=np.random.default_rng(seed))

datasets = [
    run(None, None, None, seed=0),        # healthy
    run(200, 700, "linear", seed=1),
    run(400, 900, "exponential", seed=2),
    run(300, 600, "linear", seed=3),
]

corpus = FaultDataset.concat(datasets)
train, test = corpus.train_test_split(test_frac=0.2)

# Feed directly to PyTorch / TensorFlow / scikit-learn
X_train = train.features          # (N_train, n_features)
y_rul   = train.rul               # RUL regression
y_mode  = train.failure_mode      # multi-class classification

Hybrid environment path¶

Use :class:~simweave.faults.FaultRecorder when combining a faulted system with discrete events in a :class:~simweave.core.environment.SimEnvironment:

from simweave.continuous.solver import ContinuousProcess
from simweave.faults import FaultRecorder, FaultDataset
import simweave as sw

proc = ContinuousProcess(injector)
recorder = FaultRecorder(injector)

env = sw.SimEnvironment(dt=1.0, end=1000.0)
env.register(proc)
env.register(recorder)   # register after proc for post-tick alignment
env.run(until=1000.0)

result = proc.result()
ds = FaultDataset.from_recorder(recorder, result)

Visualisation¶

Two plot helpers ship with simweave.viz:

import simweave as sw

# Sensor signals with shaded fault window
fig = sw.plot_fault_signals(result, injector)
fig.show()

# Health index + RUL on a dual y-axis
fig = sw.plot_health_index(ds, show_rul=True)
fig.show()

API reference¶

FaultProfile `dataclass` ¶

FaultProfile(onset_time: float, failure_time: float, mode: str = 'fault', shape: ProfileShape = 'linear')

Temporal evolution of a single fault from healthy to fully failed.

Parameters:

Name	Type	Description	Default
`onset_time`	`float`	Simulation time at which degradation begins. The system is fully healthy before this point (health index = 1).	required
`failure_time`	`float`	Simulation time at which the system reaches the fully-failed state (health index = 0). Must be strictly greater than onset_time.	required
`mode`	`str`	Failure mode label written into dataset labels, e.g. `"insulation_loss"` or `"bearing_wear"`. Defaults to `"fault"`.	`'fault'`
`shape`	`('linear', 'exponential', 'abrupt')`	Degradation curve shape over the window [onset_time, failure_time]: `"linear"` Constant degradation rate; health falls from 1 to 0 uniformly. `"exponential"` Convex profile: slow initial degradation that accelerates sharply near failure_time. Characteristic of fatigue and wear-out mechanisms. `"abrupt"` System is fully healthy until onset_time, then fails instantaneously. Useful for modelling sudden fracture or electrical short-circuit. callable Receives progress ∈ [0, 1] (0 = onset, 1 = failure) and must return a health value in [0, 1]. Use this for custom profiles such as S-curve, two-phase, or empirically fitted curves.	`"linear"`

Notes

All shapes are monotonically decreasing. Stochastic variation is intentionally absent at the profile level; add sensor noise through :meth:~simweave.faults.dataset.FaultDataset.from_result instead.

health_index ¶

health_index(t: float) -> float

Health index in [0, 1] at simulation time t.

Returns 1.0 before onset_time and 0.0 from failure_time onward.

rul ¶

rul(t: float) -> float

Remaining useful life in simulation time units at time t.

ParameterFault `dataclass` ¶

ParameterFault(param: str, profile: FaultProfile, max_delta: float, relative: bool = True)

A fault expressed as a perturbation of a named attribute on a DynamicSystem.

Parameters:

Name	Type	Description	Default
`param`	`str`	Name of the attribute to perturb on the wrapped system, e.g. `"R_th"` on a :class:`~simweave.continuous.systems.ThermalRC` or `"c"` on a :class:`~simweave.continuous.systems.MassSpringDamper`. The attribute must exist and be a plain `float`.	required
`profile`	`FaultProfile`	Temporal evolution of this fault.	required
`max_delta`	`float`	Magnitude of the perturbation at full failure (health index = 0). When relative is `True` this is a fractional multiplier (e.g. `2.0` doubles the nominal value at full failure). When relative is `False` it is an additive offset.	required
`relative`	`bool`	If `True` (default): `perturbed = nominal × (1 + max_delta × fault_fraction)` If `False`: `perturbed = nominal + max_delta × fault_fraction` where `fault_fraction = 1 − health_index ∈ [0, 1]`.	`True`

FaultInjector ¶

FaultInjector(system: DynamicSystem, faults: list[ParameterFault])

Bases: DynamicSystem

Wraps a :class:~simweave.continuous.solver.DynamicSystem and applies one or more :class:~simweave.faults.fault.ParameterFault objects.

The injector satisfies the :class:~simweave.continuous.solver.SupportsDynamics protocol so it can be passed directly to :func:~simweave.continuous.solver.simulate or wrapped in a :class:~simweave.continuous.solver.ContinuousProcess without any changes to the solver.

On each :meth:derivatives call the injector:

Computes the fault fraction (1 − health_index) for every fault.
Perturbs the named parameter(s) on the wrapped system proportionally.
Delegates to the underlying system's :meth:derivatives.
Restores nominal parameter values (safe with RK4's multi-eval per step).

Parameters:

Name	Type	Description	Default
`system`	`DynamicSystem`	The underlying physical model to wrap.	required
`faults`	`list[ParameterFault]`	Fault definitions to inject simultaneously. Multiple faults on different parameters are all active at once; the system-level health index is the minimum across all individual fault health indices and the active failure mode is that of the most-degraded fault.	required

Examples:

>>> from simweave.continuous.systems import ThermalRC
>>> from simweave.faults import FaultProfile, ParameterFault, FaultInjector
>>> profile = FaultProfile(onset_time=200, failure_time=800,
...                        mode="insulation_loss", shape="exponential")
>>> fault = ParameterFault(param="R_th", profile=profile, max_delta=2.0)
>>> injector = FaultInjector(ThermalRC(R_th=1.0, C_th=500.0), [fault])
>>> from simweave.continuous.solver import simulate
>>> result = simulate(injector, t_span=(0, 1000), dt=1.0)

overall_health ¶

overall_health(t: float) -> float

Minimum health index across all faults (1 = healthy, 0 = failed).

active_mode ¶

active_mode(t: float) -> str

Label of the most-degraded fault at time t, or "healthy".

overall_rul ¶

overall_rul(t: float) -> float

Minimum remaining useful life across all faults (simulation time units).

FaultRecorder ¶

FaultRecorder(injector: FaultInjector, name: str | None = None)

Bases: _Recorder

Snapshot health index, RUL, and failure-mode label each environment tick.

Register this after the :class:~simweave.continuous.solver.ContinuousProcess that wraps the injector so the labels align with the post-tick system state.

Parameters:

Name	Type	Description	Default
`injector`	`FaultInjector`	The injector whose fault profiles are sampled each tick.	required
`name`	`str`	Display name for this recorder.	`None`

Attributes:

Name	Type	Description
`times`	`list[float]`	Simulation times of each sample.
`health_index`	`list[float]`	Overall (minimum) health index in [0, 1] at each sample.
`rul`	`list[float]`	Minimum remaining useful life (simulation time units) across faults.
`is_failed`	`list[bool]`	`True` when health index has reached 0.
`failure_mode`	`list[str]`	Label of the most-degraded fault (`"healthy"` when all faults are before their onset time).

FaultDataset `dataclass` ¶

FaultDataset(time: ndarray, features: ndarray, feature_names: list[str], health_index: ndarray, rul: ndarray, is_failed: ndarray, failure_mode: ndarray)

ML-ready dataset produced from a faulted simulation run.

Columns are aligned on a shared time grid so the dataset can be fed directly into a time-series model (LSTM, TCN, etc.) or a tabular classifier after slicing.

Attributes:

Name	Type	Description
`time`	`(ndarray, shape(N))`	Simulation time at each sample.
`features`	`(ndarray, shape(N, F))`	Sensor / state observations, optionally with additive Gaussian noise.
`feature_names`	`list[str]`	Column labels for features — state labels first, then input labels.
`health_index`	`(ndarray, shape(N))`	Health index in [0, 1]: 1 = healthy, 0 = fully failed.
`rul`	`(ndarray, shape(N))`	Remaining useful life in simulation time units.
`is_failed`	`np.ndarray, shape (N,), dtype bool`	`True` from the failure time onward.
`failure_mode`	`np.ndarray, shape (N,), dtype object`	String label for the active failure mode at each sample (`"healthy"` before any fault onset).

from_result `classmethod` ¶

from_result(result: 'SimulationResult', injector: 'FaultInjector', noise_std: float | ndarray | None = None, rng: Generator | None = None) -> 'FaultDataset'

Build a dataset from a :class:~simweave.continuous.solver.SimulationResult.

Labels are computed analytically from the injector's fault profiles, so this works equally well after a standalone :func:~simweave.continuous.solver.simulate call or a hybrid :class:~simweave.continuous.solver.ContinuousProcess run.

Parameters:

Name	Type	Description	Default
`result`	`SimulationResult`	Output of :func:`~simweave.continuous.solver.simulate`.	required
`injector`	`FaultInjector`	The injector used for the simulation (provides label computation).	required
`noise_std`	`float or array - like`	Gaussian noise standard deviation added to features to simulate sensor measurement uncertainty. A scalar applies the same std to every feature column; an array of length `n_features` gives per-column control.	`None`
`rng`	`Generator`	Random generator for reproducible noise. Defaults to a fresh `np.random.default_rng()` when noise_std is provided.	`None`

from_recorder `classmethod` ¶

from_recorder(recorder: 'FaultRecorder', result: 'SimulationResult', noise_std: float | ndarray | None = None, rng: Generator | None = None) -> 'FaultDataset'

Build a dataset from a :class:~simweave.faults.recorder.FaultRecorder.

Convenience wrapper for the hybrid :class:~simweave.continuous.solver.ContinuousProcess path where a recorder was already registered with the environment.

Parameters:

Name	Type	Description	Default
`recorder`	`FaultRecorder`	Recorder attached to the running process.	required
`result`	`SimulationResult`	History extracted from the process via :meth:`~simweave.continuous.solver.ContinuousProcess.result`.	required
`noise_std`	`float or array - like`	Sensor noise std (see :meth:`from_result`).	`None`
`rng`	`Generator`	Random generator for reproducible noise.	`None`

concat `staticmethod` ¶

concat(datasets: list['FaultDataset']) -> 'FaultDataset'

Stack multiple :class:FaultDataset objects along the sample axis.

Useful for assembling a training corpus from several independent simulation runs (e.g. one healthy run + multiple fault runs with different onset times or degradation profiles).

All datasets must share the same feature_names.

Parameters:

Name	Type	Description	Default
`datasets`	`list[FaultDataset]`	Two or more datasets to concatenate.	required

Returns:

Type	Description
`FaultDataset`	A single dataset containing all samples in order.

to_dataframe ¶

to_dataframe()

Convert to a :class:pandas.DataFrame.

Requires pandas (pip install pandas).

Returns:

Type	Description
`DataFrame`	Columns: `time`, all feature columns, `health_index`, `rul`, `is_failed`, `failure_mode`.

train_test_split ¶

train_test_split(test_frac: float = 0.2, shuffle: bool = False, rng: Generator | None = None) -> 'tuple[FaultDataset, FaultDataset]'

Split into train and test subsets.

Parameters:

Name	Type	Description	Default
`test_frac`	`float`	Fraction of samples to reserve for the test set (default 0.2).	`0.2`
`shuffle`	`bool`	If `True`, shuffle indices before splitting. Set rng for reproducibility.	`False`
`rng`	`Generator`	Required when shuffle=True for reproducibility.	`None`

Returns:

Type	Description
`train, test : tuple[FaultDataset, FaultDataset]`

Fault Injection¶

Concepts¶

FaultProfile¶

ParameterFault¶

FaultInjector¶

FaultDataset¶

Quick start¶

Standalone simulate path (most common)¶

Multiple failure modes¶

Assembling a training corpus¶

Hybrid environment path¶

Visualisation¶

API reference¶

FaultProfile dataclass ¶

health_index ¶

rul ¶

ParameterFault dataclass ¶

FaultInjector ¶

overall_health ¶

active_mode ¶

overall_rul ¶

FaultRecorder ¶

FaultDataset dataclass ¶

from_result classmethod ¶

from_recorder classmethod ¶

concat staticmethod ¶

to_dataframe ¶

train_test_split ¶

FaultProfile `dataclass` ¶

ParameterFault `dataclass` ¶

FaultDataset `dataclass` ¶

from_result `classmethod` ¶

from_recorder `classmethod` ¶

concat `staticmethod` ¶