poprhythm
39fdc4276f
Wraps show/persona/schedule/trigger calls that were previously done by hand with raw docker exec curl, and documents the gotchas hit tonight (whole-object PUT/POST semantics, the frequency enum, the silent default-persona trap, and the port-not-published-to-host setup). Claude-Session: https://claude.ai/code/session_01HZQK6jHmdTpFjFZM8FUnqA
380 lines
13 KiB
Python
380 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""SUB/WAVE admin-API CLI. Auth from subwave/.env, calls proxied through
|
|
`docker exec sub-wave-controller curl ...` because the controller's admin API
|
|
(port 7701) is not published to the host -- only reachable inside the
|
|
sub-wave-controller container itself.
|
|
|
|
Usage:
|
|
subwave_api.py persona list
|
|
subwave_api.py persona create --name N --voice V --tagline T --soul S
|
|
[--frequency silent|quiet|moderate|chatty|aggressive]
|
|
[--humour 0-10] [--warmth 0-10] [--local-colour 0-10]
|
|
[--script-length concise|...] [--link-style natural|...]
|
|
subwave_api.py show list
|
|
subwave_api.py show get --id ID
|
|
subwave_api.py show create --name N --topic T --playlist-id ID
|
|
[--genres a,b,c] [--moods a,b] [--energies a,b]
|
|
[--eras 1990-2000,2005-2010] [--persona-id ID]
|
|
subwave_api.py show set-persona --id ID --persona-id ID
|
|
subwave_api.py show set-name --id ID --name N
|
|
subwave_api.py show set-topic --id ID --topic T
|
|
subwave_api.py schedule grid
|
|
subwave_api.py schedule open-slots
|
|
subwave_api.py schedule set --show-id ID --day 0-6 --hour 0-23
|
|
subwave_api.py schedule set-random --show-id ID
|
|
subwave_api.py trigger --show-id ID [--minutes 60]
|
|
"""
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import random
|
|
|
|
ENV_PATH = "/home/poprhythm/docker-infrastructure/subwave/.env"
|
|
STATE_SHOWS = "/srv/subwave/state/schedule.json"
|
|
STATE_SETTINGS = "/srv/subwave/state/settings.json"
|
|
CONTAINER = "sub-wave-controller"
|
|
BASE_URL = "http://localhost:7701"
|
|
|
|
|
|
def load_creds():
|
|
creds = {}
|
|
with open(ENV_PATH) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
creds[k] = v
|
|
return creds["ADMIN_USER"], creds["ADMIN_PASS"]
|
|
|
|
|
|
def api_call(method, path, body=None):
|
|
"""Runs curl inside the controller container. Body, if given, is written
|
|
to a temp file on the host, docker cp'd in, and cleaned up after -- avoids
|
|
quoting hell with large/nested JSON on the command line."""
|
|
user, password = load_creds()
|
|
if body is None:
|
|
cmd = [
|
|
"docker", "exec", CONTAINER, "sh", "-c",
|
|
f"curl -s -u '{user}:{password}' -X {method} {BASE_URL}{path}",
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
else:
|
|
import tempfile
|
|
import os
|
|
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
|
|
json.dump(body, f)
|
|
local_path = f.name
|
|
remote_path = f"/tmp/{os.path.basename(local_path)}"
|
|
try:
|
|
subprocess.run(["docker", "cp", local_path, f"{CONTAINER}:{remote_path}"], check=True)
|
|
cmd = [
|
|
"docker", "exec", CONTAINER, "sh", "-c",
|
|
f"curl -s -u '{user}:{password}' -X {method} {BASE_URL}{path} "
|
|
f"-H 'Content-Type: application/json' -d @{remote_path}",
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
finally:
|
|
subprocess.run(["docker", "exec", CONTAINER, "rm", "-f", remote_path],
|
|
capture_output=True)
|
|
os.unlink(local_path)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"docker exec failed: {result.stderr}")
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
raise RuntimeError(f"non-JSON response: {result.stdout[:500]}")
|
|
|
|
|
|
def load_state(path):
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def parse_eras(s):
|
|
eras = []
|
|
for part in s.split(","):
|
|
a, b = part.split("-")
|
|
eras.append({"fromYear": int(a), "toYear": int(b)})
|
|
return eras
|
|
|
|
|
|
# ---- persona ----
|
|
|
|
def cmd_persona_list(args):
|
|
d = load_state(STATE_SETTINGS)
|
|
for p in d["personas"]:
|
|
voice = p.get("tts", {}).get("voice", "")
|
|
print(f"{p['id']} | {p['name']:<12} | {voice:<12} | djMode={p.get('djMode')}")
|
|
|
|
|
|
def cmd_persona_create(args):
|
|
d = load_state(STATE_SETTINGS)
|
|
new_persona = {
|
|
"name": args.name,
|
|
"tagline": args.tagline,
|
|
"frequency": args.frequency,
|
|
"scriptLength": args.script_length,
|
|
"djMode": True,
|
|
"linkStyle": args.link_style,
|
|
"humour": args.humour,
|
|
"localColour": args.local_colour,
|
|
"warmth": args.warmth,
|
|
"soul": args.soul,
|
|
"language": "",
|
|
"avatar": "",
|
|
"tts": {"engine": "kokoro", "cloudProvider": "openai", "voice": args.voice,
|
|
"gainDb": 0, "speed": 1},
|
|
"skills": None,
|
|
"tags": [],
|
|
}
|
|
personas = d["personas"] + [new_persona]
|
|
resp = api_call("POST", "/settings", {"personas": personas})
|
|
if "error" in resp:
|
|
print(f"error: {resp['error']}", file=sys.stderr)
|
|
sys.exit(1)
|
|
d2 = load_state(STATE_SETTINGS)
|
|
for p in d2["personas"]:
|
|
if p["name"] == args.name and p not in d["personas"]:
|
|
print(f"created '{args.name}': {p['id']} (voice={args.voice})")
|
|
return
|
|
print(f"created '{args.name}' (re-read settings to confirm id)")
|
|
|
|
|
|
# ---- show ----
|
|
|
|
def cmd_show_list(args):
|
|
d = load_state(STATE_SHOWS)
|
|
for s in d["shows"]:
|
|
print(f"{s['id']} | {s['name']}")
|
|
|
|
|
|
def cmd_show_get(args):
|
|
d = load_state(STATE_SHOWS)
|
|
for s in d["shows"]:
|
|
if s["id"] == args.id:
|
|
print(json.dumps(s, indent=2))
|
|
return
|
|
print(f"no such show: {args.id}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_show_create(args):
|
|
show = {
|
|
"name": args.name,
|
|
"topic": args.topic,
|
|
"personaId": args.persona_id or "p_default0",
|
|
"guestPersonaIds": [],
|
|
"banter": False,
|
|
"pauseTalk": False,
|
|
"programme": False,
|
|
"segmentSkill": "",
|
|
"moods": args.moods.split(",") if args.moods else [],
|
|
"themeId": "",
|
|
"genres": args.genres.split(",") if args.genres else [],
|
|
"energies": args.energies.split(",") if args.energies else [],
|
|
"eras": parse_eras(args.eras) if args.eras else [],
|
|
"vocals": "",
|
|
"filtersStrict": True,
|
|
"maxTrackSeconds": args.max_track_seconds,
|
|
"minTrackLengthSeconds": None,
|
|
"fadeAtShowEnd": None,
|
|
"playlistIds": [args.playlist_id],
|
|
"playlistStrict": True,
|
|
"playlistExhaust": True,
|
|
"excludedPlaylistIds": [],
|
|
"tags": [],
|
|
}
|
|
resp = api_call("POST", "/shows", {"show": show})
|
|
if "error" in resp:
|
|
print(f"error: {resp['error']}", file=sys.stderr)
|
|
sys.exit(1)
|
|
saved = resp.get("show", {})
|
|
print(f"created '{args.name}': {saved.get('id')}")
|
|
|
|
|
|
def _update_show_field(show_id, field, value):
|
|
d = load_state(STATE_SHOWS)
|
|
for s in d["shows"]:
|
|
if s["id"] == show_id:
|
|
s = dict(s)
|
|
s[field] = value
|
|
resp = api_call("POST", "/shows", {"show": s})
|
|
if "error" in resp:
|
|
print(f"error: {resp['error']}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"{show_id}: {field} -> {value}")
|
|
return
|
|
print(f"no such show: {show_id}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_show_set_persona(args):
|
|
_update_show_field(args.id, "personaId", args.persona_id)
|
|
|
|
|
|
def cmd_show_set_name(args):
|
|
_update_show_field(args.id, "name", args.name)
|
|
|
|
|
|
def cmd_show_set_topic(args):
|
|
_update_show_field(args.id, "topic", args.topic)
|
|
|
|
|
|
# ---- schedule ----
|
|
|
|
def cmd_schedule_grid(args):
|
|
d = load_state(STATE_SHOWS)
|
|
names = {s["id"]: s["name"] for s in d["shows"]}
|
|
for day in "0123456":
|
|
row = [names.get(v, "null") if v else "null" for v in d["schedule"][day]]
|
|
print(f"day {day}:", row)
|
|
|
|
|
|
def cmd_schedule_open_slots(args):
|
|
d = load_state(STATE_SHOWS)
|
|
open_slots = []
|
|
for day in "0123456":
|
|
for h, v in enumerate(d["schedule"][day]):
|
|
if v is None:
|
|
open_slots.append((day, h))
|
|
print(f"{len(open_slots)} open slots")
|
|
for day, h in open_slots:
|
|
print(f" day {day} hour {h}")
|
|
|
|
|
|
def cmd_schedule_set(args):
|
|
d = load_state(STATE_SHOWS)
|
|
schedule = d["schedule"]
|
|
schedule[str(args.day)][args.hour] = args.show_id
|
|
resp = api_call("PUT", "/schedule", {"schedule": schedule})
|
|
if "error" in resp:
|
|
print(f"error: {resp['error']}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"set day {args.day} hour {args.hour} -> {args.show_id} (dropped={resp.get('dropped')})")
|
|
|
|
|
|
def cmd_schedule_set_random(args):
|
|
d = load_state(STATE_SHOWS)
|
|
schedule = d["schedule"]
|
|
open_slots = [(day, h) for day in "0123456" for h, v in enumerate(schedule[day]) if v is None]
|
|
if not open_slots:
|
|
print("no open slots", file=sys.stderr)
|
|
sys.exit(1)
|
|
day, h = random.choice(open_slots)
|
|
schedule[day][h] = args.show_id
|
|
resp = api_call("PUT", "/schedule", {"schedule": schedule})
|
|
if "error" in resp:
|
|
print(f"error: {resp['error']}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"set day {day} hour {h} -> {args.show_id} (dropped={resp.get('dropped')})")
|
|
|
|
|
|
# ---- trigger ----
|
|
|
|
def cmd_trigger(args):
|
|
resp = api_call("POST", "/schedule/override",
|
|
{"showId": args.show_id, "minutes": args.minutes, "until": "fixed"})
|
|
if "error" in resp:
|
|
print(f"error: {resp['error']}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(json.dumps(resp))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
|
|
persona = sub.add_parser("persona")
|
|
persona_sub = persona.add_subparsers(dest="subcmd", required=True)
|
|
|
|
p = persona_sub.add_parser("list")
|
|
p.set_defaults(func=cmd_persona_list)
|
|
|
|
p = persona_sub.add_parser("create")
|
|
p.add_argument("--name", required=True)
|
|
p.add_argument("--voice", required=True, help="Kokoro voice id, e.g. am_puck")
|
|
p.add_argument("--tagline", required=True)
|
|
p.add_argument("--soul", required=True)
|
|
p.add_argument("--frequency", default="moderate",
|
|
choices=["silent", "quiet", "moderate", "chatty", "aggressive"])
|
|
p.add_argument("--script-length", default="concise")
|
|
p.add_argument("--link-style", default="natural")
|
|
p.add_argument("--humour", type=int, default=5)
|
|
p.add_argument("--warmth", type=int, default=6)
|
|
p.add_argument("--local-colour", type=int, default=5)
|
|
p.set_defaults(func=cmd_persona_create)
|
|
|
|
show = sub.add_parser("show")
|
|
show_sub = show.add_subparsers(dest="subcmd", required=True)
|
|
|
|
p = show_sub.add_parser("list")
|
|
p.set_defaults(func=cmd_show_list)
|
|
|
|
p = show_sub.add_parser("get")
|
|
p.add_argument("--id", required=True)
|
|
p.set_defaults(func=cmd_show_get)
|
|
|
|
p = show_sub.add_parser("create")
|
|
p.add_argument("--name", required=True)
|
|
p.add_argument("--topic", required=True)
|
|
p.add_argument("--playlist-id", required=True, help="Navidrome playlist id (the anchor)")
|
|
p.add_argument("--genres", default="")
|
|
p.add_argument("--moods", default="")
|
|
p.add_argument("--energies", default="")
|
|
p.add_argument("--eras", default="", help="e.g. 1990-1995,2000-2005")
|
|
p.add_argument("--persona-id", default="")
|
|
p.add_argument("--max-track-seconds", type=int, default=None)
|
|
p.set_defaults(func=cmd_show_create)
|
|
|
|
p = show_sub.add_parser("set-persona")
|
|
p.add_argument("--id", required=True)
|
|
p.add_argument("--persona-id", required=True)
|
|
p.set_defaults(func=cmd_show_set_persona)
|
|
|
|
p = show_sub.add_parser("set-name")
|
|
p.add_argument("--id", required=True)
|
|
p.add_argument("--name", required=True)
|
|
p.set_defaults(func=cmd_show_set_name)
|
|
|
|
p = show_sub.add_parser("set-topic")
|
|
p.add_argument("--id", required=True)
|
|
p.add_argument("--topic", required=True)
|
|
p.set_defaults(func=cmd_show_set_topic)
|
|
|
|
schedule = sub.add_parser("schedule")
|
|
schedule_sub = schedule.add_subparsers(dest="subcmd", required=True)
|
|
|
|
p = schedule_sub.add_parser("grid")
|
|
p.set_defaults(func=cmd_schedule_grid)
|
|
|
|
p = schedule_sub.add_parser("open-slots")
|
|
p.set_defaults(func=cmd_schedule_open_slots)
|
|
|
|
p = schedule_sub.add_parser("set")
|
|
p.add_argument("--show-id", required=True)
|
|
p.add_argument("--day", type=int, required=True, help="0=Sunday .. 6=Saturday")
|
|
p.add_argument("--hour", type=int, required=True, help="0-23")
|
|
p.set_defaults(func=cmd_schedule_set)
|
|
|
|
p = schedule_sub.add_parser("set-random")
|
|
p.add_argument("--show-id", required=True)
|
|
p.set_defaults(func=cmd_schedule_set_random)
|
|
|
|
p = sub.add_parser("trigger")
|
|
p.add_argument("--show-id", required=True)
|
|
p.add_argument("--minutes", type=int, default=60)
|
|
p.set_defaults(func=cmd_trigger)
|
|
|
|
args = parser.parse_args()
|
|
try:
|
|
args.func(args)
|
|
except Exception as e:
|
|
print(f"error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|