diff --git a/server.py b/server.py index b9ac13d..b9fbd3a 100644 --- a/server.py +++ b/server.py @@ -17,7 +17,7 @@ from pathlib import Path from typing import Optional from fastapi import FastAPI, HTTPException, UploadFile, File -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, HTMLResponse from pydantic import BaseModel import redis @@ -255,6 +255,211 @@ async def upload_to_cache(file: UploadFile = File(...)): return {"content_hash": content_hash, "filename": file.filename, "size": len(content)} +def detect_media_type(cache_path: Path) -> str: + """Detect if file is image or video based on magic bytes.""" + with open(cache_path, "rb") as f: + header = f.read(32) + + # Video signatures + if header[:4] == b'\x1a\x45\xdf\xa3': # WebM/MKV + return "video" + if header[4:8] == b'ftyp': # MP4/MOV + return "video" + if header[:4] == b'RIFF' and header[8:12] == b'AVI ': # AVI + return "video" + + # Image signatures + if header[:8] == b'\x89PNG\r\n\x1a\n': # PNG + return "image" + if header[:2] == b'\xff\xd8': # JPEG + return "image" + if header[:6] in (b'GIF87a', b'GIF89a'): # GIF + return "image" + if header[:4] == b'RIFF' and header[8:12] == b'WEBP': # WebP + return "image" + + return "unknown" + + +UI_HTML = """ + + + + Art DAG L1 Server + + + + +

Art DAG L1 Server

+ +
+ Loading... +
+ + +""" + + +@app.get("/ui", response_class=HTMLResponse) +async def ui_index(): + """Web UI for viewing runs.""" + return UI_HTML + + +@app.get("/ui/runs", response_class=HTMLResponse) +async def ui_runs(): + """HTMX partial: list of runs.""" + runs = list_all_runs() + + if not runs: + return '

No runs yet.

' + + html_parts = ['
'] + + for run in runs[:20]: # Limit to 20 most recent + status_class = run.status + + html_parts.append(f''' +
+
+
+ {run.recipe} + {run.run_id} +
+ {run.status} +
+
+ Created: {run.created_at[:19].replace('T', ' ')} +
+ ''') + + # Show input hash + if run.inputs: + html_parts.append(f'
Input: {run.inputs[0][:32]}...
') + + # Show output if completed + if run.status == "completed" and run.output_hash: + cache_path = CACHE_DIR / run.output_hash + if cache_path.exists(): + media_type = detect_media_type(cache_path) + html_parts.append(f'
Output: {run.output_hash[:32]}...
') + html_parts.append('
') + if media_type == "video": + html_parts.append(f'') + elif media_type == "image": + html_parts.append(f'{run.output_name}') + html_parts.append('
') + + # Show error if failed + if run.status == "failed" and run.error: + html_parts.append(f'
Error: {run.error}
') + + html_parts.append('
') + + html_parts.append('
') + return '\n'.join(html_parts) + + +@app.get("/ui/run/{run_id}", response_class=HTMLResponse) +async def ui_run_detail(run_id: str): + """HTMX partial: single run (for polling updates).""" + run = load_run(run_id) + if not run: + return '
Run not found
' + + # Check Celery task status if running + if run.status == "running" and run.celery_task_id: + task = celery_app.AsyncResult(run.celery_task_id) + if task.ready(): + if task.successful(): + result = task.result + run.status = "completed" + run.completed_at = datetime.now(timezone.utc).isoformat() + run.output_hash = result.get("output", {}).get("content_hash") + output_path = Path(result.get("output", {}).get("local_path", "")) + if output_path.exists(): + cache_file(output_path) + else: + run.status = "failed" + run.error = str(task.result) + save_run(run) + + status_class = run.status + poll_attr = 'hx-get="/ui/run/{}" hx-trigger="every 2s" hx-swap="outerHTML"'.format(run_id) if run.status == "running" else "" + + html = f''' +
+
+
+ {run.recipe} + {run.run_id} +
+ {run.status} +
+
+ Created: {run.created_at[:19].replace('T', ' ')} +
+ ''' + + if run.inputs: + html += f'
Input: {run.inputs[0][:32]}...
' + + if run.status == "completed" and run.output_hash: + cache_path = CACHE_DIR / run.output_hash + if cache_path.exists(): + media_type = detect_media_type(cache_path) + html += f'
Output: {run.output_hash[:32]}...
' + html += '
' + if media_type == "video": + html += f'' + elif media_type == "image": + html += f'{run.output_name}' + html += '
' + + if run.status == "failed" and run.error: + html += f'
Error: {run.error}
' + + html += '
' + return html + + if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8100)