base

Generic no-reset metric wrappers.

class pysatl_cpd.benchmark.online.noreset.metrics.base.NoResetThresholdMetric(*args, **kwargs)

Bases: Protocol, Generic

Protocol for no-reset metrics that evaluate threshold callables over runs.

evaluate(runs)

Return a threshold-indexed evaluator.

Parameters:

runs (Sequence[SingleRun[OnlineDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any])]]) – Sequence of runs to evaluate.

Return type:

Callable[[float], TypeVar(ResultT)]

__init__(*args, **kwargs)

class pysatl_cpd.benchmark.online.noreset.metrics.base.NoResetSingleRunMetric(source, policy)

Bases: Generic

Wrap a classical single-run metric with a no-reset policy.

Parameters:

• source (ISingleRunMetric[NoResetDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultT)]) – Metric that operates on NoResetDetectionTrace.

• policy (NoResetPolicy[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any])]) – Policy that transforms raw detection traces into classified traces.

__init__(source, policy)

Parameters:

• source (ISingleRunMetric[NoResetDetectionTrace[StateT], ProviderT, ResultT])

• policy (NoResetPolicy[StateT, ProviderT])

Return type:

None

evaluate(run)

Return a callable that evaluates the wrapped metric at a threshold.

Parameters:

run (SingleRun[OnlineDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any])]) – Single run with online detection trace.

Return type:

Callable[[float], TypeVar(ResultT)]

class pysatl_cpd.benchmark.online.noreset.metrics.base.NoResetMultipleRunMetric(source, policy)

Bases: Generic

Wrap a classical multiple-run metric with a no-reset policy.

Parameters:

• source (IMultipleRunMetric[NoResetDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultT)]) – Metric that operates on NoResetDetectionTrace.

• policy (NoResetPolicy[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any])]) – Policy that transforms raw detection traces into classified traces.

__init__(source, policy)

Parameters:

• source (IMultipleRunMetric[NoResetDetectionTrace[StateT], ProviderT, ResultT])

• policy (NoResetPolicy[StateT, ProviderT])

Return type:

None

evaluate(runs)

Return a callable that evaluates the wrapped metric at a threshold.

Parameters:

runs (Sequence[SingleRun[OnlineDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any])]]) – Sequence of runs to evaluate.

Return type:

Callable[[float], TypeVar(ResultT)]

class pysatl_cpd.benchmark.online.noreset.metrics.base.NoResetDerivedMetric(source, bases)

Bases: Generic

Wrap a classical derived metric with no-reset-aware source metrics.

Parameters:

• source (DerivedMetric[NoResetDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultInT), TypeVar(ResultOutT)]) – Formula that combines source metric values.

• bases (Mapping[str, NoResetThresholdMetric[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultInT)]]) – Named no-reset metrics providing the source values.

Raises:

ValueError – If any source metric required by the derived formula is missing.

__init__(source, bases)

Parameters:

• source (DerivedMetric[NoResetDetectionTrace[StateT], ProviderT, ResultInT, ResultOutT])

• bases (Mapping[str, NoResetThresholdMetric[StateT, ProviderT, ResultInT]])

Return type:

None

evaluate(runs)

Return a callable that evaluates the wrapped derived metric at a threshold.

Parameters:

runs (Sequence[SingleRun[OnlineDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any])]]) – Sequence of runs to evaluate.

Return type:

Callable[[float], TypeVar(ResultOutT)]

pysatl_cpd.benchmark.online.noreset.metrics.base.wrap_noreset_single_run_metric(source, policy)

Construct a no-reset single-run metric wrapper.

Parameters:

• source (ISingleRunMetric[NoResetDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultT)]) – Source metric operating on NoResetDetectionTrace.

• policy (NoResetPolicy[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any])]) – No-reset policy.

Return type:

NoResetSingleRunMetric[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultT)]

pysatl_cpd.benchmark.online.noreset.metrics.base.wrap_noreset_multiple_run_metric(source, policy)

Construct a no-reset multiple-run metric wrapper.

Parameters:

• source (IMultipleRunMetric[NoResetDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultT)]) – Source metric operating on NoResetDetectionTrace.

• policy (NoResetPolicy[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any])]) – No-reset policy.

Return type:

NoResetMultipleRunMetric[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultT)]

pysatl_cpd.benchmark.online.noreset.metrics.base.wrap_noreset_derived_metric(base, bases)

Construct a no-reset derived metric wrapper.

Parameters:

• base (DerivedMetric[NoResetDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)], TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultInT), TypeVar(ResultOutT)]) – Derived metric formula.

• bases (Mapping[str, NoResetThresholdMetric[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultInT)]]) – Named no-reset metrics.

Return type:

NoResetDerivedMetric[TypeVar(StateT, bound= OnlineAlgorithmState), TypeVar(ProviderT, bound= LabeledData[Any, Any]), TypeVar(ResultInT), TypeVar(ResultOutT)]