Screen: switch off by itself when nothing is going on

With `general.screen.off_command` set, the monitor switches off once nothing has
played and nobody has done anything for `auto_off_minutes` (15 by default, 0 for
never, editable in parent mode). "Doing anything" is any intent a person could
have caused: a remote key, a web command, or a key press or tap in the kiosk,
which the page reports at most every 30s via POST /api/screen/activity. The first
such intent after an auto-off runs `on_command`, so the remote wakes the screen
too. The auto-off itself is silent; switches a person asks for still blubb.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-03 19:42:59 +02:00
parent 459599d2ba
commit 077c4e819f
16 changed files with 455 additions and 49 deletions

View File

@@ -219,6 +219,7 @@ copied to the device.
| `POST /api/volume` | `{percent}` or `{delta_percent}` |
| `POST /api/remote/press/{digit}` | what the IR remote's number key does - the keyboard's digits |
| `GET` `PUT /api/remote/mapping` | number key -> album (optionally from one `track_index`) or podcast show |
| `POST /api/screen/toggle` `/screen/activity` | the kiosk's Del key; its key presses and taps, holding off the idle screen-off |
| `GET` `PUT /api/settings` | parent mode |
Two decisions worth knowing:

View File

@@ -101,8 +101,15 @@ general:
# device without one and the key does nothing. The command runs on every press, split
# shell-style but not through a shell; it decides itself whether that means on or off
# (ansible's pi_kiosk installs kiosk-screen for this).
#
# With an off_command the screen also switches itself off once nothing has played and
# nobody has touched the remote, the web UI or the kiosk keyboard for auto_off_minutes
# (editable in parent mode; 0 = never). The next press after that runs on_command.
# screen:
# toggle_command: "/usr/local/bin/kiosk-screen toggle"
# off_command: "/usr/local/bin/kiosk-screen off"
# on_command: "/usr/local/bin/kiosk-screen on"
# auto_off_minutes: 15
# timeout: 5.0 # seconds before a hung command is killed
# Home Assistant integration. Omit the whole section to run without MQTT.

View File

@@ -381,6 +381,8 @@ def _build_services(
screen_config,
alsa_device=None if general.audio_simulated else general.alsa_device,
volume=lambda: app.player.volume / 100,
is_playing=lambda: app.player.is_playing,
clock=clock,
)
)

View File

@@ -173,6 +173,14 @@ class ScreenConfig(_Strict):
#: this is ansible's ``kiosk-screen toggle``, which asks X whether the monitor is
#: on and flips it - the backend keeps no screen state of its own.
toggle_command: str
#: Switches the monitor off unconditionally, for the idle auto-off. Without it
#: there is no auto-off.
off_command: str | None = None
#: Switches it back on when someone is about again after an auto-off.
on_command: str | None = None
#: Minutes without music or anyone at the device before ``off_command`` runs.
#: 0 means never. Parent mode edits this.
auto_off_minutes: int = Field(default=15, ge=0)
timeout: float = Field(default=5.0, gt=0)

View File

@@ -54,6 +54,7 @@ __all__ = [
"TouchButtonPressed",
"TouchButtonReleased",
"TrackChanged",
"UserActivity",
"VolumeChangeRequested",
"VolumeChanged",
]
@@ -234,6 +235,12 @@ class ToggleScreenRequested(IntentEvent):
"""Switch the attached monitor off if it is on, on if it is off (``general.screen``)."""
@dataclass(frozen=True, slots=True, kw_only=True)
class UserActivity(IntentEvent):
"""Someone is at the device - a key press or tap in the kiosk browser. Requests
nothing by itself; it keeps the screen from switching off (``ScreenService``)."""
@dataclass(frozen=True, slots=True, kw_only=True)
class LedEffectRequested(IntentEvent):
zone: LedZone

View File

@@ -1,4 +1,5 @@
"""The attached monitor: runs ``general.screen.toggle_command`` on each toggle request.
"""The attached monitor: runs ``general.screen.toggle_command`` on each toggle request,
and switches it off by itself when nothing is going on.
A service rather than a reaction because the command is a subprocess: the bus awaits
async handlers, so running it there would hold every other event up behind ``xset``.
@@ -7,6 +8,13 @@ Here each press gets its own task, and a hung or failing command is only logged.
Each switch is answered with a "blubb" (:mod:`musicmouse.blubb`) on the music's own
ALSA device, at the player's volume. The command's last word of output says which way
it went - ``on`` or ``off``, as ``kiosk-screen`` prints it; anything else gets no sound.
With an ``off_command`` it also switches the monitor off by itself once nothing has
played and nobody has done anything for ``auto_off_minutes`` - silently, since whoever
it would tell may be asleep. "Doing anything" is any intent a person could have caused:
a remote key, a web command, a key press in the kiosk browser (:class:`UserActivity`).
The first such intent after an auto-off runs ``on_command``, so the monitor wakes up
for the remote too, not only for the keyboard X already wakes it for.
"""
from __future__ import annotations
@@ -18,8 +26,15 @@ from collections.abc import Callable
from musicmouse.blubb import blubb_wav
from musicmouse.bus import EventBus
from musicmouse.clock import Clock, RealClock
from musicmouse.config import ScreenConfig
from musicmouse.events import ToggleScreenRequested
from musicmouse.events import (
Event,
IntentEvent,
LedEffectRequested,
PlaybackChanged,
ToggleScreenRequested,
)
_log = logging.getLogger(__name__)
@@ -29,6 +44,10 @@ __all__ = ["ScreenService"]
#: mastered near full scale; a notification sound sits well under it.
_BLUBB_GAIN = 0.35
#: How often the idle auto-off looks at the clock. Coarse on purpose: nobody notices
#: whether a screen went dark after 15 minutes or 15 and a half.
_IDLE_CHECK_INTERVAL = 30.0
class ScreenService:
name = "screen"
@@ -40,40 +59,97 @@ class ScreenService:
*,
alsa_device: str | None,
volume: Callable[[], float],
is_playing: Callable[[], bool] = lambda: False,
clock: Clock | None = None,
) -> None:
self.bus = bus
#: Read afresh on every check: parent mode changes ``auto_off_minutes`` on this
#: very object.
self.config = config
#: Where the blubb plays, or ``None`` for no sound (simulated audio).
self.alsa_device = alsa_device
#: The player's volume as a fraction, read at the moment the sound plays.
self.volume = volume
self.is_playing = is_playing
self._clock = clock or RealClock()
self._last_activity = self._clock.now()
#: The monitor was last switched by the idle auto-off, not by a person.
self._auto_off = False
self._running: asyncio.Task[None] | None = None
async def run(self) -> None:
unsubscribe = self.bus.subscribe(ToggleScreenRequested, self._on_toggle)
unsubscribe_toggle = self.bus.subscribe(ToggleScreenRequested, self._on_toggle)
unsubscribe_all = self.bus.subscribe_all(self._on_event)
try:
await asyncio.Event().wait()
while True:
await self._clock.sleep(_IDLE_CHECK_INTERVAL)
self.check_idle()
finally:
unsubscribe()
unsubscribe_toggle()
unsubscribe_all()
if self._running is not None:
self._running.cancel()
def _on_toggle(self, _event: ToggleScreenRequested) -> None:
# A second press while the first command is still running would read the
# Whichever way the press goes, a person decided it; nothing to wake from.
self._auto_off = False
self._start(self.config.toggle_command, sound=True)
def _on_event(self, event: Event) -> None:
if isinstance(event, PlaybackChanged):
self._last_activity = self._clock.now()
return
# LED intents arrive from Home Assistant automations as well as people, and
# "system" intents are the backend talking to itself.
if (
not isinstance(event, IntentEvent)
or isinstance(event, LedEffectRequested)
or event.source == "system"
):
return
self._last_activity = self._clock.now()
if self._auto_off and not isinstance(event, ToggleScreenRequested):
self._auto_off = False
if self.config.on_command:
self._start(self.config.on_command, sound=True)
def check_idle(self) -> None:
"""Switch the monitor off if it has been idle long enough. Called every
:data:`_IDLE_CHECK_INTERVAL` by :meth:`run`."""
now = self._clock.now()
if self.is_playing():
# The countdown starts when the music stops, not when it started.
self._last_activity = now
return
minutes = self.config.auto_off_minutes
if self._auto_off or not minutes or not self.config.off_command:
return
if now - self._last_activity < minutes * 60:
return
if self._start(self.config.off_command, sound=False):
_log.info("Nothing played for %d minutes; switching the screen off", minutes)
self._auto_off = True
def _start(self, command: str, *, sound: bool) -> bool:
# A second switch while the first command is still running would read the
# monitor's state before the first one changed it, and undo it.
if self._running is not None and not self._running.done():
_log.debug("Screen toggle still running; ignoring this press")
return
self._running = asyncio.create_task(self.toggle(), name="screen-toggle")
_log.debug("Screen command still running; ignoring this one")
return False
self._running = asyncio.create_task(self._run(command, sound=sound), name="screen")
return True
async def toggle(self) -> None:
argv = shlex.split(self.config.toggle_command)
await self._run(self.config.toggle_command, sound=True)
async def _run(self, command: str, *, sound: bool) -> None:
argv = shlex.split(command)
try:
process = await asyncio.create_subprocess_exec(
*argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT
)
except OSError as exc:
_log.warning("Screen toggle %r could not start: %s", self.config.toggle_command, exc)
_log.warning("Screen command %r could not start: %s", command, exc)
return
try:
output, _ = await asyncio.wait_for(process.communicate(), self.config.timeout)
@@ -81,23 +157,18 @@ class ScreenService:
process.kill()
await process.wait()
_log.warning(
"Screen toggle %r took longer than %gs; killed",
self.config.toggle_command,
self.config.timeout,
"Screen command %r took longer than %gs; killed", command, self.config.timeout
)
return
text = output.decode(errors="replace").strip()
if process.returncode != 0:
_log.warning(
"Screen toggle %r failed (exit %d): %s",
self.config.toggle_command,
process.returncode,
text,
"Screen command %r failed (exit %d): %s", command, process.returncode, text
)
return
_log.info("Screen toggled%s", f": {text}" if text else "")
_log.info("Screen switched%s", f": {text}" if text else "")
state = text.split()[-1] if text else ""
if state in ("on", "off"):
if sound and state in ("on", "off"):
await self.blubb(on=state == "on")
async def blubb(self, *, on: bool) -> None:

View File

@@ -32,6 +32,7 @@ from musicmouse.events import (
SeekRequested,
SetVolumeRequested,
ToggleScreenRequested,
UserActivity,
)
from musicmouse.services.web.hub import StateHub
from musicmouse.services.web.remote_settings import read_mapping, write_mapping
@@ -198,6 +199,12 @@ def build_router(
# no-op, like the remote's power key.
return emit(ToggleScreenRequested(source="web"))
@router.post("/screen/activity", status_code=204)
def screen_activity() -> Response:
# The kiosk page on the device reports key presses and taps here, so the
# screen's idle auto-off does not blank it under a child who is browsing.
return emit(UserActivity(source="web"))
@router.post("/seek", status_code=204)
def seek(body: SeekIn) -> Response:
return emit(SeekRequested(position=body.position, source="web"))
@@ -237,8 +244,13 @@ def build_router(
general = app.config.general
# Applied live as well as saved: a parent lowering the ceiling expects the next
# song to be quieter, not the next boot.
for key, value in body.model_dump().items():
values = body.model_dump()
auto_off = values.pop("screen_auto_off_minutes")
for key, value in values.items():
setattr(general, key, value)
if auto_off is not None and general.screen is not None:
# The same object ScreenService reads on every tick.
general.screen.auto_off_minutes = auto_off
app.player.set_volume_limits(general.min_volume, general.max_volume)
await asyncio.to_thread(write_settings, config_path, body)
await hub.broadcast_state()

View File

@@ -241,6 +241,9 @@ class SettingsOut(BaseModel):
initial_volume: int
volume_increment: int
button_leds_brightness: float
#: ``general.screen.auto_off_minutes``, or ``None`` when there is no screen to
#: switch off (no ``screen`` section, or one without an ``off_command``).
screen_auto_off_minutes: int | None
class SettingsIn(BaseModel):
@@ -249,6 +252,8 @@ class SettingsIn(BaseModel):
initial_volume: int = Field(ge=0, le=200)
volume_increment: int = Field(ge=1, le=100)
button_leds_brightness: float = Field(ge=0, le=1)
#: Ignored without a screen section. ``None`` leaves it as it is.
screen_auto_off_minutes: int | None = Field(default=None, ge=0, le=720)
class HaDeviceOut(BaseModel):

View File

@@ -48,12 +48,16 @@ def to_percent(volume: int, general: GeneralConfig) -> int:
def read_settings(general: GeneralConfig) -> SettingsOut:
screen = general.screen
return SettingsOut(
min_volume=general.min_volume,
max_volume=general.max_volume,
initial_volume=general.initial_volume,
volume_increment=general.volume_increment,
button_leds_brightness=general.button_leds_brightness,
screen_auto_off_minutes=(
screen.auto_off_minutes if screen is not None and screen.off_command else None
),
)
@@ -91,6 +95,11 @@ def write_settings(path: Path, settings: SettingsIn) -> None:
"""Patch the settings into ``config.yml`` in place."""
document = load_document(path)
general = document["general"]
for key, value in settings.model_dump().items():
values = settings.model_dump()
# The one setting that lives in a subsection rather than flat under `general`.
auto_off = values.pop("screen_auto_off_minutes")
for key, value in values.items():
general[key] = value
if auto_off is not None and general.get("screen") is not None:
general["screen"]["auto_off_minutes"] = auto_off
atomic_write(path, document)

View File

@@ -114,6 +114,14 @@ def test_a_series_slot_cannot_name_a_song(config_dir: Path) -> None:
assert "track_index" in message
def test_the_screen_switches_off_after_15_minutes_by_default(config_dir: Path) -> None:
data = _config(screen={"toggle_command": "kiosk-screen toggle"})
screen = load_config(write_config(config_dir, data)).general.screen
assert screen is not None
assert screen.auto_off_minutes == 15
assert screen.off_command is None # ...but only once there is a way to do it
def test_ha_device_and_scene_name_is_optional(config_dir: Path) -> None:
data = _config(
ha={

View File

@@ -15,8 +15,18 @@ import pytest
from musicmouse.blubb import RATE, blubb_wav
from musicmouse.bus import EventBus
from musicmouse.clock import FakeClock
from musicmouse.color import ColorRGBW
from musicmouse.config import ScreenConfig
from musicmouse.events import ToggleScreenRequested
from musicmouse.effects import EffectStaticConfig
from musicmouse.events import (
LedEffectRequested,
PlaybackChanged,
PlayRemoteSlotRequested,
ToggleScreenRequested,
UserActivity,
)
from musicmouse.hardware import LedZone
from musicmouse.services.screen import ScreenService
@@ -184,3 +194,157 @@ async def test_no_blubb(
task.cancel()
assert not Path(f"{fake_aplay}.wav").exists()
# ------------------------------------------------------------------- idle auto-off
class _Idle:
"""A started service with a fake clock and an ``off`` and ``on`` command that log."""
def __init__(self, bus: EventBus, tmp_path: Path, **config: object) -> None:
self.bus = bus
self.off_log = tmp_path / "off"
self.on_log = tmp_path / "on"
self.playing = False
self.clock = FakeClock(idle=bus.drain)
settings: dict[str, object] = {
"toggle_command": _python("print('toggled')"),
"off_command": _appending(self.off_log, then="print('off')"),
"on_command": _appending(self.on_log, then="print('on')"),
"auto_off_minutes": 15,
}
settings.update(config)
self.service = ScreenService(
bus,
ScreenConfig.model_validate(settings),
alsa_device=None,
volume=lambda: 0.5,
is_playing=lambda: self.playing,
clock=self.clock,
)
self.task: asyncio.Task[None] | None = None
async def __aenter__(self) -> _Idle:
self.task = asyncio.create_task(self.service.run())
await asyncio.sleep(0)
return self
async def __aexit__(self, *_: object) -> None:
assert self.task is not None
self.task.cancel()
async def wait(self, minutes: float) -> None:
await self.clock.advance(minutes * 60)
await _settle(self.service)
def offs(self) -> int:
return len(self.off_log.read_text().splitlines()) if self.off_log.exists() else 0
def ons(self) -> int:
return len(self.on_log.read_text().splitlines()) if self.on_log.exists() else 0
async def test_the_screen_goes_off_after_the_idle_time(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path) as idle:
await idle.wait(14)
assert idle.offs() == 0
await idle.wait(1)
assert idle.offs() == 1
await idle.wait(60)
assert idle.offs() == 1 # once, not every half minute after
async def test_not_while_music_plays(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path) as idle:
idle.playing = True
await idle.wait(60)
assert idle.offs() == 0
# The countdown starts when the music stops.
idle.playing = False
await bus.emit_and_wait(PlaybackChanged(playing=False, source="player"))
await idle.wait(14)
assert idle.offs() == 0
await idle.wait(1)
assert idle.offs() == 1
@pytest.mark.parametrize(
"event",
[UserActivity(source="web"), PlayRemoteSlotRequested(digit="3", source="lirc")],
)
async def test_activity_restarts_the_countdown(
bus: EventBus, tmp_path: Path, event: UserActivity | PlayRemoteSlotRequested
) -> None:
async with _Idle(bus, tmp_path) as idle:
await idle.wait(10)
await bus.emit_and_wait(event)
await idle.wait(10)
assert idle.offs() == 0
await idle.wait(5)
assert idle.offs() == 1
async def test_automations_do_not_count_as_activity(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path) as idle:
await idle.wait(10)
effect = EffectStaticConfig(ColorRGBW(0, 0, 0, 0))
automation = LedEffectRequested(zone=LedZone.MOUSE, effect=effect, source="mqtt")
await bus.emit_and_wait(automation)
await idle.wait(5)
assert idle.offs() == 1
async def test_someone_coming_back_switches_it_on(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path) as idle:
await idle.wait(15)
assert idle.offs() == 1
await bus.emit_and_wait(UserActivity(source="web"))
await _settle(idle.service)
await bus.emit_and_wait(UserActivity(source="web"))
await _settle(idle.service)
assert idle.ons() == 1 # only the first wakes it
await idle.wait(15)
assert idle.offs() == 2 # and the countdown runs again
async def test_the_power_key_after_an_auto_off_is_left_to_the_toggle(
bus: EventBus, tmp_path: Path
) -> None:
async with _Idle(bus, tmp_path) as idle:
await idle.wait(15)
await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
await _settle(idle.service)
await bus.emit_and_wait(UserActivity(source="web"))
await _settle(idle.service)
assert idle.ons() == 0
async def test_zero_minutes_means_never(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path, auto_off_minutes=0) as idle:
await idle.wait(600)
assert idle.offs() == 0
async def test_no_off_command_means_never(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path, off_command=None) as idle:
await idle.wait(600)
assert idle.offs() == 0
async def test_a_new_idle_time_applies_at_once(bus: EventBus, tmp_path: Path) -> None:
async with _Idle(bus, tmp_path) as idle:
idle.service.config.auto_off_minutes = 5 # what parent mode does
await idle.wait(5)
assert idle.offs() == 1
async def test_the_auto_off_is_silent(bus: EventBus, tmp_path: Path, fake_aplay: Path) -> None:
async with _Idle(bus, tmp_path) as idle:
idle.service.alsa_device = "softvol_effects"
await idle.wait(15)
assert idle.offs() == 1
assert not Path(f"{fake_aplay}.wav").exists()

View File

@@ -8,7 +8,7 @@ from pathlib import Path
import httpx2
import pytest
from musicmouse.config import WebConfig, load_config
from musicmouse.config import ScreenConfig, WebConfig, load_config
from musicmouse.services.web.service import build_app
from musicmouse.services.web.settings import to_device_volume, to_percent
from musicmouse.simulator.harness import Simulation, build_simulation
@@ -99,8 +99,46 @@ def test_a_degenerate_range_reads_as_full(config_dir: Path) -> None:
async def test_settings_expose_only_the_editable_subset(client: httpx2.AsyncClient) -> None:
body = (await client.get("/api/settings")).json()
assert set(body) == set(SETTINGS)
assert set(body) == {*SETTINGS, "screen_auto_off_minutes"}
assert body["max_volume"] == 60
assert body["screen_auto_off_minutes"] is None # no screen to switch off
async def test_the_screen_auto_off_is_read_and_saved_into_its_section(
client: httpx2.AsyncClient, sim: Simulation, config_dir: Path
) -> None:
path = config_dir / "config.yml"
path.write_text(
COMMENTED.replace(
"\nfigures:",
" screen:\n toggle_command: kiosk-screen toggle\n"
" off_command: kiosk-screen off\n\nfigures:",
),
encoding="utf-8",
)
sim.app.config.general.screen = ScreenConfig(
toggle_command="kiosk-screen toggle", off_command="kiosk-screen off"
)
assert (await client.get("/api/settings")).json()["screen_auto_off_minutes"] == 15
response = await client.put("/api/settings", json={**SETTINGS, "screen_auto_off_minutes": 30})
assert response.status_code == 200
assert response.json()["screen_auto_off_minutes"] == 30
assert sim.app.config.general.screen.auto_off_minutes == 30 # live, for ScreenService
reloaded = load_config(path)
assert reloaded.general.screen is not None
assert reloaded.general.screen.auto_off_minutes == 30
assert "screen_auto_off_minutes" not in path.read_text(encoding="utf-8")
async def test_the_screen_auto_off_without_a_screen_is_ignored(
client: httpx2.AsyncClient, config_dir: Path
) -> None:
response = await client.put("/api/settings", json={**SETTINGS, "screen_auto_off_minutes": 30})
assert response.status_code == 200
assert response.json()["screen_auto_off_minutes"] is None
assert "screen" not in (config_dir / "config.yml").read_text(encoding="utf-8")
# -------------------------------------------------------------------------- writing