Kiwy-Signage/linux/linux_display.py

"""linux_display.py — display power, keep-awake and rotation for Raspberry Pi OS.

Why this module exists
----------------------
On Raspberry Pi OS "Trixie" the desktop session is **Wayland/labwc**, and an
idle blanker is installed and running by default::

    swayidle -w timeout 600 'wlopm --off *' resume 'wlopm --on *'

A signage player must never blank, so that line fights the player for control of
the output. The historical implementation in ``main.py``
(``signal_screen_activity``) shells out to X11 tools that no longer exist on
Trixie (``xdotool``, ``tvservice``, ``ydotool`` are all absent) and uses a
mis-escaped ``wlopm --on \\*`` argument, so it kept the screen awake on none of
the current installs.

This module replaces that logic with the commands Trixie actually provides:

* ``wlopm``        — Wayland output power management (present, works)
* ``vcgencmd display_power`` — Raspberry Pi firmware-level display power
* ``wlr-randr``    — output configuration, used for rotation (present)

All work is best-effort and non-fatal: running the player over SSH, or on a
desktop without a compositor, must never crash or spam the log.

The platform entry point (``linux/run_linux.py``) installs
:func:`linux_screen_activity` onto ``SignagePlayer`` before playback starts.
"""

from __future__ import annotations

import os
import shutil
import subprocess
import time

# ── Tunables ─────────────────────────────────────────────────────────
#: Don't re-issue the keep-awake commands more often than this (seconds).
#: ``signal_screen_activity`` is called on a 20 s Kivy interval; the commands
#: are idempotent but spawning processes on a Pi is not free.
MIN_REASSERT_INTERVAL = 5.0

#: Set to "1" to disable every display command (diagnostics / desktop testing).
DISABLE_ENV_VAR = 'KIWY_DISPLAY_TOOLS_DISABLED'

#: Output name used when the compositor does not report one.
DEFAULT_OUTPUT = 'HDMI-A-1'

_state = {
    'last_awake_at': 0.0,
    'blinker_pids': [],
    'logged_backend': False,
    'warned_no_backend': False,
}


def _log(message, level='info'):
    """Log through Kivy when available, else print. Never raises."""
    try:
        from kivy.logger import Logger

        getattr(Logger, level, Logger.info)(f'[Display] {message}')
    except Exception:
        try:
            if level in ('error', 'warning'):
                print(f'[Display] {message}')
        except Exception:
            pass


# ── Environment detection ────────────────────────────────────────────
def is_wayland():
    """True when a Wayland compositor socket is reachable.

    Deliberately based on the socket, not on ``WAYLAND_DISPLAY``: the variable
    is empty in exactly the launch contexts this module exists to fix (systemd,
    cron, SSH), where the socket is nevertheless present and usable.
    """
    return wayland_socket_path() is not None


def is_x11():
    """True when an X11 display is reachable (and Wayland is not)."""
    return bool(os.environ.get('DISPLAY')) and not is_wayland()


def tools_disabled():
    """Honour the operator escape hatch."""
    return os.environ.get(DISABLE_ENV_VAR, '').strip().lower() in ('1', 'true', 'yes')


# ── Command helper ───────────────────────────────────────────────────
def wayland_socket_path():
    """Absolute path of the Wayland socket, or None."""
    runtime = os.environ.get('XDG_RUNTIME_DIR') or f'/run/user/{os.getuid()}'
    display = os.environ.get('WAYLAND_DISPLAY') or ''
    if display:
        # Absolute names are used as-is; relative names live in XDG_RUNTIME_DIR.
        return display if os.path.isabs(display) else os.path.join(runtime, display)
    candidate = os.path.join(runtime, 'wayland-0')
    return candidate if os.path.exists(candidate) else None


def ensure_session_environment():
    """Fill in the display environment for our child processes.

    **This is the bug that made keep-awake silently useless.** A process started
    by ``systemd``, a cron wrapper or an SSH session has ``XDG_RUNTIME_DIR`` set
    but ``WAYLAND_DISPLAY`` *empty* — the compositor only exports it inside the
    desktop session. ``wlopm`` then fails with::

        ERROR: WAYLAND_DISPLAY is not set.

    and exits 1, so the panel blanks after the idle timeout. Detecting the
    socket and exporting the name (derived from its filename, so an unusual
    ``wayland-1`` still works) makes the tools function regardless of how the
    player was launched.

    Only ever *sets* values that are missing, so an operator-provided value is
    never overridden. Returns a dict of what was changed (for logging).
    """
    changed = {}

    runtime = os.environ.get('XDG_RUNTIME_DIR')
    if not runtime:
        fallback = f'/run/user/{os.getuid()}'
        if os.path.isdir(fallback):
            os.environ['XDG_RUNTIME_DIR'] = fallback
            changed['XDG_RUNTIME_DIR'] = fallback
            runtime = fallback

    if not os.environ.get('WAYLAND_DISPLAY') and runtime:
        try:
            # Prefer wayland-0, else the lowest-numbered socket present.
            candidates = sorted(
                name for name in os.listdir(runtime)
                if name.startswith('wayland-') and not name.endswith('.lock')
            )
            if candidates:
                os.environ['WAYLAND_DISPLAY'] = candidates[0]
                changed['WAYLAND_DISPLAY'] = candidates[0]
        except Exception:
            pass

    return changed


def _run(args, timeout=5.0):
    """Run a command quietly. Returns (returncode, stdout) — never raises.

    ``stderr`` is discarded: ``wlopm``/``wlr-randr`` are chatty about compositor
    details we do not act on, and the player's log is noise-sensitive.
    """
    ensure_session_environment()
    try:
        result = subprocess.run(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            stdin=subprocess.DEVNULL,
            timeout=timeout,
            text=True,
            check=False,
        )
        return result.returncode, (result.stdout or '').strip()
    except FileNotFoundError:
        return 127, ''
    except subprocess.TimeoutExpired:
        _log(f'{args[0]} timed out after {timeout}s', 'debug')
        return 124, ''
    except Exception as exc:  # pragma: no cover - defensive
        _log(f'{args[0]} failed: {exc}', 'debug')
        return 1, ''


def _which(name):
    try:
        return shutil.which(name)
    except Exception:
        return None


# ── Output discovery ─────────────────────────────────────────────────
def list_outputs():
    """Return the connected Wayland output names (best effort).

    Uses ``wlopm`` (no arguments lists ``<name> <on|off>`` per line), falling
    back to ``wlr-randr`` so rotation works even where ``wlopm`` is missing.
    """
    if not is_wayland():
        return []
    code, out = _run(['wlopm'])
    if code == 0 and out:
        names = []
        for line in out.splitlines():
            parts = line.split()
            if parts:
                names.append(parts[0])
        if names:
            return names
    code, out = _run(['wlr-randr'])
    if code == 0 and out:
        names = []
        for line in out.splitlines():
            # Output lines are unindented: "HDMI-A-1 \"...\""
            if line and not line[0].isspace():
                names.append(line.split()[0])
        return names
    return []


def _target_output():
    """The output to address, or ``'*'`` so the compositor expands it.

    ``wlopm`` expands ``'*'`` itself, which is more robust than us guessing the
    connector name (HDMI-A-1 / HDMI-A-2 / DSI-1 differ per board and port).
    """
    return '*'


# ── Keep-awake ───────────────────────────────────────────────────────
def keep_display_awake(force=False):
    """Turn the display back on. Cheap, idempotent, rate-limited.

    Returns True when something was actually issued.
    """
    if tools_disabled():
        return False

    now = time.monotonic()
    if not force and (now - _state['last_awake_at']) < MIN_REASSERT_INTERVAL:
        return False
    _state['last_awake_at'] = now

    issued = False
    failures = []

    if is_wayland() and _which('wlopm'):
        # NOTE: the argument must stay the literal '*' — the shell must not
        # expand it (we pass a list, so no shell is involved) and wlopm does
        # the matching. The old code passed a backslash-escaped '\*' through
        # os.system(), so the compositor matched an output literally named '*'
        # and nothing happened.
        ensure_session_environment()
        code, out = _run(['wlopm', '--on', _target_output()])
        issued = issued or code == 0
        if code != 0:
            failures.append(f'wlopm --on {_target_output()} rc={code} {out}'.strip())
    elif is_x11():
        if _which('xset'):
            code, _ = _run(['xset', 's', 'reset'])
            issued = issued or code == 0
            _run(['xset', 'dpms', 'force', 'on'])
        if _which('xdotool'):
            # Nudge the pointer by a pixel and back — invisible, but enough to
            # reset the X idle counter on compositors without a Wayland path.
            code, _ = _run(['xdotool', 'mousemove_relative', '1', '1'])
            if code == 0:
                _run(['xdotool', 'mousemove_relative', '-1', '-1'])
                issued = True

    # Firmware-level backstop: covers a blanked HDMI signal even when the
    # compositor never reported the output as off.
    if _which('vcgencmd'):
        _run(['vcgencmd', 'display_power', '1'])

    if failures and not _state['warned_no_backend']:
        _state['warned_no_backend'] = True
        _log(
            'keep-awake command failed: ' + '; '.join(failures)
            + f' (WAYLAND_DISPLAY={os.environ.get("WAYLAND_DISPLAY")!r})',
            'warning',
        )
    elif not issued and not _state['warned_no_backend']:
        _state['warned_no_backend'] = True
        _log(
            'No usable display backend found (no Wayland socket and no X11 '
            'display) — keep-awake is inactive.',
            'warning',
        )
    return issued


def neutralise_idle_blanker():
    """Stop the desktop idle blanker from turning the panel off.

    On Raspberry Pi OS Trixie ``~/.config/labwc/autostart`` starts::

        swayidle -w timeout 600 'wlopm --off *' resume 'wlopm --on *'

    That is exactly the behaviour a signage player must override, so the
    ``swayidle`` process is terminated once at start-up and re-checked
    periodically (in case the session restarts it).

    Only processes owned by the current user are touched, and a failure is
    never fatal. Set ``KIWY_DISPLAY_TOOLS_DISABLED=1`` to opt out.
    """
    if tools_disabled() or not _which('swayidle'):
        return []

    try:
        result = subprocess.run(
            ['pgrep', '-x', 'swayidle'],
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
            text=True, timeout=5, check=False,
        )
        pids = [int(p) for p in (result.stdout or '').split() if p.strip().isdigit()]
    except Exception:
        return []

    own_uid = os.getuid()
    killed = []
    for pid in pids:
        try:
            # Only our own processes: swayidle is per-session and killing
            # another user's would be a surprise.
            if os.stat(f'/proc/{pid}').st_uid != own_uid:
                continue
            os.kill(pid, 15)  # SIGTERM
            killed.append(pid)
        except (ProcessLookupError, PermissionError, FileNotFoundError):
            continue
        except Exception:
            continue

    if killed:
        _log(f'Stopped the idle blanker (swayidle pids: {killed}) — the display '
             f'will stay on. Disable with {DISABLE_ENV_VAR}=1.')
    return killed


def _log_backend_once():
    if _state['logged_backend']:
        return
    _state['logged_backend'] = True
    backend = 'wayland' if is_wayland() else ('x11' if is_x11() else 'none')
    _log(f'backend={backend} outputs={list_outputs() or "unknown"} '
         f'WAYLAND_DISPLAY={os.environ.get("WAYLAND_DISPLAY") or "(unset)"} '
         f'wlopm={bool(_which("wlopm"))} vcgencmd={bool(_which("vcgencmd"))}')


# ── SignagePlayer replacement ────────────────────────────────────────
def linux_screen_activity(self, dt):
    """Drop-in replacement for ``SignagePlayer.signal_screen_activity``.

    Bound to a 20 s Kivy interval, so it must be cheap and must never raise.
    """
    try:
        _log_backend_once()
        keep_display_awake()
        # Cheap re-check: the blanker may have been restarted by the session,
        # e.g. after a compositor reload.
        if (time.monotonic() - _state.get('last_blinker_check', 0.0)) > 60.0:
            _state['last_blinker_check'] = time.monotonic()
            neutralise_idle_blanker()
    except Exception as exc:  # pragma: no cover - must never break the Clock
        _log(f'screen activity signal failed (non-fatal): {exc}', 'debug')


# ── Orientation / resolution (Wayland-native) ────────────────────────
#: Logical rotation for each supported orientation value from app_config.json.
ORIENTATION_TRANSFORMS = {
    'landscape': 'normal',
    'portrait': '90',
    'portrait-inverted': '270',
    'landscape-inverted': '180',
}


def apply_orientation(orientation):
    """Rotate the display to match the configured orientation.

    ``Window.size`` cannot rotate a fullscreen Wayland surface, so the rotation
    has to happen at the output level. ``wlr-randr`` is the tool that works on
    labwc; the call is skipped silently when it is unavailable.

    Returns True when a transform was applied.
    """
    if tools_disabled():
        return False

    key = str(orientation or '').strip().lower()
    transform = ORIENTATION_TRANSFORMS.get(key)
    if not transform or transform == 'normal':
        return False

    if not is_wayland() or not _which('wlr-randr'):
        _log(f'Orientation "{orientation}" requested but wlr-randr/Wayland is '
             f'unavailable — leaving the output unrotated.', 'warning')
        return False

    outputs = list_outputs() or [DEFAULT_OUTPUT]
    applied = False
    for name in outputs:
        code, _ = _run(['wlr-randr', '--output', name, '--transform', transform])
        applied = applied or code == 0
    if applied:
        _log(f'Applied orientation "{orientation}" (transform={transform}) to '
             f'{outputs}')
    return applied


def status():
    """Diagnostic snapshot (used by tests and the startup banner)."""
    return {
        'wayland': is_wayland(),
        'wayland_socket': wayland_socket_path(),
        'wayland_display': os.environ.get('WAYLAND_DISPLAY') or '(unset)',
        'x11': is_x11(),
        'disabled': tools_disabled(),
        'outputs': list_outputs(),
        'wlopm': bool(_which('wlopm')),
        'wlr_randr': bool(_which('wlr-randr')),
        'vcgencmd': bool(_which('vcgencmd')),
    }


if __name__ == '__main__':  # pragma: no cover - manual diagnostics
    import json
    import sys

    if len(sys.argv) > 1 and sys.argv[1] == '--awake':
        print('awake issued:', keep_display_awake(force=True))
        print('blinker killed:', neutralise_idle_blanker())
    print(json.dumps(status(), indent=2))