# -*- coding: utf-8 -*- """利用システム 画面（ブラウザUI / end-to-end）。 標準ライブラリのみ。動画ファイルをアップロードすると、上流（RF-DETR可読性・最良フレーム・ 濃度確定）→ 学習時と同じLLM画像評価 → 後段スコアリング まで通して結果を表示する。 起動: python app.py # http://127.0.0.1:8000 を開く python app.py --port 8001 --no-open 必要環境: GPU / ffmpeg(PATH) / BytePlus APIキー(secrets/byteplus.key か 環境変数 ARK_API_KEY)。 初回推論時にモデルを読み込むため、最初の1件だけ時間がかかる。 """ from __future__ import annotations import argparse import json import threading import time import traceback import urllib.parse import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path HERE = Path(__file__).resolve().parent UPLOAD_DIR = HERE / "_uploads" UPLOAD_DIR.mkdir(exist_ok=True) # 重いパイプラインは遅延初期化（サーバ起動を速く保つ） _PIPE = None _PIPE_LOCK = threading.Lock() def get_pipeline(): global _PIPE with _PIPE_LOCK: if _PIPE is None: import pipeline_full _PIPE = pipeline_full.FullPipeline() return _PIPE SLOTS = ["開店前", "17時〜19時"] SLOT_OPTS = "".join(f'' for s in SLOTS) PAGE = """ ラーメンスープ評価AI

🍜 ラーメンスープ評価AI

動画をアップロードすると、上流（RF-DETR可読性・最良フレーム・濃度）→ LLM画像評価 → 採点 まで自動で実行します。

""" class Handler(BaseHTTPRequestHandler): def _send(self, code, body, ctype="text/html; charset=utf-8"): data = body.encode("utf-8") if isinstance(body, str) else body self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _json(self, code, obj): self._send(code, json.dumps(obj, ensure_ascii=False), "application/json; charset=utf-8") def do_GET(self): if self.path in ("/", "/index.html"): self._send(200, PAGE.replace("__SLOT__", SLOT_OPTS)) else: self._send(404, "not found") def do_POST(self): if not self.path.startswith("/api/infer"): self._send(404, "not found"); return try: qs = urllib.parse.urlparse(self.path).query q = urllib.parse.parse_qs(qs) name = (q.get("name", ["upload.mov"])[0]) or "upload.mov" brix = (q.get("brix", [""])[0]).strip() slot = (q.get("slot", [""])[0]).strip() ext = Path(name).suffix or ".mov" n = int(self.headers.get("Content-Length", 0)) if n <= 0: self._json(400, {"error": "空のアップロードです"}); return data = self.rfile.read(n) # 同時アップロードの衝突回避にスレッドID＋時刻でファイル名を作る stamp = f"{threading.get_ident()}_{int(time.time()*1000)}" tmp = UPLOAD_DIR / f"upload_{stamp}{ext}" tmp.write_bytes(data) try: pipe = get_pipeline() result = pipe.run(tmp, brix=brix or None, slot=slot or None) finally: try: tmp.unlink() except OSError: pass self._json(200, result) except Exception as e: traceback.print_exc() self._json(500, {"error": str(e)}) def log_message(self, *a): pass def main(): ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8000) ap.add_argument("--no-open", action="store_true") args = ap.parse_args() url = f"http://127.0.0.1:{args.port}" print("ラーメンスープ評価AI 画面を起動:", url) print("初回推論はモデル読込で時間がかかります。停止は Ctrl+C。") if not args.no_open: try: webbrowser.open(url) except Exception: pass ThreadingHTTPServer(("127.0.0.1", args.port), Handler).serve_forever() if __name__ == "__main__": main()