160 lines
6.4 KiB
Python
160 lines
6.4 KiB
Python
"""Stand-in HEOS hardware: just enough of the CLI protocol to test against.
|
|
|
|
The grouping rules are the part worth pinning down -- what a set_group
|
|
call does to a stereo pair is the kind of thing you do not want to find
|
|
out by experimenting on the speakers at eleven at night.
|
|
"""
|
|
|
|
import json
|
|
import socket
|
|
import threading
|
|
|
|
|
|
class FakeHeos(threading.Thread):
|
|
"""A HEOS CLI server on localhost, with four players and a pair."""
|
|
|
|
NAMES = {1: "Home Cinema", 2: "Lego Room", 3: "Denon Home 200 L", 4: "Denon Home 200 R"}
|
|
|
|
# What browse/browse?sid=<AVR pid> reports: HEOS's own list of the
|
|
# AVR's local inputs, already under whatever names you gave them in
|
|
# its setup menu -- HEOS carries the renamed labels itself.
|
|
AVR_INPUTS = [
|
|
{"name": "Z30 Pro", "mid": "inputs/mediaplayer"},
|
|
{"name": "Switch", "mid": "inputs/game"},
|
|
{"name": "LG G5", "mid": "inputs/tvaudio"},
|
|
]
|
|
|
|
def __init__(self):
|
|
super().__init__(daemon=True)
|
|
self.groups = {3: [3, 4]} # gid -> pids, leader first
|
|
self.volumes = {pid: 20 for pid in self.NAMES}
|
|
self.play_states = {pid: "play" for pid in self.NAMES}
|
|
self.group_volumes = {3: 25}
|
|
self.now_playing_mid = {1: "inputs/mediaplayer"} # pid -> what get_now_playing_media reports
|
|
self.now_playing_sid = {} # pid -> its source id; 4 is Spotify
|
|
self.now_playing_track = {} # pid -> song/artist/image_url fields
|
|
self.commands = [] # everything we were asked to do
|
|
self.server = socket.socket()
|
|
self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.server.bind(("127.0.0.1", 0))
|
|
self.server.listen(8)
|
|
self.port = self.server.getsockname()[1]
|
|
self.start()
|
|
|
|
def run(self):
|
|
while True:
|
|
try:
|
|
conn, _ = self.server.accept()
|
|
except OSError:
|
|
return
|
|
threading.Thread(target=self._serve, args=(conn,), daemon=True).start()
|
|
|
|
def _serve(self, conn):
|
|
buffer = b""
|
|
with conn:
|
|
while True:
|
|
try:
|
|
chunk = conn.recv(4096)
|
|
except OSError:
|
|
return
|
|
if not chunk:
|
|
return
|
|
buffer += chunk
|
|
while b"\r\n" in buffer:
|
|
line, buffer = buffer.split(b"\r\n", 1)
|
|
reply = self.handle(line.decode().strip())
|
|
conn.sendall(json.dumps(reply).encode() + b"\r\n")
|
|
|
|
# -- command handling ------------------------------------------------
|
|
def handle(self, command: str) -> dict:
|
|
self.commands.append(command)
|
|
path, _, query = command[len("heos://"):].partition("?")
|
|
args = dict(part.split("=", 1) for part in query.split("&") if "=" in part)
|
|
|
|
if path == "player/get_players":
|
|
return self._ok(path, payload=[
|
|
{"name": name, "pid": pid, "model": "Fake"} for pid, name in self.NAMES.items()
|
|
])
|
|
|
|
if path == "group/get_groups":
|
|
payload = []
|
|
for gid, pids in self.groups.items():
|
|
payload.append({
|
|
"name": self.NAMES[gid], "gid": gid,
|
|
"players": [
|
|
{"name": self.NAMES[pid], "pid": pid,
|
|
"role": "leader" if pid == gid else "member"}
|
|
for pid in pids
|
|
],
|
|
})
|
|
return self._ok(path, payload=payload)
|
|
|
|
if path == "group/set_group":
|
|
pids = [int(p) for p in args["pid"].split(",")]
|
|
for gid in list(self.groups):
|
|
self.groups[gid] = [p for p in self.groups[gid] if p not in pids]
|
|
if len(self.groups[gid]) < 2:
|
|
del self.groups[gid] # HEOS dissolves a group of one
|
|
if len(pids) > 1:
|
|
self.groups[pids[0]] = pids
|
|
self.group_volumes.setdefault(pids[0], 25)
|
|
return self._ok(path, message=args["pid"])
|
|
|
|
if path.endswith("/get_volume"):
|
|
store, key = self._store(path, args)
|
|
return self._ok(path, message=f"{key[0]}={key[1]}&level={store[key[1]]}")
|
|
|
|
if path.endswith("/set_volume"):
|
|
store, key = self._store(path, args)
|
|
store[key[1]] = int(args["level"])
|
|
return self._ok(path, message=f"{key[0]}={key[1]}&level={args['level']}")
|
|
|
|
if path == "player/get_play_state":
|
|
pid = int(args["pid"])
|
|
return self._ok(path, message=f"pid={pid}&state={self.play_states[pid]}")
|
|
|
|
if path == "player/set_play_state":
|
|
pid = int(args["pid"])
|
|
self.play_states[pid] = args["state"]
|
|
return self._ok(path, message=f"pid={pid}&state={args['state']}")
|
|
|
|
if path == "player/get_now_playing_media":
|
|
pid = int(args["pid"])
|
|
mid = self.now_playing_mid.get(pid, "")
|
|
name = next((s["name"] for s in self.AVR_INPUTS if s["mid"] == mid), mid)
|
|
payload = {"mid": mid, "station": name} if mid else {}
|
|
if pid in self.now_playing_sid:
|
|
payload["sid"] = self.now_playing_sid[pid]
|
|
payload.update(self.now_playing_track.get(pid, {}))
|
|
return self._ok(path, payload=payload)
|
|
|
|
if path == "browse/browse":
|
|
sid = int(args["sid"])
|
|
payload = list(self.AVR_INPUTS) if sid == 1 else []
|
|
return self._ok(path, payload=payload)
|
|
|
|
if path == "browse/play_input":
|
|
pid = int(args["pid"])
|
|
self.now_playing_mid[pid] = args["input"]
|
|
return self._ok(path)
|
|
|
|
if path.endswith("/toggle_mute") or path == "system/heart_beat":
|
|
return self._ok(path)
|
|
|
|
return {"heos": {"command": path, "result": "fail", "message": "eid=2&text=Not+supported"}}
|
|
|
|
def _store(self, path, args):
|
|
if path.startswith("group/"):
|
|
gid = int(args["gid"])
|
|
if gid not in self.groups:
|
|
raise KeyError(gid)
|
|
return self.group_volumes, ("gid", gid)
|
|
return self.volumes, ("pid", int(args["pid"]))
|
|
|
|
@staticmethod
|
|
def _ok(path, message="", payload=None):
|
|
reply = {"heos": {"command": path, "result": "success", "message": message}}
|
|
if payload is not None:
|
|
reply["payload"] = payload
|
|
return reply
|