registry

In-memory registry of precomputed single runs (traces + labeled providers).

Persistence is opt-in: call export_registry() / upload_registry() explicitly.

class pysatl_cpd.benchmark.registry.BenchmarkRegistry[source]

Bases: Generic[DataT, TraceT]

Dict-like registry: SingleRunDescription -> SingleRun.

update skips recomputation when the key exists unless force_recompute is True.

__init__()[source]

Return type:: None

property executions_registry: Mapping[SingleRunDescription[TimeseriesAnnotation], SingleRun[TraceT, LabeledData[DataT, TimeseriesAnnotation]]]: Read-only view of the internal execution registry.

update(detector, providers, *, force_recompute=False, n_jobs=1, backend='loky')[source]

Register detection results for a detector against a sequence of providers.

For each provider, skips computation if a matching entry already exists in the registry (unless force_recompute is True). Supports sequential and parallel execution modes via the n_jobs parameter.

Parameters:

detector (ChangePointDetector[TypeVar(DataT)]) – Detector to execute against the providers.
providers (Sequence[LabeledData[TypeVar(DataT), TimeseriesAnnotation]]) – Providers whose data will be fed into the detector.
force_recompute (bool) – When True, re-executes detection even for already-registered entries.
n_jobs (int) – Number of parallel workers (default 1). Serial execution is used when set to 1.
backend (str) – Joblib parallel backend identifier (default "loky").

Return type:

None

__getitem__(key)[source]

Retrieve a SingleRun by its description key.

Parameters:: key (SingleRunDescription[TimeseriesAnnotation]) – Description identifying the detector-provider pair.
Returns:: The cached execution result for the given key.
Return type:: SingleRun[TypeVar(TraceT, bound= DetectionTrace), LabeledData[TypeVar(DataT), TimeseriesAnnotation]]

__len__()[source]

Return the number of registered executions.

Returns:: Registry size (number of detector-provider pairs stored).
Return type:: int

__contains__(key)[source]

Check whether a given description key exists in the registry.

Parameters:: key (SingleRunDescription[TimeseriesAnnotation]) – Description to look up.
Returns:: True if the key is present in the registry.
Return type:: bool

keys()[source]

Return a view of all registered description keys.

Returns:: Set-like view of registry keys.
Return type:: KeysView[SingleRunDescription[TimeseriesAnnotation]]

values()[source]

Return a view of all registered SingleRun values.

Returns:: Collection view of registry values.
Return type:: ValuesView[SingleRun[TypeVar(TraceT, bound= DetectionTrace), LabeledData[TypeVar(DataT), TimeseriesAnnotation]]]

items()[source]

Return a view of all (key, value) pairs in the registry.

Returns:: Set-like view of registry item pairs.
Return type:: ItemsView[SingleRunDescription[TimeseriesAnnotation], SingleRun[TypeVar(TraceT, bound= DetectionTrace), LabeledData[TypeVar(DataT), TimeseriesAnnotation]]]

export_registry(export_registry_path)[source]

Serialize the registry to a pickle file on disk.

Creates parent directories as needed before writing.

Parameters:: export_registry_path (Path) – Destination path for the serialized registry.
Return type:: None

upload_registry(upload_registry_path)[source]

Deserialize and load a registry from a pickle file.

Validates that the loaded object is a dict and that all entries conform to the expected SingleRunDescription -> SingleRun type contract before replacing the in-memory registry.

Parameters:: upload_registry_path (Path) – Path to a pickled registry file previously produced by export_registry.
Raises:: TypeError – If the file content is not a dict or contains entries of unexpected types.
Return type:: None