203 lines
7.8 KiB
Python
203 lines
7.8 KiB
Python
"""Discover and play HEOS favorite stations through the HEOS CLI service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import socket
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import urlencode
|
|
|
|
|
|
_host = "192.168.178.177"
|
|
_port = 1255
|
|
_timeout = 5.0
|
|
|
|
|
|
class HeosError(RuntimeError):
|
|
"""Raised when HEOS rejects a command or returns an invalid response."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Favorite:
|
|
position: int
|
|
name: str
|
|
media_id: str
|
|
|
|
|
|
def configure(host: str, port: int = 1255, timeout: float = 5.0) -> None:
|
|
"""Configure the HEOS device used by subsequent calls."""
|
|
global _host, _port, _timeout
|
|
_host = host
|
|
_port = port
|
|
_timeout = timeout
|
|
|
|
|
|
def _command(path: str, **parameters: object) -> dict[str, Any]:
|
|
# HEOS expects the comma in range values literally (e.g. range=0,100).
|
|
query = urlencode(parameters, safe=",")
|
|
command = f"heos://{path}" + (f"?{query}" if query else "")
|
|
try:
|
|
with socket.create_connection((_host, _port), timeout=_timeout) as connection:
|
|
connection.settimeout(_timeout)
|
|
connection.sendall((command + "\r\n").encode("utf-8"))
|
|
with connection.makefile("r", encoding="utf-8", newline="\n") as stream:
|
|
while line := stream.readline():
|
|
try:
|
|
response = json.loads(line)
|
|
except json.JSONDecodeError as error:
|
|
raise HeosError(f"Invalid HEOS response: {line.rstrip()!r}") from error
|
|
heos = response.get("heos", {})
|
|
if str(heos.get("command", "")).strip() != path:
|
|
continue
|
|
if heos.get("result") != "success":
|
|
raise HeosError(str(heos.get("message", "HEOS command failed")))
|
|
message = str(heos.get("message", ""))
|
|
if "command under process" in message:
|
|
continue
|
|
return response
|
|
except (OSError, TimeoutError) as error:
|
|
raise HeosError(f"Cannot communicate with HEOS at {_host}:{_port}: {error}") from error
|
|
raise HeosError(f"HEOS closed the connection without answering {path!r}")
|
|
|
|
|
|
def _player_id() -> str:
|
|
response = _command("player/get_players")
|
|
players = response.get("payload", [])
|
|
if not isinstance(players, list) or not players:
|
|
raise HeosError("No HEOS players found")
|
|
return str(players[0]["pid"])
|
|
|
|
|
|
def _favorites_source_id() -> str:
|
|
response = _command("browse/get_music_sources")
|
|
sources = response.get("payload", [])
|
|
if not isinstance(sources, list):
|
|
raise HeosError("HEOS returned an invalid music-source list")
|
|
for source in sources:
|
|
if str(source.get("name", "")).casefold() == "favorites":
|
|
if source.get("available") == "false":
|
|
raise HeosError("The HEOS Favorites source is unavailable")
|
|
return str(source["sid"])
|
|
raise HeosError("HEOS Favorites source not found")
|
|
|
|
|
|
def list_favorites() -> list[Favorite]:
|
|
"""Return the currently configured HEOS favorite stations."""
|
|
response = _command("browse/browse", sid=_favorites_source_id(), range="0,100")
|
|
payload = response.get("payload", [])
|
|
if not isinstance(payload, list):
|
|
raise HeosError("HEOS returned an invalid Favorites list")
|
|
return [
|
|
Favorite(position=index, name=str(item["name"]), media_id=str(item.get("mid", "")))
|
|
for index, item in enumerate(payload, start=1)
|
|
if item.get("playable") == "yes"
|
|
]
|
|
|
|
|
|
def _find_current_favorite(favorites: list[Favorite], payload: object) -> Favorite | None:
|
|
if not isinstance(payload, dict) or payload.get("type") != "station":
|
|
return None
|
|
current_ids = {
|
|
str(payload.get(key, "")).strip()
|
|
for key in ("album_id", "mid")
|
|
if payload.get(key)
|
|
}
|
|
for favorite in favorites:
|
|
if favorite.media_id and favorite.media_id in current_ids:
|
|
return favorite
|
|
station = str(payload.get("station", "")).strip().casefold()
|
|
if station:
|
|
return next((item for item in favorites if item.name.casefold() == station), None)
|
|
return None
|
|
|
|
|
|
def current_favorite() -> Favorite | None:
|
|
"""Return the playing HEOS favorite, or None for other/no playback."""
|
|
favorites = list_favorites()
|
|
player_id = _player_id()
|
|
response = _command("player/get_now_playing_media", pid=player_id)
|
|
return _find_current_favorite(favorites, response.get("payload"))
|
|
|
|
|
|
def next_favorite() -> Favorite | None:
|
|
"""Play the favorite after the current one, wrapping at the end."""
|
|
favorites = list_favorites()
|
|
if not favorites:
|
|
raise HeosError("No HEOS favorites configured")
|
|
player_id = _player_id()
|
|
response = _command("player/get_now_playing_media", pid=player_id)
|
|
current = _find_current_favorite(favorites, response.get("payload"))
|
|
if current is None:
|
|
return None
|
|
current_index = favorites.index(current)
|
|
selected = favorites[(current_index + 1) % len(favorites)]
|
|
_command("browse/play_preset", pid=player_id, preset=selected.position)
|
|
return selected
|
|
|
|
|
|
def previous_favorite() -> Favorite | None:
|
|
"""Play the favorite before the current one, wrapping at the start."""
|
|
favorites = list_favorites()
|
|
if not favorites:
|
|
raise HeosError("No HEOS favorites configured")
|
|
player_id = _player_id()
|
|
response = _command("player/get_now_playing_media", pid=player_id)
|
|
current = _find_current_favorite(favorites, response.get("payload"))
|
|
if current is None:
|
|
return None
|
|
current_index = favorites.index(current)
|
|
selected = favorites[(current_index - 1) % len(favorites)]
|
|
_command("browse/play_preset", pid=player_id, preset=selected.position)
|
|
return selected
|
|
|
|
|
|
def play_favorite(favorite: str | int) -> Favorite:
|
|
"""Play a favorite selected by 1-based position or case-insensitive name."""
|
|
favorites = list_favorites()
|
|
selected: Favorite | None = None
|
|
if isinstance(favorite, int) or str(favorite).strip().isdigit():
|
|
position = int(favorite)
|
|
selected = next((item for item in favorites if item.position == position), None)
|
|
else:
|
|
requested_name = str(favorite).strip().casefold()
|
|
selected = next((item for item in favorites if item.name.casefold() == requested_name), None)
|
|
if selected is None:
|
|
raise HeosError(f"HEOS favorite not found: {favorite!r}")
|
|
_command("browse/play_preset", pid=_player_id(), preset=selected.position)
|
|
return selected
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--host", default=_host, help=f"HEOS address (default: {_host})")
|
|
parser.add_argument("--port", type=int, default=_port, help=f"HEOS CLI port (default: {_port})")
|
|
parser.add_argument("--timeout", type=float, default=_timeout)
|
|
subparsers = parser.add_subparsers(dest="action", required=True)
|
|
subparsers.add_parser("list", help="list available HEOS favorites")
|
|
play_parser = subparsers.add_parser("play", help="play a favorite by number or name")
|
|
play_parser.add_argument("favorite")
|
|
args = parser.parse_args()
|
|
if not 1 <= args.port <= 65535:
|
|
parser.error("--port must be between 1 and 65535")
|
|
if args.timeout <= 0:
|
|
parser.error("--timeout must be positive")
|
|
configure(args.host, args.port, args.timeout)
|
|
try:
|
|
if args.action == "list":
|
|
for favorite in list_favorites():
|
|
print(f"{favorite.position:2}. {favorite.name}")
|
|
else:
|
|
selected = play_favorite(args.favorite)
|
|
print(f"Playing HEOS favorite {selected.position}: {selected.name}")
|
|
except HeosError as error:
|
|
print(f"HEOS {args.action} failed: {error}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|