diff --git a/src/anthias_server/app/static/img/black.png b/src/anthias_server/app/static/img/black.png new file mode 100644 index 00000000..f5aa6161 Binary files /dev/null and b/src/anthias_server/app/static/img/black.png differ diff --git a/src/anthias_viewer/__init__.py b/src/anthias_viewer/__init__.py index 4953fa81..179a6f26 100644 --- a/src/anthias_viewer/__init__.py +++ b/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 diff --git a/src/anthias_viewer/constants.py b/src/anthias_viewer/constants.py index 9ba5730f..621cdd68 100644 --- a/src/anthias_viewer/constants.py +++ b/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 diff --git a/tests/test_viewer.py b/tests/test_viewer.py index aff89aa8..b54f08fd 100644 --- a/tests/test_viewer.py +++ b/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)