feat(viewer): blank/unblank commands to turn the display off (#3065)

* feat(viewer): add blank/unblank commands to turn the display off

Adds a way to blank the screen on demand, parallel to the existing
next/previous/stop/play viewer commands on the anthias.viewer Redis
channel:

* Wayland boards (x86/pi5/arm64): wlr-randr powers the connector off
  (true DPMS power-off) and back on. _wlr_output_names() gained an
  include_disabled flag so unblank can re-enable a connector that's
  currently Enabled: no.
* eglfs/linuxfb boards (pi2/pi3/pi4): the Qt app owns the DRM master
  and can't be powered off externally, so the asset loop paints a new
  all-black BLACK_SCREEN image instead (same proven loadImage path as
  the standby screen). Backlight stays on; the screen goes black.

blank_display() flips display_blanked + loop_is_stopped from the
subscriber thread and runs the out-of-process wlr-randr call there; the
webview repaint is deferred to the main loop thread (start_loop), which
owns current_browser_url — mirroring the existing rotation-bounce
threading discipline.

Validated live on x86 (wayland): `blank` -> DP-1 Enabled: no, `unblank`
-> Enabled: yes. eglfs black-paint reuses the standby loadImage path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(viewer): address review on display blanking (mypy, log spam, state)

- Import get_skip_event from anthias_viewer.utils in the test instead
  of via the viewer module, which doesn't explicitly re-export it
  (mypy attr-defined under --no-implicit-reexport).
- start_loop: guard the black repaint on current_browser_url so it
  doesn't re-call view_image() — and re-log "Current url ..." at INFO —
  on every 0.1s tick while blanked (Copilot).
- Route stop/play through module-level helpers that set the
  loop_is_stopped global start_loop actually reads; the prior
  setattr(__main__, ...) wrote a dead namespace under
  `python -m anthias_viewer` and never paused the loop. play now
  implies unblank when the display is blanked (Copilot).
- Add tests for stop flag + play-implies-unblank.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* chore(viewer): mark BLACK_SCREEN http URL safe for Sonar (S5332)

http is intentional — the viewer talks to the local anthias-server over
plain HTTP (TLS is the opt-in Caddy sidecar's job), identical to the
existing STANDBY_SCREEN / SPLASH_PAGE_URL. Annotate the line # NOSONAR,
the repo's documented convention for Sonar false positives under
Automatic Analysis (see sonar-project.properties; cf. test_csrf.py).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/anthias_viewer/__init__.py

@@ -19,6 +19,7 @@ import sh as sh
 from anthias_server.settings import LISTEN, PORT, ReplySender, settings
 from anthias_viewer.constants import EMPTY_PL_DELAY as EMPTY_PL_DELAY
 from anthias_viewer.constants import SERVER_WAIT_TIMEOUT as SERVER_WAIT_TIMEOUT
+from anthias_viewer.constants import BLACK_SCREEN as BLACK_SCREEN
 from anthias_viewer.constants import SPLASH_DELAY as SPLASH_DELAY
 from anthias_viewer.constants import SPLASH_PAGE_URL as SPLASH_PAGE_URL
 from anthias_viewer.constants import STANDBY_SCREEN as STANDBY_SCREEN
@@ -72,6 +73,10 @@ current_browser_url: str | None = None
 _webview_supports_set_reload_interval: bool = True
 browser: Any = None
 loop_is_stopped: bool = False
+# Set by the ``blank`` command: pauses the asset loop (via
+# loop_is_stopped) and signals the main thread to paint a black screen.
+# Cleared by ``unblank``. See blank_display()/unblank_display().
+display_blanked: bool = False
 browser_bus: Any = None
 r = connect_to_redis()
 reply_sender = ReplySender(r)
@@ -231,8 +236,13 @@ def _wlr_transform_value(rotation_deg: int) -> str:
     )
 
 
-def _wlr_output_names() -> list[str]:
-    """List *enabled* connector names known to the wlroots compositor.
+def _wlr_output_names(include_disabled: bool = False) -> list[str]:
+    """List connector names known to the wlroots compositor.
+
+    By default only *enabled* outputs are returned. Pass
+    ``include_disabled=True`` to list every connector regardless of its
+    ``Enabled:`` state — the display-power path needs disabled outputs
+    too so it can turn a blanked screen back on.
 
     ``wlr-randr`` with no args prints one block per output. Each block
     looks like::
@@ -282,7 +292,9 @@ def _wlr_output_names() -> list[str]:
             # line, which on modern wlr-randr versions means the
             # output is implicitly enabled, so include it as a
             # conservative default).
-            if current is not None and current_enabled is not False:
+            if current is not None and (
+                include_disabled or current_enabled is not False
+            ):
                 names.append(current)
             current = line.split()[0]
             current_enabled = None
@@ -291,7 +303,9 @@ def _wlr_output_names() -> list[str]:
             if stripped.startswith('Enabled:'):
                 _, _, value = stripped.partition(':')
                 current_enabled = value.strip().lower() == 'yes'
-    if current is not None and current_enabled is not False:
+    if current is not None and (
+        include_disabled or current_enabled is not False
+    ):
         names.append(current)
     return names
 
@@ -355,6 +369,51 @@ def _apply_wlr_transform(rotation_deg: int) -> bool:
     return any_success
 
 
+def _apply_wlr_power(on: bool) -> bool:
+    """Turn every wlroots output on or off via ``wlr-randr``.
+
+    This is the Wayland half of display blanking: the compositor owns
+    the connector, so ``wlr-randr --output X --off`` puts the monitor
+    into DPMS power-off (no signal). No-op / False on non-Wayland boards
+    — eglfs/linuxfb can't be powered off externally (the Qt app holds
+    the DRM master), so the blank path paints a black screen there
+    instead. Toggling an already-on/off output is harmless, so we target
+    every connector including disabled ones (needed to turn a blanked
+    screen back on).
+    """
+    if not _is_wayland_board():
+        return False
+    names = _wlr_output_names(include_disabled=True)
+    if not names:
+        return False
+    flag = '--on' if on else '--off'
+    any_success = False
+    for name in names:
+        try:
+            result = subprocess.run(
+                ['wlr-randr', '--output', name, flag],
+                capture_output=True,
+                text=True,
+                timeout=5,
+                check=False,
+            )
+        except (FileNotFoundError, subprocess.SubprocessError) as exc:
+            logging.warning('wlr-randr %s failed for %s: %s', flag, name, exc)
+            continue
+        if result.returncode == 0:
+            logging.info('Set output %s %s', name, flag)
+            any_success = True
+        else:
+            logging.warning(
+                'wlr-randr %s on %s exited %d: %s',
+                flag,
+                name,
+                result.returncode,
+                (result.stderr or '').strip(),
+            )
+    return any_success
+
+
 def send_current_asset_id_to_server(correlation_id: str | None) -> None:
     if not correlation_id:
         logging.warning(
@@ -382,17 +441,67 @@ def send_current_asset_id_to_server(correlation_id: str | None) -> None:
     )
 
 
+def blank_display() -> None:
+    """Handle the ``blank`` command: darken the screen and pause playback.
+
+    Wayland boards DPMS-off via wlr-randr (true monitor power-off).
+    eglfs/linuxfb boards can't be powered off externally — the Qt app
+    owns the DRM master — so the asset loop paints BLACK_SCREEN there
+    instead (see start_loop). The wlr-randr call is an out-of-process
+    IPC that's safe from this subscriber thread; the webview repaint is
+    deferred to the main loop thread via the ``display_blanked`` flag.
+    """
+    global display_blanked, loop_is_stopped
+    display_blanked = True
+    loop_is_stopped = True
+    if _is_wayland_board():
+        _apply_wlr_power(False)
+    # Wake the main thread out of any in-progress asset sleep so it
+    # reaches the blanked branch (and the black repaint) promptly.
+    get_skip_event().set()
+
+
+def unblank_display() -> None:
+    """Handle the ``unblank`` command: power the display back on and resume."""
+    global display_blanked, loop_is_stopped
+    if _is_wayland_board():
+        _apply_wlr_power(True)
+    display_blanked = False
+    loop_is_stopped = False
+    get_skip_event().set()
+
+
+def stop_playback() -> None:
+    """Handle the ``stop`` command: pause the asset loop on the current
+    frame. Sets the module-level ``loop_is_stopped`` that ``start_loop``
+    actually reads — the previous ``setattr(__main__, ...)`` wrote to a
+    different namespace under ``python -m anthias_viewer`` and never took
+    effect (surfaced by the blank/unblank work; Copilot)."""
+    global loop_is_stopped
+    loop_is_stopped = stop_loop(scheduler)
+
+
+def play_playback() -> None:
+    """Handle the ``play`` command: resume the asset loop. If the display
+    is currently blanked, ``play`` implies ``unblank`` (power the
+    connector back on and clear the black paint) so we never end up
+    stopped-but-looking-live or repaint black on a later ``stop``."""
+    global loop_is_stopped
+    if display_blanked:
+        unblank_display()
+        return
+    loop_is_stopped = play_loop()
+
+
 commands = {
     'next': lambda _: skip_asset(scheduler),
     'previous': lambda _: skip_asset(scheduler, back=True),
     'asset': lambda asset_id: navigate_to_asset(scheduler, asset_id),
     'reload': lambda _: _handle_reload(),
-    'stop': lambda _: setattr(
-        __import__('__main__'), 'loop_is_stopped', stop_loop(scheduler)
-    ),
-    'play': lambda _: setattr(
-        __import__('__main__'), 'loop_is_stopped', play_loop()
-    ),
+    'stop': lambda _: stop_playback(),
+    'play': lambda _: play_playback(),
+    'blank': lambda _: blank_display(),
+    'unblank': lambda _: unblank_display(),
     'unknown': lambda _: command_not_found(),
     'current_asset_id': lambda corr: send_current_asset_id_to_server(corr),
 }
@@ -1408,6 +1517,12 @@ def start_loop() -> None:
     logging.debug('Entering infinite loop.')
     while True:
         if loop_is_stopped:
+            # Paint black once from the main thread (the owner of the
+            # webview / current_browser_url). Guard on the URL so we
+            # don't re-call view_image() — and re-log "Current url ..."
+            # at INFO — on every 0.1s tick while blanked (Copilot).
+            if display_blanked and current_browser_url != BLACK_SCREEN:
+                view_image(BLACK_SCREEN)
             sleep(0.1)
             continue
 

src/anthias_viewer/constants.py

@@ -4,6 +4,14 @@ SPLASH_DELAY = 60  # secs
 EMPTY_PL_DELAY = 5  # secs
 
 STANDBY_SCREEN = f'http://{LISTEN}:{PORT}/static/img/standby.png'
+# Solid-black image shown by the ``blank`` command. On eglfs/linuxfb
+# boards the Qt app owns the DRM master and can't be powered off
+# externally, so painting black is how those screens "blank"; Wayland
+# boards additionally DPMS-off via wlr-randr (see _apply_wlr_power).
+# http is intentional and safe: the viewer reaches the local
+# anthias-server over plain HTTP (TLS is the opt-in Caddy sidecar's
+# job), same as STANDBY_SCREEN / SPLASH_PAGE_URL. NOSONAR (S5332).
+BLACK_SCREEN = f'http://{LISTEN}:{PORT}/static/img/black.png'  # NOSONAR
 SPLASH_PAGE_URL = f'http://{LISTEN}:{PORT}/splash-page'
 
 SERVER_WAIT_TIMEOUT = 60

tests/test_viewer.py

@@ -13,6 +13,7 @@ import pytest
 import anthias_viewer as viewer
 from anthias_server.settings import settings
 from anthias_viewer.scheduling import Scheduler
+from anthias_viewer.utils import get_skip_event
 
 logging.disable(logging.CRITICAL)
 
@@ -2086,3 +2087,168 @@ class TestWaitForWaylandSocket:
         ):
             viewer_fixtures.u._spawn_webview_once(1)
         assert order == ['wait', 'spawn']
+
+
+# --- display blanking (blank / unblank commands) ----------------------
+
+
+@pytest.fixture
+def restore_blank_state() -> Iterator[None]:
+    try:
+        yield
+    finally:
+        viewer.display_blanked = False
+        viewer.loop_is_stopped = False
+        get_skip_event().clear()
+
+
+def test_apply_wlr_power_targets_all_outputs_including_disabled() -> None:
+    """Power-off/on must reach *disabled* outputs too, so unblank can
+    turn a previously blanked (Enabled: no) connector back on."""
+
+    def _fake_run(argv: Any, **kwargs: Any) -> mock.Mock:
+        result = mock.Mock()
+        result.returncode = 0
+        result.stderr = ''
+        result.stdout = (
+            'HDMI-A-1\n  Enabled: yes\nHDMI-A-2\n  Enabled: no\n'
+            if argv == ['wlr-randr']
+            else ''
+        )
+        return result
+
+    with (
+        mock.patch.dict(
+            os.environ,
+            {'DEVICE_TYPE': 'pi5', 'QT_QPA_PLATFORM': 'wayland'},
+            clear=False,
+        ),
+        mock.patch(
+            'anthias_viewer.subprocess.run', side_effect=_fake_run
+        ) as run,
+    ):
+        assert viewer._apply_wlr_power(False) is True
+
+    off_calls = [
+        c for c in run.call_args_list if c.args and '--off' in c.args[0]
+    ]
+    targeted = {c.args[0][c.args[0].index('--output') + 1] for c in off_calls}
+    assert targeted == {'HDMI-A-1', 'HDMI-A-2'}
+
+
+def test_apply_wlr_power_noop_off_wayland() -> None:
+    """eglfs/linuxfb boards can't be powered off via wlr-randr; the call
+    is a no-op there (the blank path paints black instead)."""
+    with (
+        mock.patch.dict(
+            os.environ,
+            {'DEVICE_TYPE': 'pi4-64', 'QT_QPA_PLATFORM': 'eglfs'},
+            clear=False,
+        ),
+        mock.patch('anthias_viewer.subprocess.run') as run,
+    ):
+        assert viewer._apply_wlr_power(False) is False
+    run.assert_not_called()
+
+
+def test_blank_display_powers_off_and_pauses_on_wayland(
+    restore_blank_state: None,
+) -> None:
+    with (
+        mock.patch.dict(
+            os.environ,
+            {'DEVICE_TYPE': 'pi5', 'QT_QPA_PLATFORM': 'wayland'},
+            clear=False,
+        ),
+        mock.patch(
+            'anthias_viewer._apply_wlr_power', return_value=True
+        ) as power,
+    ):
+        viewer.blank_display()
+
+    assert viewer.display_blanked is True
+    assert viewer.loop_is_stopped is True
+    power.assert_called_once_with(False)
+
+
+def test_blank_display_pauses_without_wlr_on_eglfs(
+    restore_blank_state: None,
+) -> None:
+    """On eglfs the screen blanks by the loop painting BLACK_SCREEN, so
+    blank_display() flips the state but must not invoke wlr-randr."""
+    with (
+        mock.patch.dict(
+            os.environ,
+            {'DEVICE_TYPE': 'pi4-64', 'QT_QPA_PLATFORM': 'eglfs'},
+            clear=False,
+        ),
+        mock.patch('anthias_viewer._apply_wlr_power') as power,
+    ):
+        viewer.blank_display()
+
+    assert viewer.display_blanked is True
+    assert viewer.loop_is_stopped is True
+    power.assert_not_called()
+
+
+def test_unblank_display_resumes_and_powers_on(
+    restore_blank_state: None,
+) -> None:
+    viewer.display_blanked = True
+    viewer.loop_is_stopped = True
+
+    with (
+        mock.patch.dict(
+            os.environ,
+            {'DEVICE_TYPE': 'pi5', 'QT_QPA_PLATFORM': 'wayland'},
+            clear=False,
+        ),
+        mock.patch(
+            'anthias_viewer._apply_wlr_power', return_value=True
+        ) as power,
+    ):
+        viewer.unblank_display()
+
+    assert viewer.display_blanked is False
+    assert viewer.loop_is_stopped is False
+    power.assert_called_once_with(True)
+
+
+def test_blank_and_unblank_commands_registered() -> None:
+    assert 'blank' in viewer.commands
+    assert 'unblank' in viewer.commands
+
+
+def test_stop_playback_sets_module_loop_flag(
+    restore_blank_state: None,
+) -> None:
+    """stop must flip the module-level loop_is_stopped that start_loop
+    reads (the old setattr(__main__, ...) wrote a dead namespace)."""
+    viewer.loop_is_stopped = False
+    viewer.stop_playback()
+    assert viewer.loop_is_stopped is True
+
+
+def test_play_unblanks_when_display_blanked(
+    restore_blank_state: None,
+) -> None:
+    """play on a blanked display implies unblank: power back on, clear
+    the blank, resume — never leave display_blanked set."""
+    viewer.display_blanked = True
+    viewer.loop_is_stopped = True
+
+    with (
+        mock.patch.dict(
+            os.environ,
+            {'DEVICE_TYPE': 'pi5', 'QT_QPA_PLATFORM': 'wayland'},
+            clear=False,
+        ),
+        mock.patch(
+            'anthias_viewer._apply_wlr_power', return_value=True
+        ) as power,
+    ):
+        viewer.play_playback()
+
+    assert viewer.display_blanked is False
+    assert viewer.loop_is_stopped is False
+    power.assert_called_once_with(True)