API: Artifacts Package
Artifacts are typed values exchanged between stages and returned to adapters. They are intentionally framework-neutral.
Frames and Buffers
Frame
dataclass
Video frame plus timeline and plugin metadata.
The image array is intentionally not copied by the value object. Producers
and consumers should document whether they mutate frame pixels in place or
return new arrays.
Attributes:
| Name | Type | Description |
|---|---|---|
image |
ndarray
|
Pixel array, typically an OpenCV-compatible NumPy image. |
index |
int | None
|
Optional source-order index. |
timestamp_seconds |
float | None
|
Optional source timestamp in seconds. |
metadata |
dict[str, Any]
|
Mutable metadata dictionary for plugin-specific annotations. |
frame
property
frame: ndarray
Backward-compatible alias for image used by existing OpenCV code.
FrameBuffer
Thread-safe frame buffer for batch and streaming execution.
FrameBuffer exposes a public capacity for frame items and reserves one
internal queue slot for the end-of-stream sentinel. Iteration blocks until
frames are available or the buffer is closed.
Ordering
Frames are yielded in the order accepted by put() or try_put().
Thread safety
The queue operations are thread-safe. The Frame objects themselves are
not copied; producers and consumers remain responsible for pixel mutability.
put
put(frame: Frame) -> None
Publish a frame, blocking while the public queue is full.
The method returns silently when the buffer has already been closed.
try_put
try_put(frame: Frame) -> bool
Publish a frame without blocking. Return False when the queue is full.
drop_oldest
drop_oldest() -> Frame | None
Drop and return the oldest queued frame item, if one is available.
get
get() -> Frame
Return the next frame or raise StopIteration when closed.
The end-of-stream sentinel is reinserted so repeated consumers observe completion consistently.
close
close() -> None
Close the buffer and wake consumers waiting for more frames.
abort
abort() -> None
Close the buffer without blocking, dropping queued frames if needed.
is_empty
is_empty() -> bool
Return whether no frame or sentinel item is currently queued.
size
size() -> int
Return current queue size, including a sentinel if present.
fill_ratio
fill_ratio() -> float
Return current public queue occupancy in the [0, 1] range.
clone_empty
clone_empty() -> 'FrameBuffer'
Return an empty buffer with the same public capacity.
__iter__
__iter__() -> Iterator[Frame]
Iterate frames until the buffer closes.
SignalBuffer
Multi-consumer streaming buffer for signal samples.
The buffer retains each sample until all configured subscribers have consumed it. This supports fan-out from one streaming signal stage to multiple downstream consumers without forcing materialization.
Lifecycle
Configure the consumer count before producing samples. Producers call
close() after normal completion or abort() after failure.
closed
property
closed: bool
Return True when the buffer has been closed or aborted.
put
put(sample: ISignalSample) -> None
Publish one sample and block when the bounded buffer is full.
The sample is ignored when the buffer is already closed or when the configured consumer count is zero.
close
close() -> None
Mark the stream complete and wake all subscribers.
abort
abort() -> None
Wake all consumers and discard buffered samples after an upstream failure.
set_consumer_count
set_consumer_count(consumers: int) -> None
Declare how many subscribers must consume each future sample.
subscribe
subscribe(
consumer_id: int,
) -> IBufferSubscription[ISignalSample]
Create a subscriber cursor for a configured consumer id.
SignalSubscription
Iterator returned by SignalBuffer.subscribe.
__iter__
__iter__()
Return this subscription as its own iterator.
__next__
__next__() -> ISignalSample
Return the next sample for this subscriber.
abort
abort() -> None
Abort the source buffer for cooperative downstream cancellation.
DataBuffer
Multi-consumer streaming buffer for progressive analyzer data.
Streaming analyzers publish IData values here so visualizers can consume
progressive updates without forcing a full analyzer result to materialize.
Items are retained until all configured subscribers have consumed them.
closed
property
closed: bool
Return True when the buffer has been closed or aborted.
put
put(item: IData) -> None
Publish one data item and block when the bounded buffer is full.
The item is ignored after closure or when the configured consumer count is zero.
close
close() -> None
Mark the stream complete and wake all subscribers.
abort
abort() -> None
Wake all consumers and discard buffered data after an upstream failure.
set_consumer_count
set_consumer_count(consumers: int) -> None
Declare how many subscribers must consume each future item.
subscribe
subscribe(consumer_id: int) -> IBufferSubscription[IData]
Create a subscriber cursor for a configured consumer id.
DataSubscription
Iterator returned by DataBuffer.subscribe.
__iter__
__iter__()
Return this subscription as its own iterator.
__next__
__next__() -> IData
Return the next data item for this subscriber.
abort
abort() -> None
Abort the source buffer for cooperative downstream cancellation.
Signals and Generic Data
Signal
NoData
dataclass
CategoryData
dataclass
TwoDimPointData
dataclass
TwoDimGraphData
dataclass
VectorFieldGraphData
dataclass
Bases: IData
Data container for vector field (quiver) visualization. (x, y) define the position of each vector (u, v) define the vector components
TrajectoryData
dataclass
Bases: IData
Represents trajectories of multiple tracked points over time.
Each trajectory is a sequence of (x, y) positions.
Motion and Tracking Samples
BoxSignalSample
dataclass
MultiManualSignalSample
dataclass
MultiObjectTrack
dataclass
Represents a single tracked object inside a frame.
MultiObjectSignalSample
dataclass
SparseOpticalFlowSignalSample
dataclass
DenseOpticalFlowSignalSample
dataclass
Bases: ISignalSample
Represents a dense optical flow frame aggregated into a grid. Each cell contains a motion vector (dx, dy).
TrackingPlaybackTrack
dataclass
Serializable tracking observation used to rebuild annotated video frames.
TrackingPlaybackFrame
dataclass
Tracked objects associated with a single frame index.
TrackingPlaybackData
dataclass
ArUco Data
ArucoMarkerObservation
dataclass
A single ArUco marker observation inside one frame.
center
property
center: Point2D | None
Return the marker center as (x, y) when both coordinates exist.
ArucoMarkerSignalSample
dataclass
Bases: ISignalSample
Frame-level collection of ArUco marker observations.
marker_by_id
marker_by_id(
marker_id: int,
) -> ArucoMarkerObservation | None
Return the first observation for marker_id, if present.
ArucoMarkerDisplacementObservation
dataclass
Progressive displacement value for one marker in one frame.
ArucoMarkerDisplacementSeries
dataclass
Displacement timeline for a single marker.
ArucoMarkerDisplacementFrameData
dataclass
ArucoMarkerDisplacementData
dataclass
Bases: IData
Playback-ready displacement analysis for multiple ArUco markers.
from_progressive_frames
classmethod
from_progressive_frames(
frames: list[ArucoMarkerDisplacementFrameData],
*,
title: str = "ArUco Marker Displacement",
use_timestamps: bool = True,
) -> ArucoMarkerDisplacementData
Build the final displacement result from progressive frame payloads.
from_stream_items
classmethod
from_stream_items(
data: Iterable[IData],
*,
title: str | None = None,
use_timestamps: bool | None = None,
) -> ArucoMarkerDisplacementData
Normalize progressive analyzer output into the final displacement model.
Streaming visualizers receive per-frame payloads. Batch visualizers receive the final aggregate directly. Keeping this conversion here prevents each visualizer from duplicating stream reconstruction rules.
ArucoMarkerRelativeMotionSeries
dataclass
Distance variation timeline between two markers.
ArucoMarkerRelativeMotionData
dataclass
COCO Pose Data
COCOSkeletonSignalSample
dataclass
Bases: ISignalSample
Single COCO pose observation extracted from one frame.
Expected format: skeleton: [17, 2] (x, y coordinates) confidence: [17] (per-joint confidence)
COCOPoseFrameData
dataclass
COCOPoseSequenceData
dataclass
COCOPoseTennisFrameData
dataclass
COCOPoseTennisSequenceData
dataclass
Intermediate Frame Artifacts
IntermediateFrameArtifactCollection
dataclass
Bases: IData
Immutable debug stream produced by frame processing stages.
The collection is deliberately separate from analysis results so existing visualizers keep receiving only the result types they already support.
count
property
count: int
Return the number of stored intermediate snapshots.
is_empty
property
is_empty: bool
Return whether no intermediate snapshots were captured.
stage_names
property
stage_names: tuple[str, ...]
Return stage names in first-seen order.
frame_indices
property
frame_indices: tuple[int | None, ...]
Return frame indexes in first-seen order.
empty
classmethod
empty() -> IntermediateFrameArtifactCollection
Return an empty collection for pipelines without debug capture.
by_stage_name
by_stage_name(
stage_name: str,
) -> tuple[IntermediateFrameArtifact, ...]
Return snapshots captured for a specific stage.
by_frame_index
by_frame_index(
frame_index: int | None,
) -> tuple[IntermediateFrameArtifact, ...]
Return snapshots captured for a specific frame index.
export
export(
output_directory: Path | str | None = None,
) -> tuple[Path, ...]
Persist stored frames to PNG files and return all paths written.
If output_directory is omitted, the collection must have an
export_directory entry in metadata. This supports deferred saving:
the pipeline can capture bounded snapshots now, while developers decide
later whether to write them to disk.
FrameComparisonPanel
dataclass
A normalized image panel used by comparison and grid composers.
MaskArtifact
dataclass
Bases: IData
Immutable base for binary mask artifacts.
The stored mask is always a read-only boolean copy. This keeps downstream preprocessing and debugging artifacts deterministic even when the source array is later reused or mutated by OpenCV code.
shape
property
shape: tuple[int, int]
Return the mask shape as (height, width).
height
property
height: int
Return the mask height in pixels.
width
property
width: int
Return the mask width in pixels.
active_pixel_count
property
active_pixel_count: int
Return the number of protected, target, or active mask pixels.
coverage_ratio
property
coverage_ratio: float
Return the fraction of active pixels in the mask.
is_empty
property
is_empty: bool
Return whether the mask contains no active pixels.
as_bool_array
as_bool_array(*, copy: bool = True) -> MaskArray
Return the mask as a boolean NumPy array.
By default this returns a mutable copy. Passing copy=False returns the
internal read-only view for zero-copy consumers.
as_uint8_array
as_uint8_array(
*, active_value: int = 255
) -> npt.NDArray[np.uint8]
Return an OpenCV-friendly uint8 mask copy.
ensure_compatible_with
ensure_compatible_with(
candidate: NDArray[Any] | tuple[int, ...],
) -> None
Validate that a frame, shape, or other mask shares this mask's spatial dimensions.
to_dict
to_dict(*, include_mask: bool = False) -> dict[str, Any]
Serialize lightweight artifact state for logs, tests, or debug UIs.
The mask payload is omitted by default to avoid accidentally logging
full-frame arrays. Set include_mask=True for small masks and tests.
to_json
to_json(*, include_mask: bool = False) -> str
Serialize the artifact to deterministic JSON.
debug_string
debug_string() -> str
Return a compact, array-safe debug representation.
MotionMaskArtifact
dataclass
TargetMaskArtifact
dataclass
ProtectedRegionArtifact
dataclass
FrameMaskArtifact
dataclass
IntermediateFrameArtifact
dataclass
Bases: IData
Immutable snapshot of an intermediate frame for pipeline debugging.
The image is copied and marked read-only so snapshots represent the exact state emitted by a stage, independent of later OpenCV buffer reuse.
frame
property
frame: NDArray[Any]
Alias matching the existing Frame.frame convention.
original_frame
property
original_frame: NDArray[Any] | None
Return the optional pre-processing frame snapshot.
processed_frame
property
processed_frame: NDArray[Any]
Return the processed frame snapshot, falling back to the primary image.
shape
property
shape: tuple[int, ...]
Return the full image shape.
spatial_shape
property
spatial_shape: tuple[int, int]
Return the image shape as (height, width).
height
property
height: int
Return the frame height in pixels.
width
property
width: int
Return the frame width in pixels.
channels
property
channels: int
Return the number of image channels, treating grayscale frames as one-channel.
dtype
property
dtype: dtype[Any]
Return the NumPy dtype of the stored snapshot.
as_array
as_array(*, copy: bool = True) -> npt.NDArray[Any]
Return the image snapshot.
By default this returns a mutable copy. Passing copy=False returns the
internal read-only array for zero-copy debug consumers.
ensure_mask_compatible
ensure_mask_compatible(
mask: MaskArtifact | NDArray[Any] | tuple[int, ...],
) -> None
Validate that a mask-like object can be applied to this frame snapshot.
to_dict
to_dict(*, include_image: bool = False) -> dict[str, Any]
Serialize lightweight frame snapshot state.
Pixel data is omitted by default because intermediate frames can be
large. Set include_image=True only for small fixtures or local debug.
to_json
to_json(*, include_image: bool = False) -> str
Serialize the snapshot to deterministic JSON.
debug_string
debug_string() -> str
Return a compact, pixel-safe debug representation.
IntermediateFrameOverlay
dataclass
Immutable rendered overlay associated with an intermediate frame snapshot.
Overlays are intentionally image-based rather than UI objects. This keeps frame processors free to emit masks, contours, heatmaps, or other debug images without coupling them to a specific visualization backend.
shape
property
shape: tuple[int, ...]
Return the full overlay image shape.
spatial_shape
property
spatial_shape: tuple[int, int]
Return the overlay shape as (height, width).
as_array
as_array(*, copy: bool = True) -> npt.NDArray[Any]
Return the overlay image, copying by default for caller safety.