Track Worker

app.infrastructure.tracking.track_worker

View source code

Frame-aware background tracking worker and strategy dispatch.

See Also

• TrackingStrategy — exposes strategy choices to the application
• TrackingPostProcessor — refines generated tracks

Layer: Infrastructure. Context: Implements concrete processing and I/O behind application-facing contracts.

TrackerStrategy

Bases: Protocol

Incremental strategy contract shared by frame-based trackers.

Note

Strategies may ignore frame pixels and appearance features, but must accept them to keep worker dispatch independent of the selected tracker.

Layer: Infrastructure. Context: Implements concrete processing or I/O behind application-facing contracts.

uses_appearance: bool instance-attribute

update_frame(frame_idx: int, frame_boxes: list[BBoxViewModel], frame: RGBFrame | None = None, embeddings: Sequence[np.ndarray | None] | None = None) -> list[BBoxViewModel]

Update tracker state from detections and optional appearance data.

PARAMETER	DESCRIPTION
frame_idx	Zero-based frame index. TYPE: int
frame_boxes	Detections available for association on this frame. TYPE: list[BBoxViewModel]
frame	Optional decoded frame for appearance-aware strategies. TYPE: RGBFrame | None DEFAULT: None
embeddings	Optional appearance descriptors aligned with the boxes. TYPE: Sequence[ndarray | None] | None DEFAULT: None

RETURNS	DESCRIPTION
list[BBoxViewModel]	Box view models representing the strategy's updated tracks.

TrackingWorker

Bases: QThread

Background QThread that runs multi-object tracking on frame detections.

Strategy behavior in simple words

• dummy: copy detections without real tracking.
• hungarian: match by box overlap (IoU), coast internally, and emit observed boxes only.
• bytetrack: prioritize strong matches and use weaker detections to keep IDs stable.
• deepsort: match using motion cascade, overlap, and appearance memory.
• botsort: combine direct width/height Kalman filtering with two-stage confidence matching and isolated Re-ID appearance updates.

ATTRIBUTE	DESCRIPTION
_strategy_name	Tracking strategy name, such as "dummy" or "hungarian". TYPE: str
_source_data	Deep copy of the input detections. TYPE: BoxesByFrameIndex
_tracked_data	Output tracking results protected by _mutex. TYPE: BoxesByFrameIndex
_mutex	Guards tracked output view_state. TYPE: QMutex
_stop_requested	True when early stop is requested. TYPE: bool
_is_complete	True after tracking finishes successfully. TYPE: bool
_tracker	Tracking strategy instance. TYPE: TrackerStrategy

Note

Signals emitted by this worker: progress_updated(frame_count, frame_count): Final processed frame count. finished_processing(): Emitted when the thread exits. error_occurred(msg): Emitted if tracking raises an exception.

Example

settings = ProcessingSettings(min_iou=0.3, confidence_decay=0.05, ...) worker = TrackingWorker("hungarian", source_boxes, settings, start=True) results = worker.get_tracked_data()

Layer: Infrastructure. Context: Implements concrete processing or I/O behind application-facing contracts.

error_occurred = Signal(str) class-attribute instance-attribute

finished_processing = Signal() class-attribute instance-attribute

progress_updated = Signal(int, int) class-attribute instance-attribute

__init__(strategy_name: str, source_data: ListOfBoxesByFrameIndexAsDict, session_state: SessionState, parent: QObject | None = None, start: bool = False) -> None

PARAMETER	DESCRIPTION
strategy_name	Registered tracker strategy key. TYPE: str
source_data	Per-frame detection boxes. TYPE: ListOfBoxesByFrameIndexAsDict
session_state	Settings and video metadata for this run. TYPE: SessionState
parent	Optional Qt object parent. TYPE: QObject | None DEFAULT: None
start	Whether to begin processing immediately. TYPE: bool DEFAULT: False

Raises: ModuleNotFoundError: If an optional selected tracker dependency is missing. ValueError: If a post-processing setting is outside its supported range.

Note

Detection boxes are deep-copied and nested same-class duplicates are suppressed before worker-thread association begins.

Example

worker = TrackingWorker("hungarian", detections, state)

get_tracked_data() -> ListOfBoxesByFrameIndexAsDict

Return deep copy of tracked results (thread-safe via _mutex).

RETURNS	DESCRIPTION
ListOfBoxesByFrameIndexAsDict	Deep copy of boxes grouped by frame index.

isRunning() -> bool

Return True while the underlying QThread is active.

RETURNS	DESCRIPTION
bool	True while the tracking worker thread is running.

is_complete() -> bool

Return True if tracking finished successfully.

RETURNS	DESCRIPTION
bool	True when tracking completed successfully.

run() -> None

Main thread loop — run the tracking strategy on source data and store results.

Note

Workflow: 1. Open a thread-owned VideoReader when the session path exists. 2. Read each frame, extract aligned embeddings, and call _tracker.update_frame(...). 3. Run appearance-aware offline stitching and coast generation. 4. Store the result in _tracked_data under the mutex. 5. Emit error_occurred(msg) on exceptions and always emit finished_processing().

Raises: None: Worker failures are signaled through error_occurred.

Note

Optional post-processing settings control tracklet stitching, GSI, and confidence-decayed coast tails after strategy tracking completes.

Example

worker.start()

stop() -> None

Request early stop (reserved for future use).