data_provider

Abstract data provider interface.

class pysatl_cpd.data.providers.data_provider.DataProvider

Bases: ABC, Generic[DataT, AnnotationT]

Abstract source class for sequential data providers.

property name: str

Name of the data provider.

Returns:

Name from the annotation.

Return type:

name

abstract property annotation: AnnotationT

Annotation of the data provider.

Returns:

Annotation instance for this provider.

Return type:

annotation

abstractmethod __iter__()

Iterate over data items in the provider.

Returns:

Iterator yielding data items.

Return type:

Iterator[TypeVar(DataT)]

abstractmethod __len__()

Return the number of data items in the provider.

Returns:

Number of data items.

Return type:

int

abstractmethod cut(start, stop, *, annotation=None)

Cut a slice of data from the provider.

Parameters:

• start (int) – Start index of the slice.

• stop (int) – Stop index of the slice (inclusive).

• annotation (Optional[TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]) – Optional annotation for the sliced data.

Returns:

New provider containing the sliced data.

Return type:

DataProvider[TypeVar(DataT), TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]

default_slice_annotation(start, stop)

Default annotation for sliced data.

Parameters:

• start (int) – Start index of the slice.

• stop (int) – Stop index of the slice.

Returns:

Default annotation for the sliced data.

Return type:

TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)

abstractmethod classmethod merge(providers, annotation_builder=None)

Merge multiple providers into one.

Parameters:

• providers (Sequence[Self]) – Sequence of providers to merge.

• annotation_builder (Callable[[Sequence[TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]], TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)] | None) – Optional callable to build merged annotation.

Returns:

New provider containing merged data.

Return type:

DataProvider[TypeVar(DataT), TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]

static default_merge_annotation_builder(source=None)

Build a default annotation builder for merged providers.

Parameters:

source (str | None) – Optional source string for the merged annotation.

Returns:

Callable that builds annotation from a sequence.

Return type:

Callable[[Sequence[TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]], TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]

__or__(other)

Merge two providers using the | operator.

Parameters:

other (Self) – Other provider to merge with.

Returns:

New provider containing merged data.

Return type:

DataProvider[TypeVar(DataT), TypeVar(AnnotationT, bound= TimeseriesAnnotation, covariant=True)]

Raises:

TypeError – If the other provider is not the same concrete type.