Extract shared realtime presentation helpers

[jarcherNV]

Extract shared realtime presentation helpers

Add transport-neutral presentation queue, pacing, frame materialization, and
MJPEG multipart helpers under flashdreams.serving.realtime. Migrate the
OmniDreams interactive runtime loop and MJPEG presenter to use the shared
queue/pacing and latest-frame MJPEG utilities while keeping demo-specific
routes, controls, overlays, and scene selection local.

[greptile-apps]

Greptile Summary

This PR extracts shared transport-neutral realtime presentation utilities into flashdreams.serving.realtime, covering PresentationQueue (batch drain with filtering, capacity eviction, and deferred prepare), wait_until_present_time, materialize_rgb_host_uint8, and MJPEG multipart helpers. The OmniDreams runtime loop and streaming presenter are then migrated to use these shared helpers.

• PresentationQueue.drain_nowait improves the old manual while get_nowait() pattern into a single call with include/prepare/capacity eviction handled atomically; the generation-filter + GPU-prepare pattern from loop.py is cleanly expressed as lambda callbacks.
• MJPEG helpers (format_mjpeg_part, send_mjpeg_response_headers, write_mjpeg_stream, publish_latest_jpeg, wait_for_latest_jpeg) reduce ~70 lines of inline duplication in streaming_presenter.py to a handful of call sites without any behavioral change.
• Test coverage in test_realtime_presenter.py and test_realtime_mjpeg.py is solid; capacity eviction, prepare-only-retained, and bus-close semantics are all exercised.

Confidence Score: 5/5

The changes are a clean mechanical refactor — no logic is rewritten, only extracted into shared helpers and re-called. All migrated sites are behaviorally equivalent to the code they replace.

The extraction is narrow and well-tested: MJPEG helpers are pure pass-throughs, PresentationQueue drain_nowait preserves the generation-filter and prepare ordering from the old loop, and wait_until_present_time is a direct lift of the existing sleep block. No new code paths are introduced, and the added test coverage exercises the edge cases.

presenter.py is the most novel file — the drain_nowait batch-drain + deferred-prepare pattern is new and callers with fallible prepare callbacks or capacity-limited queues should be aware of the exception-safety and resource-release characteristics noted in the review.

Important Files Changed

Filename	Overview
flashdreams/flashdreams/serving/realtime/presenter.py	New shared module: PresentationQueue with drain_nowait (batch-drain + prepare + capacity eviction), wait_until_present_time, and materialize_rgb_host_uint8. Logic is correct but drain_nowait loses all batch-pulled frames if prepare raises mid-loop.
flashdreams/flashdreams/serving/realtime/mjpeg.py	New shared module: MJPEG formatting, header sending, stream writing, and latest-frame bus helpers. Clean extraction from streaming_presenter.py with correct protocol abstractions.
integrations/omnidreams/omnidreams/interactive_drive/runtime/loop.py	Migrated deque+manual-drain to PresentationQueue.drain_nowait and wait_until_present_time; behavior is equivalent to the old inline code with no functional regressions.
integrations/omnidreams/omnidreams/interactive_drive/streaming_presenter.py	Replaced inline MJPEG header/stream/publish/wait logic with calls to shared helpers; thin shims retained for internal callers. No behavioral change.
flashdreams/tests/test_realtime_presenter.py	Good test coverage for PresentationQueue (capacity eviction, filter, drop reporting, prepare-only-retained), wait_until_present_time, and materialize_rgb_host_uint8.
flashdreams/tests/test_realtime_mjpeg.py	Covers MJPEG formatting, header sending, stream loop termination, publish gating on stop_event, and wait_for_latest_jpeg with close handling.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant RL as run_main_loop
    participant DPF as _drain_pipeline_frames
    participant PQ as PresentationQueue
    participant FQ as ChunkPipeline.frame_queue
    participant PR as PresenterBackend

    RL->>DPF: drain_pipeline_frames(pipeline, ready_frames)
    DPF->>PQ: drain_nowait(source, include, prepare)
    PQ->>FQ: get_nowait() repeated until Empty
    FQ-->>PQ: QueuedFrame
    Note over PQ: Filter by generation, evict by capacity, prepare retained new frames
    PQ->>PR: prepare_frame(retained_frame)
    PQ-->>DPF: QueueDrainResult

    RL->>RL: wait_until_present_time(next_present_time)
    alt presentation slot reached
        RL->>PQ: popleft()
        PQ-->>RL: QueuedFrame
        RL->>PR: present_frame(queued_frame)
    else still waiting
        RL->>RL: continue to next iteration
    end

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant RL as run_main_loop
    participant DPF as _drain_pipeline_frames
    participant PQ as PresentationQueue
    participant FQ as ChunkPipeline.frame_queue
    participant PR as PresenterBackend

    RL->>DPF: drain_pipeline_frames(pipeline, ready_frames)
    DPF->>PQ: drain_nowait(source, include, prepare)
    PQ->>FQ: get_nowait() repeated until Empty
    FQ-->>PQ: QueuedFrame
    Note over PQ: Filter by generation, evict by capacity, prepare retained new frames
    PQ->>PR: prepare_frame(retained_frame)
    PQ-->>DPF: QueueDrainResult

    RL->>RL: wait_until_present_time(next_present_time)
    alt presentation slot reached
        RL->>PQ: popleft()
        PQ-->>RL: QueuedFrame
        RL->>PR: present_frame(queued_frame)
    else still waiting
        RL->>RL: continue to next iteration
    end

Loading

_{Reviews (2): Last reviewed commit: "Fix realtime presentation queue review i..." | Re-trigger Greptile}

[gtong-nv]

This is also used only the omnidreams interactive demo. Do we really need it to be shared with others?
I think we should default to use the webRTC path for streaming?

[jarcherNV]

This is also used only the omnidreams interactive demo. Do we really need it to be shared with others? I think we should default to use the webRTC path for streaming?

Yeah, I think webrtc should be the default. What do you think? Should we not worry about refactoring this for potential future demos to use?

[jarcherNV]

I am abandoning this change as we have decided to remove the interactive drive demo, and will default to using the webrtc path for all demos going forward.