rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-05 06:45:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(display): run loop stage 3 -- ScreenRunner, PREEMPTED, OnDemand/Live/Rotation Sources (#762)

Browse Source 
* refactor(display): run each screen through a ScreenRunner (run loop stage 3)

The two frame loops, the make-up dwell and the dynamic-duration exit move
out of DisplayController.run() into src/screen_runner.py. ScreenRunner
paces with an injected FrameClock (production: this module's time, looked
up per call so the golden harness's fake clock still drives it) and
returns one Outcome whose ExitReason is DURATION, CYCLE_COMPLETE, EMPTY,
ERROR, DISPLAY_FALSE, RELOAD or PREEMPTED.

PREEMPTED replaces the re-checks that used to follow each frame loop and
the make-up dwell (current_display_mode != active_mode, the schedule, a
pending WiFi notice): the runner asks its host at named service points
(FRAME, AFTER_LOOP, after_dwell, FINAL), and on PREEMPTED run() goes to
the next pass without advancing the rotation, as each `continue` did.
RELOAD is the one early end that still advances, as a reload always did.
Each service point reads the WiFi notice file exactly when the loop did
(NoticeRead), because the read is throttled and deletes expired files.

The host answers still use the old checks; the following commits move
them to the Arbiter. _screen_preempted is gone (folded into the FRAME
check); _wait_frame_interval returns the preempting plan instead of a
bool. The frame pacing (8 ms deadline, 1 ms minimum yield, 1 Hz wait with
socket wake) is the same code, moved.

Golden traces unchanged. A capture of every harness run (all 67, with
every sleep, display() call, wifi read, live scan, publish and dwell
logged) is identical to origin/main except for throttled WiFi reads that
returned the cached answer (no side effect) after a notice preempted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(display): on-demand is an Arbiter Source (run loop stage 3)

Arbiter.decide() now answers for an active on-demand session itself
(Source.ON_DEMAND) instead of returning LEGACY:

- ArbiterState gains the session: its mode list, index, expiry and pin,
  plus current_mode, snapshotted from the controller's fields by
  _arbiter_state(). The controller's attributes stay the record that the
  web UI, the control socket and the cache read.
- The OnDemand plan is the session's current mode (an index past the end
  of a shortened list starts again at 0), with what is left of a timed
  session at `now` as max_duration and the expiry as deadline. A session
  with no modes left is a plan with no mode; the controller ends it and
  shows the rotation's mode, as _resolve_active_mode did.
- on_demand_bound() is _clamp_to_on_demand made pure. It is still applied
  after the first frame, with the clock read there.
- ArbiterState.next_on_demand() is the step _advance_on_demand takes.

run() asks decide() for the screen at the point it used to call
_resolve_active_mode (after any Vegas iteration, so a session that
started mid-iteration still shows next), and _take_plan() applies it.

Golden traces and the 67-run capture identical to origin/main. Adds
TestOnDemand and the bound table to test_display_arbiter.py; the stage-2
table's on-demand rows now name ON_DEMAND instead of LEGACY.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(display): live priority is an Arbiter Source (run loop stage 3)

The live-priority step of run() (step 7) and the live checks around the
Vegas iteration become the Arbiter's Live Source:

- ArbiterInputs gains live_modes (the scan, None where run() made none),
  vegas_enabled, vegas_live_in_ticker and vegas_yielded. ArbiterState
  gains the rotation and its index, the live resume point and the
  "takeover not shown yet" flag.
- Live picks the next live mode round-robin (live_pick, now also what
  _check_live_priority returns), not advancing past a mid-screen takeover
  that has not shown. It outranks Vegas unless the ticker keeps live
  content, in which case it has no say at all, as before. With nothing
  live, a plan below it carries ends_live and the interrupted rotation
  resumes.
- ArbiterState.claim_live/release_live are _apply_live_priority's
  bookkeeping made pure; _apply_live_priority applies them.

run() reads the inputs below the WiFi notice where it always did
(_arbiter_inputs_below_wifi: the Vegas check, then the scan), asks
decide() once more, and _take_plan applies the claim or the resume. The
Vegas iteration moves to _run_vegas_iteration, which re-decides with
vegas_yielded after a yield, so a game that stopped the ticker or an
on-demand session that started mid-iteration still shows next. LEGACY
now means Vegas or the rotation.

One redundant call is gone: a Vegas pass scanned the live plugins twice
at the same instant (step 7, then step 8's "is anything live?"); it scans
once. Golden traces unchanged. The 67-run capture is identical to
origin/main once that duplicate scan and _apply_live_priority(None) calls
that changed nothing are left out.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(display): the rotation is an Arbiter Source; LEGACY means Vegas (run loop stage 3)

decide() now names the screen for every pass: the Rotation Source
(Source.ROTATION) answers with the rotation's current mode, after the
resume when live priority just ended. LEGACY is left meaning only Vegas,
whose iteration is still run()'s own code until stage 4. Once this pass's
iteration has yielded (vegas_yielded), Vegas passes and the screen it fell
through to is decided like any other.

The rotation's mode is state.current_mode rather than
rotation[rotation_index]: they agree except where something moved the
panel off the list and the rotation carries on from there (a live mode no
entry names, or None after a session ended with nothing to resume to),
and run() always showed current_display_mode.

ArbiterState.after(outcome) is _advance_after_screen's step: an on-demand
session moves to its next mode; otherwise the rotation advances unless the
mode just shown is still live. The Outcome carries the two facts only the
controller can see at the end of the screen (on_demand_active, the live
hold from _still_live). Ending a session with no modes left stays in the
controller, because it is not pure.

Golden traces unchanged; the 67-run capture is identical to origin/main
with the same two exclusions as the previous commit. Adds the Vegas /
Rotation table and TestAfter; the stage-2 rows that said LEGACY for "live,
Vegas or rotation" now say ROTATION.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(display): one decide() call at each of the runner's service points (run loop stage 3)

The three mid-screen checks the frame loops made one after another --
_check_live_takeover, then _screen_preempted with _wifi_notice_pending in
it -- become one call: Arbiter.decide(state, inputs, now, running=plan).
It returns `running` itself while the screen holds, else the plan that
ends it, from these rules in the order the loops checked them:

1. Live: a game went live while a non-live screen runs. First because it
   is the one preemption that changes the state (the rotation moves to
   the live mode and remembers where it was), and it is still claimed
   when a WiFi notice is pending too; the next pass shows the notice,
   then the game, as before.
2. The panel's mode moved under the screen (on-demand started, ended or
   changed; the rotation was rebuilt).
3. The schedule turned the panel off.
4. A WiFi notice arrived (on-demand outranks it; compared with expiry).
5. A plugin reload is waiting (between frames only).

Each rule is gated by plan.preemptible_by: every screen may be preempted
by the gate, OnDemand, Wifi, Live, Rotation and a reload, except that a
live screen leaves Live out. A follower and Vegas never preempt
mid-screen. The pure helper live_takeover() is the Live rule, shared with
the dwell sleep's _check_live_takeover.

The controller only gathers and applies. _screen_service applies pending
changes and makes the live scan when one is due (_scan_for_takeover: the
same throttle and gates as before); _screen_check reads the WiFi notice
exactly where the loop did (the read is throttled and deletes an expired
file, so an extra read would move both), calls decide() once, and claims
a live takeover. _screen_preempted is gone; _check_live_takeover and
_wifi_notice_pending remain for the dwell sleep and the Vegas yield path,
built on the same rules.

Golden traces unchanged. The 67-run capture is identical to origin/main
(with the earlier two exclusions) except for one event: in the 125 Hz loop
the live scan still runs before the frame's sleep, but the claim is now
made by the service point after it, so the "live" state change is logged
8 ms later (test_live_game_cuts_a_scrolling_screen_short: 8.064 -> 8.072).
The screen still ends at the same frame (8.072) and every frame, sleep
and pass is unchanged.

Adds the mid-screen table (24 rows), live_takeover's table, and
test/test_screen_runner.py (the runner on a scripted host, plus the
controller's service point: which reads it makes at which checkpoint).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs, tests: run loop stage 3 -- doc, changelog, mutation survivors

docs/RUN_LOOP_REDESIGN.md describes run() as it now is (two decide()
calls per pass, the runner and its service points, the state snapshot and
its transitions), records what stage 3 shipped and how it was checked,
and adds one open "may be wrong" behaviour the mutation run surfaced: a
Vegas iteration stopped for a sync follower falls through to a full
rotation screen before the follower gets the panel (pinned by
test_vegas_yielding_to_a_follower_shows_a_rotation_screen_first; passes on
origin/main too). docs/IPC_CONTROL_SOCKET.md no longer names
_screen_preempted. CHANGELOG entry under Unreleased.

A mutation run broke 46 moved or new pieces once each (OnDemand, Live,
Vegas/Rotation, after(), each mid-screen rule, the runner's pacing, exits
and service points, the controller's gathering and claims). Three
survived and get a test here:

- the after-loop service point not reading the WiFi notice: the
  completed-loop checkpoint gets its own name, and a run-loop test has a
  notice pending when a later frame comes back empty;
- the Vegas yield path not marking vegas_yielded: the follower test above;
- _take_plan not writing back a reset on-demand index: a controller test
  with an index past a shortened list.

All 46 now fail at least one test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-10-04 22:15:30 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 294bd522bd
commit 26cae3e5d6
11 changed files with 2495 additions and 642 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/test_display_arbiter.py
+399 -12
View File
@@ -9,6 +9,7 @@ that overrides the schedule and ends (#714).
"""
import itertools
from dataclasses import replace
import os
from unittest.mock import MagicMock, patch
@@ -19,6 +20,7 @@ os.environ.setdefault("EMULATOR", "true")
from src import display_arbiter # noqa: E402
from src.display_arbiter import ( # noqa: E402
SCHEDULED_OFF_DWELL,
SCREEN_PREEMPTERS,
WIFI_NOTICE_DWELL,
Arbiter,
ArbiterInputs,
@@ -26,6 +28,9 @@ from src.display_arbiter import ( # noqa: E402
ScreenPlan,
Source,
WifiNotice,
live_pick,
live_takeover,
on_demand_bound,
wifi_notice_preempts,
)
@@ -34,7 +39,9 @@ NOTICE = WifiNotice(message="Connected to HomeNet", expires_at=1_000.0)
OFF = Source.SCHEDULED_OFF
FOLLOW = Source.FOLLOWER
WIFI = Source.WIFI
ONDEM = Source.ON_DEMAND
LEGACY = Source.LEGACY
ROTATION = Source.ROTATION
# (schedule_on, on_demand_active, follower_active, notice) -> Source.
# Every combination of the stage-2 inputs: 2 x 2 x 2 x 2 = 16 rows.
@@ -46,17 +53,17 @@ DECIDE_TABLE = [
(False, False, True, None, OFF),
(False, False, True, NOTICE, OFF),
# Scheduled off, but on-demand overrides the gate.
(False, True, False, None, LEGACY), # on-demand: run() decides
(False, True, False, NOTICE, LEGACY), # on-demand outranks WiFi
(False, True, False, None, ONDEM), # on-demand overrides the gate
(False, True, False, NOTICE, ONDEM), # on-demand outranks WiFi
(False, True, True, None, FOLLOW), # follower outranks on-demand
(False, True, True, NOTICE, FOLLOW),
# Scheduled on.
(True, False, False, None, LEGACY), # live / Vegas / rotation
(True, False, False, None, ROTATION), # live / Vegas / rotation
(True, False, False, NOTICE, WIFI),
(True, False, True, None, FOLLOW),
(True, False, True, NOTICE, FOLLOW), # follower outranks WiFi
(True, True, False, None, LEGACY),
(True, True, False, NOTICE, LEGACY), # on-demand outranks WiFi
(True, True, False, None, ONDEM),
(True, True, False, NOTICE, ONDEM), # on-demand outranks WiFi
(True, True, True, None, FOLLOW),
(True, True, True, NOTICE, FOLLOW),
]
@@ -95,8 +102,15 @@ class TestDecide:
elif expected is WIFI:
assert plan == ScreenPlan(WIFI, max_duration=WIFI_NOTICE_DWELL,
notice=NOTICE)
elif expected is ONDEM:
# An empty state: a session with no modes, which the
# controller ends (see TestOnDemand for real sessions).
assert plan == ScreenPlan(ONDEM)
elif expected is ROTATION:
# No live scan and Vegas off: the rotation's (empty) mode.
assert plan == ScreenPlan(ROTATION, preemptible_by=SCREEN_PREEMPTERS)
else:
# A follower paces itself; LEGACY is run()'s existing code.
# A follower paces itself.
assert plan == ScreenPlan(expected)
def test_dwells_are_todays(self):
@@ -224,16 +238,389 @@ class TestControllerSnapshot:
assert (plan.source is OFF) is (not dc.is_display_active), time_str
return plan.source
assert step("22:59:30") is LEGACY
assert step("22:59:30") is ROTATION
assert step("23:00:00") is OFF # window ends
dc.on_demand_active = True
assert step("23:00:10") is LEGACY # on-demand overrides
assert step("23:00:10") is ONDEM # on-demand overrides
assert dc.on_demand_schedule_override is True
assert step("23:01:00") is LEGACY # next minute, still on
assert step("23:01:00") is ONDEM # next minute, still on
dc._reset_on_demand_fields() # session ends
assert step("23:01:20") is OFF # same minute: blanks
dc.on_demand_active = True
assert step("06:59:00") is LEGACY
assert step("07:00:00") is LEGACY # schedule back on mid-session
assert step("06:59:00") is ONDEM
assert step("07:00:00") is ONDEM # schedule back on mid-session
dc._reset_on_demand_fields()
assert step("07:00:30") is LEGACY
assert step("07:00:30") is ROTATION
# -- OnDemand (stage 3) ---------------------------------------------------
ON = ArbiterInputs(schedule_on=True, on_demand_active=True, follower_active=False)
def _session(modes=("a", "b", "c"), index=0, expires_at=None, current=None):
return ArbiterState(current_mode=current, on_demand_modes=tuple(modes),
on_demand_index=index, on_demand_expires_at=expires_at)
class TestOnDemand:
# (modes, index, expires_at, now) -> (mode, max_duration)
TABLE = [
(("a", "b", "c"), 0, None, 100.0, "a", None), # untimed
(("a", "b", "c"), 2, None, 100.0, "c", None),
(("a", "b", "c"), 3, None, 100.0, "a", None), # past the end: 0
(("a", "b", "c"), 9, None, 100.0, "a", None),
(("a",), 0, 130.0, 100.0, "a", 30.0), # 30 s left
(("a",), 0, 130.0, 130.0, "a", 0.0), # none left
(("a",), 0, 130.0, 200.0, "a", 0.0), # never negative
]
@pytest.mark.parametrize("modes,index,expires_at,now,mode,max_duration", TABLE)
def test_current_mode_and_time_left(self, modes, index, expires_at, now, mode,
max_duration):
plan = Arbiter.decide(_session(modes, index, expires_at), ON, now)
assert plan.source is ONDEM
assert plan.mode == mode
assert plan.max_duration == max_duration
assert plan.deadline == expires_at
def test_no_modes_left_is_a_plan_with_no_mode(self):
plan = Arbiter.decide(_session(modes=()), ON, 0.0)
assert plan == ScreenPlan(ONDEM)
def test_preemptible_by_the_schedule_a_reload_and_its_own_changes(self):
plan = Arbiter.decide(_session(), ON, 0.0)
assert {Source.SCHEDULED_OFF, ONDEM, Source.RELOAD} <= plan.preemptible_by
assert Source.FOLLOWER not in plan.preemptible_by
@pytest.mark.parametrize("index,expected_index,expected_mode", [
(0, 1, "b"), (1, 2, "c"), (2, 0, "a")])
def test_next_on_demand_wraps(self, index, expected_index, expected_mode):
nxt = _session(index=index).next_on_demand()
assert (nxt.on_demand_index, nxt.current_mode) == (expected_index, expected_mode)
def test_showing_a_plan_resets_an_index_past_the_end(self):
state = _session(index=5, current="x")
plan = Arbiter.decide(state, ON, 0.0)
shown = state.showing(plan)
assert (shown.on_demand_index, shown.current_mode) == (0, "a")
assert state.on_demand_index == 5 # not mutated
# (min, max, deadline, now) -> bounds. The bound applied after the first frame.
BOUND_TABLE = [
(10.0, 20.0, None, 0.0, (10.0, 20.0)), # untimed: unchanged
(10.0, 20.0, 100.0, 50.0, (10.0, 20.0)), # plenty left
(10.0, 20.0, 100.0, 85.0, (10.0, 15.0)), # max cut to what is left
(10.0, 20.0, 100.0, 95.0, (5.0, 5.0)), # both cut
(10.0, 20.0, 100.0, 100.0, None), # nothing left
(10.0, 20.0, 100.0, 150.0, None),
]
@pytest.mark.parametrize("min_d,max_d,deadline,now,expected", BOUND_TABLE)
def test_on_demand_bound(min_d, max_d, deadline, now, expected):
assert on_demand_bound(min_d, max_d, deadline, now) == expected
# -- Live (stage 3) -------------------------------------------------------
LIVE = Source.LIVE
# (live_modes, current_mode, advance) -> pick. _check_live_priority's rule.
PICK_TABLE = [
((), "clock", True, None),
(None, "clock", True, None),
(("nfl",), "clock", True, "nfl"), # not on a live mode: first
(("nfl", "nhl"), "clock", True, "nfl"),
(("nfl", "nhl"), "clock", False, "nfl"),
(("nfl", "nhl"), "nfl", True, "nhl"), # round-robin
(("nfl", "nhl"), "nhl", True, "nfl"), # wraps
(("nfl", "nhl"), "nhl", False, "nhl"), # a peek stays put
(("nfl",), "nfl", True, "nfl"), # one game: itself
]
@pytest.mark.parametrize("live,current,advance,expected", PICK_TABLE)
def test_live_pick(live, current, advance, expected):
assert live_pick(live, current, advance) == expected
def _below(live=None, vegas=False, keeps=False, yielded=False, on_demand=False):
"""Inputs for the Sources below the notice (no notice, no follower)."""
return ArbiterInputs(schedule_on=True, on_demand_active=on_demand,
follower_active=False, live_modes=live, vegas_enabled=vegas,
vegas_live_in_ticker=keeps, vegas_yielded=yielded)
ROT = ("clock", "weather", "nfl_live", "nhl_live")
def _rot(current="clock", index=0, resume=None, unshown=False):
return ArbiterState(current_mode=current, rotation=ROT, rotation_index=index,
live_resume_index=resume, live_takeover_unshown=unshown)
class TestLive:
# (state, inputs) -> (source, mode, ends_live)
TABLE = [
# Nothing live, nothing to resume.
(_rot(), _below(live=()), "below", None, False),
# Not scanned (on-demand, or the ticker keeps live content).
(_rot(), _below(live=None), "below", None, False),
# A game is live: it takes the panel.
(_rot(), _below(live=("nfl_live",)), LIVE, "nfl_live", False),
# Two: round-robin from the one showing.
(_rot("nfl_live", 2), _below(live=("nfl_live", "nhl_live")), LIVE, "nhl_live", False),
# ... unless a mid-screen takeover chose it and it has not shown yet.
(_rot("nfl_live", 2, resume=0, unshown=True),
_below(live=("nfl_live", "nhl_live")), LIVE, "nfl_live", False),
# Vegas keeps live content in its ticker: Live has no say at all,
# not even the resume.
(_rot(), _below(live=("nfl_live",), vegas=True, keeps=True), "below", None, False),
(_rot("nfl_live", 2, resume=1),
_below(live=(), vegas=True, keeps=True), "below", None, False),
# Vegas that yields to live content: Live outranks it.
(_rot(), _below(live=("nfl_live",), vegas=True), LIVE, "nfl_live", False),
# The game ended: the interrupted rotation resumes.
(_rot("nfl_live", 2, resume=1), _below(live=()), "below", None, True),
(_rot("nfl_live", 2, resume=1), _below(live=(), vegas=True), "below", None, True),
# On-demand outranks Live.
(ArbiterState(on_demand_modes=("x",)), _below(live=("nfl_live",), on_demand=True),
ONDEM, "x", False),
]
@pytest.mark.parametrize("state,inputs,source,mode,ends_live", TABLE)
def test_decide(self, state, inputs, source, mode, ends_live):
plan = Arbiter.decide(state, inputs, 0.0)
if source == "below":
assert plan.source not in (LIVE, ONDEM, OFF, FOLLOW, WIFI)
else:
assert plan.source is source
assert plan.mode == mode
assert plan.ends_live is ends_live
def test_a_live_plan_has_no_durations_until_its_first_frame(self):
plan = Arbiter.decide(_rot(), _below(live=("nfl_live",)), 0.0)
assert (plan.min_duration, plan.max_duration, plan.frame_policy) == (None, None, None)
class TestLiveTransitions:
def test_claim_saves_where_the_rotation_was(self):
nxt = _rot("weather", 1).claim_live("nfl_live")
assert (nxt.current_mode, nxt.rotation_index, nxt.live_resume_index) == ("nfl_live", 2, 1)
def test_a_second_claim_keeps_the_first_resume_point(self):
nxt = _rot("nfl_live", 2, resume=1).claim_live("nhl_live")
assert (nxt.current_mode, nxt.rotation_index, nxt.live_resume_index) == ("nhl_live", 3, 1)
def test_claiming_the_mode_showing_changes_nothing(self):
state = _rot("nfl_live", 2, resume=1)
assert state.claim_live("nfl_live") is state
def test_a_live_mode_outside_the_rotation_keeps_the_index(self):
nxt = _rot("weather", 1).claim_live("mlb_live")
assert (nxt.current_mode, nxt.rotation_index, nxt.live_resume_index) == ("mlb_live", 1, 1)
def test_release_resumes_and_forgets(self):
nxt = _rot("nhl_live", 3, resume=1).release_live()
assert (nxt.current_mode, nxt.rotation_index, nxt.live_resume_index) == ("weather", 1, None)
def test_release_wraps_a_resume_point_past_a_shortened_rotation(self):
nxt = _rot("nhl_live", 3, resume=6).release_live()
assert (nxt.current_mode, nxt.rotation_index) == ("nfl_live", 2) # 6 % 4
def test_release_with_nothing_to_resume_changes_nothing(self):
state = _rot("clock", 0)
assert state.release_live() is state
empty = ArbiterState(current_mode="x", live_resume_index=2)
assert empty.release_live() is empty # no rotation to resume into
# -- Vegas and Rotation (stage 3) -----------------------------------------
class TestVegasAndRotation:
# (state, inputs) -> (source, mode, ends_live). LEGACY now means Vegas only.
TABLE = [
(_rot("weather", 1), _below(live=()), ROTATION, "weather", False),
(_rot("weather", 1), _below(live=None), ROTATION, "weather", False),
(_rot("weather", 1), _below(live=(), vegas=True), LEGACY, None, False),
(_rot("weather", 1), _below(live=None, vegas=True, keeps=True), LEGACY, None, False),
# The iteration yielded: the screen it fell through to.
(_rot("weather", 1), _below(live=(), vegas=True, yielded=True),
ROTATION, "weather", False),
(_rot("weather", 1), _below(live=("nfl_live",), vegas=True, yielded=True),
LIVE, "nfl_live", False),
# Live priority just ended: the rotation resumes where it was cut.
(_rot("nhl_live", 3, resume=1), _below(live=()), ROTATION, "weather", True),
# ... and Vegas carries the resume through to its own pass.
(_rot("nhl_live", 3, resume=1), _below(live=(), vegas=True), LEGACY, None, True),
# A rotation that something moved off its list carries on from there.
(ArbiterState(current_mode=None, rotation=ROT), _below(live=()), ROTATION, None, False),
]
@pytest.mark.parametrize("state,inputs,source,mode,ends_live", TABLE)
def test_decide(self, state, inputs, source, mode, ends_live):
plan = Arbiter.decide(state, inputs, 0.0)
assert (plan.source, plan.mode, plan.ends_live) == (source, mode, ends_live)
def test_a_rotation_plan_may_be_preempted_by_everything_a_screen_watches(self):
plan = Arbiter.decide(_rot(), _below(live=()), 0.0)
assert plan.preemptible_by == SCREEN_PREEMPTERS
assert {OFF, ONDEM, WIFI, LIVE, ROTATION, Source.RELOAD} == SCREEN_PREEMPTERS
class _End:
def __init__(self, on_demand_active=False, still_live=False):
self.on_demand_active = on_demand_active
self.still_live = still_live
class TestAfter:
"""ArbiterState.after: _advance_after_screen's step."""
def test_the_rotation_advances(self):
nxt = _rot("weather", 1).after(_End())
assert (nxt.current_mode, nxt.rotation_index) == ("nfl_live", 2)
def test_it_wraps(self):
nxt = _rot("nhl_live", 3).after(_End())
assert (nxt.current_mode, nxt.rotation_index) == ("clock", 0)
def test_a_live_mode_still_live_holds(self):
state = _rot("nfl_live", 2)
assert state.after(_End(still_live=True)) is state
def test_an_on_demand_session_moves_to_its_next_mode(self):
state = replace(_session(index=1, current="b"), rotation=ROT, rotation_index=1)
nxt = state.after(_End(on_demand_active=True))
assert (nxt.current_mode, nxt.on_demand_index, nxt.rotation_index) == ("c", 2, 1)
def test_a_session_with_no_modes_is_left_to_the_controller(self):
state = ArbiterState(current_mode="x", rotation=ROT)
assert state.after(_End(on_demand_active=True)) is state
def test_no_rotation_no_step(self):
state = ArbiterState(current_mode="x")
assert state.after(_End()) is state
# -- Mid-screen: decide(..., running=plan) (stage 3) ----------------------
#
# What the ScreenRunner asks at each service point. These rows are what
# _check_live_takeover, _screen_preempted and _wifi_notice_pending answered
# between frames before stage 3, written out.
CLOCK = Arbiter.decide(ArbiterState(current_mode="clock"), _below(live=()), 0.0)
NFL = Arbiter.decide(ArbiterState(current_mode="clock"), _below(live=("nfl_live",)), 0.0)
OD = Arbiter.decide(_session(modes=("x", "y")), ON, 0.0)
FRESH = WifiNotice(message="AP mode", expires_at=1_000.0)
HELD = "held"
def _mid(on_demand=False, schedule_on=True, live=None, notice=None, reload=False):
return ArbiterInputs(schedule_on=schedule_on, on_demand_active=on_demand,
follower_active=False, live_modes=live, wifi_notice=notice,
reload_pending=reload)
# (running, current_mode, inputs, now) -> HELD or (source, mode)
MID_TABLE = [
# Nothing changed.
(CLOCK, "clock", _mid(), 500.0, HELD),
(CLOCK, "clock", _mid(live=()), 500.0, HELD),
# A game went live: it takes the panel (the first live mode).
(CLOCK, "clock", _mid(live=("nfl_live", "nhl_live")), 500.0, (LIVE, "nfl_live")),
# ... even with a notice pending: the claim is made now, and the next
# pass shows the notice first (top-of-pass order), then the game.
(CLOCK, "clock", _mid(live=("nfl_live",), notice=FRESH), 500.0, (LIVE, "nfl_live")),
# ... but not over the schedule or an on-demand session.
(CLOCK, "clock", _mid(live=("nfl_live",), schedule_on=False), 500.0, (OFF, None)),
(CLOCK, "x", _mid(live=("nfl_live",), on_demand=True), 500.0, (ONDEM, "x")),
# A screen already on a live mode is not taken over by another.
(CLOCK, "nfl_live", _mid(live=("nfl_live", "nhl_live")), 500.0, (ROTATION, "nfl_live")),
(NFL, "nfl_live", _mid(live=("nhl_live",)), 500.0, HELD),
# The mode moved under the screen: on-demand started, or ended, or the
# rotation was rebuilt.
(CLOCK, "x", _mid(on_demand=True), 500.0, (ONDEM, "x")),
(OD, "weather", _mid(), 500.0, (ROTATION, "weather")),
(CLOCK, "weather", _mid(), 500.0, (ROTATION, "weather")),
# An on-demand session that ends on the same mode keeps the screen.
(OD, "x", _mid(), 500.0, HELD),
# The schedule: off ends it; an on-demand override holds.
(CLOCK, "clock", _mid(schedule_on=False), 500.0, (OFF, None)),
(OD, "x", _mid(on_demand=True, schedule_on=False), 500.0, HELD),
# A WiFi notice, compared with its expiry; on-demand outranks it.
(CLOCK, "clock", _mid(notice=FRESH), 999.9, (WIFI, None)),
(CLOCK, "clock", _mid(notice=FRESH), 1_000.0, HELD),
(NFL, "nfl_live", _mid(notice=FRESH), 500.0, (WIFI, None)),
(OD, "x", _mid(on_demand=True, notice=FRESH), 500.0, HELD),
# A plugin reload waits at the top of the loop.
(CLOCK, "clock", _mid(reload=True), 500.0, (Source.RELOAD, None)),
(OD, "x", _mid(on_demand=True, reload=True), 500.0, (Source.RELOAD, None)),
# The order between them: a moved mode before the schedule, the
# schedule before a notice, a notice before a reload.
(CLOCK, "weather", _mid(schedule_on=False), 500.0, (ROTATION, "weather")),
(CLOCK, "clock", _mid(schedule_on=False, notice=FRESH), 500.0, (OFF, None)),
(CLOCK, "clock", _mid(notice=FRESH, reload=True), 500.0, (WIFI, None)),
]
class TestMidScreen:
@pytest.mark.parametrize("running,current,inputs,now,expected", MID_TABLE)
def test_decide(self, running, current, inputs, now, expected):
state = ArbiterState(current_mode=current)
plan = Arbiter.decide(state, inputs, now, running=running)
if expected == HELD:
assert plan is running
else:
assert plan is not running
assert (plan.source, plan.mode) == expected
def test_the_running_plans(self):
assert (CLOCK.source, CLOCK.mode) == (ROTATION, "clock")
assert (NFL.source, NFL.mode) == (LIVE, "nfl_live")
assert LIVE not in NFL.preemptible_by
assert (OD.source, OD.mode) == (ONDEM, "x")
def test_a_follower_and_vegas_never_preempt(self):
for plan in (CLOCK, NFL, OD):
assert FOLLOW not in plan.preemptible_by
assert LEGACY not in plan.preemptible_by
def test_nothing_in_preemptible_by_means_nothing_preempts(self):
bare = replace(CLOCK, preemptible_by=frozenset())
inputs = _mid(live=("nfl_live",), schedule_on=False, notice=FRESH, reload=True)
assert Arbiter.decide(ArbiterState(current_mode="weather"), inputs, 0.0,
running=bare) is bare
def test_mid_screen_decide_reads_no_clock(self):
boom = MagicMock(side_effect=AssertionError("decide read the clock"))
with patch("time.time", boom), patch("time.monotonic", boom):
for running, current, inputs, now, _ in MID_TABLE:
Arbiter.decide(ArbiterState(current_mode=current), inputs, now,
running=running)
# (current, inputs) -> live_takeover. The mode a mid-screen check claims.
TAKEOVER_TABLE = [
("clock", _mid(live=("nfl_live",)), "nfl_live"),
("clock", _mid(live=()), None),
("clock", _mid(live=None), None), # no scan was due
("nfl_live", _mid(live=("nfl_live",)), None),
("clock", _mid(live=("nfl_live",), on_demand=True), None),
("clock", _mid(live=("nfl_live",), schedule_on=False), None),
("clock", replace(_mid(live=("nfl_live",)), vegas_enabled=True,
vegas_live_in_ticker=True), None),
]
@pytest.mark.parametrize("current,inputs,expected", TAKEOVER_TABLE)
def test_live_takeover(current, inputs, expected):
assert live_takeover(ArbiterState(current_mode=current), inputs) == expected
test/test_handover_scroll_state.py
+1 -1
View File
@@ -709,6 +709,6 @@ class TestTheControllersOwnScreens:
c._check_wifi_status_message.return_value = None
inputs = DisplayController._arbiter_inputs(c)
assert inputs.wifi_notice is None
assert Arbiter.decide(ArbiterState(), inputs, 0.0).source is Source.LEGACY
assert Arbiter.decide(ArbiterState(), inputs, 0.0).source is Source.ROTATION
assert dm.is_currently_scrolling()
assert dm._frame_hold == 2
test/test_ipc_display_stage2.py
+15 -5
View File
@@ -41,6 +41,15 @@ def _reload(plugin_id):
return _command(Command.PLUGIN_RELOAD, PluginReloadArgs(plugin_id))
def _screen(mode):
"""A rotation screen of ``mode``, as the ScreenRunner hands it to the
1 Hz loop's frame wait."""
from src.display_arbiter import ArbiterState, rotation_plan
from src.screen_runner import Screen
plan = rotation_plan(ArbiterState(current_mode=mode))
return Screen(plan, plugin=None, accepts_display_mode=False, start=0.0)
@pytest.fixture
def dc(test_display_controller):
c = test_display_controller
@@ -253,7 +262,8 @@ class TestRealTimeWake:
dc.on_demand_active = False
dc._activate_on_demand = MagicMock(
side_effect=lambda request: setattr(dc, 'current_display_mode', 'weather'))
dc._wifi_notice_pending = MagicMock(return_value=False)
dc._wifi_notice_pending = MagicMock(return_value=False) # the dwell's check
dc._read_wifi_notice = MagicMock(return_value=None) # the frame wait's
dc._tick_plugin_updates = MagicMock()
dc._check_live_takeover = MagicMock()
dc.cache_manager.get = MagicMock(return_value=None)
@@ -282,10 +292,10 @@ class TestRealTimeWake:
dc.current_display_mode = 'clock'
stamps = []
t = self._post_later(server, self._start_line(f's{i}'), 0.05, stamps)
ended = dc._wait_frame_interval(1.0, 'clock')
ended = dc._wait_frame_interval(1.0, _screen('clock'))
woke = time.monotonic()
t.join()
assert ended is True
assert ended is not None
latencies.append(woke - stamps[0])
latencies.sort()
print(f"static-screen wake latency: median {latencies[5] * 1000:.2f} ms, "
@@ -315,7 +325,7 @@ class TestRealTimeWake:
'args': {'brightness': 33}}).encode()
started = time.monotonic()
t = self._post_later(server, line, 0.1, stamps)
assert dc._wait_frame_interval(0.5, 'clock') is False
assert dc._wait_frame_interval(0.5, _screen('clock')) is None
assert time.monotonic() - started >= 0.49
t.join()
dc.display_manager.set_brightness.assert_called_once_with(33)
@@ -323,7 +333,7 @@ class TestRealTimeWake:
def test_without_a_socket_it_is_a_plain_sleep(self, dc):
dc._control_server = None
started = time.monotonic()
assert dc._wait_frame_interval(0.2, 'clock') is False
assert dc._wait_frame_interval(0.2, _screen('clock')) is None
assert time.monotonic() - started >= 0.19
test/test_screen_runner.py
+455
View File
@@ -0,0 +1,455 @@
"""ScreenRunner (src/screen_runner.py) on a scripted host and a fake clock.
The golden traces (test_run_loop_golden.py) run it inside the real
DisplayController; these pin down its own contract: which ExitReason each
way of ending gives, the order it calls its host in, how the 125 Hz loop
paces, and which service point asks what.
"""
from typing import Any, List, Optional, Tuple
import pytest
from src.display_arbiter import (
RELOAD_PLAN, SCREEN_PREEMPTERS, FramePolicy, ScreenPlan, Source,
)
from src.screen_runner import (
AFTER_COMPLETED_LOOP, AFTER_LOOP, DYNAMIC_GRACE, FINAL, FRAME, HIGH_FPS_INTERVAL,
Checkpoint, ExitReason, FirstFrame, NoticeRead, Screen, ScreenRunner,
)
WIFI_PLAN = ScreenPlan(Source.WIFI)
class Clock:
"""time/perf_counter read ``now``; sleep advances it."""
def __init__(self):
self.now = 0.0
self.sleeps: List[float] = []
def time(self) -> float:
return self.now
def perf_counter(self) -> float:
return self.now
def sleep(self, seconds: float) -> None:
self.sleeps.append(round(seconds, 6))
self.now += seconds
class Host:
"""A ScreenHost whose answers are scripted; records every call."""
def __init__(self, clock: Clock, *, shown=True, raised=False, minimum=10.0,
maximum=10.0, dynamic=False, policy=FramePolicy.STATIC,
completed=True, draws=None, checks=None, work=0.0,
cycle_complete_at=None, dwell_to=None):
self.clock = clock
self.calls: List[Tuple[Any, ...]] = []
self.shown, self.raised = shown, raised
self.minimum, self.maximum = minimum, maximum
self.dynamic, self.policy, self.completed = dynamic, policy, completed
#: display() results for the frames after the first, in order.
self.draws = list(draws or [])
#: (checkpoint name, frame number) -> plan, or a callable(screen, cp).
self.checks = checks or {}
self.work = work
self.cycle_complete_at = cycle_complete_at
self.dwell_to = dwell_to
self.frames = 0
def first_frame(self, plan, plugin):
self.calls.append(("first", plan.mode))
return FirstFrame(self.shown, self.raised, False)
def complete_plan(self, plan, plugin):
self.calls.append(("complete",))
if not self.completed:
return None
return ScreenPlan(plan.source, mode=plan.mode, min_duration=self.minimum,
max_duration=self.maximum, dynamic=self.dynamic,
frame_policy=self.policy, preemptible_by=plan.preemptible_by)
def draw(self, screen):
self.frames += 1
self.calls.append(("draw", self.frames))
self.clock.now += self.work
return self.draws.pop(0) if self.draws else True
def after_frame(self, screen):
self.calls.append(("after_frame",))
def tick(self):
self.calls.append(("tick",))
def service(self, screen):
self.calls.append(("service",))
return ("scan", self.frames)
def wait_frame(self, interval, screen):
self.calls.append(("wait", interval))
self.clock.sleep(interval)
return self._scripted("wait", screen)
def check(self, screen, checkpoint, live_scan=None):
self.calls.append(("check", checkpoint.name, live_scan))
return self._scripted(checkpoint.name, screen, checkpoint)
def _scripted(self, name, screen, checkpoint=None):
answer = self.checks.get((name, self.frames))
if callable(answer):
return answer(screen, checkpoint)
return answer
def dwell(self, seconds):
self.calls.append(("dwell", round(seconds, 6)))
self.clock.now = self.dwell_to if self.dwell_to is not None else self.clock.now + seconds
def cycle_complete(self, screen):
self.calls.append(("cycle?",))
return self.cycle_complete_at is not None and self.clock.now >= self.cycle_complete_at
PLAN = ScreenPlan(Source.ROTATION, mode="clock", preemptible_by=SCREEN_PREEMPTERS)
def _run(**kwargs) -> Tuple[Any, Host, Clock]:
clock = Clock()
host = Host(clock, **kwargs)
outcome = ScreenRunner(clock, host).run(PLAN, plugin=object())
return outcome, host, clock
def _names(host, *kinds):
return [c for c in host.calls if c[0] in kinds]
class TestFirstFrame:
def test_no_plugin_is_empty_without_a_dispatch(self):
clock = Clock()
host = Host(clock)
outcome = ScreenRunner(clock, host).run(PLAN, plugin=None)
assert outcome.exit_reason is ExitReason.EMPTY
assert host.calls == []
@pytest.mark.parametrize("raised,reason", [(False, ExitReason.EMPTY),
(True, ExitReason.ERROR)])
def test_nothing_shown(self, raised, reason):
outcome, host, _ = _run(shown=False, raised=raised)
assert outcome.exit_reason is reason
assert host.calls == [("first", "clock")]
def test_an_on_demand_session_with_no_time_left(self):
outcome, host, _ = _run(completed=False)
assert outcome.exit_reason is ExitReason.PREEMPTED
assert outcome.preempted_by is None
assert host.calls == [("first", "clock"), ("complete",)]
class TestStaticLoop:
def test_runs_its_duration_one_frame_a_second(self):
outcome, host, clock = _run(maximum=5.0)
assert outcome.exit_reason is ExitReason.DURATION
assert outcome.elapsed == 5.0
# Frames at 1..4 s; the wait that reaches 5 s ends it before a fifth.
assert host.frames == 4
assert clock.sleeps == [1.0] * 5
# Each frame: wait, tick, draw, follower frame, service, the check.
assert host.calls[2:9] == [("wait", 1.0), ("tick",), ("draw", 1), ("after_frame",),
("service",), ("check", "frame", ("scan", 1)), ("wait", 1.0)]
# A completed loop: the after-loop look reads no notice, then FINAL.
assert host.calls[-2:] == [("check", "after-completed-loop", None), ("check", "final", None)]
def test_preempted_between_frames(self):
by = ScreenPlan(Source.ON_DEMAND, mode="x")
outcome, host, _ = _run(maximum=30.0, checks={("frame", 3): by})
assert outcome.exit_reason is ExitReason.PREEMPTED
assert outcome.preempted_by is by
assert outcome.elapsed == 3.0
# Decided at the service point: no second look, no dwell.
assert host.calls[-1] == ("check", "frame", ("scan", 3))
def test_preempted_by_a_socket_command_in_the_frame_wait(self):
outcome, host, _ = _run(maximum=30.0, checks={("wait", 2): WIFI_PLAN})
assert outcome.exit_reason is ExitReason.PREEMPTED
assert host.calls[-1] == ("wait", 1.0)
def test_display_false_makes_up_the_minimum(self):
outcome, host, _ = _run(maximum=12.0, draws=[True, False])
assert outcome.exit_reason is ExitReason.DISPLAY_FALSE
# The loop ended early: the after-loop look may read a notice, then
# the dwell makes up the rest of the 12 s.
assert _names(host, "check", "dwell")[-4:] == [
("check", "after-loop", None), ("dwell", 10.0),
("check", "after-dwell", None), ("check", "final", None)]
assert outcome.elapsed == 12.0
def test_a_notice_that_cuts_the_dwell_short_ends_the_screen(self):
def notice(screen, checkpoint):
assert checkpoint.notice is NoticeRead.ALWAYS
return WIFI_PLAN if checkpoint.notice_counts else None
outcome, _, _ = _run(maximum=12.0, draws=[False], dwell_to=6.0,
checks={("after-dwell", 1): notice})
assert outcome.exit_reason is ExitReason.PREEMPTED
def test_a_notice_after_a_full_dwell_does_not(self):
def notice(screen, checkpoint):
return WIFI_PLAN if checkpoint.notice_counts else None
outcome, _, _ = _run(maximum=12.0, draws=[False],
checks={("after-dwell", 1): notice})
assert outcome.exit_reason is ExitReason.DISPLAY_FALSE
def test_a_reload_ends_the_loop_but_the_screen_counts(self):
outcome, host, _ = _run(maximum=30.0, checks={("frame", 2): RELOAD_PLAN})
assert outcome.exit_reason is ExitReason.RELOAD
# Not decided at the service point: the after-loop look, the dwell
# (which returns at once while a reload waits) and FINAL follow.
assert ("check", "after-loop", None) in host.calls
assert ("check", "final", None) in host.calls
def test_a_change_after_the_loop_preempts(self):
by = ScreenPlan(Source.ROTATION, mode="weather")
outcome, _, _ = _run(maximum=3.0, checks={("final", 2): by})
assert outcome.exit_reason is ExitReason.PREEMPTED
assert outcome.preempted_by is by
def test_a_dynamic_screen_keeps_going_on_false(self):
outcome, host, _ = _run(minimum=3.0, maximum=8.0, dynamic=True,
draws=[False, False, False])
assert outcome.exit_reason is ExitReason.DURATION
assert host.frames == 7
class TestHighFpsLoop:
def test_paces_to_the_deadline(self):
outcome, host, clock = _run(maximum=0.05, policy=FramePolicy.HIGH_FPS, work=0.003)
assert outcome.exit_reason is ExitReason.DURATION
# 3 ms of drawing leaves 5 ms of an 8 ms frame to sleep.
assert set(clock.sleeps) == {round(HIGH_FPS_INTERVAL - 0.003, 6)}
def test_an_overrun_frame_still_yields(self):
_, _, clock = _run(maximum=0.05, policy=FramePolicy.HIGH_FPS, work=0.02)
assert set(clock.sleeps) == {0.001}
def test_service_before_the_sleep_check_after(self):
_, host, clock = _run(maximum=0.016, policy=FramePolicy.HIGH_FPS)
first = host.calls[2:7]
assert first == [("draw", 1), ("after_frame",), ("tick",), ("service",),
("check", "frame", ("scan", 1))]
assert clock.sleeps[0] == HIGH_FPS_INTERVAL
def test_preempted_after_the_frames_sleep(self):
outcome, _, clock = _run(maximum=30.0, policy=FramePolicy.HIGH_FPS,
checks={("frame", 3): WIFI_PLAN})
assert outcome.exit_reason is ExitReason.PREEMPTED
assert clock.now == pytest.approx(3 * HIGH_FPS_INTERVAL)
def test_display_false_has_no_make_up_dwell(self):
outcome, host, _ = _run(maximum=30.0, policy=FramePolicy.HIGH_FPS,
draws=[True, False])
assert outcome.exit_reason is ExitReason.DISPLAY_FALSE
assert not _names(host, "dwell")
class TestDynamicDuration:
def test_cycle_complete_after_the_minimum_and_grace(self):
outcome, host, _ = _run(minimum=3.0, maximum=20.0, dynamic=True,
cycle_complete_at=1.0)
assert outcome.exit_reason is ExitReason.CYCLE_COMPLETE
# Not asked before minimum + grace (3.5 s): the 1 Hz loop's first
# frame at or past it is the one at 4 s.
assert outcome.elapsed == 4.0
assert DYNAMIC_GRACE == 0.5
def test_capped_at_the_maximum(self):
outcome, host, _ = _run(minimum=3.0, maximum=6.0, dynamic=True)
assert outcome.exit_reason is ExitReason.DURATION
# The end-of-screen log asks the plugin once more, before FINAL.
assert host.calls[-3:] == [("check", "after-completed-loop", None), ("cycle?",),
("check", "final", None)]
def test_high_fps_cycle_complete(self):
outcome, _, _ = _run(minimum=0.02, maximum=1.0, dynamic=True,
policy=FramePolicy.HIGH_FPS, cycle_complete_at=0.0)
assert outcome.exit_reason is ExitReason.CYCLE_COMPLETE
assert outcome.elapsed >= 0.02 + DYNAMIC_GRACE
class TestCheckpoints:
@pytest.mark.parametrize("checkpoint,notice,reload", [
(FRAME, NoticeRead.IF_UNDECIDED, True),
(AFTER_LOOP, NoticeRead.IF_UNDECIDED, False),
(AFTER_COMPLETED_LOOP, NoticeRead.NEVER, False),
(FINAL, NoticeRead.NEVER, False),
])
def test_what_each_service_point_considers(self, checkpoint, notice, reload):
"""A reload counts only between frames; the notice file is read
where the loop read it before stage 3."""
assert isinstance(checkpoint, Checkpoint)
assert (checkpoint.notice, checkpoint.reload) == (notice, reload)
def test_the_screen_carries_the_completed_plan(self):
seen: List[Optional[Screen]] = []
def grab(screen, checkpoint):
seen.append(screen)
_run(maximum=2.0, checks={("frame", 1): grab})
assert seen[0].plan.frame_policy is FramePolicy.STATIC
assert seen[0].mode == "clock"
class TestControllerServicePoint:
"""DisplayController._screen_check: the reads it makes and what it claims,
on a controller built by the run-loop harness."""
@pytest.fixture
def dc(self, tmp_path):
import os
os.environ.setdefault("EMULATOR", "true")
from test._run_loop_harness import FakePlugin, RunLoopHarness
h = RunLoopHarness(tmp_path, horizon=10)
h.add_plugin(FakePlugin("clock", ["clock"], duration=20))
h.add_plugin(FakePlugin("sports", ["sports_live"], duration=20,
live=(0, 100), live_priority=True))
dc = h.controller
dc.current_display_mode = "clock"
dc.current_mode_index = 0
dc.reads = []
def read():
dc.reads.append(1)
return {"message": "AP mode", "expires_at": 1e12}
dc._check_wifi_status_message = read
return dc
@staticmethod
def _screen(mode="clock"):
from src.display_arbiter import ArbiterState, rotation_plan
return Screen(rotation_plan(ArbiterState(current_mode=mode)), plugin=None,
accepts_display_mode=False, start=0.0)
def test_a_pending_notice_is_read_and_ends_the_screen(self, dc):
by = dc._screen_check(self._screen(), FRAME)
assert by.source is Source.WIFI and len(dc.reads) == 1
def test_not_read_once_the_mode_has_moved(self, dc):
dc.current_display_mode = "sports_live"
by = dc._screen_check(self._screen(), FRAME)
assert by.source is Source.ROTATION and dc.reads == []
def test_not_read_while_scheduled_off(self, dc):
dc.is_display_active = False
by = dc._screen_check(self._screen(), FRAME)
assert by.source is Source.SCHEDULED_OFF and dc.reads == []
def test_not_read_during_on_demand(self, dc):
dc.on_demand_active = True
dc.on_demand_schedule_override = True
assert dc._screen_check(self._screen(), FRAME) is None
assert dc.reads == []
def test_a_live_takeover_is_claimed_before_the_notice(self, dc):
by = dc._screen_check(self._screen(), FRAME, live_scan=("sports_live",))
assert by.source is Source.LIVE and dc.reads == []
assert dc.current_display_mode == "sports_live"
assert dc._live_takeover_unshown is True and dc._live_resume_index == 0
def test_after_a_completed_loop_the_notice_is_not_read(self, dc):
assert dc._screen_check(self._screen(), AFTER_COMPLETED_LOOP) is None
assert dc.reads == []
def test_after_a_full_dwell_it_is_read_but_does_not_count(self, dc):
from src.screen_runner import after_dwell
assert dc._screen_check(self._screen(), after_dwell(False)) is None
assert len(dc.reads) == 1
def test_a_reload_counts_only_between_frames(self, dc):
dc._check_wifi_status_message = lambda: None
dc._pending_plugin_reloads = ("pending",)
assert dc._screen_check(self._screen(), FRAME) is RELOAD_PLAN
assert dc._screen_check(self._screen(), AFTER_LOOP) is None
def test_an_on_demand_index_past_a_shortened_list_starts_again(self, dc):
"""_take_plan writes the shown index back, so the session's next
step goes on from the mode actually shown."""
from src.display_arbiter import Arbiter, ArbiterInputs
dc.on_demand_active = True
dc.on_demand_modes = ["clock", "sports_live"]
dc.on_demand_mode_index = 5 # the list shrank under it
inputs = ArbiterInputs(schedule_on=True, on_demand_active=True,
follower_active=False)
plan = dc._take_plan(Arbiter.decide(dc._arbiter_state(), inputs, 0.0))
assert plan.mode == "clock" and dc.on_demand_mode_index == 0
dc._advance_on_demand()
assert dc.current_display_mode == "sports_live"
@pytest.mark.parametrize("policy,report_hold", [(FramePolicy.HIGH_FPS, True),
(FramePolicy.STATIC, False)])
def test_only_the_high_fps_loop_reports_a_held_frame(policy, report_hold):
"""#758's report_hold: the 125 Hz loop times frames held by update();
the 1 Hz loop's frames are a second apart and must not."""
from unittest.mock import MagicMock
from src.display_controller import _ScreenHost
controller = MagicMock()
plan = ScreenPlan(Source.ROTATION, mode="ticker", frame_policy=policy)
plugin = object()
_ScreenHost(controller).draw(Screen(plan, plugin, True, 0.0))
controller._display_once.assert_called_once_with(plugin, "ticker", True,
report_hold=report_hold)
def _rows(tmp_path, horizon, build):
import os
os.environ.setdefault("EMULATOR", "true")
from test._run_loop_harness import RunLoopHarness
h = RunLoopHarness(tmp_path, horizon=horizon)
build(h)
return h.run()["screens"]
class TestThroughRun:
"""Service points that only the full loop reaches, on the harness."""
def test_a_notice_pending_when_a_later_frame_is_empty_ends_the_screen(self, tmp_path):
"""The after-loop look reads the notice when the loop ended early: a
1 Hz screen whose second frame has nothing to show must not sit in
its make-up dwell (which only notices a notice that arrives during
it) while the notice expires."""
from test._run_loop_harness import FakePlugin
def build(h):
h.add_plugin(FakePlugin("flaky", ["flaky"], duration=12, first_frame_only=True))
h.add_plugin(FakePlugin("clock", ["clock"], duration=12))
h.wifi_message(0.5, "Connected to HomeNet", duration=5)
rows = _rows(tmp_path, 30, build)
assert rows[0][1] == "flaky" and rows[0][2] == 1.0
assert rows[1][1] == "<wifi>"
# ... and the mode it cut short comes back, not the next one.
after = next(row for row in rows[1:] if row[1] != "<wifi>")
assert after[1] == "flaky"
def test_vegas_yielding_to_a_follower_shows_a_rotation_screen_first(self, tmp_path):
"""Pins today's behaviour (docs/RUN_LOOP_REDESIGN.md, "may be
wrong"): the interrupt check stops the ticker for a follower, but
the yield path never looks at one, so a rotation screen runs its
full duration before the next pass hands the panel to the leader."""
from test._run_loop_harness import FakePlugin
def build(h):
h.add_plugin(FakePlugin("clock", ["clock"], duration=20))
h.add_plugin(FakePlugin("weather", ["weather"], duration=20))
h.enable_vegas(cycle=30)
h.sync.follower_windows = [(10, 60)]
rows = _rows(tmp_path, 50, build)
assert rows[0][1] == "<vegas>" and rows[0][3] == "vegas-interrupt"
assert 10.0 <= rows[0][0] + rows[0][2] <= 10.2
assert rows[1][1] == "clock" and rows[1][2] == 20.0
assert rows[2][1] == "<follower>"