diff --git a/CHANGELOG.md b/CHANGELOG.md
index 135fce3c..e966511a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -368,6 +368,22 @@ read any of them:
- `scripts/check_plugin.py` checks the element contract for any plugin that
implements it (`src/plugin_system/testing/vegas.py`), and
`test/fixtures/plugins/vegas-live-stub` is a working example.
+- **Live elements update in place.** When a plugin's `update()` completes,
+ one background worker (`src/vegas_mode/live_worker.py`) redraws its live
+ elements that are on or ahead of the screen, nearest first, and hands the
+ ones whose pixels changed to the render thread, which copies them into the
+ strip between two frames (`RenderPipeline.apply_live_patches`,
+ `ScrollHelper.patch_columns`): at most four patches or two screens of bytes
+ a frame, no drawing and no locks on the render thread. Elements with
+ `refresh_hz` are redrawn that often while near the screen, through the
+ plugin's lock-free `redraw_vegas_element()`. The worker also takes over
+ group prefetching once the strip holds a live element, so one thread
+ still does all the drawing; it runs inside the render gate, starts only
+ when a live element is placed, and is restarted if it dies (three times in
+ ten minutes turns live updates off for the run). While live elements exist,
+ the Vegas update tick runs every second instead of every four.
+- Web UI: "Update live content while it scrolls" under Vegas mode's Cycle
+ Pacing (`display.vegas_scroll.live_refresh`).
### Scrolling
diff --git a/docs/OFFSCREEN_RENDERING.md b/docs/OFFSCREEN_RENDERING.md
index b7b7cc65..d84fb442 100644
--- a/docs/OFFSCREEN_RENDERING.md
+++ b/docs/OFFSCREEN_RENDERING.md
@@ -1,9 +1,10 @@
# Offscreen Rendering
-**Status (2026-09-24):** step 1, offscreen rendering, is implemented
+**Status (2026-09-30):** offscreen rendering is implemented
(`DisplayManager.offscreen()`, the adapter on the prefetch thread, the plugin
-lock). Steps 2 and 3 are proposed. When all three land, this file becomes the
-reference for how plugin content is rendered off the render thread.
+lock), and so are live elements, which grew out of steps 2 and 3 below: see
+*Live elements*. The segment strip proposed as step 2 was not needed; *Why not
+a SegmentStrip* says why.
First soak of step 1 on hdpi (50 px/s, `pwm_bits` 7, preview open, 8-minute
runs, A/B/B/A):
@@ -153,136 +154,101 @@ particular keeps presenting while a plugin draws elsewhere.
fetch left is the inline fallback when no prepared group is ready, which in
practice is the first extension. Prefetching at start removes that too.
-## Keeping live content fresh
+## Live elements: content that changes while it scrolls
-Offscreen rendering is also what makes fresh sports scores possible. Today a
-plugin's segment is drawn when its group is prefetched, and the strip carries
-7,000–10,000 px of content ahead of the viewport (hdpi logs: "7153px still
-ahead", "9842px ahead"). At ~100 px/s, a score drawn now reaches the screen
-70–100 seconds later. When a plugin reports new data, Vegas only drops its
-cache (`invalidate_pending_updates`), so the change is drawn on the plugin's
-*next* turn, several minutes later. A segment already in the strip scrolls by
-with the data it was drawn with.
+Offscreen rendering is also what makes fresh content possible. A plugin's
+segment is drawn when its group is prefetched, and the strip carries
+7,000-10,000 px of content ahead of the viewport, so at ~100 px/s a score drawn
+then reaches the screen 70-100 seconds later -- and once in the strip it never
+changed: when a plugin reported new data, Vegas only dropped its caches, so the
+change appeared on the plugin's *next* turn, minutes later.
-That was the right trade while every redraw of a canvas-bound plugin stalled
-the scroll. Off the render thread a redraw costs the scroll nothing, so the
-strip can afford three things.
+A plugin can now hand Vegas **live elements** instead of pictures
+(`BasePlugin.get_vegas_elements()`, see "Live Vegas elements" in
+[PLUGIN_API_REFERENCE.md](PLUGIN_API_REFERENCE.md#live-vegas-elements)):
+named, fixed-width pieces of content -- one per game card, one for a map. Vegas
+records where each lands in the strip and, when the plugin's data changes,
+redraws just the changed ones off the render thread and copies their pixels
+over the old ones between two frames. A card already crossing the panel
+changes; nothing next to it moves.
-### 1. Refresh at the gate
+### Why not redraw every frame
-Before a segment enters the viewport, check whether its plugin has updated
-since the segment was drawn. If it has, redraw it offscreen and replace it
-while it is still out of sight. Width changes are fine here, because
-everything from that segment onward is still invisible.
+On a Pi the render thread has about 4 ms of slack per refresh at 512x64 after
+the ~6 ms blit, and a scoreboard card is ~29 ms of Pillow work that holds the
+GIL. Drawing on the render thread is out of the question at any rate, so the
+render thread only ever *copies* pixels that are already drawn. Measured on a
+Pi 4 (ledpi): writing a 35 KB card into a 20,000 px strip takes 8.5 µs, a
+101 KB map 17 µs, four cards (the per-frame cap) 34 µs -- against 124 µs for
+the viewport slice every frame already does.
-The gate sits `lead` pixels ahead of the viewport's right edge:
-`lead = max(one screen, speed × (render time + margin))`. The render time is
-the plugin's own, measured on each render (sports cards take the longest,
-hundreds of ms up to seconds per the prefetch notes). A plugin whose render
-does not finish before its segment reaches the viewport keeps the old segment.
-The scroll never waits for it.
+### How an update reaches the screen
-Content is then at most `lead / speed` seconds old when it appears, a few
-seconds instead of minutes, without changing how far ahead the rotation
-fetches.
+1. A plugin's `update()` completes. The update worker calls
+ `PluginManager._note_update_completed`, which calls the update listeners
+ (`add_update_listener`) there and then, with the plugin's lock still held.
+ Vegas's listener moves the plugin's **epoch** on
+ (`src/vegas_mode/elements.py`, `LiveEpochs`) and wakes the live worker.
+2. The **live worker** (`src/vegas_mode/live_worker.py`), the one background
+ thread that draws for the strip once it holds a live element, finds the
+ plugin's elements whose recorded epoch is older than its current one,
+ nearest the screen first, and calls `get_vegas_elements()` under the
+ plugin's lock (0.25 s wait, then a 1 s backoff). Elements whose `version`
+ is unchanged cost nothing; the rest are pinned and checksummed, and each
+ whose pixels changed becomes a patch in a one-per-element slot (the latest
+ wins).
+3. Between two frames the render thread
+ (`RenderPipeline.apply_live_patches`, from `coordinator.run_frame`) pops at
+ most four patches or two screens of bytes and copies each into the strip
+ with `ScrollHelper.patch_columns`. It takes no lock and draws nothing. A
+ patch made for an older strip, for an element trimmed away or already
+ behind the screen, or from older data than the strip shows, is dropped.
-### 2. Replace ahead of the screen
+End to end, a new score reaches a card already on screen within one poll of
+the data source (30 s for live games) plus about a second: the listener is
+immediate, and while live elements exist the update tick that schedules
+plugins runs every second instead of every four.
-When a plugin reports new data (the Vegas update tick already names them), any
-of its segments that are **anywhere ahead of the viewport** are redrawn and
-replaced straight away, not only at the gate. That covers the long stretch of
-strip between prefetch and the gate.
+Elements that change with **time** rather than data (an aircraft moving
+between position reports) ask for `refresh_hz`; the worker calls
+`redraw_vegas_element()` -- without the plugin's lock, from state the plugin
+publishes in one assignment -- that often while the element is on or within
+`live_lead_screens` of the screen, capped by `live_max_hz` (5), at 1 Hz
+without the render gate, and halved for an element whose redraws average over
+50 ms.
-### 3. Update on screen
+### Geometry
-A segment that is already **visible** is patched in place when the redrawn
-version has the same geometry: the same total width, and the same width for
-each card (a sports plugin returns one image per game, joined with
-`intra_plugin_gap`). Scoreboard cards keep a fixed layout, so a score change
-patches in and the digits update as the card scrolls past. The patch is a
-pixel copy of one card (a 150×64 card is ~29 KB) applied by the render thread
-between frames, so a frame never shows half of a patch.
+A live element is never trimmed to its ink: the adapter pads it with
+`content_padding` black columns either side and pins its width, and a redraw
+at any other width is refused (it shows the next time the plugin comes round).
+Records keep **absolute** strip columns -- the strip column plus everything
+trimmed off the front since the strip was composed -- so a trim moves one
+origin rather than every record. Nothing on screen is ever moved, inserted or
+resized; a game added to a slate appears on the plugin's next turn.
-When the geometry differs (a game added or dropped, a card that grew), the
-visible part cannot change without a jump. Only the cards not yet on screen
-are replaced, and only if the geometry up to that point is unchanged. Otherwise
-the segment keeps its snapshot until it has scrolled off.
+### Why not a SegmentStrip
-### Avoiding wasted work
+The proposal here was to replace the single strip with a list of segments.
+In-place patching of the single strip meets every goal without that: the
+patch is O(element) and the strip layout never changes. What a segment list
+would still buy is cheaper extensions, and most of that came from making the
+strip's PIL copy lazy instead (`ScrollHelper.cached_image`: an extension used
+to rebuild it twice, 1.7-3.8 ms each on a Pi 4). The `extend` row of
+`frame_soak.py`'s "after work" table says whether the rest is worth it.
-- **Change detection.** `run_scheduled_updates_with_changes()` names a plugin
- whenever its `update()` ran, not when its data changed. On hdpi
- `clock-simple` and `ledmatrix-music` are named on every 4-second tick. A
- redraw whose pixels hash the same as the segment's is discarded without a
- swap.
-- **Redraw on real updates only.** Vegas makes no API calls. Each plugin
- fetches on its own schedule, and a redraw is triggered only when the
- plugin's `update()` has run since its segment was drawn. On hdpi live
- football, baseball and hockey poll every 30 s (live odds every 60 s,
- everything else hourly), so a live sports card is redrawn once per poll.
-- **Floor.** A plugin is redrawn at most once per
- `vegas_scroll.refresh_min_interval` (proposed 10 s), and never while its
- previous redraw is still running. The floor never holds back a sports card
- polling every 30 s. It exists for chatty plugins: `clock-simple` updates
- every second and `ledmatrix-music` polls every 2 s.
-- **One worker.** Redraws go through the same background worker as prefetch,
- one plugin at a time at `nice 10`, under the plugin's lock.
+### When it is off
-Data freshness is still bounded by each plugin's own fetch interval (how often
-it polls live scores). Drawing faster cannot beat the data source.
-
-### The strip becomes a list of segments
-
-All three need the strip to be replaceable by segment. Today it is one
-image (`ScrollHelper.cached_array`, 8,000–20,000 px wide, 1.5–3.8 MB), and
-`append_content()` rebuilds the whole thing on the render thread for every
-appended block. That is also a pause source.
-
-Proposed `SegmentStrip`, used by Vegas in place of the single image:
-
-- an ordered list of segments: plugin id, card boundaries, a pixel array, the
- render time, and the plugin data version it was drawn from, plus its
- x-offset in the strip;
-- `visible(x, width)` assembles the viewport by slicing across at most a few
- segments: the same ~100 KB copy per frame that slicing the single image
- costs today;
-- append and trim become O(block) list operations, not a copy of the strip;
-- replace swaps one list entry and shifts the offsets of the segments after it
- (dozens at most). A same-geometry patch copies pixels into the existing array.
-
-Every mutation is prepared off the render thread and applied by the render
-thread at a frame boundary, so the strip the render loop reads is never
-half-changed.
-
-### Multi-display sync
-
-The follower renders from its own copy of the strip, offset from the leader's
-scroll position. Today the leader sends that copy whole, and only in
-`start_new_cycle()` (`send_scroll_image`), plus the scroll position every
-frame. Continuous scroll, the default, extends and trims the strip without
-starting a new cycle, and nothing sends those changes. From reading the code,
-the follower therefore probably falls out of step after the first extension
-already, before any of this design. That is untested; it needs a two-Pi rig.
-
-With a segment strip, keeping the follower identical becomes **replaying the
-leader's operations**:
-
-- Every strip mutation (append, trim, replace, patch) is one operation in
- strip coordinates. The leader applies it and sends the same operation to the
- follower over the existing TCP channel. Segments are small: a card is ~29 KB
- raw and compresses well.
-- Operations on off-screen segments apply on arrival. A patch to a segment
- that is on either panel carries an *apply at scroll position X* stamp a
- couple of hundred milliseconds ahead. Both sides apply it when their scroll
- position passes X, so both panels change on the same frame, within the
- existing position-sync jitter.
-- Each operation carries a sequence number. A follower that sees a gap (a
- reconnect, a dropped message) asks for a full snapshot, which is today's
- `send_scroll_image` path.
-
-That also fixes the probable continuous-mode gap as a side effect, since
-appends and trims become operations too. Until it is in place, fresh-content
-updates are disabled while sync is active.
+- `display.vegas_scroll.live_refresh: false` (the kill switch; also in the
+ web UI), or `vegas_live: false` in one plugin's section.
+- Always under multi-display sync: the follower mirrors whole strips only, so
+ a patch would never reach it. (Continuous-mode sync has a separate problem:
+ the follower is not sent extensions or trims at all.)
+- In swap mode (`continuous_scroll: false`) and with `offscreen_prefetch:
+ false`.
+- For plugins without the hook, which are drawn and placed exactly as before,
+ and on the paths that fetch without the plugin's lock (the first strip of a
+ run, the render thread's fallback fetch), which use `get_vegas_content()`.
## Risks, and what was checked
@@ -328,11 +294,10 @@ updates are disabled while sync is active.
source of the single-refresh late frames. Holding frames for two refreshes
(≈50 px/s) doubles the budget. Cutting the blit itself is the native-presenter
step.
-- **Live refreshes pushed from `update()`.** Some sports plugins call
- `display()` and `update_display()` from inside `update()`, which runs on the
- update worker and can push to the panel mid-Vegas. That is a separate
- hazard. `offscreen()` gives a tool for it (run the update worker offscreen
- while Vegas owns the panel), but it is out of scope here.
+- **Multi-display sync in continuous mode.** The follower is sent the whole
+ strip only at a new cycle and on connect, never the extensions and trims of
+ continuous mode, so it drifts from the leader after the first extension.
+ Live elements stay off under sync for that reason.
## Test plan
@@ -348,14 +313,15 @@ updates are disabled while sync is active.
- **Emulator integration:** a stub canvas-bound plugin whose `display()` sleeps
300 ms. The Vegas render loop never goes a frame without presenting (frame
timing recorder: zero freezes).
-- **Unit, `SegmentStrip`:** the viewport assembled across segment boundaries
- matches slicing one concatenated image, pixel for pixel. Append, trim,
- replace-ahead and same-geometry patch each leave every other column
- unchanged. A geometry-changing patch of a visible segment is refused.
-- **Freshness:** a stub sports plugin whose score changes every second. The
- score on screen is never older than `lead / speed` plus the plugin's fetch
- interval. A visible card's digits change without the frame-timing recorder
- seeing a late frame. An unchanged redraw is discarded.
+- **Live elements** (`test/test_vegas_live_*.py`,
+ `test/test_vegas_elements_*.py`, `test/test_scroll_helper_patch.py`): every
+ record points at exactly its element's pixels through any sequence of
+ compose, extend and trim; a patch changes only its element's columns (a
+ property test against a twin strip that is never patched); the render
+ thread's apply takes no lock and draws nothing; the worker's priorities,
+ floors, backoff and hand-over; and, end to end on the emulator with the stub
+ plugin (`test/fixtures/plugins/vegas-live-stub`), an update changes a card
+ already in the strip and an animated element moves with no update at all.
- **Hardware:** an hdpi soak, A/B against the #628 build, alternating order.
Targets: no freezes, an empty 6+ bucket, the 3–5 bucket near zero, and the late
rate below 0.66%. Plus, for freshness: log each segment's age when it enters
@@ -363,26 +329,23 @@ updates are disabled while sync is active.
## Rollout
-Three changes, each soaked on hdpi before the next:
+1. **Offscreen rendering** (shipped): `offscreen()`, the adapter on the
+ prefetch thread, and the plugin lock. Removed the render-thread pauses.
+2. **Measurement and the lazy strip image:** late frames attributed to the
+ render-thread work before them (`FrameTimingRecorder.note_op`, the "after
+ work" table), and extensions no longer rebuilding the strip's PIL copy.
+3. **Live elements:** the plugin API, the records, the worker and in-place
+ patches, with the sports scoreboards and the flight map adopting it.
-1. **Offscreen rendering:** `offscreen()`, the adapter on the prefetch thread,
- and the plugin lock. Removes the render-thread pauses.
-2. **`SegmentStrip`:** Vegas's strip becomes a list of segments. Removes the
- whole-strip copy on append. No visible behaviour change.
-3. **Fresh content:** refresh at the gate, replace ahead, patch on screen,
- with change detection and the rate limit.
-
-`display.vegas_scroll.offscreen_prefetch` (default `true`) restores today's
+`display.vegas_scroll.offscreen_prefetch` (default `true`) restores the
deferred path when `false`, and `display.vegas_scroll.live_refresh` (default
-`true`) turns off step 3. Keep both for one release, then delete the old paths.
+`true`) turns live elements off. Keep both for one release, then delete the
+old paths.
## Open questions
-1. Keep the kill switch, or ship without one?
-2. Plugin lock timeout: skip the plugin and keep its cached segment (proposed),
- or wait longer?
-3. `refresh_min_interval`: 10 s proposed. It only limits chatty plugins;
- live sports are redrawn once per 30 s poll regardless.
-4. Multi-display sync: is there a two-Pi rig to test on? Operation replay is
- proposed as part of the segment strip (step 2), with fresh content
- disabled under sync until it has been verified on real hardware.
+1. Multi-display sync: is there a two-Pi rig to test on? Replaying strip
+ operations to the follower (append, trim, patch, in absolute columns) would
+ fix continuous-mode sync and let live elements run under it.
+2. Is the `extend` cost worth a segment list after the lazy image? The soak's
+ "after work" table answers it per rig.
diff --git a/docs/PLUGIN_API_REFERENCE.md b/docs/PLUGIN_API_REFERENCE.md
index d60ce035..92f587da 100644
--- a/docs/PLUGIN_API_REFERENCE.md
+++ b/docs/PLUGIN_API_REFERENCE.md
@@ -419,7 +419,7 @@ class MyScoreboard(BasePlugin):
|---|---|
| `key` | Names the element across redraws; unique in the list, stable for the same logical item (`"game:nfl:401547417"`, `"map"`). |
| `image` | The element now, at the display's height. A live element's **width must not depend on its data**: a redraw at another width is never swapped in (it appears the next time the plugin comes round), because nothing on screen may move. |
-| `version` | Anything hashable that changes exactly when the pixels would; lets the ticker skip unchanged elements. `None` means "compare pixels". |
+| `version` | Anything hashable that changes exactly when the pixels would. Handed back with the **same image object** as last time, it lets the ticker skip converting the element; a new image is always converted and compared by its pixels, so a redraw for new settings is never missed. `None` means "compare pixels". |
| `live` | `False` places it as plain content (trimmed, never refreshed): separators, decoration. |
| `refresh_hz` | For content that changes with **time** rather than data (an aircraft moving between position reports): the ticker calls `redraw_vegas_element()` about this often while the element is on or near the screen, capped by `vegas_scroll.live_max_hz` and at 1 Hz without the rebuilt rgbmatrix binding. |
diff --git a/mypy-clean.txt b/mypy-clean.txt
index e1381349..9564f28e 100644
--- a/mypy-clean.txt
+++ b/mypy-clean.txt
@@ -72,13 +72,17 @@ src/plugin_system/testing/loading.py
src/plugin_system/testing/mocks.py
src/plugin_system/testing/plugin_test_base.py
src/plugin_system/testing/sizes.py
+src/plugin_system/testing/vegas.py
+src/plugin_system/vegas_elements.py
src/redaction.py
src/scan_order.py
src/startup_validator.py
src/vegas_mode/__init__.py
src/vegas_mode/config.py
src/vegas_mode/coordinator.py
+src/vegas_mode/elements.py
src/vegas_mode/geometry.py
+src/vegas_mode/live_worker.py
src/vegas_mode/stream_manager.py
src/web_interface/api_helpers.py
src/web_interface/config_arrays.py
diff --git a/src/common/scroll_helper.py b/src/common/scroll_helper.py
index 2072a532..0f740abd 100644
--- a/src/common/scroll_helper.py
+++ b/src/common/scroll_helper.py
@@ -780,6 +780,43 @@ class ScrollHelper:
)
return cut
+ def patch_columns(self, x: int, pixels: np.ndarray) -> int:
+ """Overwrite the strip's columns from ``x`` with ``pixels``, in place.
+
+ What a live Vegas element update is (src/vegas_mode/elements.py): the
+ strip keeps its width, the scroll keeps its position, and only these
+ columns change. Call it between frames on the thread that draws them;
+ every frame copies its slice out of the strip (get_visible_portion),
+ so no frame already handed on can see half a patch.
+
+ Clipped to the strip at both ends. Refused (0) for an array this
+ helper may not write -- the multi-display follower adopts a read-only
+ one -- or for pixels of another height. The PIL image is deferred, so
+ a later read of cached_image shows the patch.
+
+ Args:
+ x: Strip column of the first column of ``pixels``
+ pixels: uint8 array (height, width, 3)
+
+ Returns:
+ Bytes written.
+ """
+ strip = self.cached_array
+ if strip is None or not strip.flags.writeable:
+ return 0
+ if pixels.ndim != 3 or pixels.shape[0] != strip.shape[0] \
+ or pixels.shape[2] != strip.shape[2]:
+ return 0
+ width = pixels.shape[1]
+ lo, hi = max(0, int(x)), min(strip.shape[1], int(x) + width)
+ if hi <= lo:
+ return 0
+ strip[:, lo:hi] = pixels[:, lo - int(x):hi - int(x)]
+ if self.__dict__.get('_cached_image') is not None \
+ or self.__dict__.get('_image_source') is not None:
+ self._defer_image()
+ return (hi - lo) * strip.shape[0] * strip.shape[2]
+
def remaining_unscrolled(self) -> int:
"""Columns of strip still to the right of the viewport."""
if not self.has_strip():
diff --git a/src/display_manager.py b/src/display_manager.py
index 4e32f630..4b910ef9 100644
--- a/src/display_manager.py
+++ b/src/display_manager.py
@@ -915,8 +915,7 @@ class DisplayManager:
``display.dirty_tracking: false`` if a redraw issue is ever suspected.
Serialized via ``_update_lock``: plugins can call this directly from
- background threads (e.g. sports base classes push an immediate
- "live" refresh from inside update()), so without a lock two callers
+ background threads of their own, so without a lock two callers
could both pass the digest check before either writes it back,
double-pushing a frame, or interleave the offscreen/current canvas
swap below. The lock is scoped to this method, so callers never
diff --git a/src/plugin_system/testing/vegas.py b/src/plugin_system/testing/vegas.py
index 2e44801e..d9286b4f 100644
--- a/src/plugin_system/testing/vegas.py
+++ b/src/plugin_system/testing/vegas.py
@@ -67,7 +67,7 @@ def _as_vegas_canvas(plugin: Any, display_manager: Any, width: int) -> Iterator[
def render_vegas_elements(plugin: Any, display_manager: Any,
- width: Optional[int] = None) -> Optional[list]:
+ width: Optional[int] = None) -> Any:
"""Call ``plugin.get_vegas_elements()`` as the Vegas ticker does."""
render_width = int(width or display_manager.width)
with _as_vegas_canvas(plugin, display_manager, render_width):
@@ -151,8 +151,9 @@ def check_vegas_elements(plugin: Any, display_manager: Any) -> VegasElementRepor
report.errors.append(f"{where} appears twice; keys must be unique")
continue
seen.add(key)
- if not isinstance(element.image, Image.Image):
- report.errors.append(f"{where} image is a {type(element.image).__name__}")
+ image: Any = element.image # typed Image, but a plugin may pass anything
+ if not isinstance(image, Image.Image):
+ report.errors.append(f"{where} image is a {type(image).__name__}")
continue
if element.image.height != height:
report.errors.append(
@@ -296,12 +297,12 @@ def check_plugin_vegas_elements(plugin_id: str, plugin_dir: Any, config: dict,
if run_update:
try:
plugin.update()
- except _TOLERATED_UPDATE_ERRORS as exc:
+ except Exception as exc: # noqa: BLE001 - a plugin's update can raise anything
+ if not isinstance(exc, _TOLERATED_UPDATE_ERRORS):
+ report.errors.append(f"update() raised {exc!r}")
+ return report
report.warnings.append(f"update() had no network ({exc!r}); checked "
"with whatever data the plugin starts with")
- except Exception as exc: # noqa: BLE001 - a plugin's update can raise anything
- report.errors.append(f"update() raised {exc!r}")
- return report
checked = check_vegas_elements(plugin, display_manager)
checked.warnings[:0] = report.warnings
return checked
diff --git a/src/vegas_mode/coordinator.py b/src/vegas_mode/coordinator.py
index a09ae7b2..e8071be0 100644
--- a/src/vegas_mode/coordinator.py
+++ b/src/vegas_mode/coordinator.py
@@ -387,6 +387,9 @@ class VegasModeCoordinator:
adapter = getattr(self, 'plugin_adapter', None)
if adapter is not None:
adapter.live_elements_enabled = active
+ pipeline = getattr(self, 'render_pipeline', None)
+ if pipeline is not None and hasattr(pipeline, 'set_live'):
+ pipeline.set_live(active)
plugin_manager = getattr(self, 'plugin_manager', None)
add = getattr(plugin_manager, 'add_update_listener', None)
remove = getattr(plugin_manager, 'remove_update_listener', None)
@@ -405,9 +408,11 @@ class VegasModeCoordinator:
"""Update listener: a plugin's data may have changed.
Runs on the update worker with the plugin's lock held, so it only
- moves the plugin's epoch on; whatever redraws happen later read it.
+ moves the plugin's epoch on and wakes the live-element worker, which
+ redraws once the lock is free.
"""
self.live_epochs.bump(plugin_id)
+ self.render_pipeline.notify_live_data(plugin_id)
def _install_render_gate(self) -> None:
"""Gate the prefetch thread on the render thread's swaps; see VegasModeConfig."""
@@ -504,6 +509,12 @@ class VegasModeCoordinator:
# game still shown as live the next morning.
self.render_pipeline.refresh_updated_plugins()
+ # Copy any live-element redraws the worker has finished into the
+ # strip, between this frame and the last. A deque check when there
+ # are none.
+ if self.live_active:
+ self.render_pipeline.apply_live_patches()
+
# Extend the strip before the scroll can reach its end, so the next
# group arrives from the right and motion never stops. No cycle
# boundary, so no freeze, no substitution and no restart with the
@@ -715,7 +726,12 @@ class VegasModeCoordinator:
# main loop's _tick_plugin_updates() finds all intervals already
# satisfied on return, so the inter-iteration gap is <1 ms and the
# display never shows a frozen frame between iterations.
- _UPDATE_TICK_FRAMES = max(1, int(self.render_pipeline.target_fps * 4)) # every 4 s regardless of FPS
+ # Every 4 s, or every 1 s while the strip holds live elements:
+ # plugins are only scheduled on this tick, so its period is added
+ # to how late a live update can be.
+ tick_seconds = (1.0 if self.live_active
+ and self.render_pipeline.has_live_records() else 4.0)
+ _UPDATE_TICK_FRAMES = max(1, int(self.render_pipeline.target_fps * tick_seconds))
if (self._update_callback and
frame_count % _UPDATE_TICK_FRAMES == 0 and
not self._update_tick_running):
diff --git a/src/vegas_mode/elements.py b/src/vegas_mode/elements.py
index caed7d97..04c8615e 100644
--- a/src/vegas_mode/elements.py
+++ b/src/vegas_mode/elements.py
@@ -60,6 +60,41 @@ class ElementRecord(NamedTuple):
refresh_hz: float
+class RenderedElement(NamedTuple):
+ """One live element freshly redrawn by the worker, ready to compare and swap."""
+ key: str
+ #: The plugin's data epoch it was drawn from.
+ epoch: int
+ version: object
+ #: Pinned pixels (see pin_element), read-only.
+ pixels: np.ndarray
+ digest: Tuple[Tuple[int, ...], int]
+ #: Pinned width, the width it would occupy in the strip.
+ width: int
+
+
+class LivePatch(NamedTuple):
+ """A redraw handed from the worker to the render thread for one record."""
+ seq: int
+ #: The strip generation it was made against; a patch for an older strip
+ #: is dropped.
+ strip_gen: int
+ epoch: int
+ pixels: np.ndarray
+ digest: Tuple[Tuple[int, ...], int]
+ made_at: float
+
+
+class LiveView(NamedTuple):
+ """Where the viewport is, in absolute strip columns, published every frame."""
+ abs_left: int
+ abs_right: int
+ #: The end of the strip: how far ahead content exists.
+ abs_end: int
+ #: time.monotonic() when published. An old one means frames have stopped.
+ t_mono: float
+
+
def tag(image: Image.Image, meta: ElementMeta) -> Image.Image:
"""Mark ``image`` as the live element ``meta`` describes. Returns it."""
image.info[INFO_KEY] = meta
diff --git a/src/vegas_mode/live_worker.py b/src/vegas_mode/live_worker.py
new file mode 100644
index 00000000..4dfece55
--- /dev/null
+++ b/src/vegas_mode/live_worker.py
@@ -0,0 +1,503 @@
+"""The one background worker behind live Vegas elements.
+
+Vegas draws everything the strip shows off the render thread. Until live
+elements that was one short-lived prefetch thread per group; now, once the
+strip holds a live element, it is this worker, which does three kinds of job
+one at a time, most urgent first:
+
+- **group steps**: fetching the next group of plugins for the strip, one
+ plugin per step (what the prefetch thread did in one go);
+- **data refreshes**: when a plugin's data has moved on (its epoch, see
+ elements.LiveEpochs) past what its elements in the strip were drawn from,
+ redraw them and hand over the ones whose pixels changed;
+- **ticks**: redraw an element that animates (``refresh_hz``) while it is on
+ or near the screen.
+
+Nothing here touches the strip. A finished redraw becomes a
+:class:`~src.vegas_mode.elements.LivePatch` in the pipeline's slot for that
+element (one per element, the latest wins) and the render thread copies it
+into the strip between two frames (RenderPipeline.apply_live_patches). The
+hand-over is lock-free: a dict store and a deque append here, a deque popleft
+and a dict pop there, so the render thread never waits on this thread.
+
+Every job runs inside the render gate when there is one (src/common/
+render_gate.py), so Python runs here only while the render thread is waiting
+for the panel. Without it (the stock rgbmatrix binding) animation is capped at
+:data:`UNGATED_MAX_HZ`.
+"""
+
+from __future__ import annotations
+
+import collections
+import logging
+import os
+import queue
+import threading
+import time
+from contextlib import nullcontext
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple
+
+from src.vegas_mode.elements import ElementRecord, LivePatch, LiveView
+
+logger = logging.getLogger(__name__)
+
+#: A view older than this means frames have stopped (a paused scroll, an
+#: interrupt): only group work runs, since nothing redrawn would be seen.
+VIEW_STALE_S = 0.5
+#: How long a data refresh waits for the plugin's lock before trying later.
+DATA_LOCK_TIMEOUT = 0.25
+#: ...and how much later.
+LOCK_BACKOFF_S = 1.0
+#: Animation ceiling without the render gate, where every redraw competes
+#: with the render thread for the GIL.
+UNGATED_MAX_HZ = 1.0
+#: An element whose redraws take longer than this on average is animated at
+#: half its rate, down to MIN_THROTTLED_HZ.
+SLOW_RENDER_S = 0.05
+MIN_THROTTLED_HZ = 0.5
+#: Longest the worker sleeps with nothing due, so a floor or a backoff that
+#: expires is noticed.
+IDLE_WAIT_S = 0.5
+#: How often the worker logs what it did, when it did anything.
+SUMMARY_INTERVAL_S = 300.0
+#: Weight of the newest sample in the per-element render time average.
+EWMA_ALPHA = 0.2
+#: How long the worker waits for a one-shot prefetch thread it takes over from.
+LEGACY_PREFETCH_JOIN_S = 15.0
+
+
+def _visible(record: ElementRecord, view: LiveView) -> bool:
+ return record.abs_x < view.abs_right and record.abs_x + record.width > view.abs_left
+
+
+def _behind(record: ElementRecord, view: LiveView) -> bool:
+ return record.abs_x + record.width <= view.abs_left
+
+
+class _GroupJob:
+ """A group fetch in progress, one member per step."""
+
+ def __init__(self, generation: int, plugin_ids: List[str]) -> None:
+ self.generation = generation
+ self.pending = list(plugin_ids)
+ self.group: List[Tuple[str, Any]] = []
+
+
+class VegasWorker(threading.Thread):
+ """See the module docstring. Owned by the RenderPipeline that starts it."""
+
+ def __init__(self, pipeline: Any, clock: Callable[[], float] = time.monotonic) -> None:
+ super().__init__(daemon=True, name="vegas-live-worker")
+ self.pipeline = pipeline
+ #: time.monotonic, or a fake one in tests; the pipeline's view is
+ #: stamped with time.monotonic too.
+ self._clock = clock
+ self.inbox: "queue.SimpleQueue[Tuple[str, Any]]" = queue.SimpleQueue()
+ self._stopping = False
+ self._group_wanted = False
+ self._group_job: Optional[_GroupJob] = None
+ self._strip_gen = pipeline._strip_gen
+ # Per element (record seq): the epoch this worker last handed over,
+ # or found needed nothing; the digest of its latest hand-over; when
+ # its next animation tick is due.
+ self._handled_epoch: Dict[int, int] = {}
+ self._handed_digest: Dict[int, Any] = {}
+ self._next_tick: Dict[int, float] = {}
+ # Per plugin: when its last data refresh ran, and a lock backoff.
+ self._last_data_job: Dict[str, float] = {}
+ self._backoff_until: Dict[str, float] = {}
+ # Per (plugin, key): average redraw time, for throttling.
+ self._render_ewma: Dict[Tuple[str, str], float] = {}
+ self._refused: Set[Tuple[str, str, int]] = set()
+ self.stats: collections.Counter = collections.Counter()
+ self.busy_seconds = 0.0
+ self._began = clock()
+ self._last_summary = self._began
+ self._jobs_since_prune = 0
+ # The slowest redraw since the last summary: (seconds, (plugin, key)).
+ self._slowest_redraw: Optional[Tuple[float, Tuple[str, str]]] = None
+
+ # -- control, from other threads ------------------------------------------
+
+ def request_group(self) -> None:
+ """Fetch the next group for the strip when nothing more urgent is due."""
+ self._group_wanted = True
+ self.inbox.put(("group", None))
+
+ def notify_data(self, plugin_id: str) -> None:
+ """A plugin's data moved on. Only a wake-up: its epoch is the truth."""
+ self.inbox.put(("data", plugin_id))
+
+ def stop(self) -> None:
+ """Stop after the current job. Does not wait for it."""
+ self._stopping = True
+ self.inbox.put(("stop", None))
+
+ # -- the loop -------------------------------------------------------------
+
+ def run(self) -> None:
+ try:
+ # Linux applies nice per thread: deprioritise against the render
+ # loop, as the one-shot prefetch thread always did.
+ os.nice(10)
+ except (OSError, AttributeError):
+ pass
+ self._join_legacy_prefetch()
+ while not self._should_stop():
+ self._wait(self._next_wait(self._clock()))
+ if self._should_stop():
+ break
+ job = self._pick(self._clock())
+ if job is not None:
+ self._run(job)
+ self._maybe_summarise()
+ self._hand_over_partial_group()
+ logger.debug("Vegas live worker stopped")
+
+ def _should_stop(self) -> bool:
+ # A method, not a bare attribute read: stop() sets it from another
+ # thread between two reads in run().
+ return self._stopping
+
+ def _join_legacy_prefetch(self) -> None:
+ """Let a one-shot prefetch thread, or a worker stopped earlier, finish first.
+
+ So that only one thread ever draws for the strip: measured on hdpi,
+ each extra thread competing for the GIL made the render thread late
+ more often, not less (src/common/render_gate.py).
+ """
+ for name in ('_prefetch_thread', '_retired_worker'):
+ thread = getattr(self.pipeline, name, None)
+ if thread is not None and thread is not self \
+ and thread is not threading.current_thread() and thread.is_alive():
+ thread.join(LEGACY_PREFETCH_JOIN_S)
+
+ def _wait(self, timeout: float) -> None:
+ try:
+ message = self.inbox.get(timeout=max(0.0, timeout))
+ except queue.Empty:
+ return
+ while True:
+ if message[0] == "stop":
+ self._stopping = True
+ elif message[0] == "group":
+ self._group_wanted = True
+ try:
+ message = self.inbox.get_nowait()
+ except queue.Empty:
+ return
+
+ def _next_wait(self, now: float) -> float:
+ """Seconds until something may be due: the next tick, else IDLE_WAIT_S."""
+ p = self.pipeline
+ if self._group_job is not None or (
+ self._group_wanted and p._prepared_group is None):
+ return 0.0
+ # Ticks only count while frames are flowing: during a pause _pick runs
+ # none, and a past-due tick would otherwise make this 0 and spin.
+ view = p._view
+ if view is None or now - view.t_mono > VIEW_STALE_S:
+ return IDLE_WAIT_S
+ # Only elements _due_tick would run. A tick left behind by an element
+ # trimmed away, or one no longer animated, is never run, and counting
+ # it held this at its floor: a spin at 100 wake-ups a second.
+ soonest = now + IDLE_WAIT_S
+ lead = self._tick_lead()
+ for record in p._elements:
+ if self._tickable(record, view, lead):
+ soonest = min(soonest, self._next_tick.get(record.seq, now))
+ return max(0.01, soonest - now)
+
+ # -- choosing ---------------------------------------------------------------
+
+ def _pick(self, now: float) -> Optional[Tuple[str, Any]]:
+ """The most urgent job, or None. See the module docstring for the order."""
+ p = self.pipeline
+ if p._strip_gen != self._strip_gen:
+ self._forget_everything(p._strip_gen)
+ view = p._view
+ fresh = view is not None and now - view.t_mono <= VIEW_STALE_S
+ group_ready = self._group_job is not None or (
+ self._group_wanted and p._prepared_group is None)
+ records = p._elements
+
+ if group_ready and fresh and self._group_urgent(view):
+ return ("group", None)
+ if fresh and records:
+ plugin_id = self._due_data(records, view, now, visible_only=True)
+ if plugin_id is not None:
+ return ("data", plugin_id)
+ record = self._due_tick(records, view, now)
+ if record is not None:
+ return ("tick", record)
+ if group_ready:
+ return ("group", None)
+ if fresh and records:
+ plugin_id = self._due_data(records, view, now, visible_only=False)
+ if plugin_id is not None:
+ return ("data", plugin_id)
+ return None
+
+ def _group_urgent(self, view: LiveView) -> bool:
+ width = self.pipeline.display_width
+ threshold = (self.pipeline.config.extend_threshold_screens + 1.0) * width
+ return bool(view.abs_end - view.abs_right <= threshold)
+
+ def _epoch(self, plugin_id: str) -> int:
+ epochs = getattr(self.pipeline.stream_manager.plugin_adapter, 'live_epochs', None)
+ return int(epochs.get(plugin_id)) if epochs is not None else 0
+
+ def _done_epoch(self, record: ElementRecord) -> int:
+ applied = self.pipeline._applied.get(record.seq)
+ return max(applied[0] if applied is not None else record.epoch,
+ self._handled_epoch.get(record.seq, -1))
+
+ def _due_data(self, records: Tuple[ElementRecord, ...], view: LiveView,
+ now: float, visible_only: bool) -> Optional[str]:
+ """The plugin whose stale elements are nearest the screen, if any may redraw."""
+ floor = self.pipeline.config.live_min_interval
+ best: Optional[Tuple[int, str]] = None
+ for record in records:
+ if _behind(record, view):
+ continue
+ if visible_only and not _visible(record, view):
+ continue
+ plugin_id = record.plugin_id
+ if self._epoch(plugin_id) <= self._done_epoch(record):
+ continue
+ if now < self._backoff_until.get(plugin_id, 0.0):
+ continue
+ if now - self._last_data_job.get(plugin_id, float('-inf')) < floor:
+ continue
+ distance = max(0, record.abs_x - view.abs_right)
+ if best is None or distance < best[0]:
+ best = (distance, plugin_id)
+ return best[1] if best is not None else None
+
+ def _tick_hz(self, record: ElementRecord) -> float:
+ cfg = self.pipeline.config
+ hz = min(float(record.refresh_hz), float(cfg.live_max_hz))
+ if getattr(self.pipeline.display_manager, 'render_gate', None) is None:
+ hz = min(hz, UNGATED_MAX_HZ)
+ ewma = self._render_ewma.get((record.plugin_id, record.key), 0.0)
+ if hz > 0 and ewma > SLOW_RENDER_S:
+ # Halved, but never below the floor -- nor raised to it, for an
+ # element already asking for less.
+ hz = min(hz, max(MIN_THROTTLED_HZ, hz / 2.0))
+ return hz
+
+ def _tick_lead(self) -> float:
+ return float(self.pipeline.config.live_lead_screens * self.pipeline.display_width)
+
+ def _tickable(self, record: ElementRecord, view: LiveView, lead: float) -> bool:
+ """Whether an element animates now: it has a rate, and is on or near the screen."""
+ return (record.refresh_hz > 0 and self._tick_hz(record) > 0
+ and not _behind(record, view) and record.abs_x < view.abs_right + lead)
+
+ def _due_tick(self, records: Tuple[ElementRecord, ...], view: LiveView,
+ now: float) -> Optional[ElementRecord]:
+ lead = self._tick_lead()
+ best: Optional[ElementRecord] = None
+ best_due = 0.0
+ for record in records:
+ if not self._tickable(record, view, lead):
+ self._next_tick.pop(record.seq, None)
+ continue
+ due = self._next_tick.get(record.seq, now)
+ if due <= now and (best is None or due < best_due):
+ best, best_due = record, due
+ return best
+
+ # -- running ----------------------------------------------------------------
+
+ def _run(self, job: Tuple[str, Any]) -> None:
+ kind, arg = job
+ gate = getattr(self.pipeline.display_manager, 'render_gate', None)
+ started = self._clock()
+ try:
+ with gate.yielding() if gate is not None else nullcontext():
+ if kind == "group":
+ self._group_step()
+ elif kind == "data":
+ self._data_job(arg, started)
+ else:
+ self._tick_job(arg, started)
+ self.stats[kind] += 1
+ except Exception as exc: # pylint: disable=broad-except
+ # Plugin code runs in here; one bad job must not end the worker,
+ # which also fetches every group for the strip.
+ self.stats["errors"] += 1
+ if self.stats["errors"] <= 3 or self.stats["errors"] % 100 == 0:
+ logger.exception("Vegas live worker: %s job failed (%s)", kind, exc)
+ finally:
+ self.busy_seconds += self._clock() - started
+ self._jobs_since_prune += 1
+ if self._jobs_since_prune >= 64:
+ self._prune()
+
+ def _hand_over_partial_group(self) -> None:
+ """On stopping: publish the members of a group already fetched.
+
+ Its plugins were taken from the rotation when it was planned, so a
+ group dropped here would skip them until the next cycle. The rest of
+ it is not fetched. Nothing is published over a group already waiting,
+ or into a Vegas reset since.
+ """
+ job, self._group_job = self._group_job, None
+ if job is None or not job.group:
+ return
+ p = self.pipeline
+ with p._prefetch_lock:
+ if job.generation == p._prefetch_generation and p._prepared_group is None:
+ p._prepared_group = job.group
+
+ def _group_step(self) -> None:
+ p = self.pipeline
+ job = self._group_job
+ if job is None:
+ with p._prefetch_lock:
+ if p._prepared_group is not None:
+ self._group_wanted = False
+ return
+ generation = p._prefetch_generation
+ self._group_wanted = False
+ job = self._group_job = _GroupJob(generation, p.stream_manager.plan_next_group())
+ if job.generation != p._prefetch_generation:
+ self._group_job = None # Vegas was reset meanwhile
+ return
+ if job.pending:
+ member = p.stream_manager.fetch_group_member(
+ job.pending.pop(0), offscreen_only=True)
+ if member is not None:
+ job.group.append(member)
+ if not job.pending:
+ self._group_job = None
+ with p._prefetch_lock:
+ if job.generation == p._prefetch_generation:
+ p._prepared_group = job.group
+
+ def _data_job(self, plugin_id: str, now: float) -> None:
+ p = self.pipeline
+ self._last_data_job[plugin_id] = now
+ plugin = getattr(p.stream_manager.plugin_manager, 'plugins', {}).get(plugin_id)
+ if plugin is None:
+ return
+ gen = p._strip_gen
+ batch = p.stream_manager.plugin_adapter.render_live_elements(
+ plugin, plugin_id, lock_timeout=DATA_LOCK_TIMEOUT)
+ if batch is None:
+ self.stats["lock_busy"] += 1
+ self._backoff_until[plugin_id] = now + LOCK_BACKOFF_S
+ return
+ epoch, rendered = batch
+ view = p._view
+ for record in p._elements:
+ if record.plugin_id != plugin_id:
+ continue
+ if view is not None and _behind(record, view):
+ continue
+ element = rendered.get(record.key)
+ if element is not None:
+ self._hand_over(record, element, epoch, gen)
+ # A key the plugin no longer has keeps its last pixels until it
+ # scrolls off; either way this epoch is dealt with.
+ self._handled_epoch[record.seq] = max(
+ epoch, self._handled_epoch.get(record.seq, -1))
+
+ def _tick_job(self, record: ElementRecord, now: float) -> None:
+ p = self.pipeline
+ hz = self._tick_hz(record)
+ self._next_tick[record.seq] = now + (1.0 / hz if hz > 0 else IDLE_WAIT_S)
+ plugin = getattr(p.stream_manager.plugin_manager, 'plugins', {}).get(record.plugin_id)
+ if plugin is None:
+ return
+ adapter = p.stream_manager.plugin_adapter
+ key = (record.plugin_id, record.key)
+ at = now + self._render_ewma.get(key, 0.0) + p.frame_interval
+ started = time.perf_counter()
+ if adapter.has_lock_free_redraw(plugin):
+ # None from the plugin means nothing to redraw this time.
+ element = adapter.redraw_live_element(
+ plugin, record.plugin_id, record.key, record.width, p.display_height, at)
+ else:
+ # No lock-free redraw: redraw everything, but never wait for the
+ # plugin's lock (update() may be doing network I/O under it).
+ batch = adapter.render_live_elements(plugin, record.plugin_id, lock_timeout=0.0)
+ element = batch[1].get(record.key) if batch is not None else None
+ took = time.perf_counter() - started
+ if self._slowest_redraw is None or took > self._slowest_redraw[0]:
+ self._slowest_redraw = (took, key)
+ previous = self._render_ewma.get(key)
+ self._render_ewma[key] = took if previous is None else (
+ EWMA_ALPHA * took + (1.0 - EWMA_ALPHA) * previous)
+ if element is not None:
+ self._hand_over(record, element, element.epoch, p._strip_gen)
+
+ def _hand_over(self, record: ElementRecord, element: Any, epoch: int, gen: int) -> None:
+ """Queue a redraw for the render thread, unless nothing would change."""
+ p = self.pipeline
+ if element.width != record.width:
+ marker = (record.plugin_id, record.key, element.width)
+ if marker not in self._refused:
+ self._refused.add(marker)
+ logger.info(
+ "[%s] Live element %r redrawn %dpx wide, placed at %dpx; "
+ "kept as it was (a live element's width must not change)",
+ record.plugin_id, record.key, element.width, record.width)
+ self.stats["refused"] += 1
+ return
+ last = self._handed_digest.get(record.seq)
+ if last is None:
+ applied = p._applied.get(record.seq)
+ last = applied[1] if applied is not None else record.digest
+ if element.digest == last:
+ self.stats["unchanged"] += 1
+ return
+ p._live_slots[record.seq] = LivePatch(
+ seq=record.seq, strip_gen=gen, epoch=epoch, pixels=element.pixels,
+ digest=element.digest, made_at=self._clock())
+ p._live_ready.append(record.seq)
+ self._handed_digest[record.seq] = element.digest
+ self.stats["patches"] += 1
+
+ # -- housekeeping -----------------------------------------------------------
+
+ def _forget_everything(self, gen: int) -> None:
+ self._strip_gen = gen
+ self._handled_epoch.clear()
+ self._handed_digest.clear()
+ self._next_tick.clear()
+
+ def _prune(self) -> None:
+ self._jobs_since_prune = 0
+ live = {record.seq for record in self.pipeline._elements}
+ for table in (self._handled_epoch, self._handed_digest, self._next_tick):
+ for seq in [s for s in table if s not in live]:
+ del table[seq]
+
+ def _maybe_summarise(self) -> None:
+ now = self._clock()
+ if now - self._last_summary < SUMMARY_INTERVAL_S:
+ return
+ elapsed = now - self._last_summary
+ self._last_summary = now
+ stats, self.stats = self.stats, collections.Counter()
+ busy, self.busy_seconds = self.busy_seconds, 0.0
+ slowest, self._slowest_redraw = self._slowest_redraw, None
+ if not stats:
+ return
+ redraws = ""
+ if slowest is not None:
+ # What a tick costs is the number that decides whether an
+ # animated element can keep its rate: say it for the worst one.
+ took, (plugin_id, key) = slowest
+ average = self._render_ewma.get((plugin_id, key), took)
+ redraws = "; slowest redraw %.1fms (%s %r, average %.1fms)" % (
+ took * 1000.0, plugin_id, key, average * 1000.0)
+ logger.info(
+ "Vegas live: %d group step(s), %d data refresh(es), %d tick(s); "
+ "%d patch(es) handed over, %d unchanged, %d refused, %d lock-busy, "
+ "%d error(s); worker busy %.1f%%%s",
+ stats["group"], stats["data"], stats["tick"], stats["patches"],
+ stats["unchanged"], stats["refused"], stats["lock_busy"],
+ stats["errors"], 100.0 * busy / elapsed if elapsed else 0.0, redraws)
diff --git a/src/vegas_mode/plugin_adapter.py b/src/vegas_mode/plugin_adapter.py
index cf5f65dc..3e51c208 100644
--- a/src/vegas_mode/plugin_adapter.py
+++ b/src/vegas_mode/plugin_adapter.py
@@ -9,7 +9,7 @@ import logging
import threading
import time
from contextlib import contextmanager, nullcontext
-from typing import Optional, List, Any, Tuple, Union, TYPE_CHECKING
+from typing import Dict, Optional, List, Any, Tuple, Union, TYPE_CHECKING
from PIL import Image
from src.common.scroll_helper import ScrollHelper
@@ -18,6 +18,7 @@ from src.plugin_system.vegas_elements import VegasElement
from src.vegas_mode.elements import (
ElementMeta,
LiveEpochs,
+ RenderedElement,
meta_of,
pin_element,
pixel_digest,
@@ -109,6 +110,11 @@ class PluginAdapter:
# Element problems already reported, so a plugin with a bad hook logs
# once rather than on every fetch.
self._element_warnings: set = set()
+ # (plugin_id, key) -> (version, source image, (padding, height),
+ # pinned pixels, digest) of the last conversion, so an element handed
+ # back unchanged is not converted again. Only the live-element worker
+ # reads or writes it; invalidate_cache() swaps in a fresh one.
+ self._element_memo: Dict[Tuple[str, str], Tuple[Any, ...]] = {}
logger.debug(
"PluginAdapter initialized: display=%dx%d",
@@ -198,18 +204,20 @@ class PluginAdapter:
return bool(raw)
@contextmanager
- def _plugin_lock(self, plugin_id: str):
+ def _plugin_lock(self, plugin_id: str, timeout: Optional[float] = None):
"""Hold the plugin's update/display lock, waiting a bounded time.
Yields whether it was acquired. Yields True, holding nothing, when
there is no plugin manager to ask -- the behaviour before the lock was
- taken here at all.
+ taken here at all. ``timeout`` defaults to PLUGIN_LOCK_TIMEOUT; 0
+ does not wait at all.
"""
if not hasattr(self.plugin_manager, 'get_plugin_lock'):
yield True
return
lock = self.plugin_manager.get_plugin_lock(plugin_id)
- acquired = lock.acquire(timeout=self.PLUGIN_LOCK_TIMEOUT)
+ wait = self.PLUGIN_LOCK_TIMEOUT if timeout is None else timeout
+ acquired = lock.acquire(timeout=wait) if wait > 0 else lock.acquire(blocking=False)
try:
yield acquired
finally:
@@ -455,18 +463,21 @@ class PluginAdapter:
if isinstance(plugin_cfg, dict):
raw = plugin_cfg.get('vegas_width_pct')
if raw not in (None, ''):
+ # Reported once per value: the live paths resolve the width on
+ # every redraw, several times a second for an animated element.
try:
candidate = int(raw)
except (TypeError, ValueError):
- logger.warning(
- "[%s] Invalid vegas_width_pct %r, ignoring", plugin_id, raw)
+ self._warn_element_once(
+ plugin_id, "Invalid vegas_width_pct %r, ignoring", raw,
+ once_key=repr(raw))
else:
if 10 <= candidate <= 100:
pct = candidate
else:
- logger.warning(
- "[%s] vegas_width_pct %d out of range 10-100, ignoring",
- plugin_id, candidate)
+ self._warn_element_once(
+ plugin_id, "vegas_width_pct %d out of range 10-100, ignoring",
+ candidate, once_key=repr(raw))
if pct >= 100:
return self.display_width
@@ -675,8 +686,7 @@ class PluginAdapter:
if len(images) == 1:
only = images[0]
- pad = self.config.content_padding if self.config.auto_trim else 0
- if meta_of(only) is not None and only.width - 2 * pad <= budget:
+ if meta_of(only) is not None and only.width - 2 * self._padding() <= budget:
# A live element's pinned margins are not content. One whose
# drawing fits the budget is kept whole, and live, rather than
# cut for the sake of its own blank padding.
@@ -840,9 +850,14 @@ class PluginAdapter:
)
return img.crop((start, 0, end, img.height))
- def _warn_element_once(self, plugin_id: str, problem: str, *args: Any) -> None:
- """Report a plugin's element problem once per process, then quietly."""
- key = (plugin_id, problem)
+ def _warn_element_once(self, plugin_id: str, problem: str, *args: Any,
+ once_key: Optional[str] = None) -> None:
+ """Report a plugin's problem once per process, then quietly.
+
+ Once per ``problem`` (the format string), or per ``once_key`` within it
+ when given, so a different bad value is still reported.
+ """
+ key = (plugin_id, problem, once_key)
if key in self._element_warnings:
logger.debug("[%s] " + problem, plugin_id, *args)
return
@@ -885,16 +900,98 @@ class PluginAdapter:
"be used (%r); using its get_vegas_content() instead", exc)
return None
- def _images_from_elements(
- self, result: Any, plugin_id: str, epoch: int
- ) -> Optional[List[Image.Image]]:
- """Turn get_vegas_elements()'s answer into images for the pipeline.
+ def render_live_elements(
+ self, plugin: 'BasePlugin', plugin_id: str, lock_timeout: float
+ ) -> Optional[Tuple[int, Dict[str, RenderedElement]]]:
+ """Redraw a plugin's live elements for the live-element worker.
- Live elements come out pinned (RGB, display height, content_padding
- black each side, never trimmed afterwards) and tagged with their
- ElementMeta; plain ones (``live=False``) come out as ordinary content.
- Anything that is not a usable element is dropped with a warning; a
- duplicate key keeps its first element.
+ Like the keyed fetch, but for a strip that already holds the elements:
+ no cache (the caller knows the plugin's data moved on), and each live
+ element comes back as a RenderedElement to compare with what the strip
+ shows. An element whose ``version`` is the one already redrawn reuses
+ its pinned pixels and digest, so an unchanged scoreboard costs the
+ plugin's own version check and no conversion.
+
+ Returns ``(epoch, {key: element})``, the epoch read under the lock; an
+ empty dict when the plugin had nothing (or failed, logged once). None
+ only when the lock could not be had within ``lock_timeout`` -- the
+ caller tries again later.
+ """
+ with self._plugin_lock(plugin_id, timeout=lock_timeout) as acquired:
+ if not acquired:
+ return None
+ epochs = self.live_epochs
+ epoch = epochs.get(plugin_id) if epochs is not None else 0
+ render_width = self.resolve_render_width(plugin, plugin_id)
+ plugin._vegas_render_width = render_width
+ try:
+ with self._isolated_canvas(render_width):
+ result = plugin.get_vegas_elements()
+ except Exception as exc: # pylint: disable=broad-except
+ self._warn_element_once(
+ plugin_id, "get_vegas_elements() raised %r while redrawing; "
+ "its elements keep what they show", exc)
+ return epoch, {}
+ finally:
+ plugin._vegas_render_width = None
+ return epoch, self._rendered_from_elements(result, plugin_id, epoch)
+
+ @staticmethod
+ def has_lock_free_redraw(plugin: Any) -> bool:
+ """Whether the plugin's class overrides BasePlugin.redraw_vegas_element."""
+ method = getattr(type(plugin), 'redraw_vegas_element', None)
+ return method is not None and method is not _BasePlugin.redraw_vegas_element
+
+ def redraw_live_element(
+ self, plugin: 'BasePlugin', plugin_id: str, key: str, width: int,
+ height: int, at: float
+ ) -> Optional[RenderedElement]:
+ """One element redrawn for a moment in time, without the plugin's lock.
+
+ ``width`` is the element's width in the strip (pinned); the plugin is
+ asked for that less its padding, exactly, and anything else is
+ refused. None when the plugin has no lock-free redraw, returns None,
+ or fails (logged once).
+ """
+ if not self.has_lock_free_redraw(plugin):
+ return None
+ padding = self._padding()
+ inner = width - 2 * padding
+ if inner <= 0:
+ return None
+ epochs = self.live_epochs
+ epoch = epochs.get(plugin_id) if epochs is not None else 0
+ render_width = self.resolve_render_width(plugin, plugin_id)
+ plugin._vegas_render_width = render_width
+ try:
+ with self._isolated_canvas(render_width):
+ image = plugin.redraw_vegas_element(key, inner, height, at)
+ except Exception as exc: # pylint: disable=broad-except
+ self._warn_element_once(
+ plugin_id, "redraw_vegas_element(%r) raised %r", key, exc)
+ return None
+ finally:
+ plugin._vegas_render_width = None
+ if image is None:
+ return None
+ if not isinstance(image, Image.Image) or image.size != (inner, height):
+ self._warn_element_once(
+ plugin_id, "redraw_vegas_element(%r) returned %s, expected an "
+ "image of %dx%d; ignoring it", key,
+ f"{image.width}x{image.height}" if isinstance(image, Image.Image)
+ else type(image).__name__, inner, height)
+ return None
+ _pinned, pixels = pin_element(image, padding)
+ return RenderedElement(key=key, epoch=epoch, version=None, pixels=pixels,
+ digest=pixel_digest(pixels), width=pixels.shape[1])
+
+ def _valid_elements(self, result: Any, plugin_id: str) -> Optional[List[VegasElement]]:
+ """The usable elements in a get_vegas_elements() answer, in order.
+
+ None for no answer (the plugin wants its ordinary content). Anything
+ that is not a VegasElement with a key and a non-empty image is
+ dropped, and a duplicate key keeps its first element, each reported
+ once.
"""
if result is None:
return None
@@ -903,11 +1000,8 @@ class PluginAdapter:
plugin_id, "get_vegas_elements() returned %s, expected a list "
"of VegasElement", type(result).__name__)
return None
-
- padding = self.config.content_padding if self.config.auto_trim else 0
- now = time.monotonic()
seen = set()
- images: List[Image.Image] = []
+ valid: List[VegasElement] = []
for element in result:
if not (isinstance(element, VegasElement)
and isinstance(element.key, str) and element.key
@@ -917,37 +1011,124 @@ class PluginAdapter:
"not a VegasElement with a key and an image (%s); skipping it",
type(element).__name__)
continue
+ if element.image.width <= 0 or element.image.height <= 0:
+ self._warn_element_once(
+ plugin_id, "get_vegas_elements() returned an empty image for "
+ "%r; skipping it", element.key)
+ continue
if element.key in seen:
self._warn_element_once(
plugin_id, "get_vegas_elements() returned key %r twice; "
"keeping the first", element.key)
continue
seen.add(element.key)
+ valid.append(element)
+ return valid
- image = element.image
- if image.height != self.display_height:
- image = image.resize((image.width, self.display_height),
- Image.Resampling.LANCZOS)
- if image.mode != 'RGB':
- image = image.convert('RGB')
+ def _element_image(self, element: VegasElement) -> Image.Image:
+ """An element's image at the display's height, in RGB."""
+ image = element.image
+ if image.height != self.display_height:
+ image = image.resize((image.width, self.display_height),
+ Image.Resampling.LANCZOS)
+ if image.mode != 'RGB':
+ image = image.convert('RGB')
+ return image
- if not element.live:
- # Plain content; a tag copied from a reused image must not
- # make it live by accident.
- images.append(untag(image.copy()) if meta_of(image) else image)
- continue
+ def _padding(self) -> int:
+ """Black columns a live element carries each side: what trimming would leave."""
+ return self.config.content_padding if self.config.auto_trim else 0
- pinned, pixels = pin_element(image, padding)
+ @staticmethod
+ def _refresh_hz(element: VegasElement) -> float:
+ try:
+ return max(0.0, float(element.refresh_hz or 0.0))
+ except (TypeError, ValueError):
+ return 0.0
+
+ def _images_from_elements(
+ self, result: Any, plugin_id: str, epoch: int
+ ) -> Optional[List[Image.Image]]:
+ """Turn get_vegas_elements()'s answer into images for the pipeline.
+
+ Live elements come out pinned (RGB, display height, content_padding
+ black each side, never trimmed afterwards) and tagged with their
+ ElementMeta; plain ones (``live=False``) come out as ordinary content.
+ """
+ elements = self._valid_elements(result, plugin_id)
+ if elements is None:
+ return None
+ padding = self._padding()
+ now = time.monotonic()
+ images: List[Image.Image] = []
+ for element in elements:
try:
- refresh_hz = max(0.0, float(element.refresh_hz or 0.0))
- except (TypeError, ValueError):
- refresh_hz = 0.0
+ image = self._element_image(element)
+ if not element.live:
+ # Plain content; a tag copied from a reused image must not
+ # make it live by accident.
+ images.append(untag(image.copy()) if meta_of(image) else image)
+ continue
+ pinned, pixels = pin_element(image, padding)
+ except Exception as exc: # pylint: disable=broad-except
+ # An image Pillow cannot resize or convert (an odd mode, a
+ # closed file) costs that element, not its neighbours.
+ self._warn_element_once(
+ plugin_id, "element %r could not be converted (%r); skipping it",
+ element.key, exc)
+ continue
images.append(tag(pinned, ElementMeta(
plugin_id=plugin_id, key=element.key, epoch=epoch,
digest=pixel_digest(pixels), rendered_at=now,
- refresh_hz=refresh_hz, version=element.version)))
+ refresh_hz=self._refresh_hz(element), version=element.version)))
return images or None
+ def _rendered_from_elements(
+ self, result: Any, plugin_id: str, epoch: int
+ ) -> Dict[str, RenderedElement]:
+ """RenderedElements for the live elements in a get_vegas_elements() answer.
+
+ An element handed back as the very image last converted for its key,
+ with the same ``version``, reuses that conversion's pinned pixels and
+ digest, so nothing is converted or checksummed. The image must be the
+ same object: a plugin redrawn for a new config (new colours, a
+ different font) can keep its data version, and must not keep its old
+ pixels with it.
+ """
+ elements = self._valid_elements(result, plugin_id) or []
+ padding = self._padding()
+ memo = self._element_memo
+ rendered: Dict[str, RenderedElement] = {}
+ for element in elements:
+ if not element.live:
+ continue
+ memo_key = (plugin_id, element.key)
+ cached = memo.get(memo_key)
+ if cached is not None and element.version is not None \
+ and cached[0] == element.version and cached[1] is element.image \
+ and cached[2] == (padding, self.display_height):
+ pixels, digest = cached[3], cached[4]
+ else:
+ try:
+ _pinned, pixels = pin_element(self._element_image(element), padding)
+ except Exception as exc: # pylint: disable=broad-except
+ self._warn_element_once(
+ plugin_id, "element %r could not be converted (%r); it "
+ "keeps what it shows", element.key, exc)
+ continue
+ digest = pixel_digest(pixels)
+ memo[memo_key] = (element.version, element.image,
+ (padding, self.display_height), pixels, digest)
+ rendered[element.key] = RenderedElement(
+ key=element.key, epoch=epoch, version=element.version,
+ pixels=pixels, digest=digest, width=pixels.shape[1])
+ # Forget keys the plugin no longer has, so the memo stays its size.
+ stale = [k for k in list(memo)
+ if k[0] == plugin_id and k[1] not in rendered]
+ for memo_key in stale:
+ memo.pop(memo_key, None)
+ return rendered
+
def _get_native_content(
self, plugin: 'BasePlugin', plugin_id: str, restricted: bool = False
) -> Optional[List[Image.Image]]:
@@ -1510,6 +1691,9 @@ class PluginAdapter:
self._content_cache.pop(plugin_id, None)
else:
self._content_cache.clear()
+ # A config change, most often. Swapped rather than cleared:
+ # the live-element worker may be iterating the old one.
+ self._element_memo = {}
def invalidate_plugin_scroll_cache(
self, plugin: 'BasePlugin', plugin_id: str
diff --git a/src/vegas_mode/render_pipeline.py b/src/vegas_mode/render_pipeline.py
index 62372f4c..a0ac0e50 100644
--- a/src/vegas_mode/render_pipeline.py
+++ b/src/vegas_mode/render_pipeline.py
@@ -19,7 +19,8 @@ from src.common.scroll_config import solve_crisp
from src.common.scroll_helper import ScrollHelper
from src.matrix_support import DEFAULT_REFRESH_LIMIT_HZ
from src.vegas_mode.config import VegasModeConfig
-from src.vegas_mode.elements import ElementMeta, ElementRecord, meta_of
+from src.vegas_mode.elements import ElementMeta, ElementRecord, LivePatch, LiveView, meta_of
+from src.vegas_mode.live_worker import LEGACY_PREFETCH_JOIN_S, VegasWorker
from src.vegas_mode.geometry import separation_gap
from src.vegas_mode.stream_manager import StreamManager
@@ -78,6 +79,26 @@ class RenderPipeline:
# anything computed against the old strip can tell.
_strip_gen: int = 0
+ # Live updates (see apply_live_patches and src/vegas_mode/live_worker.py).
+ # Where the viewport is, published every frame for the worker.
+ _view: Optional[LiveView] = None
+ # Set by the coordinator for a run in which live elements are on.
+ _live_enabled: bool = False
+ _live_worker: Optional[VegasWorker] = None
+ # The last worker stopped: it finishes its current job, and hands over
+ # what it has of a group, before the one-shot prefetch fetches another.
+ _retired_worker: Optional[VegasWorker] = None
+ #: Patches applied between two frames at most, and the bytes they may
+ #: copy, in screens of pixels (at least one patch is always applied).
+ LIVE_PATCHES_PER_FRAME = 4
+ LIVE_PATCH_BUDGET_SCREENS = 2
+ #: A worker that dies this often in this many seconds is not restarted
+ #: again this run; live updates stop and the one-shot prefetch returns.
+ LIVE_WORKER_MAX_DEATHS = 3
+ LIVE_WORKER_DEATH_WINDOW = 600.0
+ #: Frames between checks that the worker is still alive.
+ LIVE_SUPERVISE_FRAMES = 256
+
def __init__(
self,
config: VegasModeConfig,
@@ -129,6 +150,16 @@ class RenderPipeline:
self._cycle_complete = False
self._segments_in_scroll: List[str] = [] # Plugin IDs in current scroll
self._record_by_seq: Dict[int, ElementRecord] = {}
+ # Live updates. _applied: per record, the (epoch, digest) of the
+ # pixels the strip holds. _live_slots / _live_ready: the worker's
+ # hand-over, one slot per record (latest wins) and the order they
+ # arrived in. Written by the worker, consumed by the render thread;
+ # single-key dict operations and deque append/popleft only.
+ self._applied: Dict[int, Tuple[int, Any]] = {}
+ self._live_slots: Dict[int, LivePatch] = {}
+ self._live_ready: Deque[int] = deque()
+ self._worker_deaths: Deque[float] = deque()
+ self._live_frames = 0
# The sub-pixel path's pacing; the crisp path solves its own (frame_interval).
self._frame_interval = config.get_frame_interval()
@@ -474,23 +505,43 @@ class RenderPipeline:
if not self.config.continuous_scroll:
return
+ # With live elements in the strip, the live-element worker fetches
+ # groups too, one plugin at a time between its redraws, so that only
+ # one thread ever draws for the strip.
+ worker = self._live_worker
+ if worker is not None and worker.is_alive():
+ with self._prefetch_lock:
+ if self._prepared_group is not None:
+ return
+ worker.request_group()
+ return
+
with self._prefetch_lock:
if self._prefetch_thread is not None and self._prefetch_thread.is_alive():
return
if self._prepared_group is not None:
return # already have one waiting
generation = self._prefetch_generation
+ retired = self._retired_worker
def _work():
- # Deprioritise against the render loop. Linux applies nice
- # per-thread, and the heavy lifting here is PIL and numpy work
- # that releases the GIL, so the scheduler can actually act on
- # it — without this the prefetch competes for the same cores and
- # costs frames.
+ # Deprioritise against the render loop for CPU time (Linux
+ # applies nice per thread). Nice does nothing about the GIL,
+ # which Pillow's drawing holds (docs/OFFSCREEN_RENDERING.md,
+ # risk 5); the render gate below is what keeps this thread off
+ # it while the render thread needs it.
try:
os.nice(10)
except (OSError, AttributeError):
pass
+ # A live-element worker just stopped finishes its current job
+ # and hands over what it has of a group. Wait for it, so only
+ # one thread draws and a group it handed over is not replaced.
+ if retired is not None and retired is not threading.current_thread() and retired.is_alive():
+ retired.join(LEGACY_PREFETCH_JOIN_S)
+ with self._prefetch_lock:
+ if generation != self._prefetch_generation or self._prepared_group is not None:
+ return
# With vegas_scroll.prefetch_gate on, run only while the render
# thread waits on vsync; see src/common/render_gate.py.
gate = getattr(self.display_manager, 'render_gate', None)
@@ -503,7 +554,8 @@ class RenderPipeline:
with self._prefetch_lock:
if generation != self._prefetch_generation:
return # Vegas was reset while this was fetching
- self._prepared_group = group
+ if self._prepared_group is None:
+ self._prepared_group = group
self._prefetch_thread = threading.Thread(
target=_work, daemon=True, name="vegas-strip-prefetch")
@@ -825,8 +877,12 @@ class RenderPipeline:
if new:
self._elements = self._elements + tuple(new)
by_seq = self.__dict__.setdefault('_record_by_seq', {})
+ applied = self.__dict__.setdefault('_applied', {})
for record in new:
by_seq[record.seq] = record
+ applied[record.seq] = (record.epoch, record.digest)
+ if self._live_enabled:
+ self._ensure_live_worker()
return len(new)
def _forget_trimmed_records(self, cut: int) -> None:
@@ -841,9 +897,13 @@ class RenderPipeline:
kept = tuple(r for r in records if r.abs_x + r.width > origin)
if len(kept) != len(records):
by_seq = self.__dict__.setdefault('_record_by_seq', {})
+ applied = self.__dict__.setdefault('_applied', {})
+ slots = self.__dict__.setdefault('_live_slots', {})
for record in records:
if record.abs_x + record.width <= origin:
by_seq.pop(record.seq, None)
+ applied.pop(record.seq, None)
+ slots.pop(record.seq, None)
self._elements = kept
def _reset_records(self) -> None:
@@ -852,6 +912,145 @@ class RenderPipeline:
self._strip_origin = 0
self._elements = ()
self._record_by_seq = {}
+ self._applied = {}
+ # Patches still queued belong to the old strip; apply would drop them
+ # on their generation anyway, but there is no reason to keep them.
+ self._live_slots = {}
+ self._live_ready = deque()
+ self._view = None
+
+ def has_live_records(self) -> bool:
+ """Whether the strip holds any live element."""
+ return bool(self._elements)
+
+ # -- live updates ---------------------------------------------------------
+
+ def set_live(self, enabled: bool) -> None:
+ """Switch live updates on or off for this run (the coordinator decides)."""
+ self._live_enabled = enabled
+ if enabled:
+ if self._elements:
+ self._ensure_live_worker()
+ elif self._stop_live_worker():
+ # The worker was fetching the strip's groups as well. Hand that
+ # back to the one-shot prefetch now: nothing else asks for a group
+ # until the next extension, which would find none prepared and
+ # fetch inline, stalling the scroll.
+ self.start_prefetch()
+
+ def notify_live_data(self, plugin_id: str) -> None:
+ """A plugin's data may have changed: wake the worker, if one runs."""
+ worker = self._live_worker
+ if worker is not None:
+ worker.notify_data(plugin_id)
+
+ def _ensure_live_worker(self) -> None:
+ """Start the live-element worker, or restart one that died.
+
+ Started lazily, by the first live element placed: an install with no
+ plugin that has live elements keeps the one-shot prefetch thread and
+ never runs this worker at all. A worker that keeps dying is given up
+ on for the run; live updates stop and the one-shot prefetch returns.
+ """
+ if not self._live_enabled:
+ return
+ worker = self._live_worker
+ if worker is not None and worker.is_alive():
+ return
+ if worker is None and not self._elements:
+ # Nothing live in the strip: the one-shot prefetch does the work
+ # until a live element is placed (_register_elements).
+ return
+ deaths = self.__dict__.setdefault('_worker_deaths', deque())
+ now = time.monotonic()
+ if worker is not None:
+ deaths.append(now)
+ while deaths and now - deaths[0] > self.LIVE_WORKER_DEATH_WINDOW:
+ deaths.popleft()
+ if len(deaths) >= self.LIVE_WORKER_MAX_DEATHS:
+ logger.error(
+ "Vegas live worker stopped %d times in %.0fs; live updates "
+ "are off until Vegas restarts", len(deaths),
+ self.LIVE_WORKER_DEATH_WINDOW)
+ self._live_enabled = False
+ self._live_worker = None
+ # Whatever group the dead worker was fetching is lost.
+ self.start_prefetch()
+ return
+ logger.warning("Vegas live worker was not running; restarting it")
+ worker = VegasWorker(self)
+ self._live_worker = worker
+ worker.start()
+ # Whatever the one-shot prefetch was asked for, the worker now does.
+ with self._prefetch_lock:
+ wanted = self._prepared_group is None
+ if wanted and self.config.continuous_scroll:
+ worker.request_group()
+
+ def _stop_live_worker(self) -> bool:
+ """Ask the worker to stop after its current job. Whether one was running."""
+ worker, self._live_worker = self._live_worker, None
+ if worker is None:
+ return False
+ self._retired_worker = worker
+ worker.stop()
+ return True
+
+ def apply_live_patches(self) -> int:
+ """Copy the worker's finished redraws into the strip. Render thread only.
+
+ Called between two frames (coordinator.run_frame). The only work here
+ is popping prepared patches and a numpy slice copy per patch -- no
+ drawing, no locks, no allocation -- bounded to LIVE_PATCHES_PER_FRAME
+ patches or LIVE_PATCH_BUDGET_SCREENS screens of bytes, whichever comes
+ first (always at least one). A patch is dropped when it no longer
+ fits: made for an older strip, for an element trimmed away or already
+ behind the screen, or older than what the strip already shows.
+
+ Returns:
+ Patches applied.
+ """
+ if self._live_enabled:
+ self._live_frames = self.__dict__.get('_live_frames', 0) + 1
+ if self._live_frames % self.LIVE_SUPERVISE_FRAMES == 0:
+ self._ensure_live_worker()
+ ready = self.__dict__.get('_live_ready')
+ if not ready:
+ return 0
+ slots = self._live_slots
+ if getattr(self, 'sync_manager', None) is not None:
+ # Defensive: live elements are never on under sync, and the
+ # follower would not see a patch.
+ ready.clear()
+ slots.clear()
+ return 0
+ budget = (self.LIVE_PATCH_BUDGET_SCREENS * self.display_width
+ * self.display_height * 3)
+ helper = self.scroll_helper
+ left_edge = int(helper.scroll_position)
+ applied = 0
+ moved = 0
+ while ready and applied < self.LIVE_PATCHES_PER_FRAME \
+ and (applied == 0 or moved < budget):
+ seq = ready.popleft()
+ patch = slots.pop(seq, None)
+ if patch is None:
+ continue # a newer patch for this record already went
+ record = self._record_by_seq.get(seq)
+ if record is None or patch.strip_gen != self._strip_gen:
+ continue
+ previous = self._applied.get(seq)
+ if previous is not None and patch.epoch < previous[0]:
+ continue
+ x = record.abs_x - self._strip_origin
+ if x + record.width <= left_edge:
+ continue # scrolled past; nobody will see it
+ moved += helper.patch_columns(x, patch.pixels)
+ self._applied[seq] = (patch.epoch, patch.digest)
+ applied += 1
+ if applied:
+ self._note_op('patch', moved)
+ return applied
def live_records(self) -> Tuple[ElementRecord, ...]:
"""The live elements in the strip, in the order they were placed."""
@@ -875,6 +1074,13 @@ class RenderPipeline:
# Update scroll position
self.scroll_helper.update_scroll_position()
+ # Where the viewport is now, for the live-element worker: one
+ # tuple store, read by the worker without a lock.
+ left = self._strip_origin + int(self.scroll_helper.scroll_position)
+ self._view = LiveView(
+ abs_left=left, abs_right=left + self.display_width,
+ abs_end=self._strip_origin + self.scroll_helper.total_scroll_width,
+ t_mono=time.monotonic())
# Determine if the cycle is done.
#
@@ -1187,6 +1393,7 @@ class RenderPipeline:
self._prepared_group = None
self._deferred_queue = []
self._static_markers = ()
+ self._stop_live_worker()
self._reset_records()
self.display_manager.set_scrolling_state(False)
diff --git a/src/vegas_mode/stream_manager.py b/src/vegas_mode/stream_manager.py
index bca474a1..aa7b2978 100644
--- a/src/vegas_mode/stream_manager.py
+++ b/src/vegas_mode/stream_manager.py
@@ -717,6 +717,20 @@ class StreamManager:
A STATIC plugin is also returned with an empty list, unfetched: it
pauses the scroll instead of adding to it (see is_static_plugin).
"""
+ group: List[Tuple[str, Optional[List[Image.Image]]]] = []
+ for plugin_id in self.plan_next_group(count):
+ member = self.fetch_group_member(plugin_id, offscreen_only=offscreen_only)
+ if member is not None:
+ group.append(member)
+ return group
+
+ def plan_next_group(self, count: Optional[int] = None) -> List[str]:
+ """Which plugins the next group holds, advancing the rotation past them.
+
+ The first half of take_next_group(). The live-element worker fetches
+ a group one plugin at a time (fetch_group_member) so it can fit more
+ urgent redraws between them.
+ """
if count is None:
count = self.config.plugins_per_cycle
@@ -730,37 +744,38 @@ class StreamManager:
for _ in range(min(max(1, count), total)):
ids.append(self._ordered_plugins[self._prefetch_index])
self._prefetch_index = (self._prefetch_index + 1) % total
+ return ids
- plugins = getattr(self.plugin_manager, 'plugins', {})
- group: List[Tuple[str, Optional[List[Image.Image]]]] = []
+ def fetch_group_member(
+ self, plugin_id: str, offscreen_only: bool = False
+ ) -> Optional[Tuple[str, Optional[List[Image.Image]]]]:
+ """One plugin's entry in a group, as take_next_group() describes it.
+
+ None when the plugin is gone or its fetch raised: it is left out of
+ the group.
+ """
+ plugin = getattr(self.plugin_manager, 'plugins', {}).get(plugin_id)
+ if not plugin:
+ return None
+ if self.is_static_plugin(plugin_id):
+ # A STATIC plugin pauses the scroll rather than scrolling by, so it
+ # contributes no columns. It keeps its place in the group (empty)
+ # so the pipeline can mark where its turn falls.
+ return (plugin_id, [])
+ try:
+ images = self.plugin_adapter.get_content(
+ plugin, plugin_id, offscreen_only=offscreen_only)
+ except Exception:
+ logger.exception("[%s] ERROR fetching content", plugin_id)
+ self.stats['fetch_errors'] += 1
+ return None
+ if images:
+ self.stats['segments_fetched'] += 1
+ return (plugin_id, images)
# Only the old contract hands anything back to the render thread.
defer_empty = offscreen_only and not getattr(
self.config, 'offscreen_prefetch', True)
-
- for plugin_id in ids:
- plugin = plugins.get(plugin_id)
- if not plugin:
- continue
- if self.is_static_plugin(plugin_id):
- # A STATIC plugin pauses the scroll rather than scrolling by,
- # so it contributes no columns. It keeps its place in the
- # group (empty) so the pipeline can mark where its turn falls.
- group.append((plugin_id, []))
- continue
- try:
- images = self.plugin_adapter.get_content(
- plugin, plugin_id, offscreen_only=offscreen_only)
- except Exception:
- logger.exception("[%s] ERROR fetching content", plugin_id)
- self.stats['fetch_errors'] += 1
- continue
- if images:
- self.stats['segments_fetched'] += 1
- group.append((plugin_id, images))
- else:
- group.append((plugin_id, None if defer_empty else []))
-
- return group
+ return (plugin_id, None if defer_empty else [])
def advance_cycle(self) -> None:
"""
diff --git a/test/test_api_v3_partial_main_save.py b/test/test_api_v3_partial_main_save.py
index 85914f3f..c5feb0df 100644
--- a/test/test_api_v3_partial_main_save.py
+++ b/test/test_api_v3_partial_main_save.py
@@ -119,6 +119,21 @@ class TestJsonPartialSaves:
'continuous_scroll', 'smooth_scroll'):
assert vegas[key] is True, key
+ def test_vegas_speed_only_keeps_live_updates_as_stored(self, api_v3_client, saved):
+ resp = _post_json(api_v3_client, {'vegas_scroll_speed': 70})
+ assert resp.status_code == 200, resp.get_json()
+ assert 'live_refresh' not in saved['config']['display']['vegas_scroll']
+
+ def test_live_updates_can_be_switched_off_and_on(self, api_v3_client, saved):
+ resp = _post_json(api_v3_client, {'vegas_live_refresh': False})
+ assert resp.status_code == 200, resp.get_json()
+ assert saved['config']['display']['vegas_scroll']['live_refresh'] is False
+ resp = _post_json(api_v3_client, {'__form_section': 'display',
+ 'vegas_scroll_speed': '50',
+ 'vegas_live_refresh': 'on'})
+ assert resp.status_code == 200, resp.get_json()
+ assert saved['config']['display']['vegas_scroll']['live_refresh'] is True
+
def test_double_sided_axis_only_keeps_enabled(self, api_v3_client, saved):
resp = _post_json(api_v3_client, {'double_sided_axis': 'horizontal'})
assert resp.status_code == 200, resp.get_json()
diff --git a/test/test_scroll_helper_patch.py b/test/test_scroll_helper_patch.py
new file mode 100644
index 00000000..5b23e64a
--- /dev/null
+++ b/test/test_scroll_helper_patch.py
@@ -0,0 +1,88 @@
+"""ScrollHelper.patch_columns: changing a strip's pixels in place, between frames.
+
+The primitive a live Vegas element update is built on. It writes exactly the
+columns it is given and nothing else, never moves the scroll or resizes the
+strip, and refuses an array it may not write (the multi-display follower's).
+"""
+import sys
+from pathlib import Path
+
+import numpy as np
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.common.scroll_helper import ScrollHelper # noqa: E402
+
+W, H = 64, 16
+
+
+def _helper(width=300):
+ rng = np.random.default_rng(1)
+ pixels = rng.integers(0, 255, (H, width, 3), dtype=np.uint8)
+ helper = ScrollHelper(W, H)
+ helper.set_scrolling_image(Image.frombytes("RGB", (width, H), pixels.tobytes()))
+ helper.scroll_position = 42.0
+ return helper
+
+
+def _patch(width, value=7):
+ return np.full((H, width, 3), value, dtype=np.uint8)
+
+
+def test_writes_exactly_its_columns():
+ helper = _helper()
+ before = helper.cached_array.copy()
+ assert helper.patch_columns(100, _patch(30)) == H * 30 * 3
+ after = helper.cached_array
+ assert (after[:, 100:130] == 7).all()
+ assert np.array_equal(after[:, :100], before[:, :100])
+ assert np.array_equal(after[:, 130:], before[:, 130:])
+
+
+def test_moves_nothing_else():
+ helper = _helper()
+ width, position, distance = (helper.total_scroll_width, helper.scroll_position,
+ helper.total_distance_scrolled)
+ helper.patch_columns(10, _patch(20))
+ assert (helper.total_scroll_width, helper.scroll_position,
+ helper.total_distance_scrolled) == (width, position, distance)
+
+
+def test_clips_at_both_ends():
+ helper = _helper(width=300)
+ assert helper.patch_columns(-10, _patch(30, 1)) == H * 20 * 3
+ assert (helper.cached_array[:, :20] == 1).all()
+ assert helper.patch_columns(290, _patch(30, 2)) == H * 10 * 3
+ assert (helper.cached_array[:, 290:] == 2).all()
+ assert helper.patch_columns(400, _patch(10)) == 0
+ assert helper.patch_columns(-50, _patch(10)) == 0
+
+
+def test_the_patch_lands_in_the_next_frame():
+ helper = _helper()
+ helper.scroll_position = 100.0
+ helper.patch_columns(110, _patch(10, 200))
+ frame = np.asarray(helper.get_visible_portion())
+ assert (frame[:, 10:20] == 200).all()
+
+
+def test_refuses_a_read_only_strip():
+ # The follower adopts np.asarray(image), which is read-only.
+ helper = _helper()
+ helper.cached_array = np.asarray(helper.cached_image)
+ assert not helper.cached_array.flags.writeable
+ assert helper.patch_columns(10, _patch(10)) == 0
+
+
+def test_refuses_a_patch_of_the_wrong_height_and_no_strip():
+ helper = _helper()
+ assert helper.patch_columns(10, np.zeros((H + 1, 5, 3), dtype=np.uint8)) == 0
+ assert ScrollHelper(W, H).patch_columns(0, _patch(5)) == 0
+
+
+def test_a_deferred_image_is_built_from_the_patched_strip():
+ helper = _helper()
+ helper.append_content([Image.new("RGB", (40, H))], item_gap=0)
+ helper.patch_columns(5, _patch(5, 99))
+ assert (np.asarray(helper.cached_image)[:, 5:10] == 99).all()
diff --git a/test/test_vegas_live_adapter.py b/test/test_vegas_live_adapter.py
new file mode 100644
index 00000000..115ec3d2
--- /dev/null
+++ b/test/test_vegas_live_adapter.py
@@ -0,0 +1,120 @@
+"""The adapter's side of live redraws: render_live_elements() and its memo.
+
+The worker asks the adapter to redraw a plugin's elements whenever the plugin's
+data moves on. What comes back is compared with what the strip shows, so the
+adapter may skip converting an element it has seen before -- but only when it
+is provably the same picture, and a bad element must cost only itself.
+"""
+import sys
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from PIL import Image, ImageDraw
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.plugin_system.vegas_elements import VegasElement # noqa: E402
+from src.vegas_mode.config import VegasModeConfig # noqa: E402
+from src.vegas_mode.elements import LiveEpochs # noqa: E402
+from src.vegas_mode.plugin_adapter import PluginAdapter # noqa: E402
+
+from test.test_vegas_elements_api import _DM, _Plugin, H # noqa: E402
+
+
+def _card(width=40, colour=(255, 0, 0)):
+ image = Image.new("RGB", (width, H))
+ ImageDraw.Draw(image).rectangle([0, 0, width - 1, H - 1], outline=colour)
+ return image
+
+
+def _adapter():
+ lock = threading.Lock()
+ pm = SimpleNamespace(get_plugin_lock=lambda pid: lock)
+ adapter = PluginAdapter(_DM(), VegasModeConfig(), plugin_manager=pm)
+ adapter.live_elements_enabled = True
+ adapter.live_epochs = LiveEpochs()
+ return adapter
+
+
+def _render(adapter, plugin):
+ epoch, rendered = adapter.render_live_elements(plugin, "p", lock_timeout=1.0)
+ return rendered
+
+
+def test_the_same_image_at_the_same_version_is_not_converted_again():
+ card = _card()
+ plugin = _Plugin(lambda: [VegasElement("c", card, version=1)])
+ adapter = _adapter()
+ first = _render(adapter, plugin)["c"]
+ again = _render(adapter, plugin)["c"]
+ assert again.pixels is first.pixels
+
+
+def test_a_new_picture_at_the_same_version_is_converted():
+ # A config change (new colours) redraws the card without moving its data
+ # version; the strip must still get the new pixels.
+ colours = iter([(255, 0, 0), (0, 255, 0)])
+ plugin = _Plugin(lambda: [VegasElement("c", _card(colour=next(colours)), version=1)])
+ adapter = _adapter()
+ first = _render(adapter, plugin)["c"]
+ again = _render(adapter, plugin)["c"]
+ assert again.digest != first.digest
+
+
+def test_the_same_image_at_a_new_version_is_converted():
+ card = _card()
+ versions = iter([1, 2])
+ plugin = _Plugin(lambda: [VegasElement("c", card, version=next(versions))])
+ adapter = _adapter()
+ first = _render(adapter, plugin)["c"]
+ again = _render(adapter, plugin)["c"]
+ assert again.pixels is not first.pixels
+
+
+def test_invalidating_every_plugin_forgets_what_was_converted():
+ card = _card()
+ plugin = _Plugin(lambda: [VegasElement("c", card, version=1)])
+ adapter = _adapter()
+ first = _render(adapter, plugin)["c"]
+ adapter.invalidate_cache()
+ assert _render(adapter, plugin)["c"].pixels is not first.pixels
+
+
+def test_an_element_that_cannot_be_converted_costs_only_itself(monkeypatch):
+ plugin = _Plugin(lambda: [VegasElement("bad", _card()), VegasElement("good", _card())])
+ adapter = _adapter()
+ real = adapter._element_image
+
+ def element_image(element):
+ if element.key == "bad":
+ raise OSError("closed file")
+ return real(element)
+
+ monkeypatch.setattr(adapter, "_element_image", element_image)
+ assert set(_render(adapter, plugin)) == {"good"}
+ placed = adapter.get_content(plugin, "p", offscreen_only=True)
+ from src.vegas_mode.elements import meta_of
+ assert [meta_of(img).key for img in placed] == ["good"]
+
+
+@pytest.mark.parametrize("size", [(0, H), (40, 0)])
+def test_an_empty_image_is_dropped(size):
+ plugin = _Plugin(lambda: [VegasElement("empty", Image.new("RGB", size)),
+ VegasElement("good", _card())])
+ adapter = _adapter()
+ assert set(_render(adapter, plugin)) == {"good"}
+
+
+def test_a_bad_vegas_width_pct_is_reported_once(caplog):
+ plugin = _Plugin(lambda: [VegasElement("c", _card())], config={"vegas_width_pct": "wide"})
+ adapter = _adapter()
+ with caplog.at_level("WARNING"):
+ for _ in range(5):
+ adapter.resolve_render_width(plugin, "p")
+ assert sum("vegas_width_pct" in r.getMessage() for r in caplog.records) == 1
+ plugin.config["vegas_width_pct"] = 500
+ with caplog.at_level("WARNING"):
+ adapter.resolve_render_width(plugin, "p")
+ assert sum("out of range" in r.getMessage() for r in caplog.records) == 1
diff --git a/test/test_vegas_live_apply.py b/test/test_vegas_live_apply.py
new file mode 100644
index 00000000..1be111d6
--- /dev/null
+++ b/test/test_vegas_live_apply.py
@@ -0,0 +1,211 @@
+"""The render thread's half of a live update: RenderPipeline.apply_live_patches.
+
+Between two frames the render thread copies prepared pixels into the strip.
+It must do nothing else there -- no drawing, no locks, a bounded number of
+bytes -- and must refuse a patch that no longer fits: a strip rebuilt since,
+an element trimmed away or already behind the screen, a patch older than what
+the strip already shows.
+"""
+import sys
+import threading
+import time
+from pathlib import Path
+
+import numpy as np
+import pytest
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode import elements # noqa: E402
+from src.vegas_mode.config import VegasModeConfig # noqa: E402
+from src.vegas_mode.elements import ElementMeta, LivePatch # noqa: E402
+from src.vegas_mode.render_pipeline import RenderPipeline # noqa: E402
+
+W, H = 128, 32
+
+
+def _live(key, width, seed):
+ rng = np.random.default_rng(seed)
+ pixels = rng.integers(20, 255, (H, width, 3), dtype=np.uint8)
+ image = Image.frombytes("RGB", (width, H), pixels.tobytes())
+ pinned, array = elements.pin_element(image, 8)
+ return elements.tag(pinned, ElementMeta("p", key, 1, elements.pixel_digest(array), 0.0, 0.0))
+
+
+class _Stream:
+ def __init__(self, groups):
+ self.groups = groups
+ self.plugin_manager = type("PM", (), {"plugins": {}})()
+ self.plugin_adapter = None
+ self.i = 0
+
+ def get_grouped_content_for_composition(self):
+ return self.groups[0]
+
+ def get_active_plugin_ids(self):
+ return ["p"]
+
+ def take_next_group(self, count=None, offscreen_only=False):
+ self.i += 1
+ return self.groups[self.i] if self.i < len(self.groups) else []
+
+
+class _Timing:
+ def __init__(self):
+ self.notes = []
+
+ def note_op(self, kind, nbytes=0):
+ self.notes.append((kind, nbytes))
+
+
+class _DM:
+ width, height = W, H
+
+ def __init__(self):
+ self.image = Image.new("RGB", (W, H))
+ self.frame_timing = _Timing()
+
+ def set_scrolling_state(self, *a):
+ pass
+
+ def update_display(self):
+ pass
+
+
+def _pipeline(n=6):
+ groups = [[("p", [_live(f"k{i}", 40, i) for i in range(n)])]]
+ p = RenderPipeline(VegasModeConfig(continuous_scroll=True, lead_in_width=0),
+ _DM(), _Stream(groups))
+ assert p.compose_scroll_content()
+ p.display_manager.frame_timing.notes.clear() # the compose's own note
+ return p
+
+
+def _patch_for(p, record, value, epoch=None, gen=None):
+ pixels = np.full((H, record.width, 3), value, dtype=np.uint8)
+ pixels.setflags(write=False)
+ return LivePatch(seq=record.seq, strip_gen=p._strip_gen if gen is None else gen,
+ epoch=record.epoch if epoch is None else epoch, pixels=pixels,
+ digest=elements.pixel_digest(pixels), made_at=0.0)
+
+
+def _offer(p, patch):
+ p._live_slots[patch.seq] = patch
+ p._live_ready.append(patch.seq)
+
+
+def _columns(p, record):
+ x = record.abs_x - p._strip_origin
+ return p.scroll_helper.cached_array[:, x:x + record.width]
+
+
+def test_a_patch_is_copied_into_its_columns_and_noted():
+ p = _pipeline()
+ record = p.live_records()[2]
+ _offer(p, _patch_for(p, record, 123))
+ assert p.apply_live_patches() == 1
+ assert (_columns(p, record) == 123).all()
+ assert p._applied[record.seq][1] == elements.pixel_digest(_columns(p, record))
+ assert p.display_manager.frame_timing.notes == [("patch", record.width * H * 3)]
+
+
+def test_nothing_to_apply_costs_nothing():
+ p = _pipeline()
+ assert p.apply_live_patches() == 0
+ assert p.display_manager.frame_timing.notes == []
+
+
+def test_at_most_four_patches_a_frame_the_rest_next_frame():
+ p = _pipeline()
+ for record in p.live_records():
+ _offer(p, _patch_for(p, record, 50))
+ assert p.apply_live_patches() == 4
+ assert p.apply_live_patches() == 2
+
+
+def test_the_byte_budget_still_applies_one():
+ p = _pipeline(n=1)
+ record = p.live_records()[0]
+ big = LivePatch(seq=record.seq, strip_gen=p._strip_gen, epoch=1,
+ pixels=np.full((H, record.width, 3), 9, dtype=np.uint8),
+ digest=((H, record.width, 3), 1), made_at=0.0)
+ p.LIVE_PATCH_BUDGET_SCREENS = 0
+ _offer(p, big)
+ assert p.apply_live_patches() == 1
+
+
+@pytest.mark.parametrize("why", ["old strip", "unknown element", "older data",
+ "behind the screen"])
+def test_a_patch_that_no_longer_fits_is_dropped(why):
+ p = _pipeline()
+ record = p.live_records()[0]
+ before = _columns(p, record).copy()
+ patch = _patch_for(p, record, 77)
+ if why == "old strip":
+ patch = _patch_for(p, record, 77, gen=p._strip_gen - 1)
+ elif why == "unknown element":
+ patch = patch._replace(seq=999)
+ elif why == "older data":
+ p._applied[record.seq] = (5, p._applied[record.seq][1])
+ patch = _patch_for(p, record, 77, epoch=4)
+ else:
+ p.scroll_helper.scroll_position = record.abs_x + record.width + 1
+ _offer(p, patch)
+ assert p.apply_live_patches() == 0
+ assert np.array_equal(_columns(p, record), before)
+
+
+def test_the_latest_patch_wins_and_its_duplicate_entry_is_harmless():
+ p = _pipeline()
+ record = p.live_records()[1]
+ _offer(p, _patch_for(p, record, 10))
+ _offer(p, _patch_for(p, record, 20)) # replaces the slot, queues seq again
+ assert p.apply_live_patches() == 1
+ assert (_columns(p, record) == 20).all()
+ assert not p._live_ready
+
+
+def test_under_sync_nothing_is_applied_and_the_queue_is_emptied():
+ p = _pipeline()
+ record = p.live_records()[0]
+ before = _columns(p, record).copy()
+ p.sync_manager = object()
+ _offer(p, _patch_for(p, record, 5))
+ assert p.apply_live_patches() == 0
+ assert not p._live_ready and not p._live_slots
+ assert np.array_equal(_columns(p, record), before)
+
+
+def test_the_render_thread_takes_no_lock_and_draws_nothing(monkeypatch):
+ # Another thread holds every lock a live update could involve; the render
+ # thread's apply must not care, and must not draw or rebuild the strip.
+ p = _pipeline()
+ locks = [p._prefetch_lock, threading.Lock()]
+ for lock in locks:
+ lock.acquire()
+ for name in ("new", "fromarray"):
+ monkeypatch.setattr(Image, name, lambda *a, **k: (_ for _ in ()).throw(
+ AssertionError("drew on the render thread")))
+ monkeypatch.setattr(np, "concatenate", lambda *a, **k: (_ for _ in ()).throw(
+ AssertionError("rebuilt the strip")))
+ try:
+ for record in p.live_records():
+ _offer(p, _patch_for(p, record, 3))
+ started = time.perf_counter()
+ applied = p.apply_live_patches() + p.apply_live_patches()
+ assert time.perf_counter() - started < 0.05
+ assert applied == 6
+ finally:
+ for lock in locks:
+ lock.release()
+
+
+def test_the_view_is_published_each_frame():
+ p = _pipeline()
+ p.scroll_helper.set_pixels_per_frame(1)
+ assert p.render_frame()
+ view = p._view
+ assert view.abs_right - view.abs_left == W
+ assert view.abs_left == p._strip_origin + int(p.scroll_helper.scroll_position)
+ assert view.abs_end == p._strip_origin + p.scroll_helper.total_scroll_width
diff --git a/test/test_vegas_live_integration.py b/test/test_vegas_live_integration.py
index 5784ec2d..467f60bc 100644
--- a/test/test_vegas_live_integration.py
+++ b/test/test_vegas_live_integration.py
@@ -122,6 +122,64 @@ def test_an_update_moves_the_plugins_epoch_and_new_elements_carry_it(ticker):
assert metas and all(m.epoch == epoch for m in metas)
+def _columns(pipeline, record):
+ x = record.abs_x - pipeline._strip_origin
+ return pipeline.scroll_helper.cached_array[:, max(0, x):x + record.width].copy()
+
+
+def _upcoming(pipeline, key_prefix):
+ """A live record of this kind that is not yet behind the viewport."""
+ left = pipeline._strip_origin + int(pipeline.scroll_helper.scroll_position)
+ for record in pipeline.live_records():
+ if record.key.startswith(key_prefix) and record.abs_x + record.width > left + 40:
+ return record
+ return None
+
+
+def test_an_update_changes_cards_already_in_the_strip(ticker):
+ coordinator, plugin, pm = ticker
+ assert coordinator.start()
+ pipeline = coordinator.render_pipeline
+ assert _run_until(coordinator, lambda: _upcoming(pipeline, "card:"))
+ record = _upcoming(pipeline, "card:")
+ before = _columns(pipeline, record)
+
+ plugin.update() # new data: every card's bars change
+ pm._note_update_completed(PID) # what the update worker does after it
+ assert _run_until(
+ coordinator,
+ lambda: not np.array_equal(_columns(pipeline, record), before), seconds=5.0)
+ # The card changed in place: same columns, same width, new pixels.
+ assert pipeline._record_by_seq[record.seq] == record
+ assert pipeline._live_worker is not None and pipeline._live_worker.is_alive()
+
+
+def test_an_animated_element_moves_with_no_update_at_all(ticker):
+ coordinator, _plugin, _pm = ticker
+ assert coordinator.start()
+ pipeline = coordinator.render_pipeline
+ assert _run_until(coordinator, lambda: _upcoming(pipeline, "map"))
+ record = _upcoming(pipeline, "map")
+ before = _columns(pipeline, record)
+ assert _run_until(
+ coordinator,
+ lambda: (pipeline._record_by_seq.get(record.seq) is not None
+ and not np.array_equal(_columns(pipeline, record), before)),
+ seconds=5.0)
+
+
+def test_stopping_vegas_stops_the_worker(ticker):
+ coordinator, _plugin, _pm = ticker
+ assert coordinator.start()
+ pipeline = coordinator.render_pipeline
+ assert _run_until(coordinator, lambda: pipeline._live_worker is not None)
+ worker = pipeline._live_worker
+ coordinator.stop()
+ worker.join(3)
+ assert not worker.is_alive()
+ assert pipeline._live_worker is None
+
+
def test_with_live_refresh_off_the_same_run_is_plain_content(dm, tmp_path, ticker):
coordinator, _plugin, _pm = ticker
coordinator.vegas_config.live_refresh = False
diff --git a/test/test_vegas_live_lifecycle.py b/test/test_vegas_live_lifecycle.py
new file mode 100644
index 00000000..8729c00b
--- /dev/null
+++ b/test/test_vegas_live_lifecycle.py
@@ -0,0 +1,179 @@
+"""When the live-element worker runs, and what happens to its work when it stops.
+
+The worker fetches the strip's groups as well as redrawing live elements, so
+starting it where nothing is live costs a thread for nothing, and stopping it
+must not lose the group it was fetching: nothing else asks for one until the
+next extension, which would then fetch inline and stall the scroll.
+"""
+import sys
+import threading
+import time
+from pathlib import Path
+
+import numpy as np
+import pytest
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode import elements, render_pipeline # noqa: E402
+from src.vegas_mode.config import VegasModeConfig # noqa: E402
+from src.vegas_mode.elements import ElementMeta # noqa: E402
+from src.vegas_mode.render_pipeline import RenderPipeline # noqa: E402
+
+from test.test_vegas_live_apply import _DM # noqa: E402
+
+H = 32
+
+
+def _image(width, seed, key=None):
+ rng = np.random.default_rng(seed)
+ pixels = rng.integers(20, 255, (H, width, 3), dtype=np.uint8)
+ image = Image.frombytes("RGB", (width, H), pixels.tobytes())
+ if key is None:
+ return image
+ pinned, array = elements.pin_element(image, 8)
+ return elements.tag(pinned, ElementMeta("p", key, 1, elements.pixel_digest(array), 0.0, 0.0))
+
+
+class _Stream:
+ def __init__(self, live):
+ self.first = [("p", [_image(40, i, f"k{i}" if live else None) for i in range(6)])]
+ self.plugin_manager = type("PM", (), {"plugins": {}})()
+ self.plugin_adapter = None
+ self.taken = 0
+
+ def get_grouped_content_for_composition(self):
+ return self.first
+
+ def get_active_plugin_ids(self):
+ return ["p"]
+
+ def take_next_group(self, count=None, offscreen_only=False):
+ self.taken += 1
+ return [("q", [_image(40, 99)])]
+
+
+class _FakeWorker:
+ """Stands in for VegasWorker: no thread, and says what it was asked."""
+
+ instances = []
+ alive = True
+
+ def __init__(self, pipeline):
+ self.pipeline = pipeline
+ self.requests = 0
+ self.stopped = False
+ self.on_join = None
+ _FakeWorker.instances.append(self)
+
+ def start(self):
+ pass
+
+ def is_alive(self):
+ return _FakeWorker.alive and not self.stopped
+
+ def request_group(self):
+ self.requests += 1
+
+ def notify_data(self, plugin_id):
+ pass
+
+ def stop(self):
+ # Still finishing its job: alive until joined.
+ self.stopping = True
+
+ def join(self, timeout=None):
+ if self.on_join is not None:
+ self.on_join()
+ self.stopped = True
+
+
+@pytest.fixture(autouse=True)
+def _fake_worker(monkeypatch):
+ _FakeWorker.instances = []
+ _FakeWorker.alive = True
+ monkeypatch.setattr(render_pipeline, "VegasWorker", _FakeWorker)
+
+
+def _pipeline(live=True):
+ p = RenderPipeline(VegasModeConfig(continuous_scroll=True, lead_in_width=0),
+ _DM(), _Stream(live))
+ assert p.compose_scroll_content()
+ return p
+
+
+def _wait_for(predicate, timeout=2.0):
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ if predicate():
+ return True
+ time.sleep(0.005)
+ return False
+
+
+def test_supervision_starts_no_worker_while_nothing_is_live():
+ p = _pipeline(live=False)
+ p.set_live(True)
+ for _ in range(p.LIVE_SUPERVISE_FRAMES * 2 + 1):
+ p.apply_live_patches()
+ assert p._live_worker is None and not _FakeWorker.instances
+
+
+def test_live_elements_start_the_worker_and_it_is_asked_for_a_group():
+ p = _pipeline()
+ p.set_live(True)
+ assert isinstance(p._live_worker, _FakeWorker)
+ assert p._live_worker.requests == 1
+
+
+def test_switching_live_off_hands_group_fetching_back():
+ p = _pipeline()
+ p.set_live(True)
+ worker = p._live_worker
+ p.set_live(False)
+ assert p._live_worker is None and p._retired_worker is worker
+ assert _wait_for(lambda: p._prepared_group is not None)
+ assert worker.stopped # waited for before fetching
+ assert p._prefetch_thread.name == "vegas-strip-prefetch"
+ assert p.stream_manager.taken == 1
+
+
+def test_a_group_the_stopping_worker_hands_over_is_kept():
+ p = _pipeline()
+ p.set_live(True)
+ worker = p._live_worker
+ handed = [("p", ["from the worker"])]
+
+ def publish():
+ with p._prefetch_lock:
+ p._prepared_group = handed
+
+ worker.on_join = publish
+ p.set_live(False)
+ p._prefetch_thread.join(2)
+ assert p._prepared_group is handed
+ assert p.stream_manager.taken == 0 # nothing fetched over it
+
+
+def test_a_reset_while_stopping_fetches_nothing():
+ p = _pipeline()
+ p.set_live(True)
+ release = threading.Event()
+ p._live_worker.on_join = lambda: release.wait(2)
+ p.set_live(False)
+ p.reset()
+ release.set()
+ p._prefetch_thread.join(2)
+ assert p._prepared_group is None and p.stream_manager.taken == 0
+
+
+def test_a_worker_given_up_on_hands_group_fetching_back():
+ p = _pipeline()
+ p.set_live(True)
+ _FakeWorker.alive = False # it, and every restart, dies
+ for _ in range(p.LIVE_WORKER_MAX_DEATHS):
+ p._ensure_live_worker()
+ assert not p._live_enabled and p._live_worker is None
+ assert _wait_for(lambda: p._prepared_group is not None)
+ assert p.stream_manager.taken == 1
diff --git a/test/test_vegas_live_property.py b/test/test_vegas_live_property.py
new file mode 100644
index 00000000..86e90e24
--- /dev/null
+++ b/test/test_vegas_live_property.py
@@ -0,0 +1,146 @@
+"""Live patches change live elements' columns and nothing else, whatever happens.
+
+Two pipelines are built from the same content and put through the same random
+sequence of scrolling, extending and trimming. One of them also has random
+patches applied. After every step, the only columns allowed to differ between
+the two are those of live elements, and each of those must show exactly the
+last patch applied to it (or its original pixels).
+"""
+import collections
+import random
+import sys
+from pathlib import Path
+
+import numpy as np
+import pytest
+from PIL import Image, ImageDraw
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode import elements # noqa: E402
+from src.vegas_mode.config import VegasModeConfig # noqa: E402
+from src.vegas_mode.elements import ElementMeta, LivePatch # noqa: E402
+from src.vegas_mode.render_pipeline import RenderPipeline # noqa: E402
+
+W, H = 96, 24
+
+
+def _live(key, width, seed):
+ rng = np.random.default_rng(seed)
+ pixels = rng.integers(20, 255, (H, width, 3), dtype=np.uint8)
+ image = Image.frombytes("RGB", (width, H), pixels.tobytes())
+ pinned, array = elements.pin_element(image, 8)
+ return elements.tag(pinned, ElementMeta("p", key, 0, elements.pixel_digest(array), 0.0, 0.0))
+
+
+def _plain(width, seed):
+ image = Image.new("RGB", (width, H))
+ ImageDraw.Draw(image).rectangle([0, 0, width - 1, H - 1],
+ outline=(seed * 37 % 255, 90, 200))
+ return image
+
+
+def _groups(rng, n):
+ groups = []
+ for g in range(n):
+ group = []
+ for p in range(rng.randint(1, 3)):
+ rows = []
+ for r in range(rng.randint(1, 3)):
+ key = f"g{g}p{p}r{r}"
+ if rng.random() < 0.6:
+ rows.append(_live(key, rng.randint(10, 50), rng.randint(0, 10**6)))
+ else:
+ rows.append(_plain(rng.randint(10, 50), rng.randint(0, 255)))
+ group.append((f"pl{g}{p}", rows))
+ groups.append(group)
+ return groups
+
+
+class _Stream:
+ def __init__(self, groups):
+ self.groups = groups
+ self.plugin_manager = type("PM", (), {"plugins": {}})()
+ self.plugin_adapter = None
+ self.i = 0
+
+ def get_grouped_content_for_composition(self):
+ return self.groups[0]
+
+ def get_active_plugin_ids(self):
+ return []
+
+ def take_next_group(self, count=None, offscreen_only=False):
+ self.i += 1
+ return self.groups[self.i] if self.i < len(self.groups) else []
+
+
+class _DM:
+ width, height = W, H
+
+ def __init__(self):
+ self.image = Image.new("RGB", (W, H))
+
+ def set_scrolling_state(self, *a):
+ pass
+
+ def update_display(self):
+ pass
+
+
+def _pipeline(groups):
+ p = RenderPipeline(VegasModeConfig(continuous_scroll=True, lead_in_width=0,
+ separator_width=12), _DM(), _Stream(groups))
+ assert p.compose_scroll_content()
+ return p
+
+
+@pytest.mark.parametrize("seed", range(12))
+def test_patches_touch_only_live_columns(seed):
+ rng = random.Random(seed)
+ groups = _groups(rng, 25)
+ patched, reference = _pipeline(groups), _pipeline(groups)
+ expected = {} # seq -> pixels the patched strip must show
+ counter = collections.Counter()
+
+ for _step in range(60):
+ action = rng.random()
+ if action < 0.35:
+ advance = rng.randint(5, 60)
+ for p in (patched, reference):
+ p.scroll_helper.scroll_position += advance
+ elif action < 0.6:
+ for p in (patched, reference):
+ if p.scroll_helper.remaining_unscrolled() < 4 * W:
+ p.extend_scroll_content()
+ counter["extend"] += 1
+ else:
+ records = patched.live_records()
+ if not records:
+ continue
+ record = rng.choice(records)
+ pixels = np.full((H, record.width, 3), rng.randint(0, 255), dtype=np.uint8)
+ pixels.setflags(write=False)
+ patched._live_slots[record.seq] = LivePatch(
+ seq=record.seq, strip_gen=patched._strip_gen, epoch=0, pixels=pixels,
+ digest=elements.pixel_digest(pixels), made_at=0.0)
+ patched._live_ready.append(record.seq)
+ if patched.apply_live_patches():
+ expected[record.seq] = pixels
+ counter["patch"] += 1
+
+ # The twins agree on everything but live columns.
+ a = patched.scroll_helper.cached_array
+ b = reference.scroll_helper.cached_array
+ assert a.shape == b.shape
+ assert patched._strip_origin == reference._strip_origin
+ live = np.zeros(a.shape[1], dtype=bool)
+ for record in patched.live_records():
+ x = record.abs_x - patched._strip_origin
+ lo, hi = max(0, x), max(0, x + record.width)
+ live[lo:hi] = True
+ if record.seq in expected:
+ assert np.array_equal(a[:, lo:hi], expected[record.seq][:, lo - x:hi - x])
+ assert np.array_equal(a[:, ~live], b[:, ~live])
+
+ assert counter["patch"] > 0
diff --git a/test/test_vegas_live_worker.py b/test/test_vegas_live_worker.py
new file mode 100644
index 00000000..0abb6fd8
--- /dev/null
+++ b/test/test_vegas_live_worker.py
@@ -0,0 +1,491 @@
+"""The live-element worker's choices and hand-overs (src/vegas_mode/live_worker.py).
+
+The worker is driven here one decision at a time -- _pick() then _run() --
+against a fake pipeline, so every rule can be pinned without threads or
+timing: what runs first, what is skipped, what reaches the render thread.
+"""
+import collections
+import sys
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode import live_worker # noqa: E402
+from src.vegas_mode.config import VegasModeConfig # noqa: E402
+from src.vegas_mode.elements import ( # noqa: E402
+ ElementRecord, LiveEpochs, LiveView, RenderedElement, pixel_digest,
+)
+from src.vegas_mode.live_worker import VegasWorker # noqa: E402
+
+W, H = 128, 32
+NOW = 1000.0
+
+
+def _pixels(width, value):
+ array = np.full((H, width, 3), value, dtype=np.uint8)
+ array.setflags(write=False)
+ return array
+
+
+def _element(key, width, value, epoch=1):
+ pixels = _pixels(width, value)
+ return RenderedElement(key=key, epoch=epoch, version=value, pixels=pixels,
+ digest=pixel_digest(pixels), width=width)
+
+
+def _record(seq, key, abs_x, width=40, pid="p", epoch=0, value=0, hz=0.0):
+ return ElementRecord(seq=seq, plugin_id=pid, key=key, abs_x=abs_x, width=width,
+ epoch=epoch, digest=pixel_digest(_pixels(width, value)),
+ refresh_hz=hz)
+
+
+class _Adapter:
+ def __init__(self):
+ self.live_epochs = LiveEpochs()
+ self.batches = {} # pid -> {key: RenderedElement}
+ self.redraws = {} # key -> RenderedElement | None
+ self.busy = set()
+ self.calls = []
+
+ def render_live_elements(self, plugin, pid, lock_timeout):
+ self.calls.append(("render", pid, lock_timeout))
+ if pid in self.busy:
+ return None
+ return self.live_epochs.get(pid), self.batches.get(pid, {})
+
+ def redraw_live_element(self, plugin, pid, key, width, height, at):
+ self.calls.append(("redraw", pid, key, width, at))
+ return self.redraws.get(key)
+
+ has_redraw = True
+
+ def has_lock_free_redraw(self, plugin):
+ return self.has_redraw
+
+
+class _Stream:
+ def __init__(self, adapter):
+ self.plugin_adapter = adapter
+ self.plugin_manager = SimpleNamespace(plugins={"p": object(), "q": object()})
+ self.plans = []
+ self.fetched = []
+
+ def plan_next_group(self, count=None):
+ self.plans.append(count)
+ return ["a", "b", "c"]
+
+ def fetch_group_member(self, pid, offscreen_only=False):
+ self.fetched.append((pid, offscreen_only))
+ return (pid, [f"img-{pid}"])
+
+
+def _pipeline(gate=True, **cfg):
+ adapter = _Adapter()
+ p = SimpleNamespace(
+ _strip_gen=1, _view=None, _elements=(), _prepared_group=None,
+ _prefetch_lock=threading.Lock(), _prefetch_generation=7, _applied={},
+ _live_slots={}, _live_ready=collections.deque(),
+ display_width=W, display_height=H, frame_interval=0.01,
+ config=VegasModeConfig(**cfg),
+ display_manager=SimpleNamespace(render_gate=_Gate() if gate else None),
+ stream_manager=_Stream(adapter), _prefetch_thread=None)
+ return p, adapter
+
+
+class _Gate:
+ def __init__(self):
+ self.entered = 0
+
+ def yielding(self):
+ gate = self
+
+ class _Ctx:
+ def __enter__(self):
+ gate.entered += 1
+
+ def __exit__(self, *exc):
+ return False
+ return _Ctx()
+
+
+def _view(left=1000, end=5000, t=NOW):
+ return LiveView(abs_left=left, abs_right=left + W, abs_end=end, t_mono=t)
+
+
+def _worker(p, records=(), view=None, clock=lambda: NOW):
+ p._elements = tuple(records)
+ for r in records:
+ p._applied[r.seq] = (r.epoch, r.digest)
+ p._view = view if view is not None else _view()
+ return VegasWorker(p, clock=clock)
+
+
+# -- choosing -----------------------------------------------------------------
+
+
+def test_an_urgent_group_goes_before_anything():
+ p, adapter = _pipeline()
+ visible = _record(1, "k", 1010)
+ worker = _worker(p, [visible], _view(left=1000, end=1000 + W + 100))
+ adapter.live_epochs.bump("p")
+ worker.request_group()
+ assert worker._pick(NOW) == ("group", None)
+
+
+def test_visible_data_then_ticks_then_a_normal_group_then_data_ahead():
+ p, adapter = _pipeline()
+ visible = _record(1, "vis", 1010)
+ animated = _record(2, "map", 1060, hz=4)
+ ahead = _record(3, "far", 3000, pid="q")
+ worker = _worker(p, [visible, animated, ahead])
+ adapter.live_epochs.bump("p")
+ adapter.live_epochs.bump("q")
+ worker.request_group()
+ assert worker._pick(NOW) == ("data", "p")
+ worker._handled_epoch[1] = worker._handled_epoch[2] = adapter.live_epochs.get("p")
+ assert worker._pick(NOW) == ("tick", animated)
+ worker._next_tick[2] = NOW + 10
+ assert worker._pick(NOW) == ("group", None)
+ worker._group_wanted = False
+ assert worker._pick(NOW) == ("data", "q")
+
+
+def test_nothing_behind_the_viewport_is_redrawn():
+ p, adapter = _pipeline()
+ behind = _record(1, "gone", 900, width=40)
+ worker = _worker(p, [behind])
+ adapter.live_epochs.bump("p")
+ assert worker._pick(NOW) is None
+
+
+def test_only_group_work_while_frames_have_stopped():
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010, hz=4)], _view(t=NOW - 5))
+ adapter.live_epochs.bump("p")
+ assert worker._pick(NOW) is None
+ worker.request_group()
+ assert worker._pick(NOW) == ("group", None)
+
+
+def test_an_element_placed_from_older_data_is_caught_up():
+ # A group drawn before an update() and placed after it: its records carry
+ # the old epoch, so they are due at once.
+ p, adapter = _pipeline()
+ adapter.live_epochs.bump("p")
+ adapter.live_epochs.bump("p")
+ worker = _worker(p, [_record(1, "k", 1010, epoch=1)])
+ assert worker._pick(NOW) == ("data", "p")
+
+
+def test_the_data_floor_defers_but_does_not_drop():
+ p, adapter = _pipeline(live_min_interval=2.0)
+ worker = _worker(p, [_record(1, "k", 1010)])
+ adapter.live_epochs.bump("p")
+ worker._last_data_job["p"] = NOW - 1.0
+ assert worker._pick(NOW) is None
+ p._view = _view(t=NOW + 1.5)
+ assert worker._pick(NOW + 1.5) == ("data", "p")
+
+
+# -- data refreshes ------------------------------------------------------------
+
+
+def test_a_refresh_hands_over_only_what_changed():
+ p, adapter = _pipeline()
+ same = _record(1, "same", 1010, value=0)
+ changed = _record(2, "changed", 1060, value=0)
+ worker = _worker(p, [same, changed])
+ epoch = adapter.live_epochs.bump("p")
+ adapter.batches["p"] = {"same": _element("same", 40, 0, epoch),
+ "changed": _element("changed", 40, 9, epoch)}
+ worker._run(("data", "p"))
+ assert list(p._live_ready) == [2]
+ patch = p._live_slots[2]
+ assert patch.epoch == epoch and patch.strip_gen == 1
+ assert worker._handled_epoch == {1: epoch, 2: epoch}
+ assert worker._pick(NOW + 100) is None # nothing left due
+ assert adapter.calls[0] == ("render", "p", live_worker.DATA_LOCK_TIMEOUT)
+
+
+def test_a_redraw_of_another_width_is_refused_and_said_once(caplog):
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010, width=40)])
+ for n in range(3):
+ epoch = adapter.live_epochs.bump("p")
+ adapter.batches["p"] = {"k": _element("k", 44, n + 1, epoch)}
+ with caplog.at_level("INFO"):
+ worker._run(("data", "p"))
+ assert not p._live_ready
+ assert worker.stats["refused"] == 3
+ assert sum("must not change" in r.message for r in caplog.records) == 1
+
+
+def test_a_busy_lock_backs_off_instead_of_waiting():
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010)])
+ adapter.live_epochs.bump("p")
+ adapter.busy.add("p")
+ worker._run(("data", "p"))
+ assert worker.stats["lock_busy"] == 1
+ assert worker._backoff_until["p"] == NOW + live_worker.LOCK_BACKOFF_S
+ p._view = _view(t=NOW + 2.5)
+ assert worker._pick(NOW + 0.5) is None # backing off (and floored)
+ adapter.busy.clear()
+ assert worker._pick(NOW + 2.5) == ("data", "p") # tried again later
+
+
+def test_a_key_the_plugin_dropped_keeps_its_pixels():
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "gone", 1010)])
+ epoch = adapter.live_epochs.bump("p")
+ adapter.batches["p"] = {}
+ worker._run(("data", "p"))
+ assert not p._live_ready and worker._handled_epoch[1] == epoch
+
+
+def test_the_latest_hand_over_wins_the_slot():
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010)])
+ for value in (5, 6):
+ epoch = adapter.live_epochs.bump("p")
+ adapter.batches["p"] = {"k": _element("k", 40, value, epoch)}
+ worker._last_data_job.clear()
+ worker._run(("data", "p"))
+ assert list(p._live_ready) == [1, 1]
+ assert p._live_slots[1].pixels[0, 0, 0] == 6
+
+
+# -- ticks ----------------------------------------------------------------------
+
+
+def test_a_tick_uses_the_lock_free_redraw_and_reschedules():
+ p, adapter = _pipeline(live_max_hz=5)
+ record = _record(1, "map", 1010, width=40, hz=4)
+ worker = _worker(p, [record])
+ adapter.redraws["map"] = _element("map", 40, 3)
+ worker._run(("tick", record))
+ assert adapter.calls[0][0] == "redraw"
+ assert list(p._live_ready) == [1]
+ assert worker._next_tick[1] > 0
+
+
+def test_a_tick_without_a_redraw_never_waits_for_the_lock():
+ p, adapter = _pipeline()
+ record = _record(1, "map", 1010, hz=4)
+ worker = _worker(p, [record])
+ adapter.has_redraw = False
+ worker._run(("tick", record))
+ assert adapter.calls == [("render", "p", 0.0)]
+
+
+def test_a_redraw_that_returns_none_is_a_skip_not_a_full_redraw():
+ # None is the plugin saying "nothing new"; answering it with a locked
+ # get_vegas_elements() at the tick rate is exactly what the hook avoids.
+ p, adapter = _pipeline()
+ record = _record(1, "map", 1010, hz=4)
+ worker = _worker(p, [record])
+ adapter.redraws["map"] = None
+ worker._run(("tick", record))
+ assert [c[0] for c in adapter.calls] == ["redraw"]
+ assert not p._live_ready
+
+
+def test_animation_is_capped_without_the_gate_and_when_slow():
+ p, _ = _pipeline(gate=False, live_max_hz=5)
+ record = _record(1, "map", 1010, hz=4)
+ worker = _worker(p, [record])
+ assert worker._tick_hz(record) == live_worker.UNGATED_MAX_HZ
+ p2, _ = _pipeline(live_max_hz=5)
+ worker2 = _worker(p2, [record])
+ assert worker2._tick_hz(record) == 4
+ worker2._render_ewma[("p", "map")] = live_worker.SLOW_RENDER_S * 2
+ assert worker2._tick_hz(record) == 2
+ p3, _ = _pipeline(live_max_hz=0)
+ assert _worker(p3, [record])._tick_hz(record) == 0
+
+
+def test_throttling_never_raises_a_slow_elements_rate():
+ p, _ = _pipeline(live_max_hz=5)
+ slow = _record(1, "map", 1010, hz=live_worker.MIN_THROTTLED_HZ / 2)
+ worker = _worker(p, [slow])
+ worker._render_ewma[("p", "map")] = live_worker.SLOW_RENDER_S * 2
+ assert worker._tick_hz(slow) == live_worker.MIN_THROTTLED_HZ / 2
+
+
+def test_a_tick_left_by_an_element_trimmed_away_does_not_spin_the_worker():
+ p, _ = _pipeline()
+ worker = _worker(p, [_record(1, "map", 1010, hz=4)])
+ worker._next_tick[1] = NOW - 5 # past due
+ p._elements = () # ...and trimmed off the strip
+ assert worker._next_wait(NOW) == live_worker.IDLE_WAIT_S
+
+
+def test_a_tick_for_an_element_no_longer_animated_is_dropped():
+ p, _ = _pipeline(live_max_hz=0)
+ record = _record(1, "map", 1010, hz=4)
+ worker = _worker(p, [record])
+ worker._next_tick[1] = NOW - 5
+ assert worker._next_wait(NOW) == live_worker.IDLE_WAIT_S
+ assert worker._pick(NOW) is None
+ assert 1 not in worker._next_tick
+
+
+def test_the_wait_is_until_the_next_tick_on_screen():
+ p, _ = _pipeline(live_max_hz=5)
+ record = _record(1, "map", 1010, hz=4)
+ worker = _worker(p, [record])
+ worker._next_tick[1] = NOW + 0.2
+ assert worker._next_wait(NOW) == pytest.approx(0.2)
+
+
+def test_ticks_stop_for_an_element_far_ahead():
+ p, _ = _pipeline(live_lead_screens=1.0)
+ far = _record(1, "map", 1000 + 3 * W, hz=4)
+ worker = _worker(p, [far])
+ worker._next_tick[1] = NOW
+ assert worker._pick(NOW) is None
+ assert 1 not in worker._next_tick
+
+
+# -- groups ---------------------------------------------------------------------
+
+
+def test_a_group_is_fetched_a_member_at_a_time_and_published():
+ p, _ = _pipeline()
+ worker = _worker(p)
+ worker.request_group()
+ worker.request_group() # coalesced
+ for _ in range(3):
+ assert p._prepared_group is None
+ worker._run(worker._pick(NOW))
+ assert p._prepared_group == [("a", ["img-a"]), ("b", ["img-b"]), ("c", ["img-c"])]
+ assert p.stream_manager.plans == [None]
+ assert all(offscreen for _pid, offscreen in p.stream_manager.fetched)
+ assert worker._pick(NOW) is None # the slot is full
+
+
+def test_a_reset_mid_group_drops_it():
+ p, _ = _pipeline()
+ worker = _worker(p)
+ worker.request_group()
+ worker._run(worker._pick(NOW))
+ p._prefetch_generation += 1
+ worker._run(("group", None))
+ worker._run(("group", None))
+ assert p._prepared_group is None
+
+
+def test_a_stopped_worker_hands_over_what_it_has_of_a_group():
+ p, _ = _pipeline()
+ worker = _worker(p)
+ worker.request_group()
+ worker._run(worker._pick(NOW)) # one member of three
+ worker.stop()
+ worker._hand_over_partial_group()
+ assert p._prepared_group == [("a", ["img-a"])]
+
+
+@pytest.mark.parametrize("why", ["reset", "a group already waiting"])
+def test_a_partial_group_is_not_handed_over(why):
+ p, _ = _pipeline()
+ worker = _worker(p)
+ worker.request_group()
+ worker._run(worker._pick(NOW))
+ if why == "reset":
+ p._prefetch_generation += 1
+ else:
+ p._prepared_group = ["waiting"]
+ worker._hand_over_partial_group()
+ assert p._prepared_group == (None if why == "reset" else ["waiting"])
+
+
+def test_a_new_worker_waits_for_the_one_it_replaces():
+ import time
+ p, _ = _pipeline()
+ retired = threading.Thread(target=time.sleep, args=(0.05,))
+ retired.start()
+ p._retired_worker = retired
+ _worker(p)._join_legacy_prefetch()
+ assert not retired.is_alive()
+
+
+def test_every_job_runs_inside_the_gate():
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010)])
+ adapter.live_epochs.bump("p")
+ worker._run(("data", "p"))
+ assert p.display_manager.render_gate.entered == 1
+
+
+# -- robustness -------------------------------------------------------------------
+
+
+def test_a_job_that_raises_is_counted_and_the_worker_goes_on():
+ p, adapter = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010)])
+ adapter.live_epochs.bump("p")
+
+ def explode(*a, **k):
+ raise KeyError("plugin bug")
+
+ adapter.render_live_elements = explode
+ worker._run(("data", "p"))
+ assert worker.stats["errors"] == 1
+ worker.request_group()
+ worker._run(worker._pick(NOW))
+ assert p.stream_manager.plans
+
+
+def test_a_new_strip_forgets_the_old_ones_bookkeeping():
+ p, _ = _pipeline()
+ worker = _worker(p, [_record(1, "k", 1010)])
+ worker._handled_epoch[1] = 5
+ worker._next_tick[1] = NOW
+ p._strip_gen += 1
+ p._elements = ()
+ worker._pick(NOW)
+ assert worker._handled_epoch == {} and worker._next_tick == {}
+
+
+def test_the_thread_starts_and_stops():
+ import time
+ p, _ = _pipeline()
+ worker = _worker(p, clock=time.monotonic)
+ worker.start()
+ worker.request_group()
+ for _ in range(200):
+ if p._prepared_group is not None:
+ break
+ time.sleep(0.01)
+ worker.stop()
+ worker.join(2)
+ assert not worker.is_alive()
+ assert p._prepared_group is not None
+
+
+@pytest.mark.parametrize("view", [None])
+def test_no_view_yet_still_fetches_groups(view):
+ p, _ = _pipeline()
+ worker = VegasWorker(p)
+ worker.request_group()
+ assert worker._pick(NOW) == ("group", None)
+
+
+def test_the_summary_names_the_slowest_redraw(caplog):
+ clock = [NOW]
+ p, adapter = _pipeline()
+ record = _record(1, "map", 1010, hz=4)
+ worker = _worker(p, [record], clock=lambda: clock[0])
+ adapter.redraws["map"] = _element("map", 40, 3)
+ worker._run(("tick", record))
+ clock[0] += live_worker.SUMMARY_INTERVAL_S + 1
+ with caplog.at_level("INFO"):
+ worker._maybe_summarise()
+ line = next(r.getMessage() for r in caplog.records if "Vegas live:" in r.getMessage())
+ assert "slowest redraw" in line and "p 'map'" in line
+ assert worker._slowest_redraw is None # per summary interval
diff --git a/web_interface/blueprints/api_v3/config.py b/web_interface/blueprints/api_v3/config.py
index dd5bbe64..35515a20 100644
--- a/web_interface/blueprints/api_v3/config.py
+++ b/web_interface/blueprints/api_v3/config.py
@@ -802,7 +802,7 @@ def save_main_config():
'vegas_intra_plugin_gap', 'vegas_render_width_pct',
'vegas_min_content_separation', 'vegas_min_cut_gap',
'vegas_continuous_scroll', 'vegas_extend_threshold_screens',
- 'vegas_smooth_scroll', 'vegas_overflow_mode']
+ 'vegas_smooth_scroll', 'vegas_overflow_mode', 'vegas_live_refresh']
if any(k in data for k in vegas_fields):
if 'display' not in current_config:
@@ -822,6 +822,7 @@ def save_main_config():
_set_checkbox(vegas_config, 'dynamic_duration_enabled', 'vegas_dynamic_duration_enabled')
_set_checkbox(vegas_config, 'continuous_scroll', 'vegas_continuous_scroll')
_set_checkbox(vegas_config, 'smooth_scroll', 'vegas_smooth_scroll')
+ _set_checkbox(vegas_config, 'live_refresh', 'vegas_live_refresh')
# max_plugin_width_ratio is the one fractional setting, so it is
# handled outside the integer loop below.
diff --git a/web_interface/templates/v3/partials/display.html b/web_interface/templates/v3/partials/display.html
index 837efa5a..42044ecc 100644
--- a/web_interface/templates/v3/partials/display.html
+++ b/web_interface/templates/v3/partials/display.html
@@ -568,6 +568,17 @@
+