From 6d788ecdc830911701580cbec3bf09b0ee15a9d9 Mon Sep 17 00:00:00 2001 From: media Date: Thu, 6 Aug 2026 16:46:04 -0500 Subject: [PATCH] Version 2 Player --- README.md | 543 +++++++++++++++++ config.yaml | 22 +- homeassistant_discovery.py | 86 +++ player_service.py | 1056 +++++++++++++++++++++++----------- requirements.txt | 56 -- tests/test_player_service.py | 534 +++++++++++++++++ 6 files changed, 1886 insertions(+), 411 deletions(-) create mode 100644 README.md create mode 100644 homeassistant_discovery.py create mode 100644 tests/test_player_service.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..b8f1607 --- /dev/null +++ b/README.md @@ -0,0 +1,543 @@ +# YTPlayer Video Player + +YTPlayer is an MQTT-controlled `mpv` endpoint intended to run continuously as +the system service `ytplayer-player.service`. A controller publishes JSON +commands to play a file, playlist, or numbered media slot. The player reports +status and telemetry over MQTT and sends all audio into one virtual PipeWire +sink for the separate Audio Router to place on HDMI, analog, or neither. + +## What it does + +- Plays a direct media path or an `.m3u` playlist with `mpv`. +- Maps numbered slots to `/mnt/Media/YTPlayer/slotN/playlist.m3u`. +- Supports play, stop, pause, resume, and pause-toggle commands. +- Allows only one owned `mpv` process at a time. +- Publishes JSON status, a retained flat state, and detailed playback telemetry. +- Reconnects to MQTT automatically without exiting when the broker or network + is unavailable. +- Queues valid commands in order with bounded backpressure and acknowledges + QoS 1 commands only after main-thread execution. +- Cleans up `mpv`, its IPC socket, and its telemetry when playback stops or the + daemon shuts down. +- Isolates malformed commands, MQTT callback failures, and optional Home + Assistant discovery failures so they cannot kill command handling. +- Optionally publishes nine Home Assistant MQTT discovery sensors. Discovery is + enabled by default. +- Sends `mpv` audio only to `pipewire/ytplayer_bus`; it never connects directly + to HDMI or analog. + +## Runtime flow + +```text +Controller + | + | MQTT JSON commands + v +player_service.py ---- mpv video ----> DISPLAY=:0 + | + | pipewire/ytplayer_bus + v +YTPlayerBus monitor ports + | + v +Audio Router ----> exactly one physical YTPlayer destination, or none +``` + +The virtual sink prevents one `mpv` stream from being independently connected +to both physical outputs. Audio Router is the only component allowed to choose +the physical destination. + +## MQTT interface + +The MQTT topic names, payload shapes, QoS levels, retain flags, client ID, and +last-will behavior are infrastructure contracts. Coordinate any change with +the controller and other consumers first. + +### Topics + +| Direction | Default topic | Content | QoS | Retained | +|---|---|---|---:|---:| +| Controller → player | `youtube/player/cmd` | JSON command object | subscribed at 1 | publisher choice; 1 recommended | +| Player → consumers | `youtube/player/status` | JSON state/status | 1 | no | +| Player → consumers | `youtube/player/state` | Flat state string | 0 | yes | +| Player → consumers | `youtube/player/telemetry` | JSON playback telemetry | 0 | no | + +`cmd` and `status` are configured under `topics` in `config.yaml`. Telemetry can +also be overridden with `topics.telemetry`; its listed value is the default. +The flat `state` topic is intentionally fixed for the existing Companion +integration. + +> **Do not retain command messages.** A retained command can execute again when +> the player reconnects and resubscribes. Publish commands with `retain=false` +> (the normal `mosquitto_pub` default). + +The MQTT client ID is `-player`, currently +`ytplayer-player`. It uses a clean session and a 30-second keepalive. + +### Last will + +If the process or connection disappears, the broker publishes this exact +non-retained QoS 1 payload to `youtube/player/status`: + +```json +{"state": "offline"} +``` + +The daemon deliberately does not send an MQTT `DISCONNECT` during normal +service termination, preserving this last-will behavior. + +## Commands + +Every command must be a JSON object with a non-empty, case-sensitive string +`action`. + +### Play a numbered slot + +```json +{ + "action": "play", + "slot": 3, + "label": "Pre-service loop" +} +``` + +This resolves to: + +```text +/mnt/Media/YTPlayer/slot3/playlist.m3u +``` + +The base directory can be overridden for the process with +`YTPLAYER_MEDIA_BASE`. + +### Play an explicit playlist + +```json +{ + "action": "play", + "playlist": "/mnt/Media/YTPlayer/special/playlist.m3u", + "slot": 9, + "title": "Special event" +} +``` + +### Play a direct file + +```json +{ + "action": "play", + "path": "/mnt/Media/YTPlayer/videos/welcome.mp4", + "label": "Welcome" +} +``` + +Path selection priority is `path`, then `playlist`, then the path derived from +`slot`. The selected path is converted to an absolute path and must exist. +Files ending in lowercase `.m3u` are passed to `mpv` as playlists; other paths +are played directly. Relative paths resolve against the service working +directory (`/home/media/Scripts/Video-Player`), so absolute paths are strongly +recommended. Media must be readable by user `media`. + +`label` and `title` are aliases for the optional status label. If both are +present, `label` wins. An optional slot accompanying an explicit path is +included in status and telemetry if it can be converted to an integer. This +command label is separate from telemetry's `title`, which comes from MPV's +media metadata. + +If another owned `mpv` is still running, a new play command does not replace +it. The player reports `busy`, waits five seconds, and then reasserts `playing` +if the original process is still active. Send `stop` before switching media. + +### Playback controls + +```json +{"action": "stop"} +``` + +```json +{"action": "pause"} +``` + +```json +{"action": "resume"} +``` + +```json +{"action": "toggle"} +``` + +- `stop` terminates and reaps `mpv`, clears telemetry, and publishes `idle`. +- `pause` and `resume` set the `mpv` pause property. +- `toggle` cycles the pause property and publishes the flat state `toggled`; + use telemetry's `paused` field for the resulting pause value. +- Pause/resume/toggle status confirms that the request was issued. The current + compatibility behavior does not change that status if the IPC write fails. +- Pause/resume/toggle with nothing playing publish an `error` status. +- Unknown actions are logged and otherwise ignored. + +There are currently no MQTT actions for seeking, volume, next/previous, +fullscreen, loop, shuffle, or network-URL playback. + +### Command-line examples + +```bash +mosquitto_pub -h 10.40.0.2 -t youtube/player/cmd -q 1 \ + -m '{"action":"play","slot":1,"label":"Opening"}' +``` + +```bash +mosquitto_pub -h 10.40.0.2 -t youtube/player/cmd -q 1 \ + -m '{"action":"stop"}' +``` + +```bash +mosquitto_sub -h 10.40.0.2 \ + -t youtube/player/status \ + -t youtube/player/state \ + -t youtube/player/telemetry -v +``` + +## Status messages + +Each status update produces two publications: + +1. A JSON object on `youtube/player/status`. +2. The same state's plain string on retained `youtube/player/state`. + +Example: + +```json +{ + "state": "playing", + "path": "/mnt/Media/YTPlayer/slot3/playlist.m3u", + "slot": 3, + "label": "Pre-service loop" +} +``` + +Possible state strings include: + +| State | Meaning | +|---|---| +| `idle` | Online with no owned playback, or playback has stopped/finished. | +| `playing` | `mpv` was started or playback was resumed. | +| `paused` | A pause command was sent. | +| `toggled` | A pause-toggle command was sent. | +| `busy` | A play command arrived while another `mpv` was active. | +| `error` | Invalid input, missing media, process failure, or invalid control state. | +| `offline` | Broker-generated last will on the JSON status topic only. | + +Depending on the event, the JSON payload may also contain `path`, `slot`, +`label`, `message`, or `reason`. Common terminal reasons are `stop`, `finished`, +`mpv_error`, `sigterm`, and `sigint`. + +Normal stop publishes cleared telemetry and then `idle` with `reason:"stop"`. +Natural completion uses `reason:"finished"`. A nonzero MPV exit first publishes +`error`, then cleared telemetry, then `idle` with `reason:"mpv_error"`. Terminal +idle normally retains the previous slot but omits the previous path. Repeated +stop commands each produce their own cleared telemetry and idle publications. + +On every successful MQTT connection, including a reconnect, the player +subscribes again and publishes an `idle` status with `"message":"player +online"`. This is retained historical controller behavior even if playback is +already active; in that case the current path and slot are included. + +## Telemetry + +While `mpv` is active, the daemon observes its IPC properties and publishes at +the configured interval, defaulting to 0.5 seconds. + +```json +{ + "state": "telemetry", + "slot": 3, + "path": "/mnt/Media/YTPlayer/slot3/playlist.m3u", + "position": 12.4, + "duration": 90.0, + "remaining": 77.6, + "position_fmt": "0:12", + "duration_fmt": "1:30", + "remaining_fmt": "1:17", + "paused": false, + "title": "Welcome", + "filename": "welcome.mp4", + "playlist_pos": 0, + "playlist_count": 4, + "ts": 1785679200.0 +} +``` + +| Field | Meaning | +|---|---| +| `state` | Always `telemetry` on this topic. | +| `slot` | Integer slot supplied by the command, or `null`. | +| `path` | Active file/playlist path. | +| `position` | Current position in seconds. | +| `duration` | Current item duration in seconds. | +| `remaining` | Non-negative `duration - position`. | +| `*_fmt` | `M:SS` or `H:MM:SS` formatted values. | +| `paused` | Current pause property. | +| `title` | `mpv` media title. | +| `filename` | Current playlist entry filename. | +| `playlist_pos` | Zero-based playlist position reported by `mpv`. | +| `playlist_count` | Number of playlist entries. | +| `ts` | Unix publication timestamp. | + +When playback ends or is stopped, one cleared telemetry message is published +with a null path/title/filename, zero time values, `paused:false`, and null +playlist positions. + +## Audio routing + +`mpv` is always launched with the equivalent of: + +```text +--audio-device=pipewire/ytplayer_bus +--audio-client-name=YTPlayer +--audio-channels=stereo +--audio-fallback-to-null=no +``` + +The no-fallback setting is intentional: if the virtual bus is missing, the +player must not silently select a physical default sink. + +The persistent sink definition is maintained in the sibling Audio Router +repository: + +```text +/home/media/Scripts/Audio-Router/pipewire/20-ytplayer-bus.conf +``` + +It is installed at: + +```text +/home/media/.config/pipewire/pipewire-pulse.conf.d/20-ytplayer-bus.conf +``` + +### Current Audio Router vocabulary + +- `mixer`: Scarlett capture carrying sanctuary audio for the live audience. +- `bgmusic`: SnapBGM background music. +- `hdmi`: BMD HDMI into the ATEM, then the MacBook, then back into the mixer. +- `analog`: This computer's 3.5 mm output directly into the mixer. +- `ytplayer`: `YTPlayerBus:monitor_FL/FR`. + +### Current preset behavior + +| Preset | Existing program routes | YTPlayer route | +|---|---|---| +| `mute_all` | none | disconnected | +| `service` | mixer → HDMI | disconnected | +| `announcements` | bgmusic → HDMI and analog | analog only | +| `aux_only` | bgmusic → analog | analog only | +| `video` | bgmusic → HDMI | HDMI only | + +Audio Router rejects any preset that tries to send YTPlayer to HDMI and analog +simultaneously. It also verifies the old exclusive link is gone before making +the new one; a disconnect failure therefore leaves YTPlayer on at most the old +destination instead of bridging both. + +The player does not select an Audio Router preset. A controller must switch the +router separately by publishing a plain mode name, for example: + +```bash +mosquitto_pub -h 10.40.0.2 -t audio/player/mode/set -m video +``` + +The controller should select the desired audio preset before starting the +corresponding video. + +## Configuration + +`config.yaml` is loaded relative to `player_service.py`, not the shell's current +directory. + +```yaml +mqtt: + host: 10.40.0.2 + port: 1883 + username: null + password: null + client_id_prefix: ytplayer + +topics: + cmd: youtube/player/cmd + status: youtube/player/status + # telemetry: youtube/player/telemetry # optional override + +homeassistant: + enabled: true + discovery_prefix: homeassistant + device_id: ytplayer + device_name: YT Player + +player: + display: :0 + xdg_runtime_dir: /run/user/1000 + mpv_binary: /usr/bin/mpv + audio_device: pipewire/ytplayer_bus + audio_client_name: YTPlayer + audio_channels: stereo + # ipc_socket: /tmp/mpvsocket # optional override + # telemetry_interval: 0.5 # minimum 0.1 seconds + # commands_per_tick: 32 # minimum 1 + # command_queue_size: 512 # minimum 1 +``` + +Restart the daemon after changing configuration. + +## Home Assistant discovery + +When enabled, `homeassistant_discovery.py` reuses the player's existing MQTT +client and publishes nine retained QoS 1 sensor configurations: + +- `position` +- `duration` +- `remaining` +- `position_fmt` +- `duration_fmt` +- `remaining_fmt` +- `title` +- `paused` +- `slot` + +Their default topic pattern is: + +```text +homeassistant/sensor/ytplayer//config +``` + +To disable discovery without affecting any player MQTT topic or command: + +```yaml +homeassistant: + enabled: false +``` + +Discovery is deliberately isolated: an import or publish failure is logged but +does not stop the player from subscribing to commands. Disabling discovery +stops future discovery publications; it does not erase the nine configurations +already retained by the broker. + +## Service operation + +The installed system unit is: + +```text +/etc/systemd/system/ytplayer-player.service +``` + +It runs `player_service.py` as user/group `media`, with `DISPLAY=:0`, +`XDG_RUNTIME_DIR=/run/user/1000`, and `/home/media/.Xauthority`. It is configured +with `Restart=always`. + +```bash +sudo systemctl start ytplayer-player.service +sudo systemctl stop ytplayer-player.service +sudo systemctl restart ytplayer-player.service +systemctl status ytplayer-player.service --no-pager -l +``` + +If repeated historical failures tripped systemd's start limit: + +```bash +sudo systemctl reset-failed ytplayer-player.service +sudo systemctl start ytplayer-player.service +``` + +Follow logs with: + +```bash +journalctl -u ytplayer-player.service -f +``` + +## Dependencies and tests + +Python dependencies are intentionally limited to: + +- `paho-mqtt==2.1.0` +- `PyYAML==6.0` + +Runtime also requires `mpv`, PipeWire/PipeWire Pulse, the `ytplayer_bus` null +sink, an X display, and the sibling Audio Router for physical audio routing. + +Run the regression suite with the same interpreter used by the service: + +```bash +cd /home/media/Scripts/Video-Player +/usr/bin/python3 -m unittest discover -s tests -v +``` + +The suite covers MQTT topic/QoS/retain compatibility, Home Assistant discovery, +malformed input isolation, ordered QoS acknowledgements, reconnect generations, +MPV ownership and cleanup, stale playback generations, virtual-sink pinning, +and historical status sequences. + +## Troubleshooting + +### Service exits during boot or after a network outage + +The historical crashes were synchronous MQTT startup failures: a boot-time +`OSError: [Errno 101] Network is unreachable` exited the process repeatedly, +after which systemd latched the unit at its start limit. The current daemon uses +asynchronous MQTT connection and 1–30 second reconnect backoff, so a temporarily +unreachable broker should leave the process active and retrying. Use +`reset-failed` once if systemd is still latched from those older failures. + +### Video starts but there is no audio + +Confirm MPV can see the virtual device: + +```bash +XDG_RUNTIME_DIR=/run/user/1000 mpv --audio-device=help +``` + +Confirm the sink and JACK ports exist: + +```bash +pactl list short sinks +pw-jack jack_lsp YTPlayerBus +``` + +Then check the current Audio Router mode: + +```bash +mosquitto_sub -h 10.40.0.2 -t audio/player/mode/state -C 1 +``` + +`service` and `mute_all` intentionally disconnect YTPlayer. `video` sends it to +HDMI; `announcements` and `aux_only` send it to analog. + +### Suspected doubled YTPlayer audio + +Inspect only the YTPlayer links: + +```bash +pw-jack jack_lsp -c YTPlayerBus +``` + +Each `YTPlayerBus:monitor_*` port must have at most one physical destination. +Do not add direct MPV-to-HDMI or MPV-to-analog links back into this service. + +### Commands arrive but playback does not change + +- Verify the payload is a JSON object containing a string `action`. +- Check that the requested path exists for user `media`. +- Stop the current item before sending another play command. +- Watch `youtube/player/status` for `busy` or `error` details. +- Inspect the journal for IPC, display, or process errors. + +### Display errors + +Verify display `:0`, `/run/user/1000`, and `/home/media/.Xauthority` are valid +for the `media` account and that the graphical session is running. + +## Repository files + +| File | Purpose | +|---|---| +| `player_service.py` | Daemon, MQTT contract, MPV lifecycle, IPC, and telemetry. | +| `config.yaml` | Broker, topic, discovery, display, and audio-device settings. | +| `homeassistant_discovery.py` | Optional discovery publications using the existing MQTT client. | +| `requirements.txt` | Minimal Python dependencies. | +| `tests/test_player_service.py` | Regression and infrastructure-contract tests. | diff --git a/config.yaml b/config.yaml index 4584aad..5cdcf91 100644 --- a/config.yaml +++ b/config.yaml @@ -9,16 +9,20 @@ topics: cmd: "youtube/player/cmd" status: "youtube/player/status" +# Optional Home Assistant discovery. The player MQTT interface is unchanged; +# disabling this only skips the nine retained discovery config publications. +homeassistant: + enabled: true + discovery_prefix: "homeassistant" + device_id: "ytplayer" + device_name: "YT Player" + player: display: ":0" xdg_runtime_dir: "/run/user/1000" mpv_binary: "/usr/bin/mpv" - -video_audio: - hdmi: - left: "Built-in Audio Pro 3:playback_AUX0" - right: "Built-in Audio Pro 3:playback_AUX1" - - analog: - left: "Built-in Audio Pro:playback_AUX0" - right: "Built-in Audio Pro:playback_AUX1" \ No newline at end of file + # MPV always targets this virtual PipeWire sink. Audio Router exclusively + # decides whether its monitor is connected to HDMI, analog, or neither. + audio_device: "pipewire/ytplayer_bus" + audio_client_name: "YTPlayer" + audio_channels: "stereo" diff --git a/homeassistant_discovery.py b/homeassistant_discovery.py new file mode 100644 index 0000000..24cc6c0 --- /dev/null +++ b/homeassistant_discovery.py @@ -0,0 +1,86 @@ +"""Home Assistant MQTT discovery publishing for the video player. + +This module deliberately reuses the player's existing MQTT client. It does +not create another connection or change any of the player's MQTT topics. +""" + +import json +from typing import Any, Dict + + +def _publish_sensor( + client: Any, + *, + discovery_prefix: str, + device_id: str, + device_name: str, + telemetry_topic: str, + status_topic: str, + entity_id: str, + name: str, + value_template: str, + icon: str = "", + unit: str = "", +) -> None: + topic = f"{discovery_prefix}/sensor/{device_id}/{entity_id}/config" + payload: Dict[str, Any] = { + "name": name, + "unique_id": f"{device_id}_{entity_id}", + "state_topic": telemetry_topic, + "value_template": value_template, + "availability_topic": status_topic, + "availability_template": "{% if value_json.state != 'offline' %}online{% else %}offline{% endif %}", + "payload_available": "online", + "payload_not_available": "offline", + "device": { + "identifiers": [device_id], + "name": device_name, + "manufacturer": "Custom", + "model": "mpv mqtt player", + }, + } + + if icon: + payload["icon"] = icon + if unit: + payload["unit_of_measurement"] = unit + + client.publish(topic, json.dumps(payload), qos=1, retain=True) + + +def publish_discovery( + client: Any, + *, + discovery_prefix: str, + device_id: str, + device_name: str, + telemetry_topic: str, + status_topic: str, +) -> None: + """Publish the same nine retained discovery sensors used historically.""" + sensors = ( + ("position", "YT Position (s)", "{{ value_json.position }}", "mdi:timeline-clock", "s"), + ("duration", "YT Duration (s)", "{{ value_json.duration }}", "mdi:timer", "s"), + ("remaining", "YT Remaining (s)", "{{ value_json.remaining }}", "mdi:timer-sand", "s"), + ("position_fmt", "YT Position", "{{ value_json.position_fmt }}", "mdi:timeline-clock", ""), + ("duration_fmt", "YT Duration", "{{ value_json.duration_fmt }}", "mdi:timer", ""), + ("remaining_fmt", "YT Remaining", "{{ value_json.remaining_fmt }}", "mdi:timer-sand", ""), + ("title", "YT Title", "{{ value_json.title }}", "mdi:music", ""), + ("paused", "YT Paused", "{{ value_json.paused }}", "mdi:pause-circle", ""), + ("slot", "YT Slot", "{{ value_json.slot }}", "mdi:playlist-play", ""), + ) + + for entity_id, name, template, icon, unit in sensors: + _publish_sensor( + client, + discovery_prefix=discovery_prefix, + device_id=device_id, + device_name=device_name, + telemetry_topic=telemetry_topic, + status_topic=status_topic, + entity_id=entity_id, + name=name, + value_template=template, + icon=icon, + unit=unit, + ) diff --git a/player_service.py b/player_service.py index b6419ff..eb0f25a 100644 --- a/player_service.py +++ b/player_service.py @@ -1,17 +1,20 @@ #!/usr/bin/env python3 import json +import math import socket import os +import queue import signal import subprocess import sys import time +import traceback +from dataclasses import dataclass from pathlib import Path from typing import Optional, Dict, Any import threading import paho.mqtt.client as mqtt import yaml -import jack # ---------------------------- # Config + paths @@ -30,9 +33,23 @@ MQTT_USER = config["mqtt"].get("username") MQTT_PASS = config["mqtt"].get("password") CLIENT_ID_PREFIX = config["mqtt"].get("client_id_prefix", "ytplayer") -HA_DISCOVERY_PREFIX = config.get("homeassistant", {}).get("discovery_prefix", "homeassistant") -HA_DEVICE_ID = config.get("homeassistant", {}).get("device_id", "ytplayer") -HA_DEVICE_NAME = config.get("homeassistant", {}).get("device_name", "YT Player") +HA_CONFIG = config.get("homeassistant", {}) +HA_DISCOVERY_ENABLED = HA_CONFIG.get("enabled", True) +HA_DISCOVERY_PREFIX = HA_CONFIG.get("discovery_prefix", "homeassistant") +HA_DEVICE_ID = HA_CONFIG.get("device_id", "ytplayer") +HA_DEVICE_NAME = HA_CONFIG.get("device_name", "YT Player") + +HA_DISCOVERY_IMPORT_ERROR: Optional[Exception] = None +if HA_DISCOVERY_ENABLED: + try: + from homeassistant_discovery import publish_discovery + except Exception as exc: + # Discovery is optional. A broken/missing helper must not prevent the + # core player and its established MQTT interface from starting. + publish_discovery = None + HA_DISCOVERY_IMPORT_ERROR = exc +else: + publish_discovery = None TOPIC_CMD = config["topics"]["cmd"] # e.g. "youtube/player/cmd" TOPIC_STATUS = config["topics"]["status"] # e.g. "youtube/player/status" @@ -44,25 +61,44 @@ XDG_RUNTIME_DIR = config["player"].get("xdg_runtime_dir", "/run/user/1000") TELEMETRY_TOPIC = config.get("topics", {}).get("telemetry", "youtube/player/telemetry") MPV_SOCKET = config["player"].get("ipc_socket", "/tmp/mpvsocket") -TELEMETRY_INTERVAL = float(config["player"].get("telemetry_interval", 0.5)) # seconds - -VIDEO_AUDIO_CONNECT_DELAY = float(config["player"].get("audio_connect_delay", 1.0)) - -VIDEO_HDMI_L = config["video_audio"]["hdmi"]["left"] -VIDEO_HDMI_R = config["video_audio"]["hdmi"]["right"] -VIDEO_ANALOG_L = config["video_audio"]["analog"]["left"] -VIDEO_ANALOG_R = config["video_audio"]["analog"]["right"] +TELEMETRY_INTERVAL = max(0.1, float(config["player"].get("telemetry_interval", 0.5))) +COMMANDS_PER_TICK = max(1, int(config["player"].get("commands_per_tick", 32))) +COMMAND_QUEUE_SIZE = max(1, int(config["player"].get("command_queue_size", 512))) +MPV_AUDIO_DEVICE = config["player"].get("audio_device", "pipewire/ytplayer_bus") +MPV_AUDIO_CLIENT_NAME = config["player"].get("audio_client_name", "YTPlayer") +MPV_AUDIO_CHANNELS = config["player"].get("audio_channels", "stereo") # ---------------------------- # Runtime state # ---------------------------- -mpv_proc: Optional[subprocess.Popen] = None -current_path: Optional[str] = None -current_slot: Optional[int] = None +@dataclass +class PlaybackContext: + generation: int + process: subprocess.Popen + path: str + slot: Optional[int] + stop_event: threading.Event + listener_thread: Optional[threading.Thread] = None + + +@dataclass(frozen=True) +class QueuedCommand: + data: Dict[str, Any] + message_id: Optional[int] = None + qos: int = 0 + connection_generation: Optional[int] = None + + +current_playback: Optional[PlaybackContext] = None +playback_generation = 0 busy_until: Optional[float] = None -mpv_listener_thread: Optional["threading.Thread"] = None -mpv_listener_stop = None # will become threading.Event() +state_lock = threading.RLock() +shutdown_event = threading.Event() +mqtt_generation_lock = threading.Lock() +mqtt_connection_generation = 0 + +command_queue: "queue.Queue[QueuedCommand]" = queue.Queue(maxsize=COMMAND_QUEUE_SIZE) client = mqtt.Client( mqtt.CallbackAPIVersion.VERSION2, @@ -70,109 +106,190 @@ client = mqtt.Client( clean_session=True, ) + +def log(message: str, *, error: bool = False) -> None: + """Write each record immediately so journald timestamps are accurate.""" + print(message, file=sys.stderr if error else sys.stdout, flush=True) + + +def log_exception(message: str) -> None: + log(message, error=True) + traceback.print_exc(file=sys.stderr) + # ---------------------------- # MQTT helpers # ---------------------------- -def publish_status(state: str, **extra: Any) -> None: +_USE_CURRENT_PLAYBACK = object() + + +def get_current_playback() -> Optional[PlaybackContext]: + with state_lock: + return current_playback + + +def playback_is_current(playback: PlaybackContext) -> bool: + with state_lock: + return current_playback is playback and not playback.stop_event.is_set() + + +def publish_status( + state: str, + *, + playback: Any = _USE_CURRENT_PLAYBACK, + include_path: bool = True, + **extra: Any, +) -> None: """ Publish: - JSON status (for HA / controller) to TOPIC_STATUS - Flat string (for Companion color mapping) to TOPIC_STATE """ + if playback is _USE_CURRENT_PLAYBACK: + playback = get_current_playback() + payload: Dict[str, Any] = {"state": state} - if current_path: - payload["path"] = current_path - if current_slot is not None: - payload["slot"] = current_slot + if playback is not None: + if include_path and playback.path: + payload["path"] = playback.path + if playback.slot is not None: + payload["slot"] = playback.slot payload.update(extra) - # JSON status - client.publish(TOPIC_STATUS, json.dumps(payload), qos=1, retain=False) - - # Flat state string - client.publish(TOPIC_STATE, state, qos=0, retain=True) - -def ha_discovery_sensor( - entity_id: str, - name: str, - value_template: str, - icon: str = "", - unit: str = "", -) -> None: - topic = f"{HA_DISCOVERY_PREFIX}/sensor/{HA_DEVICE_ID}/{entity_id}/config" - - payload: Dict[str, Any] = { - "name": name, - "unique_id": f"{HA_DEVICE_ID}_{entity_id}", - "state_topic": TELEMETRY_TOPIC, - "value_template": value_template, - - "availability_topic": TOPIC_STATUS, - "availability_template": "{% if value_json.state != 'offline' %}online{% else %}offline{% endif %}", - "payload_available": "online", - "payload_not_available": "offline", - - "device": { - "identifiers": [HA_DEVICE_ID], - "name": HA_DEVICE_NAME, - "manufacturer": "Custom", - "model": "mpv mqtt player", - }, - } - - - if icon: - payload["icon"] = icon - if unit: - payload["unit_of_measurement"] = unit - - client.publish(topic, json.dumps(payload), qos=1, retain=True) + try: + # These topic names, QoS levels, retain flags, and payload shapes are + # part of the existing controller contract. Do not change them here. + client.publish(TOPIC_STATUS, json.dumps(payload), qos=1, retain=False) + client.publish(TOPIC_STATE, state, qos=0, retain=True) + except Exception: + log_exception(f"[player] failed to publish status {state!r}") # ---------------------------- # MPV control # ---------------------------- -def stop_mpv(reason: str = "stop") -> None: +def _remove_mpv_socket() -> None: + try: + os.unlink(MPV_SOCKET) + except FileNotFoundError: + pass + except OSError as exc: + log(f"[player] could not remove stale MPV socket {MPV_SOCKET}: {exc}", error=True) + + +def _join_playback_workers(playback: PlaybackContext) -> None: + for worker in (playback.listener_thread,): + if worker is None or worker is threading.current_thread(): + continue + try: + worker.join(timeout=1.5) + except RuntimeError: + # Thread construction succeeded but start() did not. + continue + if worker.is_alive(): + log( + f"[player] worker {worker.name!r} did not stop promptly " + f"for generation {playback.generation}", + error=True, + ) + + +def stop_mpv( + reason: str = "stop", + *, + expected: Optional[PlaybackContext] = None, + final_state: str = "idle", + message: Optional[str] = None, + publish_updates: bool = True, +) -> bool: """ Stop mpv if running, reset state, publish idle. Important: we keep the slot around long enough so the controller can see which slot just went idle/finished. """ - global mpv_proc, current_path, current_slot - global mpv_listener_thread, mpv_listener_stop - if mpv_listener_stop is not None: - mpv_listener_stop.set() - mpv_listener_thread = None - mpv_listener_stop = None + global current_playback, busy_until - slot_before = current_slot + with state_lock: + playback = current_playback + if expected is not None and playback is not expected: + return False + busy_until = None + if playback is not None: + playback.stop_event.set() - if mpv_proc and mpv_proc.poll() is None: + stop_error: Optional[str] = None + if playback is not None and playback.process.poll() is None: try: - mpv_proc.terminate() + playback.process.terminate() try: - mpv_proc.wait(timeout=5) + playback.process.wait(timeout=5) except subprocess.TimeoutExpired: - mpv_proc.kill() - except Exception as e: - # still publish error, with whatever slot we had - current_slot = slot_before - publish_status("error", message=f"Failed to stop mpv: {e}") - current_slot = None + playback.process.kill() + playback.process.wait(timeout=2) + except Exception as exc: + stop_error = f"Failed to stop mpv: {exc}" + log(f"[player] {stop_error}", error=True) - mpv_proc = None - current_path = None - # clear telemetry so HA doesn't keep stale values - publish_cleared_telemetry() - # temporarily restore slot so status includes it - current_slot = slot_before - publish_status("idle", reason=reason) - current_slot = None + # terminate() itself can fail. Make a separate best-effort kill so + # the daemon never abandons a live child and starts a duplicate. + if playback.process.poll() is None: + try: + playback.process.kill() + playback.process.wait(timeout=2) + except Exception as kill_exc: + log( + f"[player] Failed to kill mpv after stop error: {kill_exc}", + error=True, + ) + + if playback is not None and playback.process.poll() is None: + # Keep ownership of a child that could not be stopped. A later stop or + # systemd's cgroup cleanup can retry; handle_play will remain busy and + # cannot create a second MPV instance. + if publish_updates: + publish_status( + "error", + playback=playback, + message=stop_error or "Failed to stop mpv; process is still running", + ) + return False + + with state_lock: + if current_playback is playback: + current_playback = None + + if playback is not None: + _join_playback_workers(playback) + _remove_mpv_socket() + + if not publish_updates: + return True + + if stop_error is not None: + # Preserve the historical error-then-idle state sequence for existing + # MQTT consumers while still completing cleanup. + publish_status("error", playback=playback, message=stop_error) + + # Keep the previous slot in the clear/terminal messages, matching the + # existing controller contract, but do not carry over the old path. + publish_cleared_telemetry( + playback=None if stop_error is not None else playback, + ) + + extra: Dict[str, Any] = {"reason": reason} + if message is not None: + extra["message"] = message + publish_status( + final_state, + playback=playback, + include_path=final_state == "error", + **extra, + ) + return True -def handle_play(cmd): +def handle_play(cmd: Dict[str, Any]) -> None: """ cmd is a dict, expected keys: - action: "play" @@ -181,17 +298,41 @@ def handle_play(cmd): - slot: optional int (1..N) – controller usually sends this - label/title: optional """ - global mpv_proc, current_path, current_slot, busy_until + global current_playback, playback_generation, busy_until - print(f"[player] handle_play cmd={cmd!r}") + log(f"[player] handle_play cmd={cmd!r}") path = cmd.get("path") playlist = cmd.get("playlist") slot = cmd.get("slot") label = cmd.get("label") or cmd.get("title") - # Decide what we’re actually going to play - final_path = None + if path is not None and not isinstance(path, str): + log(f"[player] bad path value: {path!r}", error=True) + publish_status("error", message="play command path must be a string") + return + if playlist is not None and not isinstance(playlist, str): + log(f"[player] bad playlist value: {playlist!r}", error=True) + publish_status("error", message="play command playlist must be a string") + return + + slot_num: Optional[int] = None + if slot is not None: + try: + if isinstance(slot, bool): + raise ValueError("boolean slots are not valid") + slot_num = int(slot) + except (TypeError, ValueError, OverflowError): + if path or playlist: + log(f"[player] ignoring bad optional slot value: {slot!r}", error=True) + slot_num = None + else: + log(f"[player] bad slot value: {slot!r}", error=True) + publish_status("error", message=f"bad slot value: {slot!r}") + return + + # Decide what we're actually going to play. + final_path: Optional[str] = None # 1) explicit path wins if path: @@ -200,37 +341,47 @@ def handle_play(cmd): elif playlist: final_path = playlist # 3) derive from slot: /mnt/Media/YTPlayer/slotN/playlist.m3u - elif slot is not None: - try: - slot_num = int(slot) - except (TypeError, ValueError): - print(f"[player] bad slot value: {slot!r}") - publish_status("error", message=f"bad slot value: {slot!r}") - return + elif slot_num is not None: final_path = f"{MEDIA_BASE}/slot{slot_num}/playlist.m3u" if not final_path: - print("[player] no valid 'path', 'playlist' or 'slot' supplied") + log("[player] no valid 'path', 'playlist' or 'slot' supplied", error=True) publish_status("error", message="play command missing path/playlist/slot") return final_path = os.path.abspath(final_path) if not os.path.exists(final_path): - print(f"[player] path not found: {final_path}") + log(f"[player] path not found: {final_path}", error=True) publish_status("error", message=f"path not found: {final_path}") return # enforce “must be stopped before new playlist” - if mpv_proc and mpv_proc.poll() is None: - print("[player] already playing, sending busy") + existing = get_current_playback() + if existing is not None and existing.process.poll() is None: + log("[player] already playing, sending busy") publish_status("busy", message="mpv already playing; stop first") - busy_until = time.time() + 5 + busy_until = time.monotonic() + 5 return + if existing is not None: + stop_mpv( + reason="stale_process", + expected=existing, + publish_updates=False, + ) + args = [ MPV_BIN, f"--input-ipc-server={MPV_SOCKET}", - "--audio-client-name=MPV", + f"--audio-device={MPV_AUDIO_DEVICE}", + f"--audio-client-name={MPV_AUDIO_CLIENT_NAME}", + f"--audio-channels={MPV_AUDIO_CHANNELS}", + # Never fall back to a physical default sink. Audio Router is the sole + # owner of the HDMI/analog choice. + "--audio-fallback-to-null=no", + # Keep warnings/errors but suppress the continuously rewritten AV + # progress line that journald stores as very large blob records. + "--quiet", ] if final_path.endswith(".m3u"): # mpv wants --playlist=/path/to/file, not "--playlist" "/path" @@ -242,61 +393,108 @@ def handle_play(cmd): env["DISPLAY"] = DISPLAY env["XDG_RUNTIME_DIR"] = XDG_RUNTIME_DIR + _remove_mpv_socket() + process: Optional[subprocess.Popen] = None + playback: Optional[PlaybackContext] = None try: - print(f"[player] starting mpv: {args!r}") - mpv_proc = subprocess.Popen( + log(f"[player] starting mpv: {args!r}") + process = subprocess.Popen( args, env=env, stdout=None, stderr=None, ) - threading.Thread(target=connect_mpv_audio, daemon=True).start() - current_path = final_path - try: - current_slot_int = int(slot) if slot is not None else None - except (TypeError, ValueError): - current_slot_int = None - current_slot = current_slot_int - publish_status("playing", label=label) - # start telemetry observer thread - global mpv_listener_thread, mpv_listener_stop - mpv_listener_stop = threading.Event() - mpv_listener_thread = threading.Thread(target=mpv_observe_properties, daemon=True) - mpv_listener_thread.start() - except Exception as e: - print(f"[player] failed to start mpv: {e}") - mpv_proc = None - current_path = None - current_slot = None - publish_status("error", message=f"failed to start mpv: {e}") + with state_lock: + playback_generation += 1 + playback = PlaybackContext( + generation=playback_generation, + process=process, + path=final_path, + slot=slot_num, + stop_event=threading.Event(), + ) + # Establish ownership before constructing any worker. If worker + # setup raises, stop_mpv can still reap this exact child. + current_playback = playback -def mpv_ipc(command): - """Send a JSON IPC command to MPV via MPV_SOCKET.""" - try: - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - # Wait for socket to appear and accept connections - deadline = time.time() + 3.0 - while True: + playback.listener_thread = threading.Thread( + target=mpv_observe_properties, + args=(playback,), + name=f"mpv-telemetry-{playback.generation}", + daemon=True, + ) + except Exception as exc: + log(f"[player] failed to initialize mpv playback: {exc}", error=True) + if playback is not None: + stop_mpv( + reason="startup_error", + expected=playback, + final_state="error", + message=f"failed to start mpv: {exc}", + ) + elif process is not None: + # Context construction itself failed after Popen. Reap the local + # child directly because it was never safe to publish as current. try: + process.terminate() + try: + process.wait(timeout=2) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=2) + except Exception: + try: + if process.poll() is None: + process.kill() + process.wait(timeout=2) + except Exception: + log_exception("[player] failed to reap unowned mpv process") + publish_status("error", message=f"failed to start mpv: {exc}") + else: + publish_status("error", message=f"failed to start mpv: {exc}") + return + + publish_status("playing", playback=playback, label=label) + try: + playback.listener_thread.start() + except Exception as exc: + log_exception("[player] failed to start playback worker") + stop_mpv( + reason="worker_start_error", + expected=playback, + final_state="error", + message=f"failed to start playback worker: {exc}", + ) + +def mpv_ipc(command: Any, playback: Optional[PlaybackContext] = None) -> bool: + """Send a JSON IPC command to MPV via MPV_SOCKET.""" + deadline = time.monotonic() + 3.0 + while time.monotonic() < deadline: + if shutdown_event.is_set(): + return False + if playback is not None and not playback_is_current(playback): + return False + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(0.5) sock.connect(MPV_SOCKET) - break - except FileNotFoundError: - if time.time() > deadline: - raise - time.sleep(0.05) - except ConnectionRefusedError: - if time.time() > deadline: - raise - time.sleep(0.05) - sock.sendall((json.dumps({"command": command}) + "\n").encode("utf-8")) - sock.close() - return True - except Exception as e: - print(f"[player] mpv_ipc error: {e}") - return False + sock.sendall((json.dumps({"command": command}) + "\n").encode("utf-8")) + return True + except (FileNotFoundError, ConnectionRefusedError, socket.timeout): + shutdown_event.wait(0.05) + except OSError as exc: + log(f"[player] mpv_ipc error: {exc}", error=True) + return False + + log(f"[player] mpv_ipc timed out waiting for {MPV_SOCKET}", error=True) + return False def fmt_mmss(value: Optional[float]) -> Optional[str]: - if not isinstance(value, (int, float)) or value < 0: + if ( + not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + ): return None total = int(value) @@ -309,6 +507,7 @@ def fmt_mmss(value: Optional[float]) -> Optional[str]: return f"{m}:{s:02d}" def publish_telemetry( + playback: PlaybackContext, *, pos: Optional[float], dur: Optional[float], @@ -318,14 +517,17 @@ def publish_telemetry( playlist_pos: Optional[int], playlist_count: Optional[int], ) -> None: + if not playback_is_current(playback): + return + rem = None if isinstance(pos, (int, float)) and isinstance(dur, (int, float)): rem = max(0.0, float(dur) - float(pos)) payload: Dict[str, Any] = { "state": "telemetry", - "slot": current_slot, - "path": current_path, + "slot": playback.slot, + "path": playback.path, # raw seconds "position": pos, @@ -345,12 +547,15 @@ def publish_telemetry( "ts": time.time(), } - client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False) + try: + client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False) + except Exception: + log_exception("[player] failed to publish telemetry") -def publish_cleared_telemetry() -> None: +def publish_cleared_telemetry(playback: Optional[PlaybackContext] = None) -> None: payload: Dict[str, Any] = { "state": "telemetry", - "slot": current_slot, + "slot": playback.slot if playback is not None else None, "path": None, "position": 0, @@ -369,16 +574,17 @@ def publish_cleared_telemetry() -> None: "ts": time.time(), } - client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False) + try: + client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False) + except Exception: + log_exception("[player] failed to publish cleared telemetry") -def mpv_observe_properties() -> None: +def mpv_observe_properties(playback: PlaybackContext) -> None: """ Keep a persistent IPC connection to mpv and observe properties. Publishes telemetry to MQTT at TELEMETRY_INTERVAL cadence. """ - global mpv_listener_stop - # local cache pos: Optional[float] = None dur: Optional[float] = None @@ -389,19 +595,43 @@ def mpv_observe_properties() -> None: playlist_count: Optional[int] = None last_pub = 0.0 + recv_buffer = b"" + sock: Optional[socket.socket] = None + + def publish_cached() -> None: + nonlocal last_pub + if not playback_is_current(playback): + return + publish_telemetry( + playback, + pos=pos, + dur=dur, + paused=paused, + title=title, + filename=filename, + playlist_pos=playlist_pos, + playlist_count=playlist_count, + ) + last_pub = time.monotonic() try: - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - deadline = time.time() + 3.0 - while True: + deadline = time.monotonic() + 3.0 + while playback_is_current(playback): + candidate = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: - sock.connect(MPV_SOCKET) + candidate.connect(MPV_SOCKET) + sock = candidate break except (FileNotFoundError, ConnectionRefusedError): - if time.time() > deadline: + candidate.close() + if time.monotonic() > deadline: raise - time.sleep(0.05) + playback.stop_event.wait(0.05) + else: + return + if sock is None: + return sock.settimeout(1.0) @@ -416,24 +646,27 @@ def mpv_observe_properties() -> None: send(["observe_property", 5, "filename"]) # str send(["observe_property", 6, "playlist-pos"]) # int send(["observe_property", 7, "playlist-count"]) # int - print(f"[player] telemetry connected to mpv socket {MPV_SOCKET}") - publish_telemetry( - pos=pos, dur=dur, paused=paused, title=title, - filename=filename, playlist_pos=playlist_pos, playlist_count=playlist_count - ) + log(f"[player] telemetry connected to mpv socket {MPV_SOCKET}") + publish_cached() - while mpv_listener_stop is not None and not mpv_listener_stop.is_set(): + while playback_is_current(playback): try: chunk = sock.recv(65536) if not chunk: break - for line in chunk.decode("utf-8", errors="ignore").splitlines(): - if not line.strip(): + recv_buffer += chunk + lines = recv_buffer.split(b"\n") + recv_buffer = lines.pop() + for raw_line in lines: + if not raw_line.strip(): continue try: - evt = json.loads(line) - except json.JSONDecodeError: + evt = json.loads(raw_line.decode("utf-8", errors="replace")) + except (UnicodeDecodeError, json.JSONDecodeError): + continue + + if not isinstance(evt, dict): continue if evt.get("event") != "property-change": @@ -441,11 +674,16 @@ def mpv_observe_properties() -> None: name = evt.get("name") val = evt.get("data") + finite_number = ( + isinstance(val, (int, float)) + and not isinstance(val, bool) + and math.isfinite(val) + ) if name == "time-pos": - pos = float(val) if isinstance(val, (int, float)) else None + pos = float(val) if finite_number else None elif name == "duration": - dur = float(val) if isinstance(val, (int, float)) else None + dur = float(val) if finite_number else None elif name == "pause": paused = bool(val) if isinstance(val, bool) else None elif name == "media-title": @@ -453,122 +691,124 @@ def mpv_observe_properties() -> None: elif name == "filename": filename = str(val) if isinstance(val, str) else None elif name == "playlist-pos": - playlist_pos = int(val) if isinstance(val, (int, float)) else None + playlist_pos = int(val) if finite_number else None elif name == "playlist-count": - playlist_count = int(val) if isinstance(val, (int, float)) else None + playlist_count = int(val) if finite_number else None - now = time.time() + now = time.monotonic() if now - last_pub >= TELEMETRY_INTERVAL: - publish_telemetry( - pos=pos, - dur=dur, - paused=paused, - title=title, - filename=filename, - playlist_pos=playlist_pos, - playlist_count=playlist_count, - ) - last_pub = now + publish_cached() except socket.timeout: # still periodically publish even if no new events - now = time.time() + now = time.monotonic() if now - last_pub >= TELEMETRY_INTERVAL: - publish_telemetry( - pos=pos, - dur=dur, - paused=paused, - title=title, - filename=filename, - playlist_pos=playlist_pos, - playlist_count=playlist_count, - ) - last_pub = now + publish_cached() continue - except Exception as e: - print(f"[player] mpv_observe_properties error: {e}") + except Exception as exc: + if playback_is_current(playback): + log(f"[player] mpv_observe_properties error: {exc}", error=True) finally: - try: - sock.close() - except Exception: - pass - -def connect_mpv_audio(): - deadline = time.time() + 8 - - while time.time() < deadline: - try: - jack_client = jack.Client("MPVAutoRouter", no_start_server=True) - - with jack_client: - ports = [p.name for p in jack_client.get_ports()] - - left_candidates = [ - p for p in ports - if ( - p.startswith("MPV:") or p.startswith("mpv:") - ) and ( - "output_FL" in p or "out_0" in p or "output_0" in p - ) - ] - - right_candidates = [ - p for p in ports - if ( - p.startswith("MPV:") or p.startswith("mpv:") - ) and ( - "output_FR" in p or "out_1" in p or "output_1" in p - ) - ] - - if left_candidates and right_candidates: - mpv_l = left_candidates[0] - mpv_r = right_candidates[0] - - routes = [ - (mpv_l, VIDEO_HDMI_L), - (mpv_r, VIDEO_HDMI_R), - (mpv_l, VIDEO_ANALOG_L), - (mpv_r, VIDEO_ANALOG_R), - ] - - for src, dst in routes: - try: - jack_client.connect(src, dst) - print(f"[player] connected video audio: {src} -> {dst}") - except jack.JackError as e: - print(f"[player] route skipped/failed {src} -> {dst}: {e}") - - return - - print("[player] waiting for MPV audio ports...") - - except Exception as e: - print(f"[player] connect_mpv_audio retry error: {e}") - - time.sleep(0.25) - - print("[player] mpv audio ports not found after timeout") + if sock is not None: + try: + sock.close() + except OSError: + pass # ---------------------------- # Command dispatch # ---------------------------- -def handle_cmd(payload_str: str) -> None: - """Decode JSON and route to play/stop/etc.""" +def decode_command(payload_str: str) -> Optional[Dict[str, Any]]: + """Decode and validate the outer command envelope without raising.""" try: data = json.loads(payload_str or "{}") - except json.JSONDecodeError: - print(f"[player] bad JSON in cmd: {payload_str!r}") + except (TypeError, ValueError, RecursionError): + log(f"[player] bad JSON in cmd: {payload_str!r}", error=True) + return None + + if not isinstance(data, dict): + log("[player] command payload must be a JSON object", error=True) + return None + + action = data.get("action") + if not isinstance(action, str) or not action.strip(): + log("[player] cmd missing string 'action'", error=True) + return None + return data + + +def enqueue_command( + data: Dict[str, Any], + *, + message_id: Optional[int] = None, + qos: int = 0, + connection_generation: Optional[int] = None, +) -> bool: + """Hand work to the main thread with bounded, lossless backpressure.""" + queued = QueuedCommand( + data=data, + message_id=message_id, + qos=qos, + connection_generation=connection_generation, + ) + while not shutdown_event.is_set(): + try: + command_queue.put(queued, timeout=0.2) + return True + except queue.Full: + continue + return False + + +def acknowledge_command( + message_id: Optional[int], + qos: int, + connection_generation: Optional[int] = None, +) -> None: + if message_id is None or qos <= 0: + return + + if connection_generation is not None: + with mqtt_generation_lock: + if connection_generation != mqtt_connection_generation: + log( + f"[player] skipping stale MQTT ack mid={message_id} from " + f"connection {connection_generation}", + error=True, + ) + return + + try: + ack_rc = client.ack(message_id, qos) + if ack_rc != mqtt.MQTT_ERR_SUCCESS: + log( + f"[player] MQTT ack failed mid={message_id} rc={ack_rc}", + error=True, + ) + except Exception: + log_exception(f"[player] MQTT ack failed mid={message_id}") + + +def handle_cmd(command: Any) -> None: + """Route one validated command on the main player thread.""" + if isinstance(command, str): + data = decode_command(command) + if data is None: + return + elif isinstance(command, dict): + data = command + else: + log("[player] internal command was not an object", error=True) return action = data.get("action") - if not action: - print("[player] cmd missing 'action'") + if not isinstance(action, str) or not action: + log("[player] cmd missing string 'action'", error=True) return - print(f"[player] cmd received: {data!r}") + log(f"[player] cmd received: {data!r}") if action == "play": cmd = { @@ -582,93 +822,219 @@ def handle_cmd(payload_str: str) -> None: return if action == "stop": - print("[player] STOP command") + log("[player] STOP command") stop_mpv("stop") return - elif action == "pause": - if mpv_proc and mpv_proc.poll() is None: - mpv_ipc(["set_property", "pause", True]) - publish_status("paused") + playback = get_current_playback() + + if action == "pause": + if playback is not None and playback.process.poll() is None: + mpv_ipc(["set_property", "pause", True], playback) + publish_status("paused", playback=playback) else: publish_status("error", message="nothing playing to pause") + return - elif action == "resume": - if mpv_proc and mpv_proc.poll() is None: - mpv_ipc(["set_property", "pause", False]) - publish_status("playing") + if action == "resume": + if playback is not None and playback.process.poll() is None: + mpv_ipc(["set_property", "pause", False], playback) + publish_status("playing", playback=playback) else: publish_status("error", message="nothing playing to resume") + return - elif action == "toggle": - if mpv_proc and mpv_proc.poll() is None: - mpv_ipc(["cycle", "pause"]) - publish_status("toggled") + if action == "toggle": + if playback is not None and playback.process.poll() is None: + mpv_ipc(["cycle", "pause"], playback) + publish_status("toggled", playback=playback) else: publish_status("error", message="nothing playing to toggle") + return - - print(f"[player] unknown action: {action!r}") + log(f"[player] unknown action: {action!r}", error=True) # ---------------------------- # MQTT callbacks # ---------------------------- -def on_connect(client, userdata, flags, rc, properties=None): - if rc == 0: - print(f"[player] MQTT connected rc={rc}, subscribing {TOPIC_CMD}") - client.subscribe(TOPIC_CMD, qos=1) - # initial idle +def on_connect(mqtt_client, userdata, flags, rc, properties=None) -> None: + global mqtt_connection_generation + try: + if rc != 0: + log(f"[player] MQTT connect failed with code {rc}", error=True) + return + + with mqtt_generation_lock: + mqtt_connection_generation += 1 + + log(f"[player] MQTT connected rc={rc}, subscribing {TOPIC_CMD}") + mqtt_client.subscribe(TOPIC_CMD, qos=1) + publish_status("idle", message="player online") - # Home Assistant MQTT Discovery sensors - ha_discovery_sensor("position", "YT Position (s)", "{{ value_json.position }}", icon="mdi:timeline-clock", unit="s") - ha_discovery_sensor("duration", "YT Duration (s)", "{{ value_json.duration }}", icon="mdi:timer", unit="s") - ha_discovery_sensor("remaining", "YT Remaining (s)", "{{ value_json.remaining }}", icon="mdi:timer-sand", unit="s") - ha_discovery_sensor("position_fmt", "YT Position", "{{ value_json.position_fmt }}", icon="mdi:timeline-clock") - ha_discovery_sensor("duration_fmt", "YT Duration", "{{ value_json.duration_fmt }}", icon="mdi:timer") - ha_discovery_sensor("remaining_fmt", "YT Remaining", "{{ value_json.remaining_fmt }}", icon="mdi:timer-sand") + if HA_DISCOVERY_ENABLED: + try: + if publish_discovery is None: + raise RuntimeError("Home Assistant discovery module is unavailable") + publish_discovery( + mqtt_client, + discovery_prefix=HA_DISCOVERY_PREFIX, + device_id=HA_DEVICE_ID, + device_name=HA_DEVICE_NAME, + telemetry_topic=TELEMETRY_TOPIC, + status_topic=TOPIC_STATUS, + ) + except Exception: + # Discovery is optional and must never disable core MQTT + # command handling. + log_exception("[player] Home Assistant discovery publish failed") + except Exception: + log_exception("[player] unexpected error in MQTT on_connect callback") - ha_discovery_sensor("title", "YT Title", "{{ value_json.title }}", icon="mdi:music") - ha_discovery_sensor("paused", "YT Paused", "{{ value_json.paused }}", icon="mdi:pause-circle") - ha_discovery_sensor("slot", "YT Slot", "{{ value_json.slot }}", icon="mdi:playlist-play") +def on_connect_fail(mqtt_client, userdata) -> None: + log( + f"[player] MQTT connection to {MQTT_HOST}:{MQTT_PORT} failed; will retry", + error=True, + ) + + +def on_disconnect( + mqtt_client, + userdata, + disconnect_flags, + reason_code, + properties=None, +) -> None: + if shutdown_event.is_set(): + log(f"[player] MQTT disconnected rc={reason_code}") else: - sys.stderr.write(f"[player] MQTT connect failed with code {rc}\n") + log(f"[player] MQTT disconnected rc={reason_code}; will retry", error=True) -def on_message(client, userdata, msg): - if msg.topic == TOPIC_CMD: - payload = msg.payload.decode("utf-8", errors="ignore") - handle_cmd(payload) +def on_message(mqtt_client, userdata, msg) -> None: + try: + if msg.topic != TOPIC_CMD: + return + payload = msg.payload.decode("utf-8", errors="replace") + data = decode_command(payload) + message_id = getattr(msg, "mid", None) + qos = int(getattr(msg, "qos", 0)) + with mqtt_generation_lock: + connection_generation = mqtt_connection_generation + if data is None: + # Validation completed synchronously, matching the old callback's + # acknowledgement point for rejected/poison messages. + acknowledge_command(message_id, qos, connection_generation) + return + + enqueue_command( + data, + message_id=message_id, + qos=qos, + connection_generation=connection_generation, + ) + except Exception: + # Paho otherwise terminates its network thread and leaves the process + # looking healthy to systemd while it is permanently deaf to commands. + log_exception("[player] unexpected error in MQTT on_message callback") # ---------------------------- # Signal handling # ---------------------------- -def sigterm_handler(signum, frame): - print("[player] SIGTERM received, stopping mpv") - stop_mpv(reason="sigterm") - sys.exit(0) +shutdown_reason = "shutdown" + + +def sigterm_handler(signum, frame) -> None: + global shutdown_reason + shutdown_reason = "sigterm" if signum == signal.SIGTERM else "sigint" + log(f"[player] {signal.Signals(signum).name} received, shutting down") + shutdown_event.set() # ---------------------------- # Main loop # ---------------------------- -def main() -> None: - global mpv_proc, busy_until +def poll_current_playback() -> None: + playback = get_current_playback() + if playback is None: + return - signal.signal(signal.SIGTERM, sigterm_handler) - signal.signal(signal.SIGINT, sigterm_handler) + ret = playback.process.poll() + if ret is None: + return + if ret == 0: + log("[player] mpv finished normally") + stop_mpv(reason="finished", expected=playback) + else: + log(f"[player] mpv exited with code {ret}", error=True) + publish_status("error", playback=playback, message=f"mpv exited code {ret}") + stop_mpv(reason="mpv_error", expected=playback) + +def process_pending_commands() -> None: + for _ in range(COMMANDS_PER_TICK): + if shutdown_event.is_set(): + return + try: + queued = command_queue.get_nowait() + except queue.Empty: + return + + if isinstance(queued, QueuedCommand): + command = queued.data + message_id = queued.message_id + qos = queued.qos + connection_generation = queued.connection_generation + else: + # Backward-compatible for in-process callers/tests that populated + # the queue before QueuedCommand was introduced. + command = queued + message_id = None + qos = 0 + connection_generation = None + + try: + handle_cmd(command) + except Exception: + # A bad command must not terminate the daemon's main loop. + log_exception(f"[player] unexpected command failure: {command!r}") + finally: + # Match the old QoS 1 durability boundary: acknowledge only after + # main-thread execution, not merely after callback enqueueing. + acknowledge_command(message_id, qos, connection_generation) + + +def clear_expired_busy_status() -> None: + global busy_until + playback = get_current_playback() + if ( + busy_until is not None + and time.monotonic() >= busy_until + and playback is not None + and playback.process.poll() is None + ): + log("[player] clearing busy -> playing") + publish_status("playing", playback=playback) + busy_until = None + + +def configure_mqtt_client() -> None: if MQTT_USER: client.username_pw_set(MQTT_USER, MQTT_PASS or None) client.on_connect = on_connect + client.on_connect_fail = on_connect_fail + client.on_disconnect = on_disconnect client.on_message = on_message + client.suppress_exceptions = True + client.manual_ack_set(True) + client.reconnect_delay_set(min_delay=1, max_delay=30) client.will_set( TOPIC_STATUS, @@ -677,41 +1043,39 @@ def main() -> None: retain=False, ) - print(f"[player] connecting MQTT to {MQTT_HOST}:{MQTT_PORT}") - client.connect(MQTT_HOST, MQTT_PORT, keepalive=30) - client.loop_start() + +def main() -> None: + global shutdown_reason + + signal.signal(signal.SIGTERM, sigterm_handler) + signal.signal(signal.SIGINT, sigterm_handler) + shutdown_reason = "shutdown" + shutdown_event.clear() + configure_mqtt_client() + + log(f"[player] connecting MQTT to {MQTT_HOST}:{MQTT_PORT}") + client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30) + loop_rc = client.loop_start() + if loop_rc != mqtt.MQTT_ERR_SUCCESS: + raise RuntimeError(f"failed to start MQTT network loop: {loop_rc}") try: - while True: - now = time.time() - - # Watch mpv exit - if mpv_proc is not None: - ret = mpv_proc.poll() - if ret is not None: - if ret == 0: - print("[player] mpv finished normally") - stop_mpv(reason="finished") - else: - print(f"[player] mpv exited with code {ret}") - publish_status("error", message=f"mpv exited code {ret}") - stop_mpv(reason="mpv_error") - - # If we previously reported "busy", re-assert "playing" after delay - if ( - busy_until is not None - and now >= busy_until - and mpv_proc is not None - and mpv_proc.poll() is None - ): - print("[player] clearing busy → playing") - publish_status("playing") - busy_until = None - - time.sleep(0.2) + while not shutdown_event.is_set(): + poll_current_playback() + process_pending_commands() + clear_expired_busy_status() + shutdown_event.wait(0.05) finally: - client.loop_stop() - stop_mpv(reason="shutdown") + if shutdown_reason in ("sigterm", "sigint"): + # The historical signal handler published its terminal player + # state while MQTT was still running. + stop_mpv(reason=shutdown_reason) + client.loop_stop() + else: + client.loop_stop() + stop_mpv(reason=shutdown_reason) + # Do not send MQTT DISCONNECT: the broker must emit the configured + # offline LWT when the service process exits, as it did previously. if __name__ == "__main__": diff --git a/requirements.txt b/requirements.txt index fe752dd..c6ddb34 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,58 +1,2 @@ -attrs==22.2.0 -Brotli==1.0.9 -certifi==2022.9.24 -cffi==1.15.1 -chardet==5.1.0 -charset-normalizer==3.0.1 -crit==3.17.1 -cupshelpers==1.0 -dbus-python==1.3.2 -distro==1.8.0 -distro-info==1.5+deb12u1 -docker==5.0.3 -docker-compose==1.29.2 -dockerpty==0.4.1 -docopt==0.6.2 -gyp==0.1 -httplib2==0.20.4 -idna==3.3 -JACK-Client==0.5.3 -jsonpointer==2.3 -jsonschema==4.10.3 -Mako==1.2.4.dev0 -Markdown==3.4.1 -MarkupSafe==2.1.2 -mutagen==1.46.0 -numpy==1.24.2 paho-mqtt==2.1.0 -ply==3.11 -protobuf==4.21.12 -pycairo==1.20.1 -pycparser==2.21 -pycryptodomex==3.11.0 -pycups==2.0.1 -pycurl==7.45.2 -Pygments==2.14.0 -PyGObject==3.42.2 -pyparsing==3.0.9 -pyrsistent==0.18.1 -PySimpleSOAP==1.16.2 -pysmbc==1.0.23 -python-apt==2.6.0 -python-debian==0.1.49 -python-debianbts==4.0.1 -python-dotenv==0.21.0 -pyxattr==0.8.1 PyYAML==6.0 -reportbug==12.0.0 -requests==2.28.1 -rfc3987==1.3.8 -six==1.16.0 -texttable==1.6.7 -unattended-upgrades==0.1 -uritemplate==4.1.1 -urllib3==1.26.12 -webcolors==1.11.1 -websocket-client==1.2.3 -websockets==10.4 -yt-dlp==2025.9.26 diff --git a/tests/test_player_service.py b/tests/test_player_service.py new file mode 100644 index 0000000..700a880 --- /dev/null +++ b/tests/test_player_service.py @@ -0,0 +1,534 @@ +import json +import queue +import signal +import threading +import unittest +from types import SimpleNamespace +from unittest import mock + +import homeassistant_discovery +import player_service as player + + +class FakePublishClient: + def __init__(self): + self.published = [] + self.subscriptions = [] + self.acks = [] + + def publish(self, topic, payload, qos=0, retain=False): + self.published.append((topic, payload, qos, retain)) + return SimpleNamespace(rc=0) + + def subscribe(self, topic, qos=0): + self.subscriptions.append((topic, qos)) + return (0, 1) + + def ack(self, message_id, qos): + self.acks.append((message_id, qos)) + return player.mqtt.MQTT_ERR_SUCCESS + + +class FakeProcess: + def __init__(self, returncode=None): + self.returncode = returncode + self.terminate_calls = 0 + self.kill_calls = 0 + self.wait_calls = [] + + def poll(self): + return self.returncode + + def terminate(self): + self.terminate_calls += 1 + self.returncode = 0 + + def kill(self): + self.kill_calls += 1 + self.returncode = -9 + + def wait(self, timeout=None): + self.wait_calls.append(timeout) + return self.returncode + + +def playback(generation=1, returncode=None, path="/media/video.mp4", slot=1): + return player.PlaybackContext( + generation=generation, + process=FakeProcess(returncode), + path=path, + slot=slot, + stop_event=threading.Event(), + ) + + +class PlayerServiceTests(unittest.TestCase): + def setUp(self): + self.original_client = player.client + self.original_playback = player.current_playback + self.original_queue = player.command_queue + self.original_mqtt_generation = player.mqtt_connection_generation + player.current_playback = None + player.command_queue = queue.Queue() + player.mqtt_connection_generation = 0 + player.shutdown_event.clear() + + def tearDown(self): + player.client = self.original_client + player.current_playback = self.original_playback + player.command_queue = self.original_queue + player.mqtt_connection_generation = self.original_mqtt_generation + player.shutdown_event.clear() + + def test_decode_command_rejects_non_objects_and_bad_envelopes(self): + bad_payloads = ("", "not json", "null", "[]", '"text"', "1", "{}") + with mock.patch.object(player, "log"): + for payload in bad_payloads: + with self.subTest(payload=payload): + self.assertIsNone(player.decode_command(payload)) + + self.assertEqual( + player.decode_command('{"action":"stop"}'), + {"action": "stop"}, + ) + + def test_decode_command_contains_parser_resource_errors(self): + huge_integer = '{"action":"stop","value":' + ("9" * 5000) + "}" + deeply_nested = ("[" * 1200) + ("0") + ("]" * 1200) + with mock.patch.object(player, "log"): + self.assertIsNone(player.decode_command(huge_integer)) + self.assertIsNone(player.decode_command(deeply_nested)) + + def test_malformed_play_fields_do_not_escape(self): + fake_client = FakePublishClient() + player.client = fake_client + malformed = ( + {"action": "play", "path": 123}, + {"action": "play", "path": {"bad": True}}, + {"action": "play", "slot": float("inf")}, + ) + + with mock.patch.object(player, "log"): + for command in malformed: + with self.subTest(command=command): + player.handle_cmd(command) + + status_payloads = [ + json.loads(payload) + for topic, payload, _, _ in fake_client.published + if topic == player.TOPIC_STATUS + ] + self.assertEqual(len(status_payloads), len(malformed)) + self.assertTrue(all(item["state"] == "error" for item in status_payloads)) + + def test_on_message_rejects_bad_payload_then_queues_valid_command(self): + fake_client = FakePublishClient() + player.client = fake_client + with mock.patch.object(player, "log"): + player.on_message( + None, + None, + SimpleNamespace( + topic=player.TOPIC_CMD, + payload=b"[]", + mid=11, + qos=1, + ), + ) + player.on_message( + None, + None, + SimpleNamespace(topic=player.TOPIC_CMD, payload=b'{"action":"stop"}'), + ) + + self.assertEqual(fake_client.acks, [(11, 1)]) + self.assertEqual(player.command_queue.qsize(), 1) + self.assertEqual( + player.command_queue.get_nowait().data, + {"action": "stop"}, + ) + + def test_on_message_contains_unexpected_callback_exception(self): + message = SimpleNamespace(topic=player.TOPIC_CMD, payload=b'{"action":"stop"}') + with mock.patch.object(player, "decode_command", side_effect=RuntimeError("boom")): + with mock.patch.object(player, "log_exception") as logged: + player.on_message(None, None, message) + logged.assert_called_once() + + def test_repeated_commands_are_preserved_in_order(self): + command = {"action": "stop"} + for _ in range(3): + self.assertTrue(player.enqueue_command(command)) + self.assertEqual( + [player.command_queue.get_nowait().data for _ in range(3)], + [command, command, command], + ) + + def test_qos1_command_is_acked_only_after_main_thread_execution(self): + fake_client = FakePublishClient() + player.client = fake_client + message = SimpleNamespace( + topic=player.TOPIC_CMD, + payload=b'{"action":"stop"}', + mid=42, + qos=1, + ) + + player.on_message(None, None, message) + self.assertEqual(fake_client.acks, []) + + with mock.patch.object(player, "handle_cmd") as handled: + player.process_pending_commands() + + handled.assert_called_once_with({"action": "stop"}) + self.assertEqual(fake_client.acks, [(42, 1)]) + + def test_queued_command_cannot_ack_a_reused_mid_after_reconnect(self): + fake_client = FakePublishClient() + player.client = fake_client + player.mqtt_connection_generation = 3 + player.enqueue_command( + {"action": "stop"}, + message_id=42, + qos=1, + connection_generation=3, + ) + player.mqtt_connection_generation = 4 + + with mock.patch.object(player, "handle_cmd") as handled: + with mock.patch.object(player, "log"): + player.process_pending_commands() + + handled.assert_called_once_with({"action": "stop"}) + self.assertEqual(fake_client.acks, []) + + def test_status_topics_qos_retain_and_payload_are_unchanged(self): + fake_client = FakePublishClient() + player.client = fake_client + active = playback(path="/media/example.mp4", slot=4) + + player.publish_status("playing", playback=active, label="Example") + + self.assertEqual(fake_client.published[0][0], "youtube/player/status") + self.assertEqual(fake_client.published[0][2:], (1, False)) + self.assertEqual( + json.loads(fake_client.published[0][1]), + { + "state": "playing", + "path": "/media/example.mp4", + "slot": 4, + "label": "Example", + }, + ) + self.assertEqual( + fake_client.published[1], + ("youtube/player/state", "playing", 0, True), + ) + + def test_stale_generation_cannot_stop_current_playback(self): + old = playback(generation=1) + current = playback(generation=2) + player.current_playback = current + + with mock.patch.object(player, "_remove_mpv_socket"): + self.assertFalse(player.stop_mpv(expected=old)) + + self.assertIs(player.current_playback, current) + self.assertEqual(current.process.terminate_calls, 0) + self.assertFalse(current.stop_event.is_set()) + + def test_stale_generation_cannot_publish_telemetry(self): + fake_client = FakePublishClient() + player.client = fake_client + old = playback(generation=1) + player.current_playback = playback(generation=2) + + player.publish_telemetry( + old, + pos=1, + dur=2, + paused=False, + title=None, + filename=None, + playlist_pos=0, + playlist_count=1, + ) + + self.assertEqual(fake_client.published, []) + + def test_stop_failure_preserves_error_clear_idle_publish_order(self): + class FailingTerminateProcess(FakeProcess): + def terminate(self): + raise OSError("cannot terminate") + + fake_client = FakePublishClient() + player.client = fake_client + active = playback() + active.process = FailingTerminateProcess(returncode=None) + player.current_playback = active + + with mock.patch.object(player, "_remove_mpv_socket"): + with mock.patch.object(player, "log"): + player.stop_mpv("stop") + + self.assertEqual( + [topic for topic, _, _, _ in fake_client.published], + [ + player.TOPIC_STATUS, + player.TOPIC_STATE, + player.TELEMETRY_TOPIC, + player.TOPIC_STATUS, + player.TOPIC_STATE, + ], + ) + self.assertEqual(json.loads(fake_client.published[0][1])["state"], "error") + self.assertIsNone(json.loads(fake_client.published[2][1])["slot"]) + self.assertEqual(json.loads(fake_client.published[3][1])["state"], "idle") + + def test_unstoppable_process_remains_owned_and_blocks_duplicates(self): + class UnstoppableProcess(FakeProcess): + def terminate(self): + raise OSError("cannot terminate") + + def kill(self): + raise OSError("cannot kill") + + fake_client = FakePublishClient() + player.client = fake_client + active = playback() + active.process = UnstoppableProcess(returncode=None) + player.current_playback = active + + with mock.patch.object(player, "_remove_mpv_socket") as remove_socket: + with mock.patch.object(player, "log"): + self.assertFalse(player.stop_mpv("stop")) + + self.assertIs(player.current_playback, active) + remove_socket.assert_not_called() + self.assertEqual( + [json.loads(payload)["state"] for topic, payload, _, _ in fake_client.published if topic == player.TOPIC_STATUS], + ["error"], + ) + + def test_mpv_is_pinned_to_virtual_pipewire_bus(self): + fake_process = FakeProcess(returncode=None) + fake_thread = SimpleNamespace( + name="fake-listener", + start=mock.Mock(), + join=mock.Mock(), + is_alive=mock.Mock(return_value=False), + ) + + with mock.patch.object(player.os.path, "exists", return_value=True): + with mock.patch.object(player.subprocess, "Popen", return_value=fake_process) as popen: + with mock.patch.object(player.threading, "Thread", return_value=fake_thread): + with mock.patch.object(player, "_remove_mpv_socket"): + player.handle_play({"action": "play", "path": "/media/video.mp4"}) + + args = popen.call_args.args[0] + self.assertIn("--audio-device=pipewire/ytplayer_bus", args) + self.assertIn("--audio-client-name=YTPlayer", args) + self.assertIn("--audio-channels=stereo", args) + self.assertIn("--audio-fallback-to-null=no", args) + + def test_worker_construction_failure_reaps_started_mpv(self): + fake_client = FakePublishClient() + player.client = fake_client + fake_process = FakeProcess(returncode=None) + + with mock.patch.object(player.os.path, "exists", return_value=True): + with mock.patch.object(player.subprocess, "Popen", return_value=fake_process): + with mock.patch.object( + player.threading, + "Thread", + side_effect=RuntimeError("cannot construct worker"), + ): + with mock.patch.object(player, "_remove_mpv_socket"): + with mock.patch.object(player, "log"): + player.handle_play( + {"action": "play", "path": "/media/video.mp4"}, + ) + + self.assertEqual(fake_process.terminate_calls, 1) + self.assertIsNone(player.current_playback) + status_payloads = [ + json.loads(payload) + for topic, payload, _, _ in fake_client.published + if topic == player.TOPIC_STATUS + ] + self.assertEqual(status_payloads[-1]["state"], "error") + + def test_mpv_error_preserves_historical_error_then_idle_sequence(self): + fake_client = FakePublishClient() + player.client = fake_client + failed = playback(returncode=4) + player.current_playback = failed + + with mock.patch.object(player, "_remove_mpv_socket"): + with mock.patch.object(player, "log"): + player.poll_current_playback() + + state_messages = [ + (topic, json.loads(payload) if topic == player.TOPIC_STATUS else payload) + for topic, payload, _, _ in fake_client.published + if topic in (player.TOPIC_STATUS, player.TOPIC_STATE) + ] + self.assertEqual( + [ + item[1]["state"] if item[0] == player.TOPIC_STATUS else item[1] + for item in state_messages + ], + ["error", "error", "idle", "idle"], + ) + self.assertEqual(state_messages[-2][1]["reason"], "mpv_error") + + def test_shutdown_stops_command_drain_without_losing_pending_items(self): + player.command_queue.put_nowait({"action": "first"}) + player.command_queue.put_nowait({"action": "second"}) + + def handle_one(command): + player.shutdown_event.set() + + with mock.patch.object(player, "handle_cmd", side_effect=handle_one) as handled: + player.process_pending_commands() + + handled.assert_called_once_with({"action": "first"}) + self.assertEqual(player.command_queue.get_nowait(), {"action": "second"}) + + def test_main_uses_nonblocking_mqtt_connect(self): + class FakeMqttClient(FakePublishClient): + def __init__(self): + super().__init__() + self.calls = [] + self.suppress_exceptions = False + + def username_pw_set(self, *args): + self.calls.append(("username_pw_set", args)) + + def reconnect_delay_set(self, **kwargs): + self.calls.append(("reconnect_delay_set", kwargs)) + + def manual_ack_set(self, enabled): + self.calls.append(("manual_ack_set", enabled)) + + def max_queued_messages_set(self, count): + self.calls.append(("max_queued_messages_set", count)) + + def will_set(self, *args, **kwargs): + self.calls.append(("will_set", args, kwargs)) + + def connect_async(self, *args, **kwargs): + self.calls.append(("connect_async", args, kwargs)) + + def loop_start(self): + self.calls.append(("loop_start",)) + player.shutdown_event.set() + return player.mqtt.MQTT_ERR_SUCCESS + + def disconnect(self): + self.calls.append(("disconnect",)) + + def loop_stop(self): + self.calls.append(("loop_stop",)) + + fake_client = FakeMqttClient() + player.client = fake_client + + with mock.patch.object(signal, "signal"): + with mock.patch.object(player, "stop_mpv"): + player.main() + + call_names = [call[0] for call in fake_client.calls] + self.assertIn("connect_async", call_names) + self.assertIn("manual_ack_set", call_names) + self.assertNotIn("connect", call_names) + self.assertNotIn("disconnect", call_names) + self.assertLess(call_names.index("connect_async"), call_names.index("loop_start")) + + def test_discovery_failure_cannot_escape_on_connect(self): + fake_client = FakePublishClient() + player.client = fake_client + + with mock.patch.object(player, "publish_discovery", side_effect=TypeError("bad discovery")): + with mock.patch.object(player, "log_exception") as logged: + player.on_connect(fake_client, None, None, 0) + + self.assertEqual(fake_client.subscriptions, [(player.TOPIC_CMD, 1)]) + logged.assert_called_once() + self.assertEqual(fake_client.published[0][0], player.TOPIC_STATUS) + + def test_reconnect_preserves_historical_idle_payload_while_playing(self): + fake_client = FakePublishClient() + player.client = fake_client + player.current_playback = playback(path="/media/live.mp4", slot=7) + + with mock.patch.object(player, "publish_discovery"): + with mock.patch.object(player, "log"): + player.on_connect(fake_client, None, None, 0) + + self.assertEqual(fake_client.subscriptions, [(player.TOPIC_CMD, 1)]) + self.assertEqual( + json.loads(fake_client.published[0][1]), + { + "state": "idle", + "path": "/media/live.mp4", + "slot": 7, + "message": "player online", + }, + ) + self.assertEqual(fake_client.published[1], (player.TOPIC_STATE, "idle", 0, True)) + + def test_pause_status_contract_is_preserved_when_ipc_fails(self): + fake_client = FakePublishClient() + player.client = fake_client + player.current_playback = playback() + + with mock.patch.object(player, "mpv_ipc", return_value=False): + with mock.patch.object(player, "log"): + player.handle_cmd({"action": "pause"}) + + self.assertEqual(json.loads(fake_client.published[0][1])["state"], "paused") + self.assertEqual(fake_client.published[1][1], "paused") + + +class HomeAssistantDiscoveryTests(unittest.TestCase): + def test_discovery_contract_is_unchanged(self): + client = FakePublishClient() + homeassistant_discovery.publish_discovery( + client, + discovery_prefix="homeassistant", + device_id="ytplayer", + device_name="YT Player", + telemetry_topic="youtube/player/telemetry", + status_topic="youtube/player/status", + ) + + self.assertEqual(len(client.published), 9) + self.assertTrue( + all(qos == 1 and retain is True for _, _, qos, retain in client.published) + ) + self.assertEqual( + [topic for topic, _, _, _ in client.published], + [ + f"homeassistant/sensor/ytplayer/{entity}/config" + for entity in ( + "position", + "duration", + "remaining", + "position_fmt", + "duration_fmt", + "remaining_fmt", + "title", + "paused", + "slot", + ) + ], + ) + first_payload = json.loads(client.published[0][1]) + self.assertEqual(first_payload["state_topic"], "youtube/player/telemetry") + self.assertEqual(first_payload["availability_topic"], "youtube/player/status") + self.assertEqual(first_payload["unit_of_measurement"], "s") + + +if __name__ == "__main__": + unittest.main() -- 2.54.0