198 lines
8.5 KiB
Python
198 lines
8.5 KiB
Python
"""Generate the Candy camp main-menu BGM using the project's DGX music guide.
|
|
|
|
One generation only; retain request, raw response and server metadata for review.
|
|
Requires only Python's standard library. Network access to DGXA is required.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
|
|
|
|
BASE = "http://192.168.132.46:7865"
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OUTPUT = ROOT / "Tools/output/main_menu_music"
|
|
PROMPT = (
|
|
"Original instrumental background music for a cozy, colorful Candy-style "
|
|
"mobile game's main menu: a sunny miniature village camp, cream cottages, "
|
|
"green meadows, a little fountain, friendly cute animal companions and a "
|
|
"gentle sense of preparing for an adventure. Warm, welcoming, playful and "
|
|
"relaxed, cheerful without being hyperactive. A memorable but understated "
|
|
"short pentatonic melody on soft marimba and wooden xylophone, answered by "
|
|
"a mellow flute or clarinet; light nylon-string acoustic guitar arpeggios, "
|
|
"pizzicato strings, rounded upright bass and very delicate shaker and "
|
|
"woodblock percussion. A few soft celesta sparkles, warm acoustic chamber "
|
|
"arrangement, moderate gently bouncing walking pace, clear major-key "
|
|
"harmonies. Instrumental only, no singing, no humming, no choir, no speech. "
|
|
"Small recurring A and B phrases with subtle variations, stable gentle "
|
|
"dynamics and lots of breathing space for menu interactions. Designed for "
|
|
"comfortable repeated listening, a short pickup and a final phrase that "
|
|
"naturally returns to the opening harmony, loop-friendly structure, no "
|
|
"dramatic buildup, no big finale, no heavy drums, no aggressive brass, "
|
|
"no distorted guitars, no EDM drops, no environmental sound effects. "
|
|
"Polished clean stereo game soundtrack."
|
|
)
|
|
|
|
|
|
def save(name, value):
|
|
(OUTPUT / name).write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
|
|
def get_json(path, payload=None):
|
|
request = urllib.request.Request(
|
|
BASE + path,
|
|
data=None if payload is None else json.dumps(payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=60) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def main():
|
|
global OUTPUT, PROMPT
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, default=OUTPUT)
|
|
parser.add_argument("--prompt-file", type=Path)
|
|
args = parser.parse_args()
|
|
OUTPUT = args.output.resolve()
|
|
if args.prompt_file:
|
|
PROMPT = args.prompt_file.read_text(encoding="utf-8").strip()
|
|
if not PROMPT:
|
|
raise SystemExit("Music prompt must not be empty")
|
|
OUTPUT.mkdir(parents=True, exist_ok=True)
|
|
if (OUTPUT / "request.json").exists():
|
|
if not (OUTPUT / "joined.json").exists():
|
|
raise SystemExit("Existing request without receipt; inspect before submitting again.")
|
|
if (OUTPUT / "response.json").exists():
|
|
raise SystemExit("Existing response found; do not submit another generation.")
|
|
resume = json.loads((OUTPUT / "request.json").read_text(encoding="utf-8"))
|
|
else:
|
|
resume = None
|
|
config = get_json("/config")
|
|
info = get_json("/gradio_api/info")
|
|
parameters = info["named_endpoints"]["/generation_wrapper"]["parameters"]
|
|
components = {component["id"]: component for component in config["components"]}
|
|
dependency = next(
|
|
item for item in config["dependencies"] if item.get("api_name") == "generation_wrapper"
|
|
)
|
|
# Map public parameter names to live inputs, skipping hidden State entries.
|
|
# In this service parameter names retain the original component index.
|
|
overrides = {"param_0": PROMPT, "param_1": "", "param_6": 50,
|
|
"param_11": -1, "param_12": 1}
|
|
data = []
|
|
public_index = 0
|
|
for component_id in dependency["inputs"]:
|
|
component = components[component_id]
|
|
if component["type"] == "state":
|
|
data.append(None)
|
|
continue
|
|
parameter = parameters[public_index]
|
|
public_index += 1
|
|
data.append(overrides.get(parameter["parameter_name"],
|
|
component.get("props", {}).get("value")))
|
|
assert public_index == len(parameters), "Public parameter/input mismatch"
|
|
assert data[6] == 50 and data[12] == 1
|
|
session = str(uuid.uuid4())
|
|
payload = {"data": data, "fn_index": dependency["id"], "session_hash": session}
|
|
if resume:
|
|
payload = resume["payload"]
|
|
session = payload["session_hash"]
|
|
print("resuming existing job (no new generation)", flush=True)
|
|
else:
|
|
save("request.json", {"service": BASE, "protocol": config.get("protocol"),
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
|
"payload": payload, "parameters": parameters})
|
|
ready = threading.Event()
|
|
finished = threading.Event()
|
|
state = {"events": []}
|
|
|
|
def read_sse():
|
|
try:
|
|
url = BASE + "/gradio_api/queue/data?" + urllib.parse.urlencode({"session_hash": session})
|
|
deadline = time.monotonic() + 1800
|
|
while time.monotonic() < deadline:
|
|
with urllib.request.urlopen(url, timeout=1800) as response:
|
|
ready.set()
|
|
for raw_line in response:
|
|
line = raw_line.decode("utf-8").strip()
|
|
if not line.startswith("data:"):
|
|
continue
|
|
event = json.loads(line[5:])
|
|
message = event.get("msg")
|
|
if message == "heartbeat":
|
|
continue
|
|
# Gradio may reject an SSE subscription until join has
|
|
# created the session; reconnect to the SAME session.
|
|
if event.get("session_not_found"):
|
|
break
|
|
state["events"].append(event)
|
|
save("events.json", state["events"])
|
|
if message in {"estimation", "process_starts"}:
|
|
print(message, flush=True)
|
|
if message == "process_completed":
|
|
state["result"] = event
|
|
finished.set()
|
|
return
|
|
time.sleep(1)
|
|
except Exception as error:
|
|
state["error"] = repr(error)
|
|
finally:
|
|
ready.set()
|
|
finished.set()
|
|
|
|
reader = threading.Thread(target=read_sse, daemon=True)
|
|
reader.start()
|
|
if not ready.wait(60) or state.get("error"):
|
|
raise SystemExit(f"Could not open queue stream: {state.get('error')}")
|
|
if resume:
|
|
joined = json.loads((OUTPUT / "joined.json").read_text(encoding="utf-8"))
|
|
else:
|
|
joined = get_json("/gradio_api/queue/join", payload)
|
|
save("joined.json", joined)
|
|
print("submitted", joined["event_id"], flush=True)
|
|
if not finished.wait(1800):
|
|
raise SystemExit("Generation timeout; inspect the existing job before resubmitting.")
|
|
if state.get("error"):
|
|
raise SystemExit(state["error"])
|
|
result = state.get("result")
|
|
if not result or result.get("event_id") != joined["event_id"]:
|
|
raise SystemExit("No matching completion event")
|
|
save("response.json", result)
|
|
if not result.get("success", True):
|
|
raise SystemExit(json.dumps(result.get("output"), ensure_ascii=True))
|
|
|
|
files = {}
|
|
|
|
def walk(value):
|
|
if isinstance(value, dict):
|
|
url = value.get("url")
|
|
name = str(value.get("orig_name") or value.get("path") or "")
|
|
if url and name.lower().endswith((".mp3", ".json")):
|
|
files[url] = Path(name).name
|
|
for child in value.values():
|
|
walk(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
walk(child)
|
|
walk(result["output"]["data"])
|
|
if not any(name.endswith(".mp3") for name in files.values()):
|
|
raise SystemExit("No downloadable MP3 URL in response; inspect response.json")
|
|
for url, name in files.items():
|
|
# Use the URL returned by the service; never invent a file endpoint.
|
|
url = urllib.parse.urljoin(BASE, url)
|
|
urllib.request.urlretrieve(url, OUTPUT / name)
|
|
print("saved", name, flush=True)
|
|
save("downloaded.json", files)
|
|
print("Generation Complete", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|