online_wrappers

Composable wrappers for online algorithms.

class pysatl_cpd.core.online.wrappers.online_wrappers.SkippingCondition(*, name, condition)

Bases: Generic

Named condition deciding whether an observation should be skipped.

Variables:

• name – Human-readable identifier for this condition.

• condition – Callable that returns True when an observation should be skipped (not passed to the underlying algorithm).

Parameters:

• name (str)

• condition (Callable[[DataT], bool])

name: str

condition: Callable[[DataT], bool]

__hash__()

Return a stable hash based on the public wrapper name only.

Return type:

int

__post_init__()

Validate that the condition name is non-empty.

Raises:

ValueError – If name is an empty string.

Return type:

None

__init__(*, name, condition)

Parameters:

• name (str)

• condition (Callable[[DataT], bool])

Return type:

None

class pysatl_cpd.core.online.wrappers.online_wrappers.BatchReducer(*, name, reducer)

Bases: Generic

Named reducer converting a block of observations into one observation.

Variables:

• name – Human-readable identifier for this reducer.

• reducer – Callable that converts a sequence of raw observations into a single value for the wrapped algorithm.

Parameters:

• name (str)

• reducer (Callable[[Sequence], OutT])

name: str

reducer: Callable[[Sequence], OutT]

__hash__()

Return a stable hash based on the public reducer name only.

Return type:

int

__post_init__()

Validate that the reducer name is non-empty.

Raises:

ValueError – If name is an empty string.

Return type:

None

__init__(*, name, reducer)

Parameters:

• name (str)

• reducer (Callable[[Sequence[InT]], OutT])

Return type:

None

class pysatl_cpd.core.online.wrappers.online_wrappers.SkippingOnlineAlgorithmWrapper(algorithm, *, skipping_condition)

Bases: OnlineAlgorithm, Generic

Wrap an online algorithm and optionally skip observations.

When the configured skipping_condition is met for an observation, the wrapper returns the last computed detection statistic without passing the observation to the wrapped algorithm.

Parameters:

• algorithm (OnlineAlgorithm[TypeVar(DataT), TypeVar(ConfigurationT, bound= OnlineAlgorithmConfiguration), TypeVar(StateT, bound= OnlineAlgorithmState)]) – Algorithm to wrap.

• skipping_condition (SkippingCondition[TypeVar(DataT)]) – Condition that decides which observations to skip.

__init__(algorithm, *, skipping_condition)

Parameters:

• algorithm (OnlineAlgorithm[DataT, ConfigurationT, StateT])

• skipping_condition (SkippingCondition[DataT])

Return type:

None

property name: str

Return a composite name reflecting the wrapped algorithm and condition.

Returns:

Composite name in the form AlgorithmName{skip[on=condition]}.

Return type:

str

property configuration: ConfigurationT

Return the wrapped algorithm’s configuration.

Returns:

Configuration of the underlying algorithm.

Return type:

ConfigurationT

property state: StateT

Return the wrapped algorithm’s current state.

Returns:

Current state of the underlying algorithm.

Return type:

StateT

process(observation)

Process an observation, skipping it if the condition is met.

Parameters:

observation (TypeVar(DataT)) – Incoming data point.

Returns:

Detection statistic from the wrapped algorithm, or the last computed statistic when the observation is skipped.

Return type:

double | int | float

reset()

Reset the wrapper and the underlying algorithm to initial state.

Clears the cached last detection function value and delegates reset to the wrapped algorithm.

Return type:

None

recreate()

Return a fresh instance with the same skipping condition.

Returns:

A new wrapper instance sharing the same skipping condition but a freshly recreated underlying algorithm.

Return type:

SkippingOnlineAlgorithmWrapper[TypeVar(DataT), TypeVar(ConfigurationT, bound= OnlineAlgorithmConfiguration), TypeVar(StateT, bound= OnlineAlgorithmState)]

class pysatl_cpd.core.online.wrappers.online_wrappers.BatchingOnlineAlgorithmWrapper(algorithm, *, batch_size, reducer)

Bases: OnlineAlgorithm, Generic

Wrap an online algorithm and feed it reduced observation batches.

Raw observations are buffered until the batch is full, then reduced to a single value via BatchReducer and passed to the wrapped algorithm. Partial batches return the last computed statistic.

Parameters:

• algorithm (OnlineAlgorithm[TypeVar(BatchT), TypeVar(ConfigurationT, bound= OnlineAlgorithmConfiguration), TypeVar(StateT, bound= OnlineAlgorithmState)]) – Algorithm to wrap (operates on the reduced batch type).

• batch_size (int) – Number of raw observations to accumulate before feeding a reduced value to the wrapped algorithm. Must be positive.

• reducer (BatchReducer[TypeVar(DataT), TypeVar(BatchT)]) – Function that converts a sequence of raw observations into a single value for the wrapped algorithm.

Raises:

ValueError – If batch_size is not positive.

__init__(algorithm, *, batch_size, reducer)

Parameters:

• algorithm (OnlineAlgorithm[BatchT, ConfigurationT, StateT])

• batch_size (int)

• reducer (BatchReducer[DataT, BatchT])

Return type:

None

property name: str

Return a composite name reflecting the wrapped algorithm and batch config.

Returns:

Composite name in the form AlgorithmName{batch[size=N, reduce=name]}.

Return type:

str

property configuration: ConfigurationT

Return the wrapped algorithm’s configuration.

Returns:

Configuration of the underlying algorithm.

Return type:

ConfigurationT

property state: StateT

Return the wrapped algorithm’s current state.

Returns:

Current state of the underlying algorithm.

Return type:

StateT

process(observation)

Buffer an observation and process the batch once full.

Parameters:

observation (TypeVar(DataT)) – Incoming data point to buffer.

Returns:

Detection statistic. Returns the last computed value until the buffer reaches batch_size, then feeds the reduced batch to the wrapped algorithm and returns its result.

Return type:

double | int | float

reset()

Reset the buffer, cached statistic, and underlying algorithm.

Clears the observation buffer and cached detection function, then delegates reset to the wrapped algorithm.

Return type:

None

recreate()

Return a fresh instance with the same batch size and reducer.

Returns:

A new wrapper instance sharing the same batch configuration but a freshly recreated underlying algorithm.

Return type:

BatchingOnlineAlgorithmWrapper[TypeVar(DataT), TypeVar(BatchT), TypeVar(ConfigurationT, bound= OnlineAlgorithmConfiguration), TypeVar(StateT, bound= OnlineAlgorithmState)]