153 lines
5.1 KiB
Python
153 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import atexit
|
|
import logging
|
|
import math
|
|
import os
|
|
from typing import Any
|
|
|
|
from flask import Flask, jsonify, render_template, request
|
|
from werkzeug.exceptions import BadRequest, HTTPException
|
|
|
|
from .sonos import SonosError, SonosNotFound, SonosService, SonosTimeout
|
|
|
|
|
|
def create_app(config: dict[str, Any] | None = None, service: SonosService | None = None) -> Flask:
|
|
app = Flask(__name__)
|
|
app.config.from_mapping(
|
|
JSON_SORT_KEYS=False,
|
|
SONOS_DISCOVERY_TIMEOUT=float(os.getenv("SONOS_DISCOVERY_TIMEOUT", "5")),
|
|
SONOS_OPERATION_TIMEOUT=float(os.getenv("SONOS_OPERATION_TIMEOUT", "8")),
|
|
MAX_CONTENT_LENGTH=int(os.getenv("SONOS_MAX_JSON_BYTES", "16384")),
|
|
)
|
|
if config:
|
|
app.config.update(config)
|
|
for name in ("SONOS_DISCOVERY_TIMEOUT", "SONOS_OPERATION_TIMEOUT"):
|
|
value = app.config[name]
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
|
|
raise ValueError(f"{name} muss eine positive endliche Zahl sein.")
|
|
max_body = app.config["MAX_CONTENT_LENGTH"]
|
|
if isinstance(max_body, bool) or not isinstance(max_body, int) or max_body <= 0:
|
|
raise ValueError("MAX_CONTENT_LENGTH muss eine positive ganze Zahl sein.")
|
|
sonos = service or SonosService(app.config["SONOS_DISCOVERY_TIMEOUT"], app.config["SONOS_OPERATION_TIMEOUT"])
|
|
app.extensions["sonos"] = sonos
|
|
if service is None:
|
|
atexit.register(sonos.close)
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return render_template("index.html")
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return jsonify(status="ok")
|
|
|
|
@app.get("/api/zones")
|
|
def zones():
|
|
refresh = request.args.get("refresh", "1") != "0"
|
|
return jsonify(zones=sonos.zones(refresh=refresh))
|
|
|
|
@app.post("/api/zones/<uid>/transport")
|
|
def transport(uid: str):
|
|
data = body()
|
|
sonos.transport(uid, required_string(data, "action"))
|
|
return jsonify(ok=True)
|
|
|
|
@app.put("/api/zones/<uid>/volume")
|
|
def volume(uid: str):
|
|
data = body()
|
|
sonos.set_volume(uid, data.get("volume"))
|
|
return jsonify(ok=True)
|
|
|
|
@app.put("/api/zones/<uid>/eq")
|
|
def equalizer(uid: str):
|
|
data = body()
|
|
sonos.set_eq(uid, required_string(data, "control"), data.get("value"))
|
|
return jsonify(ok=True)
|
|
|
|
@app.post("/api/groups")
|
|
def group():
|
|
data = body()
|
|
coordinator = required_string(data, "coordinatorUid")
|
|
members = data.get("memberUids")
|
|
if not isinstance(members, list) or not all(isinstance(v, str) and v for v in members):
|
|
raise ValueError("memberUids muss eine Liste gültiger Zonen-IDs sein.")
|
|
sonos.group(coordinator, members)
|
|
return jsonify(ok=True)
|
|
|
|
@app.delete("/api/groups/<uid>")
|
|
def ungroup(uid: str):
|
|
sonos.ungroup(uid)
|
|
return jsonify(ok=True)
|
|
|
|
@app.get("/api/zones/<uid>/favorites")
|
|
def favorites(uid: str):
|
|
return jsonify(favorites=sonos.favorites(uid))
|
|
|
|
@app.get("/api/zones/<uid>/queue")
|
|
def queue(uid: str):
|
|
return jsonify(queue=sonos.queue(uid))
|
|
|
|
@app.post("/api/zones/<uid>/play/favorite")
|
|
def play_favorite(uid: str):
|
|
sonos.play_favorite(uid, body().get("index"))
|
|
return jsonify(ok=True)
|
|
|
|
@app.post("/api/zones/<uid>/play/queue")
|
|
def play_queue(uid: str):
|
|
sonos.play_queue(uid, body().get("index"))
|
|
return jsonify(ok=True)
|
|
|
|
@app.post("/api/zones/<uid>/play/stream")
|
|
def play_stream(uid: str):
|
|
sonos.play_stream(uid, required_string(body(), "url"))
|
|
return jsonify(ok=True)
|
|
|
|
@app.errorhandler(ValueError)
|
|
def validation_error(error):
|
|
return jsonify(error="validation_error", message=str(error)), 400
|
|
|
|
@app.errorhandler(SonosNotFound)
|
|
def not_found(error):
|
|
return jsonify(error="zone_not_found", message=str(error)), 404
|
|
|
|
@app.errorhandler(SonosTimeout)
|
|
def timeout(error):
|
|
return jsonify(error="sonos_timeout", message=str(error)), 504
|
|
|
|
@app.errorhandler(SonosError)
|
|
def sonos_error(error):
|
|
app.logger.warning("Sonos-Fehler: %s", error)
|
|
return jsonify(error="sonos_error", message=str(error)), 502
|
|
|
|
@app.errorhandler(HTTPException)
|
|
def http_error(error):
|
|
return jsonify(error="http_error", message=error.description), error.code
|
|
|
|
@app.errorhandler(Exception)
|
|
def unexpected(error):
|
|
app.logger.exception("Unerwarteter Fehler", exc_info=error)
|
|
return jsonify(error="internal_error", message="Interner Serverfehler."), 500
|
|
|
|
return app
|
|
|
|
|
|
def body() -> dict[str, Any]:
|
|
if not request.is_json:
|
|
raise BadRequest("Content-Type application/json ist erforderlich.")
|
|
data = request.get_json(silent=False)
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Der JSON-Inhalt muss ein Objekt sein.")
|
|
return data
|
|
|
|
|
|
def required_string(data: dict[str, Any], field: str) -> str:
|
|
value = data.get(field)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError(f"{field} ist erforderlich.")
|
|
return value.strip()
|
|
|
|
|
|
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
|
|
app = create_app()
|