heos/app.py

#!/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                 # http://<pi-ip>:5005/

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
from functools import wraps

from flask import Flask, jsonify, render_template, request

import config
from avr import AvrError
from controller import Controller, TargetError
from heos import HeosError

app = Flask(__name__)
controller: Controller = None


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, AvrError) 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


# --- The UI -----------------------------------------------------------
@app.route("/")
def index():
    return render_template(
        "index.html",
        app_name=config.APP_NAME,
        host=config.TARGETS[config.HOST_KEY],
        rooms=[{"key": key, **config.TARGETS[key]} for key in config.ROOM_KEYS],
        step=config.VOLUME_STEP,
    )


@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():
    return jsonify(controller.state())


@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/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():
    refresh = request.args.get("refresh") in ("1", "true", "yes")
    return jsonify(controller.avr.inputs(refresh=refresh))


@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. GAME -- see GET /api/avr/inputs")
    return jsonify(controller.avr.select_input(code))


# --- 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/raw")
@handle_errors
def legacy_avr_raw():
    cmd = request.args.get("cmd")
    if not cmd:
        raise ValueError("provide '?cmd=<raw telnet command>'")
    return jsonify({"lines": controller.avr.telnet.request(cmd, timeout=3.0)})


@app.get("/avr/inputs")
@handle_errors
def legacy_avr_inputs():
    return jsonify(controller.avr.inputs(refresh=True))


@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. GAME, TV, CD, AUX1")
    return jsonify(controller.avr.select_input(code))


def main():
    global controller
    parser = argparse.ArgumentParser(description="HEOS panel")
    parser.add_argument("--port", type=int, default=config.WEB_PORT)
    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)

    # 0.0.0.0 so your phone can reach it over the LAN.
    app.run(host="0.0.0.0", port=args.port, threaded=True)


if __name__ == "__main__":
    main()