franzz/heos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5ca504c680cd504eaeb48248ca23570d0b4f77d2
heos/app.py
T
franzz 5ca504c680
Deploy HEOS panel / deploy (push) Successful in 26s
Details
Fix account swaping
2026-09-18 02:05:48 +02:00

724 lines
27 KiB
Python
Raw Blame History

#!/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://<pi-ip>: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/<int:poster_id>")
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/<state>")
@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=<room>&members=<comma-separated rooms>'")
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=<source room>&input=<input id>'")
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/<path:subpath>")
@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=<code>', 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()
Reference in New Issue View Git Blame Copy Permalink