Tracking, motion and observations
spe.tracking, spe.motion, and spe.observations mirror the engine's temporal layer. They are pure and in-process: they operate on normalized detection mappings and never touch images or a server.
TrackingConfig
TrackingConfig(
min_iou: float = 0.01,
max_center_distance_scale: float = 2.5,
max_missing_seconds: float = 1.5,
confirmation_observations: int = 2,
)
Association thresholds in source-pixel units and seconds.
ModelAgnosticTracker
ModelAgnosticTracker(config: TrackingConfig | None = None)
tracker.update(detections, timestamp=None) -> {"detections": [...], "diagnostics": {...}}
tracker.snapshot() -> dict
tracker.reset() -> None
update assigns stable track-<n> ids to detections while rejecting duplicate and out-of-order timestamps without mutating state. Diagnostics include active_tracks, tracks_created, tracks_terminated, matched_detections, and related counters.
MotionConfig
MotionConfig(
max_trajectory_samples: int = 32,
max_velocity_gap_seconds: float = 2.0,
prediction_horizon_seconds: float = 0.5,
max_prediction_horizon_seconds: float = 2.0,
max_prediction_age_seconds: float = 1.5,
max_tracks: int = 256,
)
SpatialTrackStateManager
SpatialTrackStateManager(config: MotionConfig | None = None)
manager.update(detections, timestamp, prediction_enabled=False) -> dict
manager.snapshot(timestamp=None, prediction_enabled=False) -> dict
manager.reset() -> None
Prefers a valid detection spatial.position (keeping its coordinate system and units) and otherwise falls back to the bounding-box foot point in source pixels. Velocity is a finite difference over the last two samples; predictions are only produced when enabled and the track is fresh.
create_observation
create_observation(
detection_result,
*,
analysis_id: str,
source_id: str,
timestamp: float,
frame_index: int,
) -> dict
Builds a timestamped, provider-neutral observation without frame data, preserving calibrated spatial data and track_id when present. The SDK does not persist observations; store the returned dict however you prefer.
create_image_observation
create_image_observation(detection_result, *, source_id: str) -> dict
Wraps one static-image detection result with source context.
ObservationBatchWriter
ObservationBatchWriter(
analysis_id: str,
source_id: str,
model: Mapping[str, Any],
*,
root: str | Path,
batch_size: int = 25,
interval_ms: float | None = None,
effective_fps: float | None = None,
)
writer.append(observation) -> None
writer.flush() -> Path | None
writer.update_sampling(interval_ms, effective_fps) -> None
writer.close(status="complete") -> None
Opt-in persistence. Writes root/<analysis_id>/observations.jsonl and an adjacent manifest.json; nothing is written unless you instantiate the writer. It appends to an existing file so resumed runs keep prior records, and rejects path separators in identifiers.
Inference contexts
spe.inference validates the source-space scope used by model providers.
InferenceContext.whole_source() -> InferenceContext
InferenceContext.region_of_interest(points) -> InferenceContext
InferenceContext.from_dict(data) -> InferenceContext
context.to_dict() -> dict
resolve_inference_context(context, width, height) -> dict
resolve_inference_context accepts a mapping, an InferenceContext, or None and returns the engine-compatible dict. A ROI must be exactly four finite points inside the frame with non-degenerate area; otherwise SPEValidationError is raised.
from spe import InferenceContext, resolve_inference_context
context = InferenceContext.region_of_interest([[120, 80], [880, 110], [940, 520], [60, 540]])
payload = resolve_inference_context(context, 1000, 600)
client.detect_frame(frame, inference_context=payload)Analytics
spe.analytics builds summaries on top of tracked, spatialized detections. It is pure Python and accepts the engine's JSON shapes.
ZoneOccupancyTracker
tracker.update(detections, timestamp=None) -> dict
tracker.snapshot() -> dict
tracker.reset() -> None
Tracks per-zone occupancy, entry counts, and dwell seconds by track_id (falling back to id). Zones are read from spatial.zone, then zone.
LineCrossingCounter
LineCrossingCounter(start, end, *, name="line", max_events=1000)
counter.update(detections) -> dict
counter.snapshot() -> dict
counter.reset() -> None
Counts directional track crossings over a virtual line. Positions prefer spatial.position, then position, foot_point, center, or the bbox foot point. positive means a counter-clockwise transit; negative the reverse.
summarize_tracks
summarize_tracks(frames, *, distance_units="source_pixels") -> list[dict]
Consumes {"timestamp", "detections"} frames and returns one TrackSummary per track with duration, frames, distance, average speed, visited zones, and start/end positions. All positions must share a coordinate space.