74 lines
5.5 KiB
Python
74 lines
5.5 KiB
Python
"""LadyPoly UI -> authenticated Deck music API. No local model engine."""
|
|
import json, os, time, urllib.request, urllib.error
|
|
from pathlib import Path
|
|
from fastapi import HTTPException
|
|
from fastapi.responses import HTMLResponse
|
|
from pydantic import BaseModel
|
|
|
|
CONNECTION = Path('/data/deck-connection.json')
|
|
BASE = 'http://127.0.0.1:8120'
|
|
|
|
def request(path, data=None, connection=None):
|
|
c = connection or (json.loads(CONNECTION.read_text()) if CONNECTION.exists() else {})
|
|
if not c.get('token'): raise RuntimeError('Zuerst unter Deck-Verbindung den API-Token hinterlegen.')
|
|
req = urllib.request.Request(BASE + path, data=json.dumps(data).encode() if data is not None else None,
|
|
headers={'Authorization': 'Bearer ' + c['token'], 'Content-Type':'application/json'})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=1850) as response: return response.read()
|
|
except urllib.error.HTTPError as exc:
|
|
try: message=json.loads(exc.read()).get('error', {}).get('message', 'Deck-Anfrage abgelehnt')
|
|
except Exception: message='Deck-Anfrage abgelehnt'
|
|
raise RuntimeError(message) from None
|
|
|
|
class Loras:
|
|
def __init__(self, root):self.extra_dirs=[]
|
|
def resolve(self, values):return [], []
|
|
def status(self, values):return {'adapters':[], 'selected':[], 'missing':[], 'dirs':[]}
|
|
def discover(self):return []
|
|
|
|
class Connection(BaseModel):
|
|
token: str
|
|
model: str
|
|
|
|
PAGE='''<!doctype html><meta charset="utf-8"><style>body{background:#12151c;color:#eee;font:18px system-ui;max-width:720px;margin:50px auto}input,button{padding:12px;display:block;margin:15px 0;width:90%}a{color:#72d6ff}</style><h1>LadyPoly · Deck-Verbindung</h1><p>In Deck das YuE2-Profil am API-Endpunkt freigeben und den Musik-Modus aktivieren. Hier den API-Token aus „Zugang & API“ und den genauen Profilnamen eingeben.</p><form id="f"><input id="token" type="password" placeholder="Deck API-Token" required autocomplete="off"><input id="model" placeholder="API-Profilname, z. B. YuE2 Studio" required><button>Verbindung prüfen und speichern</button></form><p id="msg"></p><a href="/">LadyPoly öffnen →</a><p>Diese Anbindung unterstützt Liedtext, Stil, Seed und Syntheseschritte. Score, LoRA, eigener Writer und Cover-Generator sind hier noch nicht angebunden. Deck lädt und entlädt die Gewichte.</p><script>f.onsubmit=async e=>{e.preventDefault();msg.textContent='Prüfe …';try{let r=await fetch('/api/deck/connection',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({token:token.value,model:model.value})});let d=await r.json();msg.textContent=r.ok?'Verbunden. Jetzt LadyPoly öffnen.':d.detail;token.value=''}catch(e){msg.textContent='Verbindung fehlgeschlagen'}}</script>'''
|
|
|
|
def install(g):
|
|
app=g['app'];g['SETTINGS'].song_engine='gguf';g['SETTINGS'].art_auto=False;g['SETTINGS'].stall_timeout=0
|
|
@app.get('/deck-connection', response_class=HTMLResponse)
|
|
def connection_page():return PAGE
|
|
@app.post('/api/deck/connection')
|
|
def connection_save(body: Connection):
|
|
c=body.model_dump()
|
|
if not 1<=len(body.token)<=512:raise HTTPException(400,'Ungültiger Token.')
|
|
try:
|
|
models=json.loads(request('/v1/audio/music/models',connection=c))['data']
|
|
if body.model not in [x['id'] for x in models]:raise RuntimeError('Profil nicht freigegeben oder nicht ausführbar. Bitte in Deck prüfen.')
|
|
except Exception as e:raise HTTPException(400,str(e)) from None
|
|
CONNECTION.write_text(json.dumps(c));CONNECTION.chmod(0o600)
|
|
return {'ok':True}
|
|
def render(job,spec):
|
|
if spec.get('abc') or spec.get('cfg_scale') is not None or spec.get('abc_sampling') or spec.get('semantic_sampling') or g['SETTINGS'].loras:
|
|
raise RuntimeError('Score, Sampling-Overrides und LoRA sind in der Deck-Anbindung noch nicht verfügbar.')
|
|
c=json.loads(CONNECTION.read_text()) if CONNECTION.exists() else {}
|
|
spec['cot']='off'
|
|
job.setup='Deck lädt YuE2 und erzeugt Musik · Musik-Modus erforderlich';job.setup_at=time.time();job.push(force=True)
|
|
start=time.monotonic()
|
|
audio=request('/v1/audio/music',dict(model=c.get('model',''),lyrics=spec['lyrics'],style=spec['style'],seed=spec['seed'],steps=g['SETTINGS'].ode_steps,max_tokens=9000))
|
|
if job.cancel_flag.is_set():raise InterruptedError('UI-Auftrag abgebrochen. Laufenden Deck-Job in Deck stoppen.')
|
|
work=g['OUTPUTS']/('.pending-'+job.id);work.mkdir(parents=True,exist_ok=True)
|
|
wav=work/'audio.wav';wav.write_bytes(audio)
|
|
return g['GgufSong'](spec,wav,{'e2e_seconds':time.monotonic()-start},'',{'runtime':'Athena Deck audio.cpp','profile':c.get('model')})
|
|
g['render_gguf']=render;g['render_torch']=render
|
|
g['ENGINE'].detail='Remote-Laufzeit in Athena Deck · Musik-Modus erforderlich'
|
|
@app.middleware('http')
|
|
async def deck_guard(req,call_next):
|
|
path=req.url.path
|
|
origin=req.headers.get('Origin')
|
|
if req.method!='GET' and origin and origin not in ('http://'+req.headers.get('Host',''),'https://'+req.headers.get('Host','')):
|
|
return g['JSONResponse']({'detail':'Cross-Origin-Zugriff nicht erlaubt.'},status_code=403)
|
|
if req.method!='GET' and not (path in {'/api/generate','/api/deck/connection','/api/unlock'} or (req.method=='DELETE' and path.startswith('/api/library/') and path.count('/')==3)):
|
|
return g['JSONResponse']({'detail':'Laufzeiten, Modelle und Einstellungen werden in Athena Deck verwaltet.'},status_code=409)
|
|
response=await call_next(req)
|
|
response.headers['Cache-Control']='no-store'
|
|
return response
|