Recipe — Python live camera / video streaming

For LYNX SDK 1.0. Complete, end-to-end. Install: install/python.md. API: api/python.md. Conventions: api/conventions.md.

Goal: open a webcam, run tracked detection on every frame in a loop, and print the detections (with stable track ids) per frame. Model: lynx-basic (keyless, 80 COCO classes — see models/catalog.md).

Detection with model.predict is stateless — each frame is independent, track ids are always 0. For a live source you want a Tracker: it carries state across frames so the same object keeps the same id. Get one from model.tracker() and feed it frames with tracker.update(frame).

The whole thing

lynx.camera_open(0) opens the default webcam through the C core (Media Foundation on Windows, V4L2 on Linux, AVFoundation on macOS). cam.read() returns one (H, W, 3) uint8 RGB ndarray — exactly what tracker.update() wants. Open the model once (the first open downloads + verifies + caches it), warm it up, then loop.

1import lynx
2 
3def stream(max_frames=600):
4    """Open lynx-basic (keyless), track objects from the default webcam, print
5    per-frame detections with stable track ids."""
6    with lynx.open("lynx-basic") as model:
7        # Build the size-1 engine up front so the first real frame doesn't stall
8        # on cold-start engine compilation (see Notes -> warmup).
9        model.prepare([1])
10 
11        with lynx.camera_open(0) as cam:          # 0 = default device, native format
12            w, h, fps = cam.actual_format
13            print(f"camera: {w}x{h} @ {fps:.3g} fps")
14 
15            # A Tracker keeps object identity across frames. window is the temporal
16            # history (seconds) Tracker.value() reduces over; 0.0 = core default.
17            with model.tracker(window=0.0) as tracker:
18                for i in range(max_frames):
19                    frame_rgb = cam.read()        # (H, W, 3) uint8 RGB; blocks for one frame
20                    frame = tracker.update(frame_rgb)
21 
22                    dets = frame.detections
23                    if len(dets) == 0:
24                        continue
25 
26                    parts = []
27                    for d in dets:
28                        name = model.classes(d.class_id).name    # e.g. "PERSON"
29                        tid = d.tracker_id                        # stable id (0 = untracked)
30                        parts.append(f"{name}#{tid}:{d.confidence:.2f}")
31                    print(f"frame {i:5d}  {len(dets):2d} obj  " + "  ".join(parts))
32 
33 
34if __name__ == "__main__":
35    stream()

d.tracker_id is the stable identity — the same physical object keeps the same id across frames while the Tracker sees it. d.track_state (lynx.TrackState.NEW/TENTATIVE/CONFIRMED/LOST) and d.track_age (seconds since first seen) round out the per-object track info; frame.tracks gives the full active-track snapshot (Track.id, .state, .age_s, .box, .matched) including tracks with no detection this frame.

Iterating a source instead of a hand loop

model.track(source, ...) is sugar for the loop above: it builds a Tracker internally and yields one Frame per item of any iterable of frames (RGB arrays or image paths). Feed it the camera, a list of frames, or your own video-decode generator:

1def camera_frames(cam):
2    while True:
3        yield cam.read()                          # (H, W, 3) uint8 RGB
4 
5with lynx.open("lynx-basic") as model, lynx.camera_open(0) as cam:
6    for frame in model.track(camera_frames(cam), window=0.0):
7        print(len(frame.detections), "objects",
8              [t.id for t in frame.tracks])

There is no built-in video-file decoder in the SDK — to stream a .mp4, decode it yourself (OpenCV / imageio / PyAV) and yield (H, W, 3) uint8 RGB arrays into model.track(...), same as camera_frames above.

Drop-frames-friendly: capture off the inference thread

cam.read() blocks for one frame, and tracker.update() blocks for the whole forward pass. Chaining them serially means a slow inference frame backs up the camera and you accumulate latency — the tracker falls further behind real time. The live-video fix is to run capture on its own thread that keeps only the newest frame (older un-consumed frames are dropped), so inference always works on the freshest pixels and slow frames drop instead of queueing.

1import threading
2import lynx
3 
4 
5class LatestFrame:
6    """Single-slot frame buffer: capture thread overwrites, consumer takes the
7    newest. Stale frames are dropped, never queued (bounded latency)."""
8    def __init__(self):
9        self._frame = None
10        self._lock = threading.Lock()
11        self._new = threading.Condition(self._lock)
12        self._stop = False
13 
14    def put(self, frame):
15        with self._lock:
16            self._frame = frame                   # overwrite -> drops any un-taken frame
17            self._new.notify()
18 
19    def get(self):
20        with self._lock:
21            while self._frame is None and not self._stop:
22                self._new.wait()
23            frame, self._frame = self._frame, None
24            return frame
25 
26    def stop(self):
27        with self._lock:
28            self._stop = True
29            self._new.notify()
30 
31 
32def capture_loop(cam, buf):
33    try:
34        while not buf._stop:
35            buf.put(cam.read())                   # keep only the freshest frame
36    except Exception:
37        buf.stop()
38 
39 
40def stream(max_frames=600):
41    with lynx.open("lynx-basic") as model:
42        model.prepare([1])                        # warm up before the clock starts
43        with lynx.camera_open(0) as cam:
44            buf = LatestFrame()
45            grabber = threading.Thread(
46                target=capture_loop, args=(cam, buf), daemon=True)
47            grabber.start()
48 
49            with model.tracker(window=0.0) as tracker:
50                for i in range(max_frames):
51                    frame_rgb = buf.get()         # newest available; older ones dropped
52                    if frame_rgb is None:
53                        break
54                    frame = tracker.update(frame_rgb)
55 
56                    for d in frame.detections:
57                        name = model.classes(d.class_id).name
58                        print(f"frame {i:5d}  {name}#{d.tracker_id} {d.confidence:.2f}")
59 
60            buf.stop()
61            grabber.join(timeout=1.0)
62 
63 
64if __name__ == "__main__":
65    stream()

The camera thread only ever holds one frame; whenever inference is busy, incoming frames overwrite each other and the SDK never sees the backlog. Latency stays bounded to roughly one inference pass regardless of how far behind the camera would otherwise fall.

METHODOverlapping capture with inferenceclick to expand

Overlapping capture with inference (submit / process_next)

If you'd rather hand frames to the SDK and let it overlap capture with the forward pass, Tracker also has a non-blocking submit() + process_next() pair: submit(frame) enqueues without blocking, process_next() blocks until the next result is ready.

1with model.tracker(window=0.0) as tracker:
2    tracker.submit(cam.read())                    # prime the pipeline
3    for i in range(max_frames):
4        tracker.submit(cam.read())                # enqueue frame i+1 (non-blocking)
5        frame = tracker.process_next()            # get result for frame i (blocks)
6        print(i, len(frame.detections), "objects")

Keep submit() and process_next() balanced (submit one ahead, then one-in / one-out) so the internal queue doesn't grow unbounded — for a hard latency cap the single-slot LatestFrame pattern above is stricter.

Notes

Warmup. The first predict/update on a fresh model builds the native inference engine and is much slower than steady state. Call model.prepare([1]) once at startup (size 1 is the cold-start warm-up bucket) so the first live frame runs at full speed. Do the open + prepare off your UI/event thread — open downloads on first use.
Throughput. Steady-state frame rate is bounded by the forward pass, not by capture. On CPU that can be well under camera fps — the drop-frames pattern above is what keeps a live feed real-time (you process the newest frame and skip the rest). To use a GPU, call lynx.set_workers(gpus=1) once before open (see api/python.md); check the active provider with model.execution_provider.
Track ids. d.tracker_id is 0 for an untracked detection (e.g. a stateless model.predict() result); it's a stable non-zero id once the Tracker confirms the object. Use d.track_state / d.track_age (or the frame.tracks snapshot) to filter tentative vs. confirmed tracks.
Frames are RGB. cam.read() returns (H, W, 3) uint8 RGB (top-left origin). Boxes from d.box are [x1, y1, x2, y2] in those same input pixels.
Cleanup. Camera, Tracker, and the Model are context managers — the with blocks close the capture device, the stream, and the model handle even on Ctrl-C or an exception. If you record annotated output, add a lynx.video_writer(path, w, h, fps) and writer.write(frame.plot(image=frame_rgb)) inside the loop (the SDK's examples/record_depth.py is the recorder pattern).