API: Pipeline Package

Public execution facade for a single pipeline run.

Pipeline intentionally coordinates collaborators instead of owning the workflow details. Frame execution, signal analysis, visualization, event injection, and output assembly are delegated to focused collaborators in this package. This keeps the public API stable while avoiding a god object.

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

run() -> PipelineOutputs

execution_plan() -> PipelineExecutionPlan

Application-facing pipeline execution facade.

The orchestrator is the single public access point for execution: synchronous runs go through run(context), background runs go through submit(context), and event-driven integrations are optional adapters around the same submit path.

run(
    context: PipelineContext,
    pipeline_id: str | None = None,
    execution_metadata: Mapping[str, Any] | None = None,
) -> PipelineOutputs

submit(
    context: PipelineContext,
    pipeline_id: str | None = None,
    execution_metadata: Mapping[str, Any] | None = None,
) -> Future[PipelineOutputs]

terminate(pipeline_id: str) -> bool

active_ids() -> list[str]

shutdown(wait: bool = True) -> None

Bases: IPipelineRunner

Execute pipelines on a ThreadPoolExecutor.

run() and submit() share active-id tracking: the same pipeline id cannot be executed concurrently through either entry point. submit() returns the underlying Future so callers can observe asynchronous results or failures.

cancel() is best-effort: it can cancel queued work that has not started, but it cannot interrupt a pipeline that is already running.

Lifecycle events are dispatched synchronously to the configured IEventBus from within the worker thread.

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

Immutable dependency graph for one pipeline execution unit.

Attributes:

Raises:

Programmatic builder for PipelineContext.

The builder only collects pipeline components and creates a validated context. Execution belongs to Pipeline, PipelineOrchestrator, or a custom runner.

with_frame_extractor(
    extractor: IFrameExtractor,
) -> FluentPipelineBuilder

with_signal_extractor(
    extractor: ISignalExtractor,
) -> FluentPipelineBuilder

with_frame_processors(
    processors: Iterable[IFrameBufferProcessor],
) -> FluentPipelineBuilder

add_frame_processor(
    processor: IFrameBufferProcessor,
) -> FluentPipelineBuilder

with_frame_exporters(
    exporters: Iterable[IFrameExporter],
) -> FluentPipelineBuilder

add_frame_exporter(
    exporter: IFrameExporter,
) -> FluentPipelineBuilder

with_signal_cleaners(
    cleaners: Iterable[ISignalCleaner],
) -> FluentPipelineBuilder

add_signal_cleaner(
    cleaner: ISignalCleaner,
) -> FluentPipelineBuilder

with_analyzers(
    analyzers: Iterable[IAnalyzer],
) -> FluentPipelineBuilder

add_analyzer(analyzer: IAnalyzer) -> FluentPipelineBuilder

with_visualizers(
    visualizers: Iterable[IVisualizer],
) -> FluentPipelineBuilder

add_visualizer(
    visualizer: IVisualizer,
) -> FluentPipelineBuilder

with_visualizer_bindings(
    bindings: Iterable[VisualizerBinding],
) -> FluentPipelineBuilder

add_visualizer_for_results(
    visualizer: IVisualizer, result_indices: Iterable[int]
) -> FluentPipelineBuilder

with_intermediate_frame_capture(
    config: IntermediateFrameCaptureConfig | dict | None,
) -> FluentPipelineBuilder

with_intermediate_frame_visualizers(
    visualizers: Iterable[IVisualizer],
) -> FluentPipelineBuilder

add_intermediate_frame_visualizer(
    visualizer: IVisualizer,
) -> FluentPipelineBuilder

with_stream_runtime(
    config: StreamRuntimeConfig | dict | None,
) -> FluentPipelineBuilder

build_context() -> PipelineContext

Build a validated PipelineContext from a declarative configuration.

schema_version: "1.0"
pipeline:
  frame_extractor:
    name: opencv_buffered
    params:
      path: /path/to/video.mp4

  frame_processors:
    - name: opencv_gray
      processor_type: single_frame
    - name: motion_magnification
      processor_type: frame_buffer

  signal_extractor:
    name: opencv_tracker
    params:
      roi: [100, 200, 50, 80]

  signal_cleaners:
    - name: moving_average
      params:
        window: 5

  analyzers:
    - name: vertical_position

  visualizers:
    - name: matplotlib
      result_indices: [0]

  intermediate_frames:
    enabled: true
    sampling_interval: 10
    max_stored_frames: 20
    export_directory: artifacts/debug
    lazy_saving: true
    visualizers:
      - name: intermediate_frames_grid

  runtime:
    frame_buffer_size: 8
    signal_buffer_size: 8
    data_buffer_size: 8
    latency_policy:
      name: blocking
      params: {}

Raises:

build_context(config: Mapping[str, Any]) -> PipelineContext

Binds a visualizer to selected analyzer result indexes.

result_indices=None preserves the default behaviour: the visualizer is applied to every analyzer result. Providing indexes makes the binding selective without changing the visualizer contract.

target_indexes(result_count: int) -> tuple[int, ...]

Public description of one stage in an execution plan.

The value is generated before execution starts and is safe to display in UIs, logs, or diagnostics. It contains decisions and rationale, not runtime executor objects.

Attributes:

streams: bool

as_dict() -> dict[str, Any]

Public execution plan generated before a pipeline run starts.

The plan is an observability contract. It explains which stages are streamable, where the runtime will materialize data, and why each decision was selected by the execution policy.

streamable_end_to_end: bool

materialization_boundaries: tuple[ExecutionPlanStage, ...]

by_group(group: str) -> tuple[ExecutionPlanStage, ...]

as_dict() -> dict[str, Any]

as_text() -> str

Bases: str, Enum

Execution modes that can be selected for one pipeline stage.

Policy decision for one stage.

The decision carries both the selected mode and the reason that will appear in execution plans. Keeping the explanation in the policy prevents planner and runtime from inventing separate narratives for the same choice.

streams: bool

Inputs available to execution-mode policies for one stage.

The context contains facts, not decisions: capabilities, current stream state, downstream demand and optional cost estimates. This keeps policies replaceable and testable without depending on executor internals.

Bases: Protocol

Strategy interface for batch/streaming execution decisions.

Custom policies can implement latency-first, memory-first, observability or domain-specific rules without changing the execution planner or runtime. Implementations must never select streaming when stage_streamable is false.

decide_source(
    context: PipelineStagePolicyContext,
) -> PipelineExecutionDecision

decide_stage(
    context: PipelineStagePolicyContext,
) -> PipelineExecutionDecision

decide_analyzer(
    context: PipelineStagePolicyContext,
) -> PipelineExecutionDecision

Default cost-aware streaming policy.

The policy keeps streaming when it avoids materializing an active stream, starts new streaming segments only when there is downstream demand, and uses available size estimates to prefer bounded queues over full materialization when that is clearly cheaper.

decide_source(
    context: PipelineStagePolicyContext,
) -> PipelineExecutionDecision

decide_stage(
    context: PipelineStagePolicyContext,
) -> PipelineExecutionDecision

decide_analyzer(
    context: PipelineStagePolicyContext,
) -> PipelineExecutionDecision

Bounded-buffer and latency settings for adaptive streaming execution.

The config controls queue capacities between streaming stages and selects the frame latency policy used by streaming frame extractors. It is immutable and safe to store in PipelineContext or reproducibility metadata.

Attributes:

from_mapping(
    value: Mapping[str, Any] | None,
) -> StreamRuntimeConfig

validate() -> None

as_dict() -> dict[str, Any]

Serializable selector for frame-stream latency behavior.

The config is stored in pipeline configs and run metadata. Calling create() returns a fresh policy instance because concrete policies keep per-run counters and, for adaptive sampling, per-run control state.

Attributes:

from_mapping(
    value: Mapping[str, Any] | None,
) -> LatencyPolicyConfig

create() -> FrameLatencyPolicy

as_dict() -> dict[str, Any]

Bases: ABC

Strategy used by streaming frame extractors under queue pressure.

Implementations decide whether an incoming frame should be published, dropped, or used to replace an older queued frame. Instances may keep counters, so they are per-run runtime objects rather than reusable config values.

publish(frame: Frame, output_buffer: IFrameBuffer) -> bool

metrics() -> dict[str, Any]

Bases: FrameLatencyPolicy

Preserve every frame by blocking upstream when the queue is full.

Use this policy for offline reproducibility and deterministic frame coverage. It can increase latency in realtime pipelines when downstream inference or visualization is slower than the source.

Bases: FrameLatencyPolicy

Drop the incoming frame when the downstream queue is full.

This policy protects downstream stages from backlog growth while preserving already queued frames. It is useful when continuity of accepted frames is more important than visual freshness.

Bases: FrameLatencyPolicy

Keep preview latency low by replacing stale queued frames.

This policy favors the most recent source frame and is usually the best default for realtime camera previews where freshness is more important than processing every frame.

Bases: FrameLatencyPolicy

Reduce processed FPS by increasing the sampling interval under queue pressure.

The policy is intentionally simple and deterministic: it observes the target queue fill ratio before each publish and adjusts the sampling interval within configured bounds. This stabilizes live preview latency without adding worker coordination complexity.

from_mapping(
    params: Mapping[str, Any],
) -> AdaptiveSamplingFrameLatencyPolicy

Declarative migration step between two config schema versions.

Migrations are intentionally pure: they receive a mapping and return a new mapping. This keeps future schema evolution testable and prevents builders from mutating user-provided configuration objects.

apply(config: Mapping[str, Any]) -> dict[str, Any]

Normalized pipeline configuration plus schema-version metadata.

Attributes:

source_config() -> dict[str, Any]

Validate and normalize declarative pipeline config versions.

The current implementation supports the first public schema, 1.0. The migration list is an explicit extension point for future schema changes without pushing version conditionals into ConfigPipelineBuilder.

normalize(
    config: Mapping[str, Any],
) -> VersionedPipelineConfig

Bases: StrEnum

Observable lifecycle states for a pipeline run.

Immutable snapshot of a pipeline run.

Timestamps use time.time() seconds to stay consistent with the existing event timestamp model.