import argparse import os import httpx import uvicorn import json import urllib.request import time import subprocess import sys from datetime import datetime from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, StreamingResponse, Response from pydantic import BaseModel from typing import List # Command Line Arguments Configuration parser = argparse.ArgumentParser(description="VLM Edge Studio WebApp Bridge") parser.add_argument("--video-dir", required=True, help="Directory path where video MP4 files are hosted") parser.add_argument("--aaf-server", default="http://127.0.0.1:8000", help="AAF Server backend Base URL") parser.add_argument("--host", default="0.0.0.0", help="Host binding address for this web application") parser.add_argument("--port", type=int, default=8080, help="Port binding for this web application") parser.add_argument("--verbose", action="store_true", default=False, help="Enable verbose terminal dumping") args, _ = parser.parse_known_args() TARGET_MODEL = "Qwen2.5-VL-7B-Instruct" CONFIG_PATH = "/usr/share/eiq/aaf-connector/server_config.json" SERVICE_NAME = "eiq-aaf-connector.service" # --- MODEL CONFIG & SERVICE CONTROL FUNCTIONS --- def ensure_model_enabled(target_model: str, config_path: str) -> bool: """Updates server_config.json so only the target model is set to enabled=True.""" if not os.path.exists(config_path): print(f"[Warning] Config file not found at {config_path}.") return False try: with open(config_path, "r") as f: config = json.load(f) config_changed = False available_models = config.get("available_models", []) if isinstance(available_models, list): for model_info in available_models: model_id = model_info.get("name") or model_info.get("id") should_be_enabled = (model_id == target_model) if model_info.get("enabled") != should_be_enabled: model_info["enabled"] = should_be_enabled config_changed = True if config_changed: print(f"[Config] Enabling '{target_model}' in server_config.json...") with open(config_path, "w") as f: json.dump(config, f, indent=4) return True print(f"[Config] '{target_model}' is already set as enabled.") except Exception as e: print(f"[Error] Failed updating config file: {e}") return False def is_service_running(service_name: str) -> bool: try: res = subprocess.run( ["systemctl", "is-active", service_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) return res.stdout.strip() == "active" except Exception: return False def restart_aaf_service(service_name: str): print(f"[Service] Restarting '{service_name}'...") try: subprocess.run(["systemctl", "restart", service_name], check=True) except subprocess.CalledProcessError as e: print(f"[Error] Failed to restart service: {e}") sys.exit(1) def wait_for_model_ready(target_model: str, server_base_url: str, timeout: int = 300) -> bool: models_url = f"{server_base_url}/v1/models" print(f"[Service] Waiting for '{target_model}' to load on NPU (can take several minutes)...", end="", flush=True) start_time = time.time() while time.time() - start_time < timeout: try: req = urllib.request.Request(models_url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=2.0) as resp: if resp.status == 200: data = json.loads(resp.read().decode('utf-8')) loaded_models = [m.get("id") or m.get("name") for m in data.get("data", [])] if target_model in loaded_models or len(loaded_models) > 0: print("\n[Service] AAF Connector is ready and Vision model endpoint is active!") return True except Exception: pass print(".", end="", flush=True) time.sleep(2.0) print("\n[Error] Timed out waiting for AAF Connector model initialization.") return False def prepare_aaf_backend(): config_updated = ensure_model_enabled(TARGET_MODEL, CONFIG_PATH) service_active = is_service_running(SERVICE_NAME) if config_updated or not service_active: if not service_active: print(f"[Service] '{SERVICE_NAME}' is not active. Starting service...") restart_aaf_service(SERVICE_NAME) if not wait_for_model_ready(TARGET_MODEL, args.aaf_server, timeout=600): print(f"[Error] Unable to reach target model '{TARGET_MODEL}' on AAF Server ({args.aaf_server}).") sys.exit(1) # --- FASTAPI WEB APPLICATION --- app = FastAPI(title="Gateworks AI VLM Chat") if not os.path.isdir(args.video_dir): raise RuntimeError(f"Provided video directory target does not exist: {args.video_dir}") class ChatMessage(BaseModel): role: str content: str class MultiTurnPayload(BaseModel): video_name: str history: List[ChatMessage] def get_timestamp(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] # --- STREAMING VIDEO ROUTE WITH HTTP RANGE SUPPORT --- # Change "async def" to "def" @app.get("/stream/video/{filename}", tags=["Media"]) def stream_video(filename: str, request: Request): video_path = os.path.join(args.video_dir, filename) if not os.path.isfile(video_path): raise HTTPException(status_code=404, detail="Video file not found.") file_size = os.path.getsize(video_path) range_header = request.headers.get("range") if not range_header: def iterfile(): with open(video_path, mode="rb") as file_like: yield from file_like return StreamingResponse( iterfile(), media_type="video/mp4", headers={"Accept-Ranges": "bytes", "Content-Length": str(file_size)} ) try: range_str = range_header.replace("bytes=", "") start_str, end_str = range_str.split("-") start = int(start_str) if start_str else 0 end = int(end_str) if end_str else file_size - 1 except Exception: raise HTTPException(status_code=416, detail="Invalid byte range requested") if start >= file_size or end >= file_size: raise HTTPException(status_code=416, detail="Requested range not satisfiable") chunk_size = (end - start) + 1 def send_bytes(): with open(video_path, "rb") as f: f.seek(start) bytes_left = chunk_size while bytes_left > 0: read_bytes = min(1024 * 512, bytes_left) data = f.read(read_bytes) if not data: break bytes_left -= len(data) yield data headers = { "Content-Range": f"bytes {start}-{end}/{file_size}", "Accept-Ranges": "bytes", "Content-Length": str(chunk_size), "Content-Type": "video/mp4", } return StreamingResponse(send_bytes(), status_code=206, headers=headers) @app.get("/api/videos", tags=["Media"]) async def get_available_videos(): try: if not os.path.exists(args.video_dir): return [] files = os.listdir(args.video_dir) valid_extensions = (".mp4", ".mov", ".mkv", ".avi") return [f for f in files if f.lower().endswith(valid_extensions)] except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/metrics") async def proxy_metrics(): headers = {"Accept": "application/json", "User-Agent": "AAFConnector/1.0"} async with httpx.AsyncClient() as client: try: url = f"{args.aaf_server}/metrics/" response = await client.get(url, params={"model_name": TARGET_MODEL}, headers=headers, timeout=3.0) return response.json() except Exception: return { TARGET_MODEL: { "llm_average_token_per_second": 0.0, "llm_first_infer_duration": 0.0, "generated_token_num": 0 } } @app.post("/api/analyze") async def proxy_analysis_stream(payload: MultiTurnPayload): absolute_video_target_path = os.path.join(args.video_dir, payload.video_name) start_time = time.time() if len(payload.history) > 1: flattened_text = "Here is the conversation history context for this execution sequence:\n" for msg in payload.history[:-1]: label = "User Question" if msg.role == "user" else "Your Previous Response" flattened_text += f"[{label}]: {msg.content}\n" flattened_text += f"\n[New Follow-up Question to Answer]: {payload.history[-1].content}" else: flattened_text = payload.history[0].content aaf_payload = { "model": TARGET_MODEL, "stream": True, "messages": [{ "role": "user", "content": [ {"type": "text", "text": flattened_text}, {"type": "video_url", "video_url": {"url": absolute_video_target_path}} ] }] } print("\n" + "═"*70) print(f"[{get_timestamp()}] [CONVERSATIONAL INFERENCE DISPATCH]") print(f" Model ID : {TARGET_MODEL}") print(f" Target Path : {absolute_video_target_path}") print(f" Turn Count : {len(payload.history)} turns processed in session state.") if args.verbose: print("─"*70) print("[RAW OUTGOING JSON PAYLOAD SENT TO AAF SERVER]:") print(json.dumps(aaf_payload, indent=2)) print("═"*70 + "\n") def raw_socket_generator(): target_endpoint = f"{args.aaf_server}/v1/chat/completions" data_bytes = json.dumps(aaf_payload).encode('utf-8') req = urllib.request.Request( target_endpoint, data=data_bytes, headers={ "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "AAFConnector/1.0" }, method="POST" ) first_token_received = False try: with urllib.request.urlopen(req, timeout=300.0) as response: while True: line_bytes = response.readline() if not line_bytes: break line_str = line_bytes.decode('utf-8', errors='ignore') trimmed = line_str.strip() if trimmed: yield f"{trimmed}\n".encode('utf-8') if trimmed.startswith('data: '): data_content = trimmed[5:].strip() if data_content == "[DONE]": continue try: parsed = json.loads(data_content) token = parsed["choices"][0]["delta"].get("content", "") if token: if not first_token_received: ttft_duration = time.time() - start_time print(f"[{get_timestamp()}] [TTFT / DECODE PHASE]: {ttft_duration:.2f}s.") print(f"[{get_timestamp()}] [STREAMING TEXT TOKENS]: ", end="") first_token_received = True print(token, end="", flush=True) except Exception: pass except urllib.error.HTTPError as http_err: err_body = http_err.read().decode('utf-8', errors='ignore') yield f"data: {{\"error\": \"AAF Server Engine error {http_err.code}: {err_body}\"}}\n\n".encode('utf-8') except Exception as e: yield f"data: {{\"error\": \"Direct socket pipeline fault: {str(e)}\"}}\n\n".encode('utf-8') finally: duration = time.time() - start_time print("\n" + "═"*70) print(f"[{get_timestamp()}] [INFERENCE COMPLETED] Turn Runtime: {duration:.2f}s") print("═"*70 + "\n") return StreamingResponse(raw_socket_generator(), media_type="text/event-stream") # User Interface (HTML Layer with Debug Output) @app.get("/", response_class=HTMLResponse) async def serve_index(): html_content = r""" VLM Edge Studio Analyzer

VLM Edge Platform Interface

Metrics: Waiting for pipeline activity...

Target Profile: __MODEL_NAME_PLACEHOLDER__
No Active Video Stream Sample Loaded
🔍 Video Player Debug Console:
Initializing video pipeline...
URL: None

Conversational History Log

System initialized. Awaiting prompt loop...
""" return HTMLResponse(content=html_content.replace("__MODEL_NAME_PLACEHOLDER__", TARGET_MODEL)) if __name__ == "__main__": prepare_aaf_backend() uvicorn.run(app, host=args.host, port=args.port)