#!/usr/bin/env python3 """HEOS panel: a phone-sized web remote plus the HTTP bridge it runs on. pip3 install -r requirements.txt python3 app.py # dev server, http://:5443/ The service instead runs this under gunicorn -- see deploy/heos-panel.service -- which imports `app` without ever calling main(), so the controller below is built at import time rather than from main()'s argparse. Everything the UI does goes through /api/*. The flatter, query-string endpoints from the original heos_bridge.py (/volume/up?target=..., and friends) are still here so existing Shortcuts and scripts keep working. """ import argparse import os import time from functools import wraps from flask import Flask, jsonify, render_template, request, url_for from werkzeug.middleware.proxy_fix import ProxyFix import config import spotify_zc from controller import Controller, TargetError from heos import HeosError from spotify import SpotifyClient, SpotifyDeviceUnavailable, SpotifyError from spotify_zc import ZeroconfError from zidoo import ZidooError app = Flask(__name__) # Served straight from port 5443 this changes nothing. Behind a reverse # proxy that mounts us on a sub-path (Apache at /heos, say) it reads the # X-Forwarded-Prefix that proxy sets, so every URL the app generates is # /heos/... instead of /..., and the page works either way. app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1) controller: Controller = None spotify: dict = {} # account key -> SpotifyClient, configured accounts only # HEOS merges any In-Room Group into one player at the hardware level (see # controller.py's module docstring), so its API has no field that says how # many units heos_name actually names -- that has to come from config.json's # "in_room_group" instead (config.IN_ROOM_GROUPS). The AVR/speaker split # needs no such flag: this app's own design always makes HOST_KEY the AVR # and ROOM_KEYS plain speakers, so it is known before any HEOS call is made. # # "kind" is passed straight through to the template as-is (one of # config.IN_ROOM_GROUPS, or "avr") -- which icon that draws is the # interface's business, not ours: see ICONS in static/app.js. def _room_meta(key: str, kind: str) -> dict: return {"key": key, "kind": kind, **config.TARGETS[key]} def _build_spotify() -> dict: if not (config.SPOTIFY_CLIENT_ID and config.SPOTIFY_CLIENT_SECRET): return {} return { key: SpotifyClient(config.SPOTIFY_CLIENT_ID, config.SPOTIFY_CLIENT_SECRET, account["refresh_token"]) for key, account in config.SPOTIFY_ACCOUNTS.items() if account.get("refresh_token") } if __name__ != "__main__": # Imported by a WSGI server rather than run as a script, so main()'s # argparse never executes -- build the one controller instance here # instead. HEOS_DEMO lets the fake-speakers mode work this way too. if os.environ.get("HEOS_DEMO", "").lower() in ("1", "true", "yes"): from demo import DemoController controller = DemoController(config) else: controller = Controller(config) spotify = _build_spotify() def handle_errors(view): """One place to turn our three failure modes into sensible JSON.""" @wraps(view) def wrapped(*args, **kwargs): try: return view(*args, **kwargs) except (TargetError, ValueError) as exc: return jsonify({"error": str(exc)}), 400 except (HeosError, SpotifyError, ZeroconfError, ZidooError) as exc: return jsonify({"error": str(exc)}), 502 return wrapped @app.after_request def no_store(response): """It is a LAN remote: never let a phone show a cached volume, and never let iOS pin an old copy of the UI to the home screen.""" response.headers["Cache-Control"] = "no-store" return response def _payload() -> dict: return request.get_json(silent=True) or request.form.to_dict() or request.args.to_dict() def _target_from(data: dict, field: str = "target") -> str: key = data.get(field) if not key: raise ValueError(f"Missing '{field}'. Valid rooms: {', '.join(config.TARGETS)}") if key not in config.TARGETS: raise ValueError(f"Unknown room '{key}'. Valid rooms: {', '.join(config.TARGETS)}") return key def _spotify_from(data: dict) -> SpotifyClient: if not spotify: raise ValueError("Spotify isn't configured -- see the README's Spotify section") key = data.get("account") if not key: raise ValueError(f"Missing 'account'. Spotify accounts: {', '.join(spotify)}") if key not in spotify: raise ValueError(f"Unknown Spotify account '{key}'. Spotify accounts: {', '.join(spotify)}") return spotify[key] def _spotify_name(key: str) -> str: target = config.TARGETS[key] return target.get("spotify_name", target["heos_name"]) def _spotify_playing_on() -> dict: """Device name -> the key of the account playing on it. An account Spotify refuses (a revoked token, say) just plays on nothing, instead of failing whatever asked.""" playing_on = {} for account, client in spotify.items(): try: player = client.playback() except (SpotifyError, OSError): continue device = (player.get("device") or {}).get("name") # Should two accounts both claim a device, the one actually playing wins. if device and (device not in playing_on or player.get("is_playing")): playing_on[device] = account return playing_on def _spotify_account_seeing(device_name: str, besides: str) -> str: """Which OTHER configured account can currently see that Connect receiver, if any. A speaker is logged into one Spotify account at a time and Spotify shows it to that account alone, so when a room goes missing this -- not the room's power or its spotify_name -- is very often the whole answer.""" for key, client in spotify.items(): if key == besides: continue try: if any(d.get("name") == device_name for d in client.devices()): return key except (SpotifyError, OSError): continue # an account we cannot ask is not an account holding it return None def _spotify_pause_on(key: str, account: str) -> bool: """Pause an account, but only if this room is where it is playing -- an account that has already moved on is playing somewhere else now, and stopping it there is nobody's intention. Best effort: an account Spotify will not answer for is no reason to leave the room alone.""" client = spotify.get(account) if client is None: return False try: player = client.playback() if (player.get("device") or {}).get("name") == _spotify_name(key) and player.get("is_playing"): client.pause() return True except (SpotifyError, OSError): pass return False # How long to let a speaker finish changing source before stopping it. RELEASE_TRIES = 6 RELEASE_WAIT = 0.5 def _is_demo() -> bool: """Demo mode's rooms are invented, but the real speakers are very likely on the same LAN -- and signing one in or out would physically change it. So the zeroconf side sits out a demo.""" return bool(getattr(controller, "demo", False)) def _find_speaker(device_name: str): """A room's own LAN endpoint, or None when it cannot be reached.""" try: return spotify_zc.find(device_name) except (ZeroconfError, OSError): return None def _spotify_release(key: str, account: str = None) -> dict: """Take a room off Spotify: pause the account playing there, then sign the speaker out over the LAN. Reports what it managed: {"paused", "signed_out"}. Signing out is the whole disconnect -- the speaker stops within a second and drops out of the account's device list -- and it leaves the room belonging to nobody, which costs nothing now that either button can claim it back with a sign-in (see _spotify_take_over). Pausing first is manners: it leaves that account where it was rather than cut off mid-song. The HEOS fallback below is for a speaker that does not answer on the LAN. It can end the stream but not the sign-in, and it cannot do it the obvious way either: player/set_play_state with stop is accepted for a Connect stream and then quietly leaves the player paused, session intact. Putting the speaker on one of its own inputs does end the stream, and stopping *that* is honoured, so the room is left idle rather than sitting on a live input. """ released = {"paused": _spotify_pause_on(key, account), "signed_out": False} speaker = None if _is_demo() else _find_speaker(_spotify_name(key)) if speaker is not None: speaker.reset_users() released["signed_out"] = True return released _release_over_heos(key) return released def _release_over_heos(key: str): """Stop a room's Spotify stream without the speaker's help -- see _spotify_release for why it goes the long way round.""" inputs = controller.heos_inputs(key) if not inputs: raise ValueError( f"'{config.TARGETS[key]['label']}' did not answer on the network, and HEOS lists " "no input on it to fall back to -- so there is no way from here to take this room " "off Spotify.") # An AUX jack with nothing in it is the quietest thing a speaker can be # switched to, and the stop below cuts even that short. quiet = next((i for i in inputs if "aux" in i["input_id"]), inputs[0]) controller.play_heos_input(key, quiet["input_id"]) # A speaker takes a moment to change source, and a stop that arrives # while it is still on the Spotify stream is exactly the one HEOS turns # into a pause -- so wait for the input to be what is playing. for _ in range(RELEASE_TRIES): time.sleep(RELEASE_WAIT) if not controller.on_spotify(key): break controller.toggle_play(key, "stop") # Signing a speaker in is a LAN round trip, but Spotify's cloud hearing # about it is not: the room only turns up in the account's device list a # moment later, and a resume before then looks exactly like a failure. HANDOVER_TRIES = 8 HANDOVER_WAIT = 1.2 def _spotify_take_over(device_name: str, client: SpotifyClient) -> str: """Sign a room's speaker in to this account over the LAN, the way the Spotify app does -- see spotify_zc. Returns None when it worked, or what went wrong, since that is what the caller has to tell the user. This is the only thing that moves a speaker between accounts: Spotify's cloud will not, and until the speaker itself is signed in, the Web API does not admit the room exists.""" try: # Checked before knocking, not after: a speaker takes the sign-in, # drops whoever was on it, and only then finds out Spotify will not # have the token -- which leaves the room signed in to nobody. A # login too old to carry the scope must not cost you the room. if not client.has_scope("streaming"): return ("that account's Spotify login has no 'streaming' scope, which is the one a " "speaker asks for -- re-run tools/spotify_auth.py for it (see the README's " "Spotify section), put the new refresh token in .env and restart the panel") if _is_demo(): return "the panel is in demo mode, which leaves real speakers alone" speaker = _find_speaker(device_name) if speaker is None: return f"no speaker calling itself '{device_name}' answered on the network" speaker.add_user(client.me()["id"], client.access_token()) except (ZeroconfError, SpotifyError, OSError) as exc: return str(exc) return None def _resume_once_signed_in(client: SpotifyClient, device_name: str) -> dict: """Resume as soon as the room turns up in this account's device list, rather than once, at the one moment it is least likely to be there.""" for attempt in range(HANDOVER_TRIES): if attempt: time.sleep(HANDOVER_WAIT) try: return client.resume(device_name) except SpotifyDeviceUnavailable: continue raise SpotifyError( f"'{device_name}' took the sign-in, but Spotify never offered the room to this account " "-- which means the speaker could not log in with the token it was given, and is now " "signed in to nobody. Pick the room once in that account's Spotify app to put it back.") def _mark_spotify_accounts(rooms: list): """Tag each room HEOS says is on Spotify with the account playing it, so its card can pick out that account's button. Spotify is only asked when some room is on Spotify at all.""" on_spotify = [room for room in rooms if room.get("spotify")] if not (spotify and on_spotify): return playing_on = _spotify_playing_on() for room in on_spotify: room["spotify_account"] = playing_on.get(_spotify_name(room["key"])) def _link_zidoo_poster(avr: dict): """Point the AVR's now-playing cover at our own copy of the Zidoo's poster: the panel is served over https, and the Zidoo only speaks plain http, which a phone will not load into an https page.""" track = avr.get("now_playing") if not track: return poster_id = track.pop("poster_id", None) if poster_id is not None: track["image"] = url_for("api_zidoo_poster", poster_id=poster_id) # --- The UI ----------------------------------------------------------- @app.route("/") def index(): return render_template( "index.html", app_name=config.APP_NAME, host=_room_meta(config.HOST_KEY, "avr"), rooms=[ _room_meta(key, config.TARGETS[key].get("in_room_group", "none")) for key in config.ROOM_KEYS ], step=config.VOLUME_STEP, spotify_accounts=[{"key": key, "label": config.SPOTIFY_ACCOUNTS[key]["label"]} for key in spotify], ) @app.get("/manifest.webmanifest") def manifest(): """Rendered rather than static, so APP_NAME is only written down once.""" return app.response_class( render_template("manifest.webmanifest", app_name=config.APP_NAME), mimetype="application/manifest+json", ) # --- API the UI talks to ---------------------------------------------- @app.get("/api/state") @handle_errors def api_state(): data = controller.state() # Demo rooms are not on anyone's real Spotify, so they bring their own account. if not data.get("demo"): _mark_spotify_accounts(data["rooms"]) _link_zidoo_poster(data["avr"]) return jsonify(data) @app.get("/api/zidoo/poster/") def api_zidoo_poster(poster_id): """A film's poster, fetched from the Zidoo on the phone's behalf -- see _link_zidoo_poster().""" zidoo = getattr(controller, "zidoo", None) image = zidoo.poster(poster_id) if zidoo else None if image is None: return "", 404 body, mimetype = image return app.response_class(body, mimetype=mimetype) @app.get("/api/targets") @handle_errors def api_targets(): """Diagnostic: every player and group HEOS can see, with its exact name -- this is what you copy into config.py.""" found = controller.scan() return jsonify({ "players": [ {"kind": "player", "name": p.get("name"), "pid": p.get("pid"), "model": p.get("model", "")} for p in found["players"] ], "groups": [ {"kind": "group", "name": g.get("name"), "gid": g.get("gid"), "players": [{"name": m.get("name"), "pid": m.get("pid"), "role": m.get("role")} for m in g.get("players", [])]} for g in found["groups"] ], }) @app.post("/api/volume") @handle_errors def api_volume(): data = _payload() key = _target_from(data) if data.get("level") is not None: return jsonify({"target": key, "level": controller.set_volume(key, int(data["level"]))}) if data.get("steps") is not None: # Taps, not points: each one lands on the next multiple of VOLUME_STEP. return jsonify({"target": key, "level": controller.step_volume(key, int(data["steps"]))}) if data.get("delta") is not None: return jsonify({"target": key, "level": controller.nudge_volume(key, int(data["delta"]))}) raise ValueError("Provide one of 'steps', 'delta' or 'level'") @app.post("/api/mute") @handle_errors def api_mute(): key = _target_from(_payload()) controller.toggle_mute(key) return jsonify({"target": key, "ok": True}) @app.post("/api/playback") @handle_errors def api_playback(): """Start or stop a room. Send 'state' to be explicit, or leave it out to flip whatever the speakers are actually doing.""" data = _payload() key = _target_from(data) state = data.get("state") if state is not None and state not in ("play", "pause", "stop"): raise ValueError("'state' must be play, pause or stop -- or left out to toggle") return jsonify({"target": key, "state": controller.toggle_play(key, state)}) @app.post("/api/skip") @handle_errors def api_skip(): """Jump to the next (or previous) track in a room's queue.""" data = _payload() key = _target_from(data) direction = data.get("direction", "next") if direction not in ("next", "previous"): raise ValueError("'direction' must be next or previous") controller.skip(key, direction) return jsonify({"target": key, "direction": direction}) @app.post("/api/seek") @handle_errors def api_seek(): """Jump to a point in what a card is playing. HEOS itself cannot seek, so this goes around it: to the Zidoo for the AVR's card, and for a room, to Spotify, through whichever of your accounts is playing there.""" data = _payload() key = _target_from(data) if data.get("position_ms") is None: raise ValueError("Provide 'position_ms', where to jump to") position = max(0, int(data["position_ms"])) if key == config.HOST_KEY: controller.zidoo_seek(position) else: account = _spotify_playing_on().get(_spotify_name(key)) if account is None: raise ValueError("Only a Spotify stream one of your accounts is playing can seek -- HEOS itself cannot") spotify[account].seek(position) return jsonify({"target": key, "position_ms": position}) @app.post("/api/group") @handle_errors def api_group(): """Join or leave the AVR's group. The AVR is always the host, so its content is what the joined rooms start playing.""" data = _payload() key = _target_from(data) joined = data.get("joined") if isinstance(joined, str): joined = joined.lower() in ("1", "true", "yes", "on") if joined is None: raise ValueError("Provide 'joined': true to merge with the AVR, false to split off") return jsonify({"joined": controller.join(key) if joined else controller.leave(key)}) @app.post("/api/group/none") @handle_errors def api_group_none(): """Every room back on its own.""" return jsonify({"joined": controller.set_membership([])}) @app.get("/api/avr/inputs") @handle_errors def api_avr_inputs(): return jsonify(controller.avr_inputs()) @app.post("/api/avr/input") @handle_errors def api_avr_set_input(): code = _payload().get("code") if not code: raise ValueError("Provide 'code', e.g. inputs/aux_in_1 -- see GET /api/avr/inputs") return jsonify(controller.avr_select_input(code)) @app.get("/api/spotify/devices") @handle_errors def api_spotify_devices(): """Diagnostic: every Spotify Connect receiver one account currently sees (?account=account1) -- use this to fill in a target's spotify_name if it differs from heos_name.""" return jsonify(_spotify_from(request.args).devices()) @app.post("/api/spotify/resume") @handle_errors def api_spotify_resume(): """Ask a room's own Spotify Connect receiver to resume one account's playback -- the reverse of connecting to it from the Spotify app.""" data = _payload() client = _spotify_from(data) key = _target_from(data) try: device = client.resume(_spotify_name(key)) except SpotifyDeviceUnavailable as exc: # A room this account cannot see is nearly always one another # account is signed in to -- which is worth naming if the hand-over # below cannot go through. holder = _spotify_account_seeing(exc.device_name, besides=data["account"]) # Leave whoever is there paused where they were, rather than cut # off: signing the speaker in to someone else ends their stream. if holder: _spotify_pause_on(key, holder) trouble = _spotify_take_over(exc.device_name, client) if trouble: whose = (f" It is signed in to {config.SPOTIFY_ACCOUNTS[holder]['label']}'s Spotify," " and Spotify's own API will not move a speaker between accounts." if holder else "") raise SpotifyError( f"Could not sign '{exc.device_name}' in to " f"{config.SPOTIFY_ACCOUNTS[data['account']]['label']}'s Spotify: {trouble}.{whose} " "Picking the room once in that account's Spotify app does the same thing by " "hand.") from exc device = _resume_once_signed_in(client, exc.device_name) return jsonify({"target": key, "account": data["account"], "device": device.get("name")}) @app.post("/api/spotify/disconnect") @handle_errors def api_spotify_disconnect(): """Let go of a room: pause what the account is playing there, then sign the speaker out of it, which stops the music and leaves the room free for either account's button to claim -- see _spotify_release.""" data = _payload() _spotify_from(data) # the account has to be one of ours to let go of key = _target_from(data) released = _spotify_release(key, data["account"]) return jsonify({"target": key, "account": data["account"], "device": _spotify_name(key), **released}) # --- The original bridge's API, unchanged ----------------------------- @app.get("/targets") @handle_errors def legacy_targets(): found = controller.scan() return jsonify( [{"kind": "player", "name": p.get("name"), "id": p.get("pid"), "model": p.get("model", "")} for p in found["players"]] + [{"kind": "group", "name": g.get("name"), "id": g.get("gid"), "members": [m.get("name") for m in g.get("players", [])]} for g in found["groups"]] ) @app.get("/volume") @handle_errors def legacy_get_volume(): return jsonify({"level": controller.volume(_target_from(request.args))}) @app.post("/volume/set") @handle_errors def legacy_set_volume(): level = request.args.get("level", type=int) if level is None or not 0 <= level <= 100: raise ValueError("provide integer 'level' between 0 and 100") return jsonify({"level": controller.set_volume(_target_from(request.args), level)}) @app.post("/volume/up") @handle_errors def legacy_volume_up(): key = _target_from(request.args) step = request.args.get("step", type=int) # No ?step= means one tap of the panel's own button, snapping to the # next multiple; an explicit ?step= stays a raw number of points. level = controller.step_volume(key, 1) if step is None else controller.nudge_volume(key, step) return jsonify({"level": level}) @app.post("/volume/down") @handle_errors def legacy_volume_down(): key = _target_from(request.args) step = request.args.get("step", type=int) level = controller.step_volume(key, -1) if step is None else controller.nudge_volume(key, -step) return jsonify({"level": level}) @app.post("/volume/mute") @handle_errors def legacy_mute(): controller.toggle_mute(_target_from(request.args)) return jsonify({"ok": True}) @app.post("/playback/") @handle_errors def legacy_playback(state): key = _target_from(request.args) if state in ("play", "pause", "stop"): controller.play_state(key, state) return jsonify({"state": state}) if state in ("next", "previous"): controller.skip(key, state) return jsonify({"ok": True}) raise ValueError(f"Unknown playback action '{state}'") @app.post("/group/create") @handle_errors def legacy_group_create(): host = _target_from(request.args, "host") members = [m.strip() for m in request.args.get("members", "").split(",") if m.strip()] if not members: raise ValueError("provide '?host=&members='") for key in members: _target_from({"target": key}) controller.group_targets(host, members) return jsonify({"host": host, "members": members}) @app.post("/group/remove") @handle_errors def legacy_group_remove(): controller.ungroup(_target_from(request.args)) return jsonify({"ok": True}) @app.get("/inputs") @handle_errors def legacy_inputs(): return jsonify(controller.heos_inputs(_target_from(request.args))) @app.post("/input/set") @handle_errors def legacy_input_set(): input_id = request.args.get("input") if not input_id: raise ValueError("provide '?input=' -- see GET /inputs for valid values") controller.play_heos_input(_target_from(request.args), input_id) return jsonify({"input": input_id}) @app.post("/input/relay") @handle_errors def legacy_input_relay(): input_id = request.args.get("input") source = request.args.get("from") if not input_id or not source: raise ValueError("provide '?from=&input='") controller.play_heos_input(_target_from(request.args), input_id, _target_from(request.args, "from")) return jsonify({"input": input_id, "from": source}) @app.get("/raw/") @handle_errors def legacy_raw(subpath): """Diagnostic: forward any heos:// command as-is. e.g. GET /raw/browse/browse?sid=1027""" return jsonify(controller.heos.command(subpath, **request.args.to_dict())) @app.get("/avr/inputs") @handle_errors def legacy_avr_inputs(): return jsonify(controller.avr_inputs()) @app.route("/avr/input", methods=["GET", "POST"]) @handle_errors def legacy_avr_input(): if request.method == "GET": return jsonify(controller.avr_current_input() or {"error": "AVR not reachable"}) code = request.args.get("input") if not code: raise ValueError("provide '?input=', e.g. inputs/aux_in_1 -- see GET /avr/inputs") return jsonify(controller.avr_select_input(code)) def main(): global controller, spotify parser = argparse.ArgumentParser(description="HEOS panel") parser.add_argument("--port", type=int, default=config.WEB_PORT) parser.add_argument("--host", default="0.0.0.0", help="0.0.0.0 so your phone can reach it directly; " "127.0.0.1 to allow only a local reverse proxy") parser.add_argument("--demo", action="store_true", help="run with fake speakers, for working on the UI away from the kit") args = parser.parse_args() if args.demo: from demo import DemoController controller = DemoController(config) else: controller = Controller(config) spotify = _build_spotify() app.run(host=args.host, port=args.port, threaded=True) if __name__ == "__main__": main()