API: Interfaces Package

Stable execution contract declared by every pipeline stage.

The pipeline runtime uses these flags to build an execution plan without relying on implementation details such as ad-hoc attributes. Concrete stages should be conservative: declare streaming only when they can produce output progressively without requiring the full upstream sequence.

batch(
    *,
    stateful: bool = True,
    preserves_order: bool = True,
    supports_frame_parallelism: bool = False,
    realtime_safe: bool = False,
) -> StageCapabilities

streaming(
    *,
    stateful: bool = False,
    preserves_order: bool = True,
    supports_frame_parallelism: bool = False,
    realtime_safe: bool = True,
) -> StageCapabilities

as_dict() -> dict[str, bool]

Bases: ABC

Batch source contract for frame-producing plugins.

A frame extractor is the pipeline entry point. Batch extractors return a closed FrameBuffer; streaming extractors should additionally implement IStreamingFrameExtractor so the planner can avoid materializing source frames.

__init__(config: dict[str, Any] | None = None)

extract() -> FrameBuffer

Bases: ABC

Contract for stateless or local frame transformations.

Single-frame processors are adapted into the frame-buffer pipeline by the core runtime. Use this contract when processing one frame does not require future or past frames.

__init__(config: dict[str, Any] | None = None)

process(frame: Frame) -> Frame

Bases: ABC

Buffer-level frame processing contract.

Implementations receive the whole frame sequence so both stateless per-frame operations and temporal algorithms can participate in the same frame processing pipeline.

Streaming implementations should also implement IStreamingFrameBufferProcessor when they can publish processed frames progressively without requiring the complete sequence first.

__init__(config: dict[str, Any] | None = None)

process(buffer: FrameBuffer) -> FrameBuffer

Runtime metadata passed to frame-buffer exporters.

Attributes:

Exporter result that preserves frames for downstream stages.

Exporters may create artifacts but must also return a frame buffer so signal extraction can continue after export.

Bases: ABC

Export final artifacts from processed frames without being a frame processor.

Exporters may write file-backed artifacts and must return a buffer that can still be consumed by signal extractors.

export(
    buffer: FrameBuffer, context: FrameExportContext
) -> FrameExportResult

Bases: ABC

Batch contract for converting frames into signal samples.

Signal extractors consume processed frames and produce an ISignal that analyzers can consume. Streaming implementations should also implement IStreamingSignalExtractor and declare streaming capabilities.

__init__(config: dict[str, Any] | None = None)

extract(buffer: FrameBuffer) -> ISignal

Bases: ABC

Batch contract for transforming signal samples before analysis.

Cleaners should preserve signal semantics unless their documentation states otherwise. Typical implementations smooth, normalize, filter, or repair samples produced by a signal extractor.

__init__(config: Dict[str, Any] | None = None)

clean(signal: ISignal) -> ISignal

Bases: ABC

Batch contract for turning a signal into analytical data.

An analyzer is the boundary between signal processing and visualization. Streaming analyzers should also implement IStreamingAnalyzer to publish progressive IData values while still returning a final result.

__init__(config: dict[str, Any] | None = None)

analyze(signal: ISignal) -> IData

Bases: ABC

Batch contract for converting analytical data into visual artifacts.

Visualizers must return UI-agnostic VisualArtifact values. They should not require Streamlit containers, OpenCV windows, notebook globals, or web framework state unless they are explicitly adapter implementations outside the core package.

__init__(config: dict[str, Any] | None = None)

render(
    data: IData, context: VisualizationContext | None = None
) -> tuple[VisualArtifact, ...]

Bases: ABC

Marker base class for chart-ready analytical data.

Bases: ABC

Base contract for ordered signal samples.

Signals are analyzer input values. Batch signal extractors return this complete collection, while streaming extractors publish ISignalSample values progressively into a buffer.

signal: list[ISignalSample]

__init__(
    samples: Sequence[ISignalSample],
    config: dict[str, Any] | None = None,
)

__iter__() -> Iterator[ISignalSample]

__len__() -> int

Bases: ABC

Base contract for one time-indexed signal sample.

Attributes:

__init__(
    frame_index: int,
    timestamp_seconds: float | None = None,
    metadata: dict[str, Any] | None = None,
)

Bases: Protocol[T]

Minimal producer-side contract shared by pipeline buffers.

closed: bool

put(item: T) -> None

close() -> None

Bases: IBuffer[T], Protocol[T]

Buffer that can interrupt a stream immediately.

abort() -> None

Bases: Iterator[T], Protocol[T]

Consumer-side stream cursor that can cancel its upstream buffer.

abort() -> None

Bases: IAbortableBuffer[T], Protocol[T]

Multi-consumer buffer with explicit fan-out configuration.

set_consumer_count(consumers: int) -> None

subscribe(consumer_id: int) -> IBufferSubscription[T]

Bases: IAbortableBuffer[Frame], Iterable[Frame], Protocol

Frame stream contract required by realtime latency policies.

This intentionally lives outside IBuffer because capacity and dropping are frame-queue concerns, not universal buffer responsibilities.

capacity: int

try_put(frame: Frame) -> bool

drop_oldest() -> Frame | None

fill_ratio() -> float

Bases: IFrameExtractor

Frame extractor that can publish frames into a bounded output stream.

Implementations must close output_buffer on normal completion and should cooperate with the supplied latency policy instead of writing directly with unbounded buffering.

extract_into(
    output_buffer: IFrameBuffer,
    latency_policy: FrameLatencyPolicy,
) -> None

Bases: IFrameBufferProcessor

Frame processor that can transform a stream without materializing it.

Implementations should publish outputs progressively and preserve ordering unless their plugin documentation explicitly states otherwise.

process_into(
    input_buffer: Iterable[Frame],
    output_buffer: IBuffer[Frame],
    *,
    processor_index: int,
    intermediate_store: IntermediateFrameArtifactStore
    | None,
) -> None

Bases: IFrameExporter

Frame exporter that writes artifacts while forwarding the frame stream.

This contract lets exporters produce file-backed artifacts without forcing the frame stream to be fully materialized before signal extraction.

export_into(
    frames: Iterable[Frame],
    output_buffer: IBuffer[Frame],
    context: FrameExportContext,
) -> tuple[VisualArtifact, ...]

Bases: ISignalExtractor

Signal extractor that emits samples progressively from frame input.

extract_into(
    frames: IFrameBuffer,
    output_buffer: IBuffer[ISignalSample],
) -> None

Bases: ISignalCleaner

Signal cleaner that can transform samples progressively.

clean_into(
    input_signal: Iterable[ISignalSample],
    output_buffer: IBuffer[ISignalSample],
) -> None

Bases: IAnalyzer

Analyzer that can publish progressive data and still return a final result.

Progressive values feed streaming visualizers during execution. The returned final IData remains the authoritative analyzer result included in PipelineOutputs.

analyze_into(
    signal: Iterable[ISignalSample],
    output_buffer: IBuffer[IData],
) -> IData

Bases: IVisualizer

Visualizer that can consume progressive analyzer data.

Streaming visualizers should avoid toolkit-specific blocking calls in core execution paths. For live UI updates, publish through adapter-safe contracts such as realtime sinks or return final artifacts after the stream closes.

render_stream(
    data: Iterable[IData],
    context: VisualizationContext | None = None,
) -> tuple[VisualArtifact, ...]

Bases: ABC

Factory port that creates executable Pipeline instances.

Applications can replace this port to inject custom execution policy, event buses, metadata propagation, or pipeline subclasses without changing the orchestrator.

create(
    context: PipelineContext,
    event_bus: IEventBus | None = None,
    pipeline_id: str | None = None,
    execution_metadata: Mapping[str, Any] | None = None,
) -> Pipeline

Bases: ABC

Executes Pipeline instances synchronously or asynchronously.

Implementations own execution concerns such as active-run tracking, retry, lifecycle events, and monitor registration/completion.

run(
    pipeline_id: str, pipeline: Pipeline
) -> PipelineOutputs

submit(
    pipeline_id: str, pipeline: Pipeline
) -> Future[PipelineOutputs]

cancel(pipeline_id: str) -> bool

active_ids() -> list[str]

snapshot(pipeline_id: str) -> PipelineRunSnapshot | None

snapshots() -> list[PipelineRunSnapshot]

shutdown(wait: bool = True) -> None

Bases: ABC

Port for observable pipeline execution state.

Monitors are read by UIs, APIs, and operators. Implementations should keep state transitions deterministic and should not expose mutable internal storage through snapshots.

register(pipeline_id: str) -> None

mark_running(pipeline_id: str, attempt: int) -> None

complete(pipeline_id: str) -> None

fail(
    pipeline_id: str, error: Exception | str, attempt: int
) -> None

terminate(pipeline_id: str) -> None

active_ids() -> list[str]

snapshot(pipeline_id: str) -> PipelineRunSnapshot | None

snapshots() -> list[PipelineRunSnapshot]

Bases: ABC

Persistence port for completed pipeline outputs.

Stores may be in-memory, filesystem-backed, database-backed, or service backed. Implementations should treat PipelineOutputs as immutable values.

save(pipeline_id: str, outputs: PipelineOutputs) -> None

get(pipeline_id: str) -> PipelineOutputs | None

delete(pipeline_id: str) -> None

Bases: ABC

Validation port for executable pipeline instances.

Validators are application-level checks. They should not mutate the pipeline and should raise typed configuration or validation errors when a run cannot be accepted.

validate(pipeline: 'Pipeline') -> None

Bases: ABC

Typed async-first pub/sub interface.

subscribe(event_type: str, handler: EventHandler) -> None

unsubscribe(event_type: str, handler: EventHandler) -> None

dispatch(event: Event) -> None

publish(event: Event) -> None

Bases: ABC

Strategy interface that governs retry behavior for runners.

should_retry(attempt: int, error: Exception) -> bool

wait_seconds(attempt: int) -> float

Bases: ABC

Strategy that decides IF and HOW to create a secondary pipeline in response to a domain event.

matches(event: Event) -> bool

build_context(event: Event) -> PipelineContext