scriptosandClaude Opus 5.5
67a687f4e6
Build & Push Docker Image / Build & push image (push) Successful in 6m49s
LOSSLESS_ONLY (default true): follow-up tracks are only converted when nothing is lost (MP3 into a FLAC stream). FLAC after MP3 or higher resolution FLAC plays in original quality with a normal track change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2104 lines
94 KiB
Python
2104 lines
94 KiB
Python
#!/usr/bin/env python3
|
||
"""bluos-upnp-bridge - exposes a Bluesound/BluOS player as a UPnP/DLNA MediaRenderer.
|
||
|
||
A UPnP control point (e.g. Symfonium) hands a stream URL to this bridge, the
|
||
bridge forwards it to the BluOS HTTP API (port 11000) and the player pulls the
|
||
stream directly from the media server - no transcoding, no AirPlay.
|
||
|
||
Subsonic stream URLs are rewritten to a dedicated service account (token auth)
|
||
so the control point's own credentials never reach the player.
|
||
|
||
Stdlib only, Python >= 3.9.
|
||
"""
|
||
import base64
|
||
import collections
|
||
import hashlib
|
||
import hmac
|
||
import html
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import secrets
|
||
import signal
|
||
import socket
|
||
import struct
|
||
import sys
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
import xml.etree.ElementTree as ET
|
||
from email.utils import formatdate
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
VERSION = "0.7.1"
|
||
STARTED = time.time()
|
||
|
||
# --------------------------------------------------------------------------- config
|
||
# Values come from the environment; settings saved in the web UI (DATA_DIR/settings.json)
|
||
# take precedence.
|
||
|
||
DATA_DIR = os.environ.get("DATA_DIR", "/data")
|
||
SETTINGS_FILE = os.path.join(DATA_DIR, "settings.json")
|
||
|
||
|
||
def _load_overrides():
|
||
try:
|
||
with open(SETTINGS_FILE, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except FileNotFoundError:
|
||
return {}
|
||
except Exception as exc:
|
||
print(f"WARNING: could not read {SETTINGS_FILE}: {exc}", file=sys.stderr)
|
||
return {}
|
||
|
||
|
||
OVERRIDES = _load_overrides()
|
||
|
||
|
||
def cfg(key, default=""):
|
||
return str(OVERRIDES.get(key, os.environ.get(key, default)))
|
||
|
||
|
||
def as_bool(value):
|
||
return str(value).lower() in ("1", "true", "yes", "on")
|
||
|
||
|
||
BLUOS_HOST = cfg("BLUOS_HOST")
|
||
BLUOS_PORT = int(cfg("BLUOS_PORT", "11000"))
|
||
ADVERTISE_IP = cfg("ADVERTISE_IP")
|
||
HTTP_PORT = int(cfg("HTTP_PORT", "49494"))
|
||
FRIENDLY_NAME = cfg("FRIENDLY_NAME")
|
||
SUBSONIC_URL = cfg("SUBSONIC_URL").rstrip("/")
|
||
SUBSONIC_USER = cfg("SUBSONIC_USER")
|
||
SUBSONIC_PASSWORD = cfg("SUBSONIC_PASSWORD")
|
||
SERVER_TYPE = cfg("SERVER_TYPE", "subsonic").lower()
|
||
SUBSONIC_API_KEY = cfg("SUBSONIC_API_KEY")
|
||
SUBSONIC_ALIASES = cfg("SUBSONIC_ALIASES")
|
||
CLIENT_NAME = cfg("CLIENT_NAME", "bluos-upnp-bridge")
|
||
PLAYER_NAME = "" # BluOS player name, filled in at start for {player} in CLIENT_NAME
|
||
SSDP_INTERVAL = int(cfg("SSDP_INTERVAL", "60"))
|
||
LOG_LEVEL = cfg("LOG_LEVEL", "INFO").upper()
|
||
REPEAT_OFF = as_bool(cfg("REPEAT_OFF", "true"))
|
||
PROXY_STREAMS = as_bool(cfg("PROXY_STREAMS", "true"))
|
||
GAPLESS = as_bool(cfg("GAPLESS", "false"))
|
||
FADE_MS = int(cfg("FADE_MS", "500") or 0)
|
||
OPENHOME = as_bool(cfg("OPENHOME", "true"))
|
||
UI_LANG = cfg("UI_LANG", "en").lower()
|
||
LOSSLESS_ONLY = as_bool(cfg("LOSSLESS_ONLY", "true"))
|
||
|
||
|
||
def L(de, en):
|
||
"""Pick the UI language for user-facing messages."""
|
||
return de if UI_LANG == "de" else en
|
||
WEB_USER = os.environ.get("WEB_USER", "admin")
|
||
WEB_PASSWORD = os.environ.get("WEB_PASSWORD", "")
|
||
|
||
if not BLUOS_HOST or not ADVERTISE_IP:
|
||
sys.exit("BLUOS_HOST and ADVERTISE_IP must be set")
|
||
|
||
|
||
class RingHandler(logging.Handler):
|
||
"""Keeps the latest log lines for the web UI."""
|
||
|
||
def __init__(self, size=300):
|
||
super().__init__()
|
||
self.lines = collections.deque(maxlen=size)
|
||
self.counter = 0
|
||
|
||
def emit(self, record):
|
||
self.counter += 1
|
||
self.lines.append({"id": self.counter, "t": record.created, "level": record.levelname,
|
||
"msg": record.getMessage()})
|
||
|
||
|
||
RING = RingHandler()
|
||
logging.basicConfig(level=LOG_LEVEL, format="%(asctime)s %(levelname)-7s %(name)s: %(message)s")
|
||
logging.getLogger().addHandler(RING)
|
||
log = logging.getLogger("bridge")
|
||
|
||
DEVICE_UUID = "uuid:" + str(uuid.uuid5(uuid.NAMESPACE_URL, f"bluos-upnp-bridge://{BLUOS_HOST}"))
|
||
DEVICE_TYPE = "urn:schemas-upnp-org:device:MediaRenderer:1"
|
||
SERVICES = {
|
||
"AVTransport": "urn:schemas-upnp-org:service:AVTransport:1",
|
||
"RenderingControl": "urn:schemas-upnp-org:service:RenderingControl:1",
|
||
"ConnectionManager": "urn:schemas-upnp-org:service:ConnectionManager:1",
|
||
}
|
||
SERVER_HEADER = f"Linux/1.0 UPnP/1.0 bluos-upnp-bridge/{VERSION}"
|
||
SINK_PROTOCOLS = ",".join(
|
||
f"http-get:*:{m}:*" for m in (
|
||
"audio/mpeg", "audio/mp3", "audio/flac", "audio/x-flac", "audio/mp4", "audio/x-m4a",
|
||
"audio/aac", "audio/ogg", "audio/opus", "audio/wav", "audio/x-wav", "audio/L16",
|
||
"audio/x-ms-wma", "audio/aiff", "audio/x-aiff", "audio/dsd", "audio/x-dsf",
|
||
)
|
||
)
|
||
|
||
NS_DIDL = {
|
||
"didl": "urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/",
|
||
"dc": "http://purl.org/dc/elements/1.1/",
|
||
"upnp": "urn:schemas-upnp-org:metadata-1-0/upnp/",
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- helpers
|
||
|
||
def hms(secs):
|
||
secs = max(0, int(secs or 0))
|
||
return f"{secs // 3600}:{secs % 3600 // 60:02d}:{secs % 60:02d}"
|
||
|
||
|
||
def parse_hms(value):
|
||
if not value:
|
||
return 0
|
||
parts = value.split(".")[0].split(":")
|
||
try:
|
||
nums = [int(p) for p in parts]
|
||
except ValueError:
|
||
return 0
|
||
total = 0
|
||
for n in nums:
|
||
total = total * 60 + n
|
||
return total
|
||
|
||
|
||
def mask_url(url):
|
||
"""Hide credentials in logs."""
|
||
return re.sub(r"([?&](?:p|t|s|u|apiKey)=)[^&]*", r"\1***", url or "")
|
||
|
||
|
||
def parse_didl(meta):
|
||
"""Extract title/artist/album/art/duration from DIDL-Lite metadata."""
|
||
info = {"title": "", "artist": "", "album": "", "art": "", "duration": 0}
|
||
if not meta or "<" not in meta:
|
||
return info
|
||
try:
|
||
root = ET.fromstring(meta)
|
||
except ET.ParseError:
|
||
log.debug("could not parse DIDL metadata")
|
||
return info
|
||
item = root.find("didl:item", NS_DIDL)
|
||
if item is None:
|
||
return info
|
||
info["title"] = item.findtext("dc:title", "", NS_DIDL)
|
||
info["artist"] = (item.findtext("upnp:artist", "", NS_DIDL)
|
||
or item.findtext("dc:creator", "", NS_DIDL))
|
||
info["album"] = item.findtext("upnp:album", "", NS_DIDL)
|
||
info["art"] = item.findtext("upnp:albumArtURI", "", NS_DIDL)
|
||
res = item.find("didl:res", NS_DIDL)
|
||
if res is not None:
|
||
info["duration"] = parse_hms(res.get("duration"))
|
||
return info
|
||
|
||
|
||
# --------------------------------------------------------------------------- subsonic
|
||
|
||
class Subsonic:
|
||
"""Rewrites Subsonic URLs to the bridge's own service account.
|
||
|
||
server_type "subsonic": classic token auth (u + t + s).
|
||
server_type "opensubsonic": OpenSubsonic servers (Airsonic Pulse, Navidrome, ...) - uses the
|
||
apiKey extension if an API key is set, otherwise the (hex encoded) password. Airsonic Pulse
|
||
stores passwords hashed, so token auth can't work there. The password never reaches the
|
||
player as long as streams are proxied.
|
||
aliases: further hostnames under which the control point reaches the same server
|
||
(e.g. a public domain while the bridge talks to the internal address).
|
||
"""
|
||
|
||
AUTH_PARAMS = {"u", "p", "t", "s", "apiKey", "c", "v", "f"}
|
||
|
||
def __init__(self, base, user, password, server_type="subsonic", api_key="", aliases="",
|
||
client="bluos-upnp-bridge"):
|
||
self.client = client or "bluos-upnp-bridge"
|
||
self.base = base
|
||
self.user = user
|
||
self.password = password
|
||
self.server_type = server_type if server_type in ("subsonic", "opensubsonic") else "subsonic"
|
||
self.api_key = api_key if self.server_type == "opensubsonic" else ""
|
||
self.host = (urllib.parse.urlsplit(base).hostname or "").lower() if base else ""
|
||
self.hosts = {self.host} | {h.strip().lower() for h in (aliases or "").replace(";", ",").split(",") if h.strip()}
|
||
self.hosts.discard("")
|
||
|
||
@property
|
||
def enabled(self):
|
||
return bool(self.base and (self.api_key or (self.user and self.password)))
|
||
|
||
@property
|
||
def password_auth(self):
|
||
return self.server_type == "opensubsonic" and not self.api_key
|
||
|
||
@property
|
||
def auth_label(self):
|
||
if self.api_key:
|
||
return "API-Key"
|
||
return f"{self.user} ({'Passwort' if self.password_auth else 'Token'})"
|
||
|
||
@staticmethod
|
||
def has_credentials(url):
|
||
q = urllib.parse.parse_qs(urllib.parse.urlsplit(url or "").query)
|
||
return any(k in q for k in ("p", "t", "apiKey"))
|
||
|
||
def _auth(self):
|
||
if self.api_key:
|
||
return {"apiKey": self.api_key, "v": "1.16.1", "c": self.client}
|
||
if self.password_auth:
|
||
return {"u": self.user, "p": "enc:" + self.password.encode("utf-8").hex(), "v": "1.16.1",
|
||
"c": self.client}
|
||
salt = secrets.token_hex(6)
|
||
token = hashlib.md5((self.password + salt).encode("utf-8")).hexdigest()
|
||
return {"u": self.user, "t": token, "s": salt, "v": "1.13.0", "c": self.client}
|
||
|
||
def is_subsonic(self, url):
|
||
parts = urllib.parse.urlsplit(url or "")
|
||
if "/rest/" not in parts.path:
|
||
return False
|
||
return not self.hosts or (parts.hostname or "").lower() in self.hosts
|
||
|
||
def song_id(self, url):
|
||
q = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
|
||
return (q.get("id") or [""])[0]
|
||
|
||
def rewrite(self, url, path="stream", offset=0):
|
||
"""Return the URL with the bridge's credentials, or the URL unchanged.
|
||
|
||
offset > 0 asks Subsonic to start the stream at that position. Subsonic only
|
||
honours timeOffset while transcoding, so a seeked stream is MP3 320k.
|
||
"""
|
||
if not (self.enabled and url and self.is_subsonic(url)):
|
||
return url
|
||
parts = urllib.parse.urlsplit(url)
|
||
q = [(k, v) for k, v in urllib.parse.parse_qsl(parts.query) if k not in self.AUTH_PARAMS]
|
||
if path == "stream":
|
||
if offset:
|
||
q = [(k, v) for k, v in q if k not in ("format", "maxBitRate", "timeOffset")]
|
||
q += [("format", "mp3"), ("maxBitRate", "320"), ("timeOffset", str(int(offset)))]
|
||
keys = {k for k, _ in q}
|
||
# no explicit transcoding request -> hand the original file to the player
|
||
if "format" not in keys and "maxBitRate" not in keys:
|
||
q.append(("format", "raw"))
|
||
if "estimateContentLength" not in keys:
|
||
q.append(("estimateContentLength", "true"))
|
||
q.extend(self._auth().items())
|
||
base = urllib.parse.urlsplit(self.base)
|
||
# Symfonium hands out download.view (Content-Type application/x-download),
|
||
# which BluOS refuses - always use stream.view / the requested endpoint instead
|
||
endpoint = re.sub(r"/rest/[^/]+$", f"/rest/{path}.view", parts.path)
|
||
return urllib.parse.urlunsplit((base.scheme, base.netloc, endpoint, urllib.parse.urlencode(q), ""))
|
||
|
||
def song_info(self, song_id):
|
||
"""Length and content type from getSong - unlike a stream request, this doesn't count as playback."""
|
||
info = {"duration": 0, "content_type": ""}
|
||
if not (self.enabled and song_id):
|
||
return info
|
||
try:
|
||
q = urllib.parse.urlencode({**self._auth(), "id": song_id})
|
||
with urllib.request.urlopen(f"{self.base}/rest/getSong.view?{q}", timeout=10) as r:
|
||
song = ET.fromstring(r.read()).find(".//{http://subsonic.org/restapi}song")
|
||
if song is not None:
|
||
info = {"duration": int(song.get("duration", "0") or 0), "content_type": song.get("contentType", "")}
|
||
except Exception as exc:
|
||
log.debug("getSong %s failed: %s", song_id, exc)
|
||
return info
|
||
|
||
def duration(self, song_id):
|
||
"""Track length in seconds from getSong (0 if unknown)."""
|
||
return self.song_info(song_id)["duration"]
|
||
|
||
def cover_url(self, song_id, size=600):
|
||
if not (self.enabled and song_id):
|
||
return ""
|
||
q = {**self._auth(), "id": song_id, "size": str(size)}
|
||
return f"{self.base}/rest/getCoverArt.view?{urllib.parse.urlencode(q)}"
|
||
|
||
|
||
# --------------------------------------------------------------------------- BluOS
|
||
|
||
class BluOS:
|
||
def __init__(self, host, port):
|
||
self.base = f"http://{host}:{port}"
|
||
|
||
def get(self, path, timeout=10):
|
||
with urllib.request.urlopen(self.base + path, timeout=timeout) as r:
|
||
return r.read().decode("utf-8", "replace")
|
||
|
||
def xml(self, path, timeout=10):
|
||
return ET.fromstring(self.get(path, timeout))
|
||
|
||
def name(self):
|
||
try:
|
||
return self.xml("/SyncStatus").get("name", "")
|
||
except Exception:
|
||
return ""
|
||
|
||
def play_url(self, url, title="", image="", seek=None):
|
||
params = {"url": url}
|
||
if title:
|
||
params["title1"] = title
|
||
if image:
|
||
params["image"] = image
|
||
if seek:
|
||
params["seek"] = str(int(seek))
|
||
return self.get("/Play?" + urllib.parse.urlencode(params))
|
||
|
||
def repeat_off(self):
|
||
"""BluOS repeat: 0 = queue, 1 = track, 2 = off."""
|
||
st = self.xml("/Status")
|
||
if st.findtext("repeat", "2") != "2":
|
||
log.info("switching BluOS repeat off (was %s)", st.findtext("repeat"))
|
||
self.get("/Repeat?state=2")
|
||
|
||
def resume(self):
|
||
return self.get("/Play")
|
||
|
||
def pause(self):
|
||
return self.get("/Pause")
|
||
|
||
def stop(self):
|
||
return self.get("/Stop")
|
||
|
||
def seek(self, secs):
|
||
return self.get(f"/Play?seek={int(secs)}")
|
||
|
||
def volume(self, level=None, mute=None):
|
||
params = {}
|
||
if level is not None:
|
||
params["level"] = str(int(level))
|
||
if mute is not None:
|
||
params["mute"] = "1" if mute else "0"
|
||
return self.xml("/Volume" + ("?" + urllib.parse.urlencode(params) if params else ""))
|
||
|
||
|
||
# --------------------------------------------------------------------------- renderer state
|
||
|
||
class Renderer:
|
||
"""Holds UPnP renderer state and translates actions to BluOS."""
|
||
|
||
def __init__(self, bluos, subsonic):
|
||
self.bluos = bluos
|
||
self.subsonic = subsonic
|
||
self.lock = threading.RLock()
|
||
self.events = None # set by EventManager
|
||
self.transport_state = "NO_MEDIA_PRESENT"
|
||
self.current_uri = ""
|
||
self.current_meta = ""
|
||
self.current_info = parse_didl("")
|
||
self.next_uri = ""
|
||
self.next_meta = ""
|
||
self.player_url = "" # URL we gave the player (rewritten)
|
||
self.owned = False # player is currently playing our URL
|
||
self.started_at = 0.0
|
||
self.expect_stop = False
|
||
self.position = 0
|
||
self.seek_offset = 0
|
||
self.player_duration = 0
|
||
self.volume = 0
|
||
self.mute = False
|
||
self.player = {} # last BluOS /Status values (web UI)
|
||
self.player_name = ""
|
||
self.fading = False # volume ramp in progress (ignore volume changes from the player)
|
||
self.on_change = None # OpenHome hook, called on every state change
|
||
self.hold_state_until = 0.0 # ignore play/pause from status polls started before our own action
|
||
self.session = None # gapless.Session while gapless playback runs
|
||
self.seg_index = 0
|
||
self._last_secs = -1 # stall detection for gapless streams
|
||
self._last_secs_at = 0.0
|
||
self.player_ok = 0.0
|
||
self.player_error = ""
|
||
self.control_points = {} # ip -> last seen (web UI)
|
||
|
||
def snapshot(self):
|
||
"""State for the web UI - no credentials, no player URLs."""
|
||
with self.lock:
|
||
nxt = parse_didl(self.next_meta)
|
||
info = self.current_info
|
||
return {
|
||
"transport_state": self.transport_state,
|
||
"owned": self.owned,
|
||
"position": self.position,
|
||
"duration": self.duration(),
|
||
"current": {"title": info["title"], "artist": info["artist"], "album": info["album"],
|
||
"subsonic": self.subsonic.is_subsonic(self.current_uri),
|
||
"song_id": self.subsonic.song_id(self.current_uri)
|
||
if self.subsonic.is_subsonic(self.current_uri) else ""} if self.current_uri else None,
|
||
"next": {"title": nxt["title"], "artist": nxt["artist"]} if self.next_uri else None,
|
||
"volume": self.volume,
|
||
"mute": self.mute,
|
||
"player": {**self.player, "name": self.player_name},
|
||
"player_ok": self.player_ok,
|
||
"player_error": self.player_error,
|
||
"control_points": dict(self.control_points),
|
||
}
|
||
|
||
# ---- helpers
|
||
def _title(self, info):
|
||
if info["title"] and info["artist"]:
|
||
return f"{info['title']} – {info['artist']}"
|
||
return info["title"] or info["artist"]
|
||
|
||
def _art(self, uri, info):
|
||
if self.subsonic.is_subsonic(uri):
|
||
song_id = self.subsonic.song_id(uri)
|
||
if song_id and self.subsonic.enabled:
|
||
# served by the bridge, so no server credentials end up on the player
|
||
# ".jpg" matters: BluOS only takes an ICY StreamUrl as artwork if it looks like an image
|
||
return f"http://{ADVERTISE_IP}:{HTTP_PORT}/cover/{urllib.parse.quote(song_id)}.jpg"
|
||
return ""
|
||
art = info.get("art", "")
|
||
return "" if self.subsonic.has_credentials(art) else art
|
||
|
||
def _is_ours(self, stream_url, st):
|
||
"""BluOS may re-encode the URL, so compare the significant query params only."""
|
||
if not stream_url:
|
||
# no stream URL reported: assume ours unless a named service is playing
|
||
return st.findtext("service", "") in ("", "http", "https")
|
||
if stream_url == self.player_url:
|
||
return True
|
||
a = urllib.parse.parse_qs(urllib.parse.urlsplit(stream_url).query)
|
||
b = urllib.parse.parse_qs(urllib.parse.urlsplit(self.player_url).query)
|
||
keys = [k for k in ("id", "s") if k in b]
|
||
if keys:
|
||
return all(a.get(k) == b.get(k) for k in keys)
|
||
return urllib.parse.urlsplit(stream_url).path == urllib.parse.urlsplit(self.player_url).path
|
||
|
||
def duration(self):
|
||
return self.current_info["duration"] or self.player_duration
|
||
|
||
def _hold_state(self):
|
||
self.hold_state_until = time.time() + 1.5
|
||
|
||
def _set_state(self, state, from_player=False):
|
||
if from_player and time.time() < self.hold_state_until:
|
||
return
|
||
if state != self.transport_state:
|
||
log.info("transport state %s -> %s", self.transport_state, state)
|
||
self.transport_state = state
|
||
self._notify_avt()
|
||
|
||
def _notify_avt(self):
|
||
if self.events:
|
||
self.events.notify_avt()
|
||
if self.on_change:
|
||
self.on_change()
|
||
|
||
def _proxied(self, uri=None):
|
||
uri = self.current_uri if uri is None else uri
|
||
return PROXY_STREAMS and self.subsonic.enabled and self.subsonic.is_subsonic(uri)
|
||
|
||
def _segment(self, uri, meta):
|
||
info = parse_didl(meta)
|
||
song = self.subsonic.song_info(self.subsonic.song_id(uri))
|
||
info["duration"] = info["duration"] or song["duration"]
|
||
info["content_type"] = song["content_type"]
|
||
return gapless.Segment(self.subsonic.rewrite(uri), uri, meta, info, self._title(info), self._art(uri, info))
|
||
|
||
def _close_session(self):
|
||
if self.session:
|
||
self.session.close()
|
||
self.session = None
|
||
|
||
def _start_gapless(self, seek):
|
||
info = self.current_info
|
||
self._close_session()
|
||
self.session = gapless.Session(self._segment(self.current_uri, self.current_meta), start_secs=seek)
|
||
if self.next_uri and self._proxied(self.next_uri):
|
||
self.session.offer_next(self._segment(self.next_uri, self.next_meta), 0)
|
||
self.seg_index = 0
|
||
self.player_url = STREAMS.add_session(self.session, self._title(info))
|
||
log.info("play %r at %s (gapless via %s)", self._title(info) or "?", hms(seek), self.player_url)
|
||
self.expect_stop = False
|
||
self.started_at = time.time()
|
||
self.seek_offset = 0
|
||
self.position = seek
|
||
if REPEAT_OFF:
|
||
try:
|
||
self.bluos.repeat_off()
|
||
except Exception as exc:
|
||
log.warning("could not switch repeat off: %s", exc)
|
||
self.bluos.play_url(self.player_url, self._title(info), self._art(self.current_uri, info))
|
||
self.owned = True
|
||
self._set_state("PLAYING")
|
||
|
||
def _start(self, seek=None):
|
||
"""Start playback of current_uri on the player.
|
||
|
||
Subsonic streams are served through the bridge's own stream proxy, which adds
|
||
Content-Length and byte-range support. Without it BluOS treats the file like
|
||
internet radio: it can't seek and reconnects at EOF.
|
||
"""
|
||
info = self.current_info
|
||
seek = int(seek or 0)
|
||
proxied = self._proxied()
|
||
if proxied and GAPLESS:
|
||
return self._start_gapless(seek)
|
||
self._close_session()
|
||
if not proxied and self.subsonic.password_auth and self.subsonic.is_subsonic(self.current_uri):
|
||
# without the proxy the player would see (and store) the service account password
|
||
log.error("password auth needs PROXY_STREAMS=true or an API key - refusing to play")
|
||
self._set_state("STOPPED")
|
||
raise UPnPError(716, "Resource not found (enable stream proxy or set an API key)")
|
||
offset_supported = not proxied and self.subsonic.is_subsonic(self.current_uri) and self.subsonic.enabled
|
||
upstream = self.subsonic.rewrite(self.current_uri, offset=seek if offset_supported else 0)
|
||
if upstream == self.current_uri and self.subsonic.has_credentials(upstream):
|
||
# never hand a control point's credentials to the player (it shows and stores the URL)
|
||
log.error("refusing to play URL with foreign credentials: %s", mask_url(upstream))
|
||
self._set_state("STOPPED")
|
||
raise UPnPError(716, "Resource not found (credentials could not be replaced)")
|
||
self.player_url = STREAMS.add(upstream, self._title(info)) if proxied else upstream
|
||
log.info("play %r at %s (%s)", self._title(info) or "?", hms(seek),
|
||
f"via proxy {self.player_url}" if proxied else mask_url(self.player_url))
|
||
self.expect_stop = False
|
||
self.started_at = time.time()
|
||
self.seek_offset = seek if offset_supported else 0
|
||
self.position = seek
|
||
if REPEAT_OFF:
|
||
# the control point handles repeat itself; BluOS repeat would loop the URL
|
||
try:
|
||
self.bluos.repeat_off()
|
||
except Exception as exc:
|
||
log.warning("could not switch repeat off: %s", exc)
|
||
self.bluos.play_url(self.player_url, self._title(info), self._art(self.current_uri, info),
|
||
None if (offset_supported or proxied) else seek)
|
||
if proxied and seek:
|
||
# /Play with url+seek is unreliable; seek once the stream is running
|
||
threading.Timer(1.5, self._late_seek, args=(self.player_url, seek)).start()
|
||
self.owned = True
|
||
self._set_state("PLAYING")
|
||
|
||
def _late_seek(self, url, secs):
|
||
with self.lock:
|
||
if self.player_url == url and self.owned:
|
||
try:
|
||
self.bluos.seek(secs)
|
||
except Exception as exc:
|
||
log.warning("seek after start failed: %s", exc)
|
||
|
||
# ---- AVTransport
|
||
def set_uri(self, uri, meta):
|
||
with self.lock:
|
||
was_playing = self.transport_state == "PLAYING" and self.owned
|
||
self.current_uri, self.current_meta = uri, meta
|
||
self.current_info = parse_didl(meta)
|
||
self.player_duration = 0
|
||
self.position = 0
|
||
log.info("SetAVTransportURI %r %s", self._title(self.current_info), mask_url(uri))
|
||
if was_playing:
|
||
self._switch(self._start)
|
||
else:
|
||
self._set_state("STOPPED")
|
||
self._notify_avt()
|
||
|
||
def set_next_uri(self, uri, meta):
|
||
with self.lock:
|
||
self.next_uri, self.next_meta = uri, meta
|
||
log.info("SetNextAVTransportURI %r", self._title(parse_didl(meta)) if uri else "(none)")
|
||
if self.session and self.owned and not self.session.closed:
|
||
if uri and self._proxied(uri):
|
||
ok = self.session.offer_next(self._segment(uri, meta), self.seg_index)
|
||
else:
|
||
ok = self.session.clear_next(self.seg_index)
|
||
if not ok:
|
||
log.warning("gapless: next track already sent to the player, queue change applies after it")
|
||
self._notify_avt()
|
||
|
||
# ---- fades (volume ramps via the BluOS API)
|
||
def _fade_steps(self):
|
||
return max(2, min(20, FADE_MS // 50))
|
||
|
||
def _fade_out(self):
|
||
"""Ramp the volume down; returns the level to restore afterwards (None if no fade)."""
|
||
if FADE_MS <= 0 or not self.owned or self.transport_state != "PLAYING" or self.mute:
|
||
return None
|
||
level = self.volume
|
||
if level <= 0:
|
||
return None
|
||
self.fading = True
|
||
try:
|
||
steps = self._fade_steps()
|
||
for i in range(1, steps + 1):
|
||
self.bluos.volume(level=round(level * (1 - i / steps)))
|
||
time.sleep(FADE_MS / 1000 / steps)
|
||
except Exception as exc:
|
||
log.warning("fade out failed: %s", exc)
|
||
return level
|
||
|
||
def _restore_volume(self, level):
|
||
if level is None:
|
||
return
|
||
try:
|
||
self.bluos.volume(level=level)
|
||
except Exception as exc:
|
||
log.warning("restoring volume %s failed: %s", level, exc)
|
||
finally:
|
||
self.fading = False
|
||
|
||
def _fade_in_async(self, level, wait_for_audio=False):
|
||
"""Ramp the volume up from 0 in the background (player must already be at 0)."""
|
||
def run():
|
||
try:
|
||
if wait_for_audio:
|
||
deadline = time.time() + 4
|
||
while time.time() < deadline:
|
||
st = self.bluos.xml("/Status", timeout=3).findtext("state", "")
|
||
if st in ("play", "stream"):
|
||
break
|
||
time.sleep(0.2)
|
||
steps = self._fade_steps()
|
||
for i in range(1, steps + 1):
|
||
self.bluos.volume(level=round(level * i / steps))
|
||
time.sleep(FADE_MS / 1000 / steps)
|
||
except Exception as exc:
|
||
log.warning("fade in failed: %s", exc)
|
||
finally:
|
||
self._restore_volume(level)
|
||
threading.Thread(target=run, name="fade-in", daemon=True).start()
|
||
|
||
def _silence_for_fade_in(self):
|
||
"""Set the volume to 0 before audio starts; returns the target level (None if no fade)."""
|
||
if FADE_MS <= 0 or self.mute or self.volume <= 0:
|
||
return None
|
||
level = self.volume
|
||
self.fading = True
|
||
try:
|
||
self.bluos.volume(level=0)
|
||
except Exception as exc:
|
||
log.warning("fade preparation failed: %s", exc)
|
||
self.fading = False
|
||
return None
|
||
return level
|
||
|
||
def play(self):
|
||
with self.lock:
|
||
if not self.current_uri:
|
||
raise UPnPError(701, "Transition not available")
|
||
if self.transport_state == "PAUSED_PLAYBACK" and self.owned:
|
||
level = self._silence_for_fade_in()
|
||
self.bluos.resume()
|
||
self._hold_state()
|
||
self._set_state("PLAYING")
|
||
if level is not None:
|
||
self._fade_in_async(level)
|
||
elif self.transport_state != "PLAYING" or not self.owned:
|
||
level = self._silence_for_fade_in()
|
||
try:
|
||
self._start()
|
||
finally:
|
||
if level is not None:
|
||
self._fade_in_async(level, wait_for_audio=True)
|
||
|
||
def pause(self):
|
||
with self.lock:
|
||
if self.owned:
|
||
level = self._fade_out()
|
||
self.bluos.pause()
|
||
self._restore_volume(level)
|
||
self._hold_state()
|
||
self._set_state("PAUSED_PLAYBACK")
|
||
|
||
def stop(self):
|
||
with self.lock:
|
||
self.expect_stop = True
|
||
if self.owned:
|
||
level = self._fade_out()
|
||
self.bluos.pause() # /Stop would clear the queue display; pause is enough
|
||
self._restore_volume(level)
|
||
self._hold_state()
|
||
self._close_session()
|
||
self.position = 0
|
||
self._set_state("STOPPED")
|
||
|
||
def seek(self, unit, target):
|
||
with self.lock:
|
||
if unit not in ("REL_TIME", "ABS_TIME"):
|
||
raise UPnPError(710, "Seek mode not supported")
|
||
secs = parse_hms(target)
|
||
log.info("seek to %s", hms(secs))
|
||
if not self.current_uri:
|
||
raise UPnPError(701, "Transition not available")
|
||
paused = self.transport_state == "PAUSED_PLAYBACK"
|
||
if self._proxied() and self.owned and not self.session and \
|
||
self.transport_state in ("PLAYING", "PAUSED_PLAYBACK"):
|
||
# the proxy supports byte ranges, so BluOS can seek natively
|
||
self.bluos.seek(secs)
|
||
self.position = secs
|
||
self.started_at = time.time()
|
||
self._notify_avt()
|
||
return
|
||
self._start(seek=secs)
|
||
if paused:
|
||
self.bluos.pause()
|
||
self._set_state("PAUSED_PLAYBACK")
|
||
self._notify_avt()
|
||
|
||
def next(self):
|
||
with self.lock:
|
||
if not self.next_uri:
|
||
raise UPnPError(711, "No next track")
|
||
self._switch(self._advance)
|
||
|
||
def _switch(self, start):
|
||
"""Track change by the user: fade out, start the new track, fade in."""
|
||
level = self._fade_out()
|
||
try:
|
||
start()
|
||
finally:
|
||
if level is not None:
|
||
self._fade_in_async(level, wait_for_audio=True)
|
||
|
||
def _advance(self):
|
||
self.current_uri, self.current_meta = self.next_uri, self.next_meta
|
||
self.current_info = parse_didl(self.current_meta)
|
||
self.next_uri, self.next_meta = "", ""
|
||
self.player_duration = 0
|
||
self._start()
|
||
|
||
# ---- RenderingControl
|
||
def set_volume(self, level):
|
||
with self.lock:
|
||
v = self.bluos.volume(level=level)
|
||
self.volume = int(v.text or level)
|
||
if self.events:
|
||
self.events.notify_rcs()
|
||
|
||
def set_mute(self, mute):
|
||
with self.lock:
|
||
self.bluos.volume(mute=mute)
|
||
self.mute = bool(mute)
|
||
if self.events:
|
||
self.events.notify_rcs()
|
||
|
||
# ---- player status
|
||
def on_status(self, st):
|
||
"""Called by the poller with each /Status document."""
|
||
with self.lock:
|
||
self.player = {k: st.findtext(k, "") for k in (
|
||
"state", "service", "title1", "title2", "secs", "totlen", "repeat", "shuffle",
|
||
"streamFormat", "volume", "mute", "name")}
|
||
self.player_ok = time.time()
|
||
self.player_error = ""
|
||
state = st.findtext("state", "")
|
||
stream_url = st.findtext("streamUrl", "")
|
||
vol = int(st.findtext("volume", "0") or 0)
|
||
mute = st.findtext("mute", "0") == "1"
|
||
if not self.fading and (vol, mute) != (self.volume, self.mute):
|
||
self.volume, self.mute = vol, mute
|
||
if self.events:
|
||
self.events.notify_rcs()
|
||
if not self.owned:
|
||
return
|
||
settling = time.time() - self.started_at < 4
|
||
ours = self._is_ours(stream_url, st)
|
||
if self.session and ours and not settling:
|
||
self._gapless_status(st, state)
|
||
return
|
||
if ours and not settling:
|
||
new_pos = self.seek_offset + int(float(st.findtext("secs", "0") or 0))
|
||
duration = self.duration()
|
||
if duration and self.position >= duration - 10 and new_pos < 10 and not self.expect_stop:
|
||
# BluOS "repeat one" restarts the URL instead of stopping
|
||
log.info("player restarted the same track (repeat?), treating as track end")
|
||
self._track_ended()
|
||
return
|
||
# Without Content-Length BluOS treats the stream like internet radio: at EOF it keeps
|
||
# reporting "stream"/"connecting" and the counter runs on. Use the known duration.
|
||
known = self.current_info["duration"]
|
||
if known and new_pos >= known + 1 and not self.expect_stop and state in ("stream", "play", "connecting"):
|
||
log.info("position %s passed track length %s, treating as track end", hms(new_pos), hms(known))
|
||
self._track_ended()
|
||
return
|
||
self.position = new_pos
|
||
totlen = int(float(st.findtext("totlen", "0") or 0))
|
||
self.player_duration = self.seek_offset + totlen if totlen else 0
|
||
|
||
if self.player_url and not settling and not ours:
|
||
# At the end of a URL stream BluOS falls back to its own play queue
|
||
# (e.g. Deezer), which looks exactly like someone taking over.
|
||
duration = self.duration()
|
||
if duration and self.position >= duration - 8 and not self.expect_stop:
|
||
self._track_ended()
|
||
else:
|
||
log.info("player was taken over by another source")
|
||
self.owned = False
|
||
self._set_state("STOPPED")
|
||
return
|
||
if state in ("play", "stream"):
|
||
self._set_state("PLAYING", from_player=True)
|
||
elif state == "pause":
|
||
self._set_state("PAUSED_PLAYBACK", from_player=True)
|
||
elif state == "stop" and not settling:
|
||
if self.expect_stop:
|
||
self._set_state("STOPPED")
|
||
else:
|
||
self._track_ended()
|
||
|
||
def rebase_session(self, session):
|
||
with self.lock:
|
||
if session is not self.session:
|
||
return
|
||
log.info("gapless: player reconnected, continuing %r at %s",
|
||
self._title(self.current_info), hms(self.position))
|
||
session.rebase(self.seg_index, self.position)
|
||
self.seg_index = 0
|
||
self.started_at = time.time()
|
||
self._last_secs, self._last_secs_at = -1, time.time()
|
||
|
||
def _gapless_status(self, st, state):
|
||
"""Map BluOS' stream position to the track that is audible right now."""
|
||
sess = self.session
|
||
secs = int(float(st.findtext("secs", "0") or 0))
|
||
if state in ("pause", "stop", "connecting") and secs == 0 and self.position > 0:
|
||
# BluOS resets the counter when it drops a paused stream - keep the known position
|
||
return
|
||
sess.playback_secs = secs
|
||
starts = sess.starts()
|
||
streamed = [i for i, s in enumerate(sess.segments) if s.streamed or i == 0]
|
||
last = max(streamed)
|
||
idx = max(i for i in streamed if secs >= starts[i])
|
||
while self.seg_index < idx:
|
||
self.seg_index += 1
|
||
seg = sess.segments[self.seg_index]
|
||
self.current_uri, self.current_meta, self.current_info = seg.uri, seg.meta, seg.info
|
||
nxt = sess.segments[self.seg_index + 1] if self.seg_index + 1 < len(sess.segments) else None
|
||
self.next_uri, self.next_meta = (nxt.uri, nxt.meta) if nxt else ("", "")
|
||
log.info("gapless: now playing %r", seg.label)
|
||
self._notify_avt()
|
||
seg = sess.segments[self.seg_index]
|
||
self.position = secs - starts[self.seg_index] + (sess.start_secs if self.seg_index == 0 else 0)
|
||
self.player_duration = 0
|
||
# BluOS freezes its counter (or shows "connecting") once the sent audio has run out
|
||
now = time.time()
|
||
if secs != self._last_secs:
|
||
self._last_secs, self._last_secs_at = secs, now
|
||
stalled = state == "connecting" or (state in ("play", "stream") and now - self._last_secs_at >= 2.5)
|
||
drained = seg.duration and self.position >= seg.duration - 2 and stalled
|
||
if state in ("play", "stream"):
|
||
self._set_state("PLAYING", from_player=True)
|
||
elif state == "pause":
|
||
self._set_state("PAUSED_PLAYBACK", from_player=True)
|
||
if self.seg_index == last and not self.expect_stop and (
|
||
(seg.duration and self.position >= seg.duration + 1) or state == "stop" or drained):
|
||
# end of the stitched stream; a follow-up that couldn't be stitched starts normally
|
||
log.info("gapless stream finished at %r", seg.label)
|
||
old = self.session
|
||
# pause (or start the unstitched follow-up) before closing the stream - otherwise BluOS
|
||
# reconnects to the ended stream and briefly plays its beginning again
|
||
self._track_ended()
|
||
if self.session is old:
|
||
self._close_session()
|
||
|
||
def _track_ended(self):
|
||
if self.next_uri:
|
||
log.info("track ended, advancing to next")
|
||
self._advance()
|
||
return
|
||
log.info("track ended, nothing queued")
|
||
try:
|
||
self.bluos.pause() # keep BluOS from continuing with its own queue
|
||
except Exception as exc:
|
||
log.warning("pause after track end failed: %s", exc)
|
||
self.position = 0
|
||
self.expect_stop = True
|
||
self._set_state("STOPPED")
|
||
|
||
|
||
class UPnPError(Exception):
|
||
def __init__(self, code, desc):
|
||
super().__init__(desc)
|
||
self.code, self.desc = code, desc
|
||
|
||
|
||
def poll_status(renderer, bluos, stop_event):
|
||
"""Long-poll BluOS /Status and feed it into the renderer."""
|
||
etag = ""
|
||
while not stop_event.is_set():
|
||
try:
|
||
path = "/Status" + (f"?timeout=30&etag={etag}" if etag else "")
|
||
st = bluos.xml(path, timeout=40)
|
||
etag = st.get("etag", "")
|
||
renderer.on_status(st)
|
||
if renderer.on_change:
|
||
renderer.on_change()
|
||
if renderer.owned and renderer.transport_state == "PLAYING":
|
||
# long-poll returns on change only; refresh position regularly while playing
|
||
time.sleep(1)
|
||
etag = ""
|
||
except (socket.timeout, TimeoutError):
|
||
etag = "" # long-poll ran out, just ask again
|
||
except Exception as exc:
|
||
log.warning("status poll failed: %s", exc)
|
||
renderer.player_error = str(exc)
|
||
etag = ""
|
||
stop_event.wait(5)
|
||
|
||
|
||
# --------------------------------------------------------------------------- SCPD
|
||
|
||
def _arg(name, direction, var):
|
||
return name, direction, var
|
||
|
||
|
||
AVT_VARS = {
|
||
"TransportState": ("string", "no", ["STOPPED", "PLAYING", "PAUSED_PLAYBACK", "TRANSITIONING", "NO_MEDIA_PRESENT"]),
|
||
"TransportStatus": ("string", "no", ["OK", "ERROR_OCCURRED"]),
|
||
"PlaybackStorageMedium": ("string", "no", ["NETWORK", "NONE"]),
|
||
"RecordStorageMedium": ("string", "no", ["NOT_IMPLEMENTED"]),
|
||
"PossiblePlaybackStorageMedia": ("string", "no", None),
|
||
"PossibleRecordStorageMedia": ("string", "no", None),
|
||
"CurrentPlayMode": ("string", "no", ["NORMAL"]),
|
||
"TransportPlaySpeed": ("string", "no", ["1"]),
|
||
"RecordMediumWriteStatus": ("string", "no", ["NOT_IMPLEMENTED"]),
|
||
"CurrentRecordQualityMode": ("string", "no", ["NOT_IMPLEMENTED"]),
|
||
"PossibleRecordQualityModes": ("string", "no", None),
|
||
"NumberOfTracks": ("ui4", "no", None),
|
||
"CurrentTrack": ("ui4", "no", None),
|
||
"CurrentTrackDuration": ("string", "no", None),
|
||
"CurrentMediaDuration": ("string", "no", None),
|
||
"CurrentTrackMetaData": ("string", "no", None),
|
||
"CurrentTrackURI": ("string", "no", None),
|
||
"AVTransportURI": ("string", "no", None),
|
||
"AVTransportURIMetaData": ("string", "no", None),
|
||
"NextAVTransportURI": ("string", "no", None),
|
||
"NextAVTransportURIMetaData": ("string", "no", None),
|
||
"RelativeTimePosition": ("string", "no", None),
|
||
"AbsoluteTimePosition": ("string", "no", None),
|
||
"RelativeCounterPosition": ("i4", "no", None),
|
||
"AbsoluteCounterPosition": ("i4", "no", None),
|
||
"CurrentTransportActions": ("string", "no", None),
|
||
"LastChange": ("string", "yes", None),
|
||
"A_ARG_TYPE_SeekMode": ("string", "no", ["REL_TIME", "ABS_TIME", "TRACK_NR"]),
|
||
"A_ARG_TYPE_SeekTarget": ("string", "no", None),
|
||
"A_ARG_TYPE_InstanceID": ("ui4", "no", None),
|
||
}
|
||
I = _arg("InstanceID", "in", "A_ARG_TYPE_InstanceID")
|
||
AVT_ACTIONS = {
|
||
"SetAVTransportURI": [I, _arg("CurrentURI", "in", "AVTransportURI"),
|
||
_arg("CurrentURIMetaData", "in", "AVTransportURIMetaData")],
|
||
"SetNextAVTransportURI": [I, _arg("NextURI", "in", "NextAVTransportURI"),
|
||
_arg("NextURIMetaData", "in", "NextAVTransportURIMetaData")],
|
||
"GetMediaInfo": [I, _arg("NrTracks", "out", "NumberOfTracks"),
|
||
_arg("MediaDuration", "out", "CurrentMediaDuration"),
|
||
_arg("CurrentURI", "out", "AVTransportURI"),
|
||
_arg("CurrentURIMetaData", "out", "AVTransportURIMetaData"),
|
||
_arg("NextURI", "out", "NextAVTransportURI"),
|
||
_arg("NextURIMetaData", "out", "NextAVTransportURIMetaData"),
|
||
_arg("PlayMedium", "out", "PlaybackStorageMedium"),
|
||
_arg("RecordMedium", "out", "RecordStorageMedium"),
|
||
_arg("WriteStatus", "out", "RecordMediumWriteStatus")],
|
||
"GetTransportInfo": [I, _arg("CurrentTransportState", "out", "TransportState"),
|
||
_arg("CurrentTransportStatus", "out", "TransportStatus"),
|
||
_arg("CurrentSpeed", "out", "TransportPlaySpeed")],
|
||
"GetPositionInfo": [I, _arg("Track", "out", "CurrentTrack"),
|
||
_arg("TrackDuration", "out", "CurrentTrackDuration"),
|
||
_arg("TrackMetaData", "out", "CurrentTrackMetaData"),
|
||
_arg("TrackURI", "out", "CurrentTrackURI"),
|
||
_arg("RelTime", "out", "RelativeTimePosition"),
|
||
_arg("AbsTime", "out", "AbsoluteTimePosition"),
|
||
_arg("RelCount", "out", "RelativeCounterPosition"),
|
||
_arg("AbsCount", "out", "AbsoluteCounterPosition")],
|
||
"GetDeviceCapabilities": [I, _arg("PlayMedia", "out", "PossiblePlaybackStorageMedia"),
|
||
_arg("RecMedia", "out", "PossibleRecordStorageMedia"),
|
||
_arg("RecQualityModes", "out", "PossibleRecordQualityModes")],
|
||
"GetTransportSettings": [I, _arg("PlayMode", "out", "CurrentPlayMode"),
|
||
_arg("RecQualityMode", "out", "CurrentRecordQualityMode")],
|
||
"GetCurrentTransportActions": [I, _arg("Actions", "out", "CurrentTransportActions")],
|
||
"Stop": [I],
|
||
"Play": [I, _arg("Speed", "in", "TransportPlaySpeed")],
|
||
"Pause": [I],
|
||
"Seek": [I, _arg("Unit", "in", "A_ARG_TYPE_SeekMode"), _arg("Target", "in", "A_ARG_TYPE_SeekTarget")],
|
||
"Next": [I],
|
||
"Previous": [I],
|
||
}
|
||
|
||
RCS_VARS = {
|
||
"PresetNameList": ("string", "no", None),
|
||
"Mute": ("boolean", "no", None),
|
||
"Volume": ("ui2", "no", (0, 100, 1)),
|
||
"LastChange": ("string", "yes", None),
|
||
"A_ARG_TYPE_Channel": ("string", "no", ["Master"]),
|
||
"A_ARG_TYPE_InstanceID": ("ui4", "no", None),
|
||
"A_ARG_TYPE_PresetName": ("string", "no", ["FactoryDefaults"]),
|
||
}
|
||
CH = _arg("Channel", "in", "A_ARG_TYPE_Channel")
|
||
RCS_ACTIONS = {
|
||
"ListPresets": [I, _arg("CurrentPresetNameList", "out", "PresetNameList")],
|
||
"SelectPreset": [I, _arg("PresetName", "in", "A_ARG_TYPE_PresetName")],
|
||
"GetMute": [I, CH, _arg("CurrentMute", "out", "Mute")],
|
||
"SetMute": [I, CH, _arg("DesiredMute", "in", "Mute")],
|
||
"GetVolume": [I, CH, _arg("CurrentVolume", "out", "Volume")],
|
||
"SetVolume": [I, CH, _arg("DesiredVolume", "in", "Volume")],
|
||
}
|
||
|
||
CM_VARS = {
|
||
"SourceProtocolInfo": ("string", "yes", None),
|
||
"SinkProtocolInfo": ("string", "yes", None),
|
||
"CurrentConnectionIDs": ("string", "yes", None),
|
||
"A_ARG_TYPE_ConnectionStatus": ("string", "no", ["OK", "ContentFormatMismatch", "InsufficientBandwidth",
|
||
"UnreliableChannel", "Unknown"]),
|
||
"A_ARG_TYPE_ConnectionManager": ("string", "no", None),
|
||
"A_ARG_TYPE_Direction": ("string", "no", ["Input", "Output"]),
|
||
"A_ARG_TYPE_ProtocolInfo": ("string", "no", None),
|
||
"A_ARG_TYPE_ConnectionID": ("i4", "no", None),
|
||
"A_ARG_TYPE_AVTransportID": ("i4", "no", None),
|
||
"A_ARG_TYPE_RcsID": ("i4", "no", None),
|
||
}
|
||
CM_ACTIONS = {
|
||
"GetProtocolInfo": [_arg("Source", "out", "SourceProtocolInfo"), _arg("Sink", "out", "SinkProtocolInfo")],
|
||
"GetCurrentConnectionIDs": [_arg("ConnectionIDs", "out", "CurrentConnectionIDs")],
|
||
"GetCurrentConnectionInfo": [_arg("ConnectionID", "in", "A_ARG_TYPE_ConnectionID"),
|
||
_arg("RcsID", "out", "A_ARG_TYPE_RcsID"),
|
||
_arg("AVTransportID", "out", "A_ARG_TYPE_AVTransportID"),
|
||
_arg("ProtocolInfo", "out", "A_ARG_TYPE_ProtocolInfo"),
|
||
_arg("PeerConnectionManager", "out", "A_ARG_TYPE_ConnectionManager"),
|
||
_arg("PeerConnectionID", "out", "A_ARG_TYPE_ConnectionID"),
|
||
_arg("Direction", "out", "A_ARG_TYPE_Direction"),
|
||
_arg("Status", "out", "A_ARG_TYPE_ConnectionStatus")],
|
||
}
|
||
|
||
SCPD_SPECS_AV = {
|
||
"AVTransport": (AVT_ACTIONS, AVT_VARS),
|
||
"RenderingControl": (RCS_ACTIONS, RCS_VARS),
|
||
"ConnectionManager": (CM_ACTIONS, CM_VARS),
|
||
}
|
||
|
||
|
||
SCPD_SPECS = dict(SCPD_SPECS_AV)
|
||
|
||
|
||
def build_scpd(actions, variables):
|
||
out = ['<?xml version="1.0" encoding="utf-8"?>',
|
||
'<scpd xmlns="urn:schemas-upnp-org:service-1-0">',
|
||
'<specVersion><major>1</major><minor>0</minor></specVersion><actionList>']
|
||
for name, args in actions.items():
|
||
out.append(f"<action><name>{name}</name><argumentList>")
|
||
for arg, direction, var in args:
|
||
out.append(f"<argument><name>{arg}</name><direction>{direction}</direction>"
|
||
f"<relatedStateVariable>{var}</relatedStateVariable></argument>")
|
||
out.append("</argumentList></action>")
|
||
out.append("</actionList><serviceStateTable>")
|
||
for name, (dtype, events, allowed) in variables.items():
|
||
out.append(f'<stateVariable sendEvents="{events}"><name>{name}</name><dataType>{dtype}</dataType>')
|
||
if isinstance(allowed, list):
|
||
out.append("<allowedValueList>" + "".join(f"<allowedValue>{v}</allowedValue>" for v in allowed)
|
||
+ "</allowedValueList>")
|
||
elif isinstance(allowed, tuple):
|
||
out.append(f"<allowedValueRange><minimum>{allowed[0]}</minimum><maximum>{allowed[1]}</maximum>"
|
||
f"<step>{allowed[2]}</step></allowedValueRange>")
|
||
out.append("</stateVariable>")
|
||
out.append("</serviceStateTable></scpd>")
|
||
return "".join(out)
|
||
|
||
|
||
def build_description(name):
|
||
services = "".join(
|
||
f"<service><serviceType>{st}</serviceType><serviceId>"
|
||
f"{openhome.SERVICE_ID_PREFIX if sid in openhome.SERVICES else 'urn:upnp-org:serviceId:'}{sid}</serviceId>"
|
||
f"<SCPDURL>/{sid}.xml</SCPDURL><controlURL>/ctl/{sid}</controlURL>"
|
||
f"<eventSubURL>/evt/{sid}</eventSubURL></service>"
|
||
for sid, st in SERVICES.items()
|
||
)
|
||
return (
|
||
'<?xml version="1.0" encoding="utf-8"?>'
|
||
'<root xmlns="urn:schemas-upnp-org:device-1-0" xmlns:dlna="urn:schemas-dlna-org:device-1-0">'
|
||
"<specVersion><major>1</major><minor>0</minor></specVersion><device>"
|
||
f"<deviceType>{DEVICE_TYPE}</deviceType>"
|
||
f"<friendlyName>{html.escape(name)}</friendlyName>"
|
||
"<manufacturer>bluos-upnp-bridge</manufacturer>"
|
||
"<manufacturerURL>https://git.techniverse.net/scriptos/bluos-upnp-bridge</manufacturerURL>"
|
||
"<modelDescription>UPnP/DLNA bridge for BluOS players</modelDescription>"
|
||
f"<modelName>BluOS UPnP Bridge</modelName><modelNumber>{VERSION}</modelNumber>"
|
||
f"<UDN>{DEVICE_UUID}</UDN>"
|
||
"<dlna:X_DLNADOC>DMR-1.50</dlna:X_DLNADOC>"
|
||
f"<serviceList>{services}</serviceList>"
|
||
f"<presentationURL>http://{BLUOS_HOST}/</presentationURL>"
|
||
"</device></root>"
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------- GENA eventing
|
||
|
||
class EventManager:
|
||
def __init__(self, renderer):
|
||
self.renderer = renderer
|
||
self.subs = {} # sid -> dict(service, callbacks, seq, expires)
|
||
self.lock = threading.Lock()
|
||
|
||
def subscribe(self, service, callbacks, timeout):
|
||
sid = "uuid:" + str(uuid.uuid4())
|
||
with self.lock:
|
||
self.subs[sid] = {"service": service, "callbacks": callbacks, "seq": 0,
|
||
"expires": time.time() + timeout}
|
||
log.info("SUBSCRIBE %s from %s (%s)", service, callbacks[0] if callbacks else "?", sid)
|
||
threading.Timer(0.3, self._send_initial, args=(sid,)).start()
|
||
return sid
|
||
|
||
def renew(self, sid, timeout):
|
||
with self.lock:
|
||
if sid not in self.subs:
|
||
return False
|
||
self.subs[sid]["expires"] = time.time() + timeout
|
||
return True
|
||
|
||
def unsubscribe(self, sid):
|
||
with self.lock:
|
||
return self.subs.pop(sid, None) is not None
|
||
|
||
def _send_initial(self, sid):
|
||
with self.lock:
|
||
sub = self.subs.get(sid)
|
||
if sub:
|
||
self._send(sid, sub, self._body(sub["service"]))
|
||
|
||
def _body(self, service):
|
||
r = self.renderer
|
||
if service in openhome.SERVICES:
|
||
props = OH.state(service) if OH else {}
|
||
inner = "".join(f"<e:property><{k}>{html.escape(v)}</{k}></e:property>" for k, v in props.items())
|
||
return ('<?xml version="1.0" encoding="utf-8"?>'
|
||
f'<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">{inner}</e:propertyset>')
|
||
if service == "AVTransport":
|
||
vals = avt_state(r)
|
||
inner = "".join(f'<{k} val="{html.escape(str(v))}"/>' for k, v in vals.items())
|
||
last = f'<Event xmlns="urn:schemas-upnp-org:metadata-1-0/AVT/"><InstanceID val="0">{inner}</InstanceID></Event>'
|
||
props = {"LastChange": last}
|
||
elif service == "RenderingControl":
|
||
last = ('<Event xmlns="urn:schemas-upnp-org:metadata-1-0/RCS/"><InstanceID val="0">'
|
||
f'<Volume channel="Master" val="{r.volume}"/><Mute channel="Master" val="{int(r.mute)}"/>'
|
||
'<PresetNameList val="FactoryDefaults"/></InstanceID></Event>')
|
||
props = {"LastChange": last}
|
||
else:
|
||
props = {"SourceProtocolInfo": "", "SinkProtocolInfo": SINK_PROTOCOLS, "CurrentConnectionIDs": "0"}
|
||
inner = "".join(f"<e:property><{k}>{html.escape(v)}</{k}></e:property>" for k, v in props.items())
|
||
return ('<?xml version="1.0" encoding="utf-8"?>'
|
||
f'<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">{inner}</e:propertyset>')
|
||
|
||
def _send(self, sid, sub, body):
|
||
data = body.encode("utf-8")
|
||
for cb in sub["callbacks"]:
|
||
try:
|
||
req = urllib.request.Request(cb, data=data, method="NOTIFY", headers={
|
||
"Content-Type": 'text/xml; charset="utf-8"', "NT": "upnp:event",
|
||
"NTS": "upnp:propchange", "SID": sid, "SEQ": str(sub["seq"])})
|
||
urllib.request.urlopen(req, timeout=3).close()
|
||
break
|
||
except Exception as exc:
|
||
log.debug("NOTIFY to %s failed: %s", cb, exc)
|
||
sub["seq"] += 1
|
||
|
||
def _notify(self, service):
|
||
now = time.time()
|
||
with self.lock:
|
||
for sid in [s for s, v in self.subs.items() if v["expires"] < now]:
|
||
del self.subs[sid]
|
||
targets = [(sid, sub) for sid, sub in self.subs.items() if sub["service"] == service]
|
||
if not targets:
|
||
return
|
||
body = self._body(service)
|
||
for sid, sub in targets:
|
||
threading.Thread(target=self._send, args=(sid, sub, body), daemon=True).start()
|
||
|
||
def notify_avt(self):
|
||
self._notify("AVTransport")
|
||
|
||
def notify_rcs(self):
|
||
self._notify("RenderingControl")
|
||
|
||
|
||
def transport_actions(r):
|
||
if r.transport_state == "PLAYING":
|
||
acts = ["Pause", "Stop", "Seek"]
|
||
elif r.transport_state == "PAUSED_PLAYBACK":
|
||
acts = ["Play", "Stop", "Seek"]
|
||
elif r.current_uri:
|
||
acts = ["Play", "Seek"]
|
||
else:
|
||
acts = []
|
||
if r.next_uri:
|
||
acts.append("Next")
|
||
return ",".join(acts)
|
||
|
||
|
||
def avt_state(r):
|
||
return {
|
||
"TransportState": r.transport_state,
|
||
"TransportStatus": "OK",
|
||
"TransportPlaySpeed": "1",
|
||
"CurrentPlayMode": "NORMAL",
|
||
"NumberOfTracks": 1 if r.current_uri else 0,
|
||
"CurrentTrack": 1 if r.current_uri else 0,
|
||
"CurrentTrackDuration": hms(r.duration()),
|
||
"CurrentMediaDuration": hms(r.duration()),
|
||
"AVTransportURI": r.current_uri,
|
||
"AVTransportURIMetaData": r.current_meta,
|
||
"CurrentTrackURI": r.current_uri,
|
||
"CurrentTrackMetaData": r.current_meta,
|
||
"NextAVTransportURI": r.next_uri,
|
||
"NextAVTransportURIMetaData": r.next_meta,
|
||
"CurrentTransportActions": transport_actions(r),
|
||
"PlaybackStorageMedium": "NETWORK" if r.current_uri else "NONE",
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- SOAP dispatch
|
||
|
||
OH = None # openhome.OpenHome instance (if enabled)
|
||
|
||
|
||
def handle_action(renderer, service, action, args):
|
||
r = renderer
|
||
if service in openhome.SERVICES:
|
||
if not OH:
|
||
raise UPnPError(401, "Invalid Action")
|
||
try:
|
||
return OH.handle(service, action, args)
|
||
except openhome.OHError as exc:
|
||
raise UPnPError(exc.code, exc.desc)
|
||
if service == "AVTransport" and action in ("SetAVTransportURI", "SetNextAVTransportURI") and OH and OH.active:
|
||
log.info("AVTransport control took over from OpenHome")
|
||
OH.active = False
|
||
if service == "AVTransport":
|
||
if action == "SetAVTransportURI":
|
||
r.set_uri(args.get("CurrentURI", ""), args.get("CurrentURIMetaData", ""))
|
||
return {}
|
||
if action == "SetNextAVTransportURI":
|
||
r.set_next_uri(args.get("NextURI", ""), args.get("NextURIMetaData", ""))
|
||
return {}
|
||
if action == "Play":
|
||
r.play()
|
||
return {}
|
||
if action == "Pause":
|
||
r.pause()
|
||
return {}
|
||
if action == "Stop":
|
||
r.stop()
|
||
return {}
|
||
if action == "Seek":
|
||
r.seek(args.get("Unit", ""), args.get("Target", ""))
|
||
return {}
|
||
if action == "Next":
|
||
r.next()
|
||
return {}
|
||
if action == "Previous":
|
||
r.seek("REL_TIME", "0:00:00")
|
||
return {}
|
||
if action == "GetTransportInfo":
|
||
return {"CurrentTransportState": r.transport_state, "CurrentTransportStatus": "OK", "CurrentSpeed": "1"}
|
||
if action == "GetPositionInfo":
|
||
return {"Track": "1" if r.current_uri else "0", "TrackDuration": hms(r.duration()),
|
||
"TrackMetaData": r.current_meta, "TrackURI": r.current_uri,
|
||
"RelTime": hms(r.position), "AbsTime": hms(r.position),
|
||
"RelCount": "2147483647", "AbsCount": "2147483647"}
|
||
if action == "GetMediaInfo":
|
||
return {"NrTracks": "1" if r.current_uri else "0", "MediaDuration": hms(r.duration()),
|
||
"CurrentURI": r.current_uri, "CurrentURIMetaData": r.current_meta,
|
||
"NextURI": r.next_uri, "NextURIMetaData": r.next_meta,
|
||
"PlayMedium": "NETWORK", "RecordMedium": "NOT_IMPLEMENTED", "WriteStatus": "NOT_IMPLEMENTED"}
|
||
if action == "GetDeviceCapabilities":
|
||
return {"PlayMedia": "NETWORK", "RecMedia": "NOT_IMPLEMENTED", "RecQualityModes": "NOT_IMPLEMENTED"}
|
||
if action == "GetTransportSettings":
|
||
return {"PlayMode": "NORMAL", "RecQualityMode": "NOT_IMPLEMENTED"}
|
||
if action == "GetCurrentTransportActions":
|
||
return {"Actions": transport_actions(r)}
|
||
elif service == "RenderingControl":
|
||
if action == "GetVolume":
|
||
return {"CurrentVolume": str(r.volume)}
|
||
if action == "SetVolume":
|
||
r.set_volume(int(args.get("DesiredVolume", "0")))
|
||
return {}
|
||
if action == "GetMute":
|
||
return {"CurrentMute": "1" if r.mute else "0"}
|
||
if action == "SetMute":
|
||
r.set_mute(args.get("DesiredMute", "0") in ("1", "true", "True"))
|
||
return {}
|
||
if action == "ListPresets":
|
||
return {"CurrentPresetNameList": "FactoryDefaults"}
|
||
if action == "SelectPreset":
|
||
return {}
|
||
elif service == "ConnectionManager":
|
||
if action == "GetProtocolInfo":
|
||
return {"Source": "", "Sink": SINK_PROTOCOLS}
|
||
if action == "GetCurrentConnectionIDs":
|
||
return {"ConnectionIDs": "0"}
|
||
if action == "GetCurrentConnectionInfo":
|
||
return {"RcsID": "0", "AVTransportID": "0", "ProtocolInfo": "", "PeerConnectionManager": "",
|
||
"PeerConnectionID": "-1", "Direction": "Input", "Status": "OK"}
|
||
raise UPnPError(401, "Invalid Action")
|
||
|
||
|
||
def soap_response(service, action, values):
|
||
body = "".join(f"<{k}>{html.escape(str(v))}</{k}>" for k, v in values.items())
|
||
return ('<?xml version="1.0" encoding="utf-8"?>'
|
||
'<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
|
||
's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body>'
|
||
f'<u:{action}Response xmlns:u="{SERVICES[service]}">{body}</u:{action}Response>'
|
||
"</s:Body></s:Envelope>")
|
||
|
||
|
||
def soap_fault(code, desc):
|
||
return ('<?xml version="1.0" encoding="utf-8"?>'
|
||
'<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
|
||
's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><s:Fault>'
|
||
"<faultcode>s:Client</faultcode><faultstring>UPnPError</faultstring><detail>"
|
||
'<UPnPError xmlns="urn:schemas-upnp-org:control-1-0">'
|
||
f"<errorCode>{code}</errorCode><errorDescription>{html.escape(desc)}</errorDescription>"
|
||
"</UPnPError></detail></s:Fault></s:Body></s:Envelope>")
|
||
|
||
|
||
# --------------------------------------------------------------------------- web UI
|
||
|
||
import gapless # noqa: E402
|
||
import openhome # noqa: E402
|
||
import web_ui # noqa: E402 (templates only)
|
||
|
||
CSRF_TOKEN = secrets.token_urlsafe(24)
|
||
|
||
# key, label, kind, needs restart, help text
|
||
SETTING_FIELDS = [
|
||
("UI_LANG", "Sprache der Weboberfläche", "select:en,de", False, "en = English, de = Deutsch."),
|
||
("FRIENDLY_NAME", "Name im Netzwerk", "text", True,
|
||
"So erscheint die Brücke in Symfonium. Leer = Playername + „(BluOS Bridge)“."),
|
||
("BLUOS_HOST", "BluOS-Player (IP oder Hostname)", "text", True, ""),
|
||
("BLUOS_PORT", "BluOS-API-Port", "number", True, "Standard: 11000"),
|
||
("ADVERTISE_IP", "IP-Adresse der Brücke", "text", True,
|
||
"Unter dieser Adresse kündigt sich die Brücke per SSDP im Netz an."),
|
||
("SSDP_INTERVAL", "SSDP-Ankündigung alle … Sekunden", "number", True, ""),
|
||
("SERVER_TYPE", "Servertyp", "select:subsonic,opensubsonic", False,
|
||
"subsonic = klassischer Subsonic (Token-Anmeldung). opensubsonic = Airsonic Pulse, Navidrome & Co. "
|
||
"(API-Key, ohne Key Passwort-Anmeldung – dann muss „Streams über die Brücke ausliefern“ an sein)."),
|
||
("SUBSONIC_URL", "Musikserver (von der Brücke aus)", "text", False,
|
||
"z. B. http://192.168.1.20:4040 oder https://music.example.com"),
|
||
("SUBSONIC_ALIASES", "Weitere Hostnamen des Servers", "text", False,
|
||
"Kommagetrennt. Hostnamen, unter denen Symfonium denselben Server anspricht, z. B. die öffentliche Domain."),
|
||
("CLIENT_NAME", "Client-Name am Musikserver", "text", False,
|
||
"So erscheint die Brücke beim Server, z. B. „benutzer@BluOS Player“. {player} = Name des BluOS-Players."),
|
||
("SUBSONIC_USER", "Service-Benutzer", "text", False,
|
||
"Lokaler Benutzer (kein LDAP), der nur streamen darf."),
|
||
("SUBSONIC_PASSWORD", "Passwort des Service-Benutzers", "password", False, "Leer lassen = unverändert."),
|
||
("SUBSONIC_API_KEY", "API-Key des Service-Benutzers (nur opensubsonic)", "password", False,
|
||
"Wird statt Benutzer/Passwort genutzt. Leer lassen = unverändert."),
|
||
("REPEAT_OFF", "BluOS-Wiederholung beim Abspielen ausschalten", "bool", False,
|
||
"Empfohlen: Symfonium kümmert sich selbst um Wiederholen und Zufallswiedergabe."),
|
||
("PROXY_STREAMS", "Subsonic-Streams über die Brücke ausliefern", "bool", False,
|
||
"Empfohlen: ermöglicht Spulen und ein sauberes Songende. Der Player muss die Brücke erreichen können."),
|
||
("GAPLESS", "Lückenlose Wiedergabe (experimentell)", "bool", False,
|
||
"Fügt MP3- bzw. FLAC-Titel zu einem durchgehenden Stream zusammen – für Live-Alben, Mixe und Hörspiele. "
|
||
"Spulen startet den Stream neu, das Cover in der BluOS-App bleibt beim ersten Titel."),
|
||
("LOSSLESS_ONLY", "Lückenlos nur ohne Qualitätsverlust", "bool", False,
|
||
"Wandelt Folgetitel nur um, wenn nichts verloren geht (MP3 → FLAC). Ein FLAC nach einem MP3 wird nicht zu "
|
||
"MP3, sondern im Original mit normalem Titelwechsel gespielt."),
|
||
("FADE_MS", "Ein-/Ausblenden (Millisekunden)", "number", False,
|
||
"Sanftes Ein- und Ausblenden bei Pause, Stopp und Titelwechsel. 0 = aus."),
|
||
("OPENHOME", "OpenHome anbieten", "bool", True,
|
||
"Die App übergibt die ganze Warteschlange an die Brücke – die Musik läuft weiter, auch wenn das Handy schläft."),
|
||
("LOG_LEVEL", "Log-Level", "select:DEBUG,INFO,WARNING", False, ""),
|
||
]
|
||
RUNNING = {k: str(globals()[k]) for k, *_ in SETTING_FIELDS} # values active since start
|
||
|
||
|
||
def make_subsonic(values):
|
||
return Subsonic(values["SUBSONIC_URL"].rstrip("/"), values["SUBSONIC_USER"], values["SUBSONIC_PASSWORD"],
|
||
values["SERVER_TYPE"].lower(), values["SUBSONIC_API_KEY"], values["SUBSONIC_ALIASES"],
|
||
values["CLIENT_NAME"].replace("{player}", PLAYER_NAME or "BluOS").strip())
|
||
|
||
|
||
def current_settings():
|
||
return {k: cfg(k, RUNNING[k]) for k, *_ in SETTING_FIELDS}
|
||
|
||
|
||
def render_settings(notice=""):
|
||
values = current_settings()
|
||
parts = []
|
||
t = web_ui.texts(UI_LANG)
|
||
for key, label, kind, restart, help_text in SETTING_FIELDS:
|
||
if UI_LANG != "de" and key in web_ui.FIELDS_EN:
|
||
label, help_text = web_ui.FIELDS_EN[key]
|
||
value = values.get(key, "")
|
||
badge = f' <span class="restart">{t["restart"]}</span>' if restart else ""
|
||
if restart and value != RUNNING[key]:
|
||
badge = f' <span class="restart">{t["restart_pending"]}</span>'
|
||
help_html = f"<small>{html.escape(help_text)}</small>" if help_text else ""
|
||
if kind == "bool":
|
||
checked = " checked" if as_bool(value) else ""
|
||
parts.append(f'<label class="check"><input type="checkbox" name="{key}" value="true"{checked}>'
|
||
f'<span>{html.escape(label)}{badge}<br>{help_html}</span></label>')
|
||
elif kind.startswith("select:"):
|
||
opts = "".join(f'<option{" selected" if o.upper() == value.upper() else ""}>{o}</option>'
|
||
for o in kind[7:].split(","))
|
||
parts.append(f'<label>{html.escape(label)}{badge}<select name="{key}">{opts}</select>{help_html}</label>')
|
||
else:
|
||
shown = "" if kind == "password" else html.escape(value)
|
||
placeholder = ' placeholder="••••••••"' if kind == "password" and value else ""
|
||
parts.append(f'<label>{html.escape(label)}{badge}<input type="{kind}" name="{key}" '
|
||
f'value="{shown}"{placeholder} autocomplete="off">{help_html}</label>')
|
||
if not os.access(DATA_DIR, os.W_OK):
|
||
notice += ('<div class="notice err">' + html.escape(t["not_writable"]).replace(
|
||
"{dir}", f'<span class="mono">{html.escape(DATA_DIR)}</span>') + "</div>")
|
||
page = web_ui.localize(web_ui.SETTINGS_HTML, UI_LANG)
|
||
return (page.replace("__FIELDS__", "".join(parts)).replace("__NOTICE__", notice)
|
||
.replace("__CSRF__", CSRF_TOKEN).replace("__FILE__", html.escape(SETTINGS_FILE)))
|
||
|
||
|
||
def save_settings(form, renderer):
|
||
"""Validate, persist and apply settings. Returns (ok, message)."""
|
||
old = current_settings()
|
||
new = {}
|
||
for key, label, kind, *_ in SETTING_FIELDS:
|
||
value = (form.get(key) or [""])[0].strip()
|
||
if kind == "bool":
|
||
value = "true" if value else "false"
|
||
elif kind == "password" and not value:
|
||
value = old.get(key, "")
|
||
elif kind == "number":
|
||
if not value.isdigit():
|
||
return False, f"{label}: " + L("bitte eine Zahl eingeben.", "please enter a number.")
|
||
elif kind.startswith("select:") and value not in kind[7:].split(","):
|
||
return False, f"{label}: " + L("ungültiger Wert.", "invalid value.")
|
||
new[key] = value
|
||
if not new["BLUOS_HOST"] or not new["ADVERTISE_IP"]:
|
||
return False, L("Player-Adresse und IP der Brücke dürfen nicht leer sein.",
|
||
"Player address and bridge IP must not be empty.")
|
||
if not 1 <= int(new["BLUOS_PORT"]) <= 65535 or not 10 <= int(new["SSDP_INTERVAL"]) <= 3600:
|
||
return False, L("Port oder SSDP-Intervall außerhalb des gültigen Bereichs.",
|
||
"Port or SSDP interval out of range.")
|
||
if new["SUBSONIC_URL"] and not re.match(r"^https?://", new["SUBSONIC_URL"]):
|
||
return False, L("Musikserver muss mit http:// oder https:// beginnen.",
|
||
"Music server must start with http:// or https://.")
|
||
new["SUBSONIC_URL"] = new["SUBSONIC_URL"].rstrip("/")
|
||
|
||
os.makedirs(DATA_DIR, exist_ok=True)
|
||
tmp = SETTINGS_FILE + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(new, f, indent=2, ensure_ascii=False)
|
||
os.chmod(tmp, 0o600)
|
||
if os.path.exists(SETTINGS_FILE):
|
||
os.replace(SETTINGS_FILE, SETTINGS_FILE + ".bak")
|
||
os.replace(tmp, SETTINGS_FILE)
|
||
OVERRIDES.clear()
|
||
OVERRIDES.update(new)
|
||
|
||
# apply what can change at runtime
|
||
global REPEAT_OFF, PROXY_STREAMS, GAPLESS, FADE_MS, UI_LANG, LOSSLESS_ONLY
|
||
REPEAT_OFF = as_bool(new["REPEAT_OFF"])
|
||
PROXY_STREAMS = as_bool(new["PROXY_STREAMS"])
|
||
GAPLESS = as_bool(new["GAPLESS"])
|
||
FADE_MS = min(int(new["FADE_MS"] or 0), 5000)
|
||
UI_LANG = new["UI_LANG"].lower()
|
||
LOSSLESS_ONLY = gapless.LOSSLESS_ONLY = as_bool(new["LOSSLESS_ONLY"])
|
||
logging.getLogger().setLevel(new["LOG_LEVEL"])
|
||
with renderer.lock:
|
||
renderer.subsonic = make_subsonic(new)
|
||
changed = [k for k in new if new[k] != old.get(k)]
|
||
log.info("settings saved via web UI (changed: %s)", ", ".join(changed) or "nothing")
|
||
restart = [label for key, label, _, needs, _ in SETTING_FIELDS if needs and new[key] != RUNNING[key]]
|
||
msg = L("Gespeichert.", "Saved.")
|
||
if restart:
|
||
msg += L(" Für " + ", ".join(restart) + " ist ein Neustart der Brücke nötig.",
|
||
" Restart the bridge to apply: " + ", ".join(restart) + ".")
|
||
return True, msg
|
||
|
||
|
||
def run_action(action, renderer):
|
||
"""Tools on the settings page. Returns (ok, message)."""
|
||
if action == "test_player":
|
||
st = renderer.bluos.xml("/SyncStatus", timeout=5)
|
||
return True, (f"{st.get('name')} · {st.get('brand')} {st.get('modelName')} · "
|
||
f"BluOS {st.get('version')} · Lautstärke {st.get('volume')}")
|
||
if action == "test_subsonic":
|
||
sub = renderer.subsonic
|
||
if not sub.enabled:
|
||
return False, L("Musikserver ist nicht konfiguriert.", "Music server is not configured.")
|
||
q = urllib.parse.urlencode(sub._auth())
|
||
with urllib.request.urlopen(f"{sub.base}/rest/ping.view?{q}", timeout=10) as r:
|
||
root = ET.fromstring(r.read())
|
||
if root.get("status") == "ok":
|
||
server = root.get("type") or "subsonic"
|
||
extra = f", {server} {root.get('serverVersion')}" if root.get("serverVersion") else ""
|
||
hint = ""
|
||
if root.get("openSubsonic") == "true" and sub.server_type == "subsonic":
|
||
hint = L(" Hinweis: Der Server kann OpenSubsonic – Servertyp „opensubsonic“ wählen.",
|
||
" Note: the server supports OpenSubsonic – choose server type \"opensubsonic\".")
|
||
return True, L(f"Anmeldung mit {sub.auth_label} ok", f"Login as {sub.auth_label} ok") + f" (API {root.get('version')}{extra}).{hint}"
|
||
err = root.find("{http://subsonic.org/restapi}error")
|
||
return False, L("Server meldet: ", "Server says: ") + (err.get("message") if err is not None else "?")
|
||
if action == "pause":
|
||
renderer.pause()
|
||
return True, L("Pausiert.", "Paused.")
|
||
if action == "play":
|
||
renderer.play()
|
||
return True, L("Wiedergabe fortgesetzt.", "Playback resumed.")
|
||
if action == "restart":
|
||
log.info("restart requested via web UI")
|
||
threading.Timer(1.0, lambda: os._exit(0)).start() # restart policy brings the container back
|
||
return True, L("Brücke startet neu …", "Bridge is restarting …")
|
||
return False, L("Unbekannte Aktion.", "Unknown action.")
|
||
|
||
|
||
class CoverCache:
|
||
"""Serves covers to the browser and the player without exposing server credentials."""
|
||
|
||
def __init__(self, size=20):
|
||
self.items = collections.OrderedDict()
|
||
self.size = size
|
||
self.lock = threading.Lock()
|
||
|
||
def get(self, subsonic, song_id):
|
||
with self.lock:
|
||
if song_id in self.items:
|
||
self.items.move_to_end(song_id)
|
||
return self.items[song_id]
|
||
with urllib.request.urlopen(subsonic.cover_url(song_id, 600), timeout=10) as r:
|
||
ctype = r.headers.get("Content-Type", "image/jpeg")
|
||
if not ctype.startswith("image/"):
|
||
raise ValueError(f"no image ({ctype})")
|
||
data = (r.read(), ctype)
|
||
with self.lock:
|
||
self.items[song_id] = data
|
||
while len(self.items) > self.size:
|
||
self.items.popitem(last=False)
|
||
return data
|
||
|
||
|
||
COVERS = CoverCache()
|
||
|
||
|
||
class Monitor:
|
||
"""Health data for Uptime Kuma, PRTG and Prometheus. The server check is cached."""
|
||
|
||
def __init__(self, renderer, ttl=30):
|
||
self.renderer = renderer
|
||
self.ttl = ttl
|
||
self.lock = threading.Lock()
|
||
self.server = {"checked": 0.0, "ok": None, "latency_ms": None, "error": ""}
|
||
|
||
def _check_server(self):
|
||
sub = self.renderer.subsonic
|
||
if not sub.enabled:
|
||
return {"checked": time.time(), "ok": None, "latency_ms": None, "error": "not configured"}
|
||
start = time.time()
|
||
try:
|
||
q = urllib.parse.urlencode(sub._auth())
|
||
with urllib.request.urlopen(f"{sub.base}/rest/ping.view?{q}", timeout=8) as r:
|
||
ok = ET.fromstring(r.read()).get("status") == "ok"
|
||
return {"checked": time.time(), "ok": ok, "latency_ms": int((time.time() - start) * 1000),
|
||
"error": "" if ok else "ping failed"}
|
||
except Exception as exc:
|
||
return {"checked": time.time(), "ok": False, "latency_ms": None, "error": str(exc)}
|
||
|
||
def health(self):
|
||
with self.lock:
|
||
if time.time() - self.server["checked"] > self.ttl:
|
||
self.server = self._check_server()
|
||
server = dict(self.server)
|
||
r = self.renderer
|
||
player_age = time.time() - r.player_ok if r.player_ok else None
|
||
player_ok = player_age is not None and player_age < 90 and not r.player_error
|
||
problems = []
|
||
if not player_ok:
|
||
problems.append("player unreachable")
|
||
if server["ok"] is False:
|
||
problems.append("music server unreachable")
|
||
return {
|
||
"status": "ok" if not problems else "degraded",
|
||
"problems": problems,
|
||
"version": VERSION,
|
||
"uptime": int(time.time() - STARTED),
|
||
"player": {"ok": player_ok, "last_seen_s": None if player_age is None else int(player_age),
|
||
"error": r.player_error},
|
||
"server": {"ok": server["ok"], "latency_ms": server["latency_ms"], "error": server["error"]},
|
||
"playback": {"state": r.transport_state, "owned": r.owned, "gapless": bool(r.session)},
|
||
}
|
||
|
||
def prtg(self):
|
||
h = self.health()
|
||
channels = [
|
||
{"channel": "Brücke", "value": 1, "valuelookup": "prtg.standardlookups.offon.stateonok"},
|
||
{"channel": "Player erreichbar", "value": int(h["player"]["ok"]),
|
||
"valuelookup": "prtg.standardlookups.offon.stateonok"},
|
||
{"channel": "Musikserver erreichbar", "value": int(bool(h["server"]["ok"])),
|
||
"valuelookup": "prtg.standardlookups.offon.stateonok"},
|
||
{"channel": "Musikserver Antwortzeit", "value": h["server"]["latency_ms"] or 0, "unit": "TimeResponse"},
|
||
{"channel": "Spielt", "value": int(h["playback"]["state"] == "PLAYING" and h["playback"]["owned"])},
|
||
{"channel": "Laufzeit", "value": h["uptime"], "unit": "TimeSeconds"},
|
||
]
|
||
text = f"bluos-upnp-bridge {VERSION}: " + (", ".join(h["problems"]) or "OK")
|
||
return {"prtg": {"result": channels, "text": text}}
|
||
|
||
def prometheus(self):
|
||
h = self.health()
|
||
lines = [
|
||
"# HELP bluos_bridge_up Bridge is running.", "# TYPE bluos_bridge_up gauge", "bluos_bridge_up 1",
|
||
"# HELP bluos_bridge_player_up BluOS player answered within the last 90 s.",
|
||
"# TYPE bluos_bridge_player_up gauge", f"bluos_bridge_player_up {int(h['player']['ok'])}",
|
||
"# HELP bluos_bridge_server_up Music server answers ping.", "# TYPE bluos_bridge_server_up gauge",
|
||
f"bluos_bridge_server_up {int(bool(h['server']['ok']))}",
|
||
"# HELP bluos_bridge_server_latency_seconds Music server ping latency.",
|
||
"# TYPE bluos_bridge_server_latency_seconds gauge",
|
||
f"bluos_bridge_server_latency_seconds {(h['server']['latency_ms'] or 0) / 1000:.3f}",
|
||
"# HELP bluos_bridge_playing Bridge is playing on the player.", "# TYPE bluos_bridge_playing gauge",
|
||
f"bluos_bridge_playing {int(h['playback']['state'] == 'PLAYING' and h['playback']['owned'])}",
|
||
"# HELP bluos_bridge_uptime_seconds Bridge uptime.", "# TYPE bluos_bridge_uptime_seconds counter",
|
||
f"bluos_bridge_uptime_seconds {h['uptime']}",
|
||
f'bluos_bridge_info{{version="{VERSION}"}} 1',
|
||
]
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
class StreamProxy:
|
||
"""Short-lived, unguessable stream URLs for the player.
|
||
|
||
The player only ever sees http://<bridge>/stream/<token>; the upstream URL with the
|
||
service account token stays inside the bridge.
|
||
"""
|
||
|
||
def __init__(self, keep=30):
|
||
self.items = collections.OrderedDict()
|
||
self.keep = keep
|
||
self.lock = threading.Lock()
|
||
|
||
def add(self, upstream, label=""):
|
||
token = secrets.token_urlsafe(18)
|
||
with self.lock:
|
||
self.items[token] = (upstream, label)
|
||
while len(self.items) > self.keep:
|
||
self.items.popitem(last=False)
|
||
return f"http://{ADVERTISE_IP}:{HTTP_PORT}/stream/{token}"
|
||
|
||
def add_session(self, session, label=""):
|
||
return self.add(session, label)
|
||
|
||
def get(self, token):
|
||
with self.lock:
|
||
return self.items.get(token)
|
||
|
||
|
||
STREAMS = StreamProxy()
|
||
|
||
|
||
# --------------------------------------------------------------------------- HTTP server
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = f"bluos-upnp-bridge/{VERSION}"
|
||
protocol_version = "HTTP/1.1"
|
||
renderer = None
|
||
events = None
|
||
monitor = None
|
||
description = ""
|
||
|
||
def log_message(self, fmt, *args):
|
||
log.debug("%s %s", self.client_address[0], fmt % args)
|
||
|
||
def _send(self, code, body=b"", ctype='text/xml; charset="utf-8"', headers=None):
|
||
if isinstance(body, str):
|
||
body = body.encode("utf-8")
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Server", SERVER_HEADER)
|
||
for k, v in (headers or {}).items():
|
||
self.send_header(k, v)
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
# ---- web UI helpers
|
||
def _authorized(self):
|
||
if not WEB_PASSWORD:
|
||
self._send(403, "Einstellungen sind deaktiviert (WEB_PASSWORD ist nicht gesetzt).\n",
|
||
"text/plain; charset=utf-8")
|
||
return False
|
||
header = self.headers.get("Authorization", "")
|
||
if header.startswith("Basic "):
|
||
try:
|
||
user, _, pw = base64.b64decode(header[6:]).decode("utf-8").partition(":")
|
||
if hmac.compare_digest(user, WEB_USER) and hmac.compare_digest(pw, WEB_PASSWORD):
|
||
return True
|
||
except Exception:
|
||
pass
|
||
log.warning("failed web login from %s", self.client_address[0])
|
||
self._send(401, "Anmeldung erforderlich\n", "text/plain; charset=utf-8",
|
||
{"WWW-Authenticate": 'Basic realm="BluOS-Bruecke", charset="UTF-8"'})
|
||
return False
|
||
|
||
def _json(self, data, code=200):
|
||
self._send(code, json.dumps(data, ensure_ascii=False), "application/json; charset=utf-8",
|
||
{"Cache-Control": "no-store"})
|
||
|
||
def _html(self, body, code=200):
|
||
self._send(code, body, "text/html; charset=utf-8", {"Cache-Control": "no-store"})
|
||
|
||
def _status_json(self):
|
||
r = self.renderer
|
||
data = r.snapshot()
|
||
data.update({
|
||
"version": VERSION, "uptime": int(time.time() - STARTED), "name": self.name,
|
||
"bluos_host": BLUOS_HOST, "bluos_port": BLUOS_PORT,
|
||
"location": f"http://{ADVERTISE_IP}:{HTTP_PORT}/description.xml",
|
||
"subsonic": {"enabled": r.subsonic.enabled, "user": r.subsonic.auth_label, "url": r.subsonic.base,
|
||
"type": r.subsonic.server_type},
|
||
"repeat_off": REPEAT_OFF, "subscriptions": len(self.events.subs),
|
||
"openhome": bool(OH), "gapless": GAPLESS,
|
||
})
|
||
return data
|
||
|
||
def do_GET(self):
|
||
parts = urllib.parse.urlsplit(self.path)
|
||
path, query = parts.path, urllib.parse.parse_qs(parts.query)
|
||
if path == "/description.xml":
|
||
return self._send(200, self.description)
|
||
name = path.strip("/").removesuffix(".xml")
|
||
if name in SCPD_SPECS:
|
||
return self._send(200, build_scpd(*SCPD_SPECS[name]))
|
||
if path == "/health":
|
||
return self._send(200, f"ok {self.renderer.transport_state}\n", "text/plain")
|
||
if path == "/api/health":
|
||
h = self.monitor.health()
|
||
return self._json(h, 200 if h["status"] == "ok" else 503)
|
||
if path == "/api/prtg":
|
||
return self._json(self.monitor.prtg())
|
||
if path == "/metrics":
|
||
return self._send(200, self.monitor.prometheus(), "text/plain; version=0.0.4; charset=utf-8")
|
||
if path in ("/", "/index.html"):
|
||
return self._html(web_ui.localize(web_ui.STATUS_HTML, UI_LANG))
|
||
if path == "/api/status":
|
||
return self._json(self._status_json())
|
||
if path == "/api/log":
|
||
after = int((query.get("after") or ["0"])[0] or 0)
|
||
return self._json([line for line in list(RING.lines) if line["id"] > after])
|
||
if path.startswith("/stream/"):
|
||
return self._stream(path[len("/stream/"):])
|
||
if path == "/cover" or (path.startswith("/cover/") and path.endswith(".jpg")):
|
||
song_id = (query.get("id") or [""])[0] if path == "/cover" else path[len("/cover/"):-len(".jpg")]
|
||
if not song_id or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", song_id):
|
||
return self._send(404, "", "text/plain")
|
||
try:
|
||
data, ctype = COVERS.get(self.renderer.subsonic, song_id)
|
||
return self._send(200, data, ctype, {"Cache-Control": "max-age=3600"})
|
||
except Exception as exc:
|
||
log.debug("cover fetch failed: %s", exc)
|
||
return self._send(404, "", "text/plain")
|
||
if path == "/settings":
|
||
if self._authorized():
|
||
notice = ""
|
||
if "saved" in query:
|
||
notice = f'<div class="notice ok">{html.escape(query.get("msg", [L("Gespeichert.", "Saved.")])[0])}</div>'
|
||
self._html(render_settings(notice))
|
||
return
|
||
self._send(404, "not found", "text/plain")
|
||
|
||
do_HEAD = do_GET
|
||
|
||
def _stream(self, token):
|
||
"""Proxy a Subsonic stream with Content-Length and byte-range support."""
|
||
item = STREAMS.get(token)
|
||
if not item:
|
||
return self._send(404, "unknown stream\n", "text/plain")
|
||
upstream, label = item
|
||
if isinstance(upstream, gapless.Session):
|
||
return self._stream_session(upstream, label)
|
||
client_range = self.headers.get("Range")
|
||
try:
|
||
up = urllib.request.urlopen(urllib.request.Request(
|
||
upstream, headers={"Range": client_range or "bytes=0-"}), timeout=20)
|
||
except urllib.error.HTTPError as exc:
|
||
log.warning("stream %r: upstream answered %s", label, exc.code)
|
||
return self._send(502 if exc.code != 416 else 416, "", "text/plain")
|
||
except Exception as exc:
|
||
log.warning("stream %r: upstream failed: %s", label, exc)
|
||
return self._send(502, "", "text/plain")
|
||
headers = {"Content-Type": up.headers.get("Content-Type", "audio/mpeg"), "Accept-Ranges": "bytes",
|
||
"Cache-Control": "no-store", "Connection": "close"}
|
||
content_range = up.headers.get("Content-Range")
|
||
length = up.headers.get("Content-Length")
|
||
code = 200
|
||
if up.status == 206 and content_range:
|
||
if client_range:
|
||
code = 206
|
||
headers["Content-Range"] = content_range
|
||
elif not length:
|
||
length = content_range.rsplit("/", 1)[-1]
|
||
if length and length.isdigit():
|
||
headers["Content-Length"] = length
|
||
log.info("stream %r -> player (%s%s)", label, f"range {client_range}" if client_range else "full",
|
||
f", {int(length) // 1024} KiB" if length and length.isdigit() else "")
|
||
self.close_connection = True
|
||
self.send_response(code)
|
||
for k, v in headers.items():
|
||
self.send_header(k, v)
|
||
self.end_headers()
|
||
if self.command == "HEAD":
|
||
up.close()
|
||
return
|
||
try:
|
||
while True:
|
||
chunk = up.read(65536)
|
||
if not chunk:
|
||
break
|
||
self.wfile.write(chunk)
|
||
except (BrokenPipeError, ConnectionResetError):
|
||
pass # player closed the connection (seek, skip, stop)
|
||
finally:
|
||
up.close()
|
||
|
||
def _stream_session(self, session, label):
|
||
"""Serve a gapless session as one endless stream (optionally with ICY titles)."""
|
||
if session.closed or session.ended or (
|
||
session.served and session.playback_secs >= session.end_of_streamed() - 3):
|
||
# BluOS reconnects to finished / drained "radio" streams - don't replay the session from the start
|
||
log.info("gapless stream %r already finished, refusing reconnect", label)
|
||
return self._send(410, "", "text/plain")
|
||
if session.served:
|
||
# reconnect (e.g. resume after a long pause): continue where playback stopped
|
||
self.renderer.rebase_session(session)
|
||
session.served += 1
|
||
# content type from getSong: a probe request would show up as a second playback on the server
|
||
ctype = session.segments[0].info.get("content_type") or "audio/mpeg"
|
||
icy = self.headers.get("Icy-MetaData") == "1"
|
||
log.info("gapless stream %r -> player (%s%s)", label, ctype, ", ICY titles" if icy else "")
|
||
self.close_connection = True
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("Connection", "close")
|
||
if icy:
|
||
self.send_header("icy-metaint", "16000")
|
||
self.send_header("icy-name", self.name)
|
||
self.end_headers()
|
||
if self.command == "HEAD":
|
||
return
|
||
try:
|
||
gapless.stream_session(session, gapless.IcyWriter(self.wfile, 16000 if icy else 0))
|
||
except (BrokenPipeError, ConnectionResetError):
|
||
log.info("gapless stream %r: player closed the connection", label)
|
||
|
||
def _web_post(self, path):
|
||
if not self._authorized():
|
||
return
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
raw = self.rfile.read(min(length, 65536)).decode("utf-8", "replace")
|
||
if path == "/settings":
|
||
form = urllib.parse.parse_qs(raw, keep_blank_values=True)
|
||
if not hmac.compare_digest((form.get("csrf") or [""])[0], CSRF_TOKEN):
|
||
return self._html(render_settings(
|
||
f'<div class="notice err">{web_ui.texts(UI_LANG)["session_expired"]}</div>'), 403)
|
||
try:
|
||
ok, msg = save_settings(form, self.renderer)
|
||
except Exception as exc:
|
||
log.exception("saving settings failed")
|
||
ok, msg = False, L("Speichern fehlgeschlagen: ", "Saving failed: ") + str(exc)
|
||
if ok:
|
||
loc = "/settings?saved=1&msg=" + urllib.parse.quote(msg)
|
||
return self._send(303, "", "text/plain", {"Location": loc})
|
||
return self._html(render_settings(f'<div class="notice err">{html.escape(msg)}</div>'), 400)
|
||
# /api/action
|
||
if not hmac.compare_digest(self.headers.get("X-CSRF-Token", ""), CSRF_TOKEN):
|
||
return self._json({"ok": False, "message": L("Sitzung abgelaufen, bitte Seite neu laden.",
|
||
"Session expired, please reload the page.")}, 403)
|
||
try:
|
||
action = json.loads(raw or "{}").get("action", "")
|
||
ok, msg = run_action(action, self.renderer)
|
||
except UPnPError as exc:
|
||
ok, msg = False, exc.desc
|
||
except Exception as exc:
|
||
ok, msg = False, L("Fehler: ", "Error: ") + str(exc)
|
||
self._json({"ok": ok, "message": msg})
|
||
|
||
def do_POST(self):
|
||
path = self.path.split("?")[0]
|
||
if path in ("/settings", "/api/action"):
|
||
return self._web_post(path)
|
||
service = path.removeprefix("/ctl/")
|
||
if service not in SERVICES:
|
||
return self._send(404, "not found", "text/plain")
|
||
self.renderer.control_points[self.client_address[0]] = time.time()
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
raw = self.rfile.read(length)
|
||
soap_action = self.headers.get("SOAPACTION", "").strip('"')
|
||
action = soap_action.split("#")[-1]
|
||
args = {}
|
||
try:
|
||
root = ET.fromstring(raw)
|
||
body = root.find("{http://schemas.xmlsoap.org/soap/envelope/}Body")
|
||
call = list(body)[0]
|
||
action = action or call.tag.split("}")[-1]
|
||
args = {child.tag.split("}")[-1]: (child.text or "") for child in call}
|
||
except Exception:
|
||
log.warning("could not parse SOAP request for %s", action)
|
||
if action not in ("GetPositionInfo", "GetTransportInfo", "GetVolume", "GetMute",
|
||
"GetMediaInfo", "GetCurrentTransportActions"):
|
||
log.info("%s %s.%s", self.client_address[0], service, action)
|
||
try:
|
||
values = handle_action(self.renderer, service, action, args)
|
||
self._send(200, soap_response(service, action, values))
|
||
except UPnPError as exc:
|
||
log.info("%s.%s -> UPnP error %s %s", service, action, exc.code, exc.desc)
|
||
self._send(500, soap_fault(exc.code, exc.desc))
|
||
except Exception as exc:
|
||
log.exception("%s.%s failed", service, action)
|
||
self._send(500, soap_fault(501, f"Action failed: {exc}"))
|
||
|
||
def do_SUBSCRIBE(self):
|
||
service = self.path.split("?")[0].removeprefix("/evt/")
|
||
if service not in SERVICES:
|
||
return self._send(404, "", "text/plain")
|
||
self.renderer.control_points[self.client_address[0]] = time.time()
|
||
timeout = 1800
|
||
m = re.search(r"Second-(\d+)", self.headers.get("TIMEOUT", ""))
|
||
if m:
|
||
timeout = min(int(m.group(1)), 3600)
|
||
sid = self.headers.get("SID")
|
||
if sid:
|
||
if not self.events.renew(sid, timeout):
|
||
return self._send(412, "", "text/plain")
|
||
else:
|
||
callbacks = re.findall(r"<([^>]+)>", self.headers.get("CALLBACK", ""))
|
||
if not callbacks:
|
||
return self._send(412, "", "text/plain")
|
||
sid = self.events.subscribe(service, callbacks, timeout)
|
||
self._send(200, "", "text/plain", {"SID": sid, "TIMEOUT": f"Second-{timeout}"})
|
||
|
||
def do_UNSUBSCRIBE(self):
|
||
self.events.unsubscribe(self.headers.get("SID", ""))
|
||
self._send(200, "", "text/plain")
|
||
|
||
|
||
# --------------------------------------------------------------------------- SSDP
|
||
|
||
SSDP_ADDR, SSDP_PORT = "239.255.255.250", 1900
|
||
|
||
|
||
class SSDP:
|
||
def __init__(self, location):
|
||
self.location = location
|
||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
if hasattr(socket, "SO_REUSEPORT"):
|
||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
|
||
self.sock.bind(("", SSDP_PORT))
|
||
mreq = struct.pack("4s4s", socket.inet_aton(SSDP_ADDR), socket.inet_aton(ADVERTISE_IP))
|
||
self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
|
||
self.out = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||
self.out.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(ADVERTISE_IP))
|
||
self.out.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 4)
|
||
|
||
def targets(self):
|
||
t = [("upnp:rootdevice", f"{DEVICE_UUID}::upnp:rootdevice"),
|
||
(DEVICE_UUID, DEVICE_UUID),
|
||
(DEVICE_TYPE, f"{DEVICE_UUID}::{DEVICE_TYPE}")]
|
||
t += [(st, f"{DEVICE_UUID}::{st}") for st in SERVICES.values()]
|
||
return t
|
||
|
||
def notify(self, nts):
|
||
for nt, usn in self.targets():
|
||
lines = ["NOTIFY * HTTP/1.1", f"HOST: {SSDP_ADDR}:{SSDP_PORT}", f"NT: {nt}", f"NTS: {nts}",
|
||
f"USN: {usn}"]
|
||
if nts == "ssdp:alive":
|
||
lines += ["CACHE-CONTROL: max-age=1800", f"LOCATION: {self.location}",
|
||
f"SERVER: {SERVER_HEADER}"]
|
||
self.out.sendto(("\r\n".join(lines) + "\r\n\r\n").encode(), (SSDP_ADDR, SSDP_PORT))
|
||
|
||
def respond(self, st_wanted, addr):
|
||
for st, usn in self.targets():
|
||
if st_wanted not in ("ssdp:all", st):
|
||
continue
|
||
msg = "\r\n".join([
|
||
"HTTP/1.1 200 OK", "CACHE-CONTROL: max-age=1800", f"DATE: {formatdate(usegmt=True)}",
|
||
"EXT:", f"LOCATION: {self.location}", f"SERVER: {SERVER_HEADER}", f"ST: {st}",
|
||
f"USN: {usn}"]) + "\r\n\r\n"
|
||
self.out.sendto(msg.encode(), addr)
|
||
|
||
def serve(self, stop_event):
|
||
self.sock.settimeout(1.0)
|
||
while not stop_event.is_set():
|
||
try:
|
||
data, addr = self.sock.recvfrom(4096)
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
break
|
||
text = data.decode("utf-8", "replace")
|
||
if not text.startswith("M-SEARCH"):
|
||
continue
|
||
headers = {k.strip().upper(): v.strip() for k, _, v in
|
||
(line.partition(":") for line in text.split("\r\n")[1:]) if k}
|
||
st = headers.get("ST", "")
|
||
if st == "ssdp:all" or st in dict(self.targets()):
|
||
log.debug("M-SEARCH %s from %s", st, addr[0])
|
||
self.respond(st, addr)
|
||
|
||
def announce(self, stop_event):
|
||
while not stop_event.is_set():
|
||
try:
|
||
self.notify("ssdp:alive")
|
||
except OSError as exc:
|
||
log.warning("SSDP notify failed: %s", exc)
|
||
stop_event.wait(SSDP_INTERVAL)
|
||
|
||
|
||
# --------------------------------------------------------------------------- main
|
||
|
||
def main():
|
||
bluos = BluOS(BLUOS_HOST, BLUOS_PORT)
|
||
subsonic = make_subsonic(current_settings())
|
||
global PLAYER_NAME
|
||
player_name = bluos.name()
|
||
PLAYER_NAME = player_name
|
||
name = FRIENDLY_NAME or (player_name or BLUOS_HOST) + " (BluOS Bridge)"
|
||
|
||
renderer = Renderer(bluos, subsonic)
|
||
renderer.player_name = player_name
|
||
events = EventManager(renderer)
|
||
renderer.events = events
|
||
|
||
gapless.LOSSLESS_ONLY = LOSSLESS_ONLY
|
||
global OH
|
||
if OPENHOME:
|
||
SERVICES.update(openhome.SERVICES)
|
||
SCPD_SPECS.update(openhome.SPECS)
|
||
OH = openhome.OpenHome(renderer, name, player_name or name, SINK_PROTOCOLS, VERSION)
|
||
OH.notify = events._notify
|
||
renderer.on_change = OH.on_renderer_change
|
||
|
||
Handler.renderer = renderer
|
||
Handler.events = events
|
||
Handler.description = build_description(name)
|
||
Handler.name = name
|
||
Handler.monitor = Monitor(renderer)
|
||
httpd = ThreadingHTTPServer(("0.0.0.0", HTTP_PORT), Handler)
|
||
httpd.daemon_threads = True
|
||
|
||
location = f"http://{ADVERTISE_IP}:{HTTP_PORT}/description.xml"
|
||
ssdp = SSDP(location)
|
||
stop_event = threading.Event()
|
||
|
||
threads = [
|
||
threading.Thread(target=httpd.serve_forever, name="http", daemon=True),
|
||
threading.Thread(target=ssdp.serve, args=(stop_event,), name="ssdp", daemon=True),
|
||
threading.Thread(target=ssdp.announce, args=(stop_event,), name="ssdp-notify", daemon=True),
|
||
threading.Thread(target=poll_status, args=(renderer, bluos, stop_event), name="poll", daemon=True),
|
||
]
|
||
for t in threads:
|
||
t.start()
|
||
|
||
log.info("bluos-upnp-bridge %s: %r -> BluOS %s:%s, %s", VERSION, name, BLUOS_HOST, BLUOS_PORT, location)
|
||
log.info("Subsonic rewrite: %s", f"enabled ({SUBSONIC_USER}@{SUBSONIC_URL})" if subsonic.enabled else "disabled")
|
||
log.info("web UI: http://%s:%s/ (settings %s)", ADVERTISE_IP, HTTP_PORT,
|
||
"enabled" if WEB_PASSWORD else "disabled - set WEB_PASSWORD")
|
||
|
||
def shutdown(*_):
|
||
log.info("shutting down")
|
||
stop_event.set()
|
||
try:
|
||
ssdp.notify("ssdp:byebye")
|
||
except OSError:
|
||
pass
|
||
httpd.shutdown()
|
||
|
||
signal.signal(signal.SIGTERM, shutdown)
|
||
signal.signal(signal.SIGINT, shutdown)
|
||
while not stop_event.is_set():
|
||
stop_event.wait(1)
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|