commit d595d07752aa7e4a8fcd6155afa7bb52e21a8e65 Author: Eiichi Shimazu Date: Sat Nov 14 20:52:50 2026 +0900 推論の記録に成功/失敗を付ける判定の画面を作る (依頼 183). pi05/verdict_viewer/ を足す. 判定は KIOXIA の runs/pi05_policy_verdicts_v1.csv (MASTER の保存先テーブルに行を足した) に追記だけで残し, 同じ run_stem はいちばん新しい行が正・labeler=ai は未確定. verdicts.py (列・追記・読み. 書く前に KIOXIA のマウントを確かめ, なければ止める. 追記の前に今のファイルを確かめる (改行・見出し・列の数・閉じていない引用符). シンボリックリンク・元の記録と datasets の中には書かない. 記録の外へ出るリンクは読まない), seed_verdicts.py (--run-input で {run_stem}.json の outcome を labeler=run_input として写す. --ai で印のない回を vlm_judge と同じモデルで判定. 素材は top.mp4 の方策が動いていた区間の後ろ寄り 6 コマ, なければ frames/ の最後を含む 6 枚. 2 枚未満は not_judgeable. API が使えない回は飛ばす. 人の行がある回には足さない (ロックの中で確かめる)), server.py + index.html (python3 一発で localhost:8765. 左に一覧 (既定は人の確定がまだの回・全部・失敗だけ・AI と人が違う回), 右に top.mp4 か frames/ のコマ送り・関節 6 本と爪のグラフ・AI の判定と理由・今の判定・この回の行. y/n/u/x で 1 行追記して次へ, ←→ で前後, pick/place はボタン. 読み込み中・押しっぱなしは書かない. 操作ごとの op_id で応答が届かず押し直しても 2 回書かない. ダイアログは使わない). テスト test_verdicts 31 (依頼 183. codex 7 回) Co-Authored-By: Claude Opus 5.5 --- pi05/verdict_viewer/index.html | 329 ++++++++++++++++++++++++++++++ pi05/verdict_viewer/seed_verdicts.py | 300 +++++++++++++++++++++++++++ pi05/verdict_viewer/server.py | 322 +++++++++++++++++++++++++++++ pi05/verdict_viewer/test_verdicts.py | 378 +++++++++++++++++++++++++++++++++++ pi05/verdict_viewer/verdicts.py | 251 +++++++++++++++++++++++ 5 files changed, 1580 insertions(+) diff --git a/pi05/verdict_viewer/index.html b/pi05/verdict_viewer/index.html new file mode 100644 index 0000000..461bab8 --- /dev/null +++ b/pi05/verdict_viewer/index.html @@ -0,0 +1,329 @@ + + + + +推論の記録の判定 + + + +

推論の記録の判定

+ +

—

判定する (キー)

+ + + + +

pick + + 選ばなければ成功→yes / ほか→unknown

place +

← → 前後の回・ , . コマ送り / 動画は space で再生

AI の判定 (未確定)

今の判定 (いちばん新しい行)

この回の行 (古い順)

+ + + diff --git a/pi05/verdict_viewer/seed_verdicts.py b/pi05/verdict_viewer/seed_verdicts.py new file mode 100644 index 0000000..9deec01 --- /dev/null +++ b/pi05/verdict_viewer/seed_verdicts.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python3 +"""推論の記録の判定ファイルに, 今ある判定の写しと AI の先回り判定を足す (依頼 183). + + python3 pi05/verdict_viewer/seed_verdicts.py --run-input # {run_stem}.json の outcome を写す (labeler=run_input) + python3 pi05/verdict_viewer/seed_verdicts.py --ai # 印のない回を AI が判定 (labeler=ai) + python3 pi05/verdict_viewer/seed_verdicts.py --ai --dry-run # API を呼ばずに素材だけ数える + +- 元の記録は読むだけ. 判定ファイルには追記だけ (verdicts.append_rows) +- すでに同じ種類の行がある回は足さない (何度流しても同じ行が増えない) +- AI の素材: recording/top.mp4 があれば, 推論が動いていた区間 (top_frame_log.csv の mode が wait / home + でないコマ) の終わりまでから 6 コマ (後ろ寄り). なければ frames/ (各チャンクの入力画像) から最後を含む 6 枚. + 2 枚に満たなければ API を呼ばずに not_judgeable + (終わりの wait / home では人がブロックを箱から出していることがあるので, そこは使わない) +- API は今の vlm_judge (dynamic_tracking_limits/place/place_modules/vlm_judge.py) と同じモデル・キーの読み方. + 使えない (キーがない・通信やクレジットのエラー) 回は行を書かずに飛ばし, 最後に数を出す +""" +import argparse +import base64 +import csv +import json +import os +import sys +import time +import urllib.error +import urllib.request +from collections import Counter +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) +sys.path.insert(0, str(HERE.parent.parent / "dynamic_tracking_limits" / "place" / "place_modules")) +import verdicts as V # noqa: E402 +import vlm_judge # noqa: E402 + +N_IMAGES = 6 +MIN_IMAGES = 2 +MP4_FRACTIONS = (0.3, 0.5, 0.65, 0.8, 0.9, 1.0) +MP4_WIDTH = 960 +API_TIMEOUT_SEC = 90.0 +API_MAX_TOKENS = 600 + +PROMPT = """画像は mechArm 270 (ロボットアーム) の作業台を真上から見たもので, 時間の順に並んでいる ({n} 枚・{source}). +課題は「赤いブロックをつかみ, 白い箱の中に置く」(put the red block in the white box). AI の方策が動かしている. +{source_note} +この回について, 次を判定してください. +- pick: 赤いブロックを爪でつかんで持ち上げられたか (yes / no / unknown) +- place: 最後に赤いブロックが白い箱の中にあるか (yes / no / unknown) +- outcome: success (最後に赤いブロックが白い箱の中にある) / failure (箱の中にない・つかめていない・落とした) / + unknown (画像はあるが判別できない) / not_judgeable (画像が終わりまで写っておらず判定できない) +以下の JSON だけで返答してください. 前置きや後置きの説明は一切含めないこと. +{{"pick": "yes|no|unknown", "place": "yes|no|unknown", "outcome": "success|failure|unknown|not_judgeable", "reason": "日本語で 1〜2 文の短い理由"}}""" + +NOTE_MP4 = ("上カメラの録画から, 方策が動いていた区間の後ろ寄りのコマを選んだ. 最後の 1 枚は方策が止まる直前. " + "人の手が写ることがある (補正や置き直し).") +NOTE_FRAMES = ("各画像は方策に渡した入力画像で, チャンク (動きのひとまとまり) を送る前に撮ったもの. " + "最後の 1 枚は最後のチャンクを送る前なので, その後の動きは写っていない. " + "赤いブロックが写っていない画像があってもよい (爪の中や箱の中にあることがある).") + + +def mp4_active_end(run_dir, n_frames): + """top_frame_log.csv で mode が wait / home でない最後のコマ. 読めなければ最後のコマ.""" + p = V.safe_file(run_dir, "recording", "top_frame_log.csv") + last = None + try: + if p is None: + raise FileNotFoundError + with open(p, newline="") as f: + for r in csv.DictReader(f): + if r.get("mode") not in ("wait", "home"): + last = int(r["frame_idx"]) + except Exception: + last = None + if last is None: + return n_frames - 1 + return max(0, min(last, n_frames - 1)) + + +def mp4_images(run_dir): + import cv2 + cap = cv2.VideoCapture(str(Path(run_dir) / "recording" / "top.mp4")) + try: + n = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + if n <= 0: + return [], {} + end = mp4_active_end(run_dir, n) + idxs = sorted(set(int(round(end * fr)) for fr in MP4_FRACTIONS)) + out = [] + for i in idxs: + cap.set(cv2.CAP_PROP_POS_FRAMES, i) + ok, img = cap.read() + if not ok or img is None: + continue + h, w = img.shape[:2] + if w > MP4_WIDTH: + img = cv2.resize(img, (MP4_WIDTH, int(h * MP4_WIDTH / w))) + ok, buf = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 85]) + if ok: + out.append(buf.tobytes()) + return out, {"mp4_frames": n, "active_end": end, "picked": idxs} + finally: + cap.release() + + +def frames_images(run_dir): + names = V.frame_files(run_dir) + if not names: + return [], {} + if len(names) <= N_IMAGES: + pick = names + else: + step = (len(names) - 1) / (N_IMAGES - 1) + pick = [names[int(round(i * step))] for i in range(N_IMAGES)] + out = [(Path(run_dir) / "frames" / nm).read_bytes() for nm in pick] + return out, {"frames": len(names), "picked": pick} + + +def material(run_dir): + """(evidence, 画像の bytes のリスト, メモ). 素材が足りなければ evidence='' で画像は空.""" + if V.has_top_mp4(run_dir): + imgs, info = mp4_images(run_dir) + if len(imgs) >= MIN_IMAGES: + return "top.mp4", imgs, info + imgs, info = frames_images(run_dir) + if len(imgs) >= MIN_IMAGES: + return "frames", imgs, info + return "", [], info + + +def api_key(): + k = os.environ.get("ANTHROPIC_API_KEY") + if not k: + p = os.path.expanduser("~/.anthropic_key") + if os.path.isfile(p): + k = open(p).read().strip() + # 「export ANTHROPIC_API_KEY=...」の形で置かれていることがある (2026-11-14 に確かめた) + if k.startswith("export "): + k = k[len("export "):].strip() + if k.startswith("ANTHROPIC_API_KEY="): + k = k[len("ANTHROPIC_API_KEY="):].strip() + k = k.strip("'\"") + return k or None + + +class ApiUnavailable(RuntimeError): + pass + + +def call_api(key, imgs, evidence): + source = "上カメラの録画 top.mp4 のコマ" if evidence == "top.mp4" else "frames/ の入力画像" + note = NOTE_MP4 if evidence == "top.mp4" else NOTE_FRAMES + content = [{"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", + "data": base64.b64encode(b).decode("ascii")}} for b in imgs] + content.append({"type": "text", "text": PROMPT.format(n=len(imgs), source=source, source_note=note)}) + payload = {"model": vlm_judge.VLM_MODEL, "max_tokens": API_MAX_TOKENS, + "messages": [{"role": "user", "content": content}]} + req = urllib.request.Request(vlm_judge.VLM_API_URL, data=json.dumps(payload).encode("utf-8"), + headers={"x-api-key": key, "anthropic-version": vlm_judge.VLM_ANTHROPIC_VERSION, + "content-type": "application/json"}, method="POST") + last_err = "" + for attempt in range(3): + try: + with urllib.request.urlopen(req, timeout=API_TIMEOUT_SEC) as resp: + body = json.loads(resp.read().decode("utf-8")) + text = "".join(c.get("text", "") for c in body.get("content", []) if c.get("type") == "text") + return vlm_judge._parse_json_from_text(text), text + except urllib.error.HTTPError as e: + last_err = f"http_{e.code}" + if e.code in (429, 500, 502, 503, 504, 529): + time.sleep(5 * (attempt + 1)) + continue + raise ApiUnavailable(last_err) + except Exception as e: + last_err = type(e).__name__ + time.sleep(3) + raise ApiUnavailable(last_err) + + +def ai_row(run_stem, parsed, evidence, n_exec): + def pv(v): + return v if v in V.PICK_PLACE else "unknown" + oc = parsed.get("outcome") if isinstance(parsed, dict) else None + oc = oc if oc in V.OUTCOMES else "unknown" + reason = str(parsed.get("reason", "")).replace("\n", " ").strip() if isinstance(parsed, dict) else "" + return {"labeled_at": V.now_iso(), "run_stem": run_stem, + "pick": pv(parsed.get("pick")), "place": pv(parsed.get("place")), "outcome": oc, "labeler": "ai", + "ai_outcome": oc, "ai_reason": f"[送ったチャンク {n_exec}] {reason}", "evidence": evidence, "note": ""} + + +def do_run_input(csv_path, runs_dir, dry_run): + have = V.latest_by_stem(V.read_rows(csv_path), labelers=("run_input",) + V.HUMAN_LABELERS) + rows = [] + for stem in V.list_run_stems(runs_dir): + if stem in have: + continue + r = V.run_input_row(stem, V.read_run_json(Path(runs_dir) / stem)) + if r: + rows.append(r) + c = Counter(r["outcome"] for r in rows) + print(f"[COPY ] 写す行 {len(rows)} (success {c['success']} / failure {c['failure']} / unknown {c['unknown']})" + + (" ※ dry-run なので書いていません" if dry_run else "")) + if not dry_run: + written = V.append_rows(rows, csv_path, skip_if_labeled_by=("run_input",) + V.HUMAN_LABELERS, + forbid_dirs=(runs_dir,)) + if len(written) != len(rows): + print(f"[COPY ] 書くまでの間に人か写しの行が足されていた {len(rows) - len(written)} 回は足さなかった") + return c + + +def do_ai(csv_path, runs_dir, dry_run, limit, only): + rows = V.read_rows(csv_path) + marked = set(V.latest_by_stem(rows, labelers=("run_input",))) + have_ai = set(V.latest_by_stem(rows, labelers=("ai",))) + human = set(V.latest_by_stem(rows, labelers=V.HUMAN_LABELERS)) # 人が確定した回には AI の行を足さない (codex R2) + targets = [] + for stem in V.list_run_stems(runs_dir): + j = V.read_run_json(Path(runs_dir) / stem) + if j and j.get("outcome") not in (None, ""): + continue # 印がある回は写しの対象 + if stem in marked or stem in have_ai or stem in human: + continue + if only and stem not in only: + continue + targets.append(stem) + if limit: + targets = targets[:limit] + key = None if dry_run else api_key() + counts = Counter() + print(f"[AI ] 対象 {len(targets)} 回 (モデル {vlm_judge.VLM_MODEL})") + for i, stem in enumerate(targets, 1): + rd = Path(runs_dir) / stem + n_exec = V.executed_chunks(V.read_log(rd)) + evidence, imgs, info = material(rd) + if not imgs: + row = {"labeled_at": V.now_iso(), "run_stem": stem, "pick": "unknown", "place": "unknown", + "outcome": "not_judgeable", "labeler": "ai", "ai_outcome": "not_judgeable", + "ai_reason": f"[送ったチャンク {n_exec}] 素材が足りない (top.mp4 なし・frames/ {len(V.frame_files(rd))} 枚). API は呼んでいない", + "evidence": "", "note": ""} + elif dry_run: + print(f" {i:3d} {stem} {evidence} {len(imgs)} 枚送ったチャンク {n_exec}") + counts["dry_run"] += 1 + continue + elif not key: + counts["skipped"] += 1 + print(f" {i:3d} {stem} 飛ばした (API キーがない)") + continue + else: + try: + parsed, text = call_api(key, imgs, evidence) + except ApiUnavailable as e: + counts["skipped"] += 1 + print(f" {i:3d} {stem} 飛ばした (API: {e})") + continue + if not isinstance(parsed, dict): + counts["skipped"] += 1 + print(f" {i:3d} {stem} 飛ばした (答えを読めない: {text[:80]!r})") + continue + row = ai_row(stem, parsed, evidence, n_exec) + if dry_run: + print(f" {i:3d} {stem} not_judgeable (素材なし) 送ったチャンク {n_exec}") + counts["not_judgeable"] += 1 + continue + # 1 回ずつ書く (途中で止まっても済んだ回は残る). API を待つ間に人が確定していたら足さない + if not V.append_rows([row], csv_path, skip_if_labeled_by=("ai", "run_input") + V.HUMAN_LABELERS, + forbid_dirs=(runs_dir,)): + counts["skipped_human"] += 1 + print(f" {i:3d} {stem} 足さなかった (待つ間に人か写しの行が足されていた)") + continue + counts[row["outcome"]] += 1 + print(f" {i:3d} {stem} {row['outcome']:13s} pick={row['pick']} place={row['place']} {evidence or '-'} {row['ai_reason'][:70]}") + print("[AI ] " + " / ".join(f"{k} {counts[k]}" for k in + ("success", "failure", "unknown", "not_judgeable", "skipped", "skipped_human", "dry_run") if counts[k] or k not in ("dry_run", "skipped_human"))) + return counts + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--run-input", action="store_true", help="{run_stem}.json の outcome を写す") + ap.add_argument("--ai", action="store_true", help="印のない回を AI が判定する") + ap.add_argument("--dry-run", action="store_true", help="書かない・API を呼ばない") + ap.add_argument("--limit", type=int, default=0, help="AI の対象を先頭から N 回まで") + ap.add_argument("--stems", nargs="*", default=None, help="AI の対象をこの run_stem だけにする") + ap.add_argument("--runs-dir", default=str(V.RUNS_DIR)) + a = ap.parse_args() + if not (a.run_input or a.ai): + ap.error("--run-input か --ai を付けてください") + try: + csv_path = V.VERDICTS_CSV # 書き先は保存先テーブルの 1 か所だけ (codex R3. テストは関数に渡す) + V.ensure_ssd_mounted(csv_path) + if a.run_input: + do_run_input(csv_path, a.runs_dir, a.dry_run) + if a.ai: + do_ai(csv_path, a.runs_dir, a.dry_run, a.limit, set(a.stems) if a.stems else None) + except (V.NotMounted, V.BrokenCsv) as e: + print(f"[ABORT] {e}") + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pi05/verdict_viewer/server.py b/pi05/verdict_viewer/server.py new file mode 100644 index 0000000..7e67f1e --- /dev/null +++ b/pi05/verdict_viewer/server.py @@ -0,0 +1,322 @@ +#!/usr/bin/env python3 +"""推論の記録に成功/失敗を付ける判定の画面 (依頼 183). + + python3 pi05/verdict_viewer/server.py # http://localhost:8765/ をブラウザで開く + python3 pi05/verdict_viewer/server.py --no-open # ブラウザを開かない + +- 元の記録 (runs/pi05_policy/) は読むだけ. 判定は pi05_policy_verdicts_v1.csv に 1 行ずつ追記するだけ +- 書く前に KIOXIA のマウントを確かめ, なければ書かずに画面へエラーを返す +- localhost だけで待ち受ける +""" +import argparse +import csv +import json +import mimetypes +import re +import sys +import threading +import webbrowser +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import unquote, urlparse + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) +import verdicts as V # noqa: E402 + +STEM_RE = re.compile(r"^pi05_[0-9_]+$") +FRAME_RE = re.compile(r"^[0-9A-Za-z_]+\.jpg$") +MAX_STATE_POINTS = 1500 + + +def state_series(run_dir): + """関節 6 本と爪の時系列. recording/states.csv があればそれ, なければ log.jsonl の state.""" + rd = Path(run_dir) + p = V.safe_file(rd, "recording", "states.csv") + t, joints, grip = [], [], [] + stages = [] + if p is not None: + t0 = None + with open(p, newline="") as f: + for r in csv.DictReader(f): + try: + ts = float(r["ts"]) + except (TypeError, ValueError): + continue + t0 = ts if t0 is None else t0 + ev = r.get("event_type") + if ev == "actual": + try: + joints.append([float(r[f"j{i}"]) for i in range(6)]) + except (TypeError, ValueError): + continue + t.append(round(ts - t0, 3)) + elif ev == "gripper_cmd" and r.get("gripper_state") not in (None, ""): + try: + grip.append([round(ts - t0, 3), float(r["gripper_state"])]) + except ValueError: + pass + elif ev == "stage": + stages.append([round(ts - t0, 3), r.get("stage", "")]) + if len(t) > MAX_STATE_POINTS: + k = len(t) / MAX_STATE_POINTS + idx = sorted(set(int(i * k) for i in range(MAX_STATE_POINTS)) | {len(t) - 1}) + t = [t[i] for i in idx] + joints = [joints[i] for i in idx] + return {"source": "recording/states.csv", "t": t, "joints": joints, "grip": grip, + "grip_label": "爪の命令 (0 開 / 1 閉)", "stages": stages, "t0": t0} + t0 = None + for x in V.read_log(rd): + if "chunk" not in x or "state" not in x: + continue + try: + st = json.loads(x["state"]) if isinstance(x["state"], str) else x["state"] + ts = float(x["t"]) + except Exception: + continue + if not isinstance(st, list) or len(st) < 6: + continue + t0 = ts if t0 is None else t0 + t.append(round(ts - t0, 3)) + joints.append([float(v) for v in st[:6]]) + if len(st) >= 7: + grip.append([round(ts - t0, 3), float(st[6])]) + return {"source": "log.jsonl の state (推論の前に読んだ値)", "t": t, "joints": joints, "grip": grip, + "grip_label": "state の爪 (7 番目の値)", "stages": [], "t0": t0} + + +def video_times(run_dir, t0): + """top.mp4 の各コマの時刻 (states の始まりからの秒). 再生位置のしるしに使う.""" + p = V.safe_file(run_dir, "recording", "top_frame_log.csv") + if t0 is None or p is None: + return None + out = [] + try: + with open(p, newline="") as f: + for r in csv.DictReader(f): + out.append(round(float(r["capture_ts"]) - t0, 3)) + except Exception: + return None + return out + + +class App: + def __init__(self, runs_dir, csv_path): + self.runs_dir = Path(runs_dir) + self.csv_path = Path(csv_path) + self.lock = threading.Lock() + self.done_ops = {} # 画面の 1 回の操作 (op_id) → 書いた行. 応答が届かず押し直しても 2 回書かない (codex R7) + + def runs_summary(self): + rows = V.read_rows(self.csv_path) + latest = V.latest_by_stem(rows) + ai = V.latest_by_stem(rows, labelers=("ai",)) + human = V.latest_by_stem(rows, labelers=V.HUMAN_LABELERS) + out = [] + for stem in V.list_run_stems(self.runs_dir): + rd = self.runs_dir / stem + lt, a, h = latest.get(stem), ai.get(stem), human.get(stem) + out.append({ + "run_stem": stem, + "has_mp4": V.has_top_mp4(rd), + "n_frames": len(V.frame_files(rd)), + "outcome": lt["outcome"] if lt else "", + "labeler": lt["labeler"] if lt else "", + "ai_outcome": a["outcome"] if a else "", + "human_outcome": h["outcome"] if h else "", + "confirmed": bool(lt and lt["labeler"] in V.HUMAN_LABELERS), + }) + return out + + def run_detail(self, stem): + rd = self.runs_dir / stem + rows = [r for r in V.read_rows(self.csv_path) if r["run_stem"] == stem] + ai = [r for r in rows if r["labeler"] == "ai"] + j = V.read_run_json(rd) + log = V.read_log(rd) + st = state_series(rd) + return { + "run_stem": stem, + "has_mp4": V.has_top_mp4(rd), + "frames": V.frame_files(rd), + "states": st, + "video_t": video_times(rd, st.get("t0")) if V.has_top_mp4(rd) else None, + "ai": ai[-1] if ai else None, + "latest": rows[-1] if rows else None, + "history": rows, + "executed_chunks": V.executed_chunks(log), + "end_reason": (j or {}).get("end_reason", ""), + "model": next((x.get("model") for x in log if x.get("type") == "model"), "") + or ((j or {}).get("args") or {}).get("model", ""), + } + + def add_verdict(self, body): + stem = body.get("run_stem", "") + if (not STEM_RE.match(stem) or not (self.runs_dir / stem).is_dir() + or not V.within(self.runs_dir, self.runs_dir / stem)): + raise ValueError(f"run_stem がありません: {stem!r}") + outcome = body.get("outcome") + if outcome not in V.OUTCOMES: + raise ValueError(f"outcome が正しくありません: {outcome!r}") + rd = self.runs_dir / stem + op_id = str(body.get("op_id") or "") + with self.lock: + if op_id and op_id in self.done_ops: + print(f"[SAVE ] {stem} 同じ操作の送り直し. 書いていない (op_id {op_id})") + return self.done_ops[op_id] + ai = V.latest_by_stem(V.read_rows(self.csv_path), labelers=("ai",)).get(stem) + default = "yes" if outcome == "success" else "unknown" + pick = body.get("pick") or default + place = body.get("place") or default + row = {"labeled_at": V.now_iso(), "run_stem": stem, "pick": pick, "place": place, + "outcome": outcome, "labeler": V.human_labeler_for(outcome, ai), + "ai_outcome": ai["outcome"] if ai else "", "ai_reason": ai["ai_reason"] if ai else "", + "evidence": "top.mp4" if V.has_top_mp4(rd) else ("frames" if V.frame_files(rd) else ""), + "note": str(body.get("note", "")).replace("\n", " ").strip()} + written = V.append_rows([row], self.csv_path, forbid_dirs=(self.runs_dir,))[0] + if op_id: + self.done_ops[op_id] = written + print(f"[SAVE ] {written['run_stem']} {written['outcome']} pick={written['pick']} " + f"place={written['place']} labeler={written['labeler']}") + return written + + +def make_handler(app): + class H(BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def _send(self, code, data, ctype="application/json; charset=utf-8", extra=None): + if not isinstance(data, (bytes, bytearray)): + data = json.dumps(data, ensure_ascii=False).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(data))) + self.send_header("Cache-Control", "no-store") + for k, v in (extra or {}).items(): + self.send_header(k, v) + self.end_headers() + if self.command != "HEAD": + self.wfile.write(data) + + def _file(self, stem, *rel): + # リンクでその回の記録の外へ出ない (codex R6・3 回目 R2) + path = V.safe_file(app.runs_dir / stem, *rel) if V.within(app.runs_dir, app.runs_dir / stem) else None + if path is None: + return self._send(404, {"error": "not found"}) + size = path.stat().st_size + ctype = mimetypes.guess_type(str(path))[0] or "application/octet-stream" + rng = self.headers.get("Range") + start, end = 0, size - 1 + if rng: + m = re.match(r"bytes=(\d*)-(\d*)", rng) + if m: + if m.group(1): + start = int(m.group(1)) + if m.group(2): + end = min(int(m.group(2)), size - 1) + elif m.group(2): + start = max(0, size - int(m.group(2))) + if start > end or start >= size: + self.send_response(416) + self.send_header("Content-Range", f"bytes */{size}") + self.end_headers() + return + length = end - start + 1 + self.send_response(206 if rng else 200) + self.send_header("Content-Type", ctype) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Content-Length", str(length)) + if rng: + self.send_header("Content-Range", f"bytes {start}-{end}/{size}") + self.end_headers() + if self.command == "HEAD": + return + with open(path, "rb") as f: + f.seek(start) + left = length + while left > 0: + chunk = f.read(min(1 << 20, left)) + if not chunk: + break + try: + self.wfile.write(chunk) + except (BrokenPipeError, ConnectionResetError): + return + left -= len(chunk) + + def do_HEAD(self): + self.do_GET() + + def do_GET(self): + u = urlparse(self.path) + parts = [unquote(p) for p in u.path.split("/") if p] + try: + if not parts: + return self._send(200, (HERE / "index.html").read_bytes(), "text/html; charset=utf-8") + if parts == ["api", "runs"]: + return self._send(200, {"runs": app.runs_summary(), "csv": str(app.csv_path)}) + if (len(parts) == 3 and parts[:2] == ["api", "run"] and STEM_RE.match(parts[2]) + and (app.runs_dir / parts[2]).is_dir() and V.within(app.runs_dir, app.runs_dir / parts[2])): + return self._send(200, app.run_detail(parts[2])) + if len(parts) == 3 and parts[0] == "media" and STEM_RE.match(parts[1]) and parts[2] == "top.mp4": + return self._file(parts[1], "recording", "top.mp4") + if (len(parts) == 4 and parts[0] == "media" and STEM_RE.match(parts[1]) and parts[2] == "frames" + and FRAME_RE.match(parts[3])): + return self._file(parts[1], "frames", parts[3]) + return self._send(404, {"error": "not found"}) + except Exception as e: # 画面に理由を返す + return self._send(500, {"error": f"{type(e).__name__}: {e}"}) + + def do_POST(self): + if urlparse(self.path).path != "/api/verdict": + return self._send(404, {"error": "not found"}) + try: + n = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(n).decode("utf-8") or "{}") + row = app.add_verdict(body) + return self._send(200, {"ok": True, "row": row}) + except (V.NotMounted, V.BrokenCsv) as e: + print(f"[ABORT] {e}") + return self._send(503, {"error": str(e)}) + except ValueError as e: + return self._send(400, {"error": str(e)}) + except Exception as e: + return self._send(500, {"error": f"{type(e).__name__}: {e}"}) + return H + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--port", type=int, default=8765) + ap.add_argument("--no-open", action="store_true", help="ブラウザを開かない") + ap.add_argument("--runs-dir", default=str(V.RUNS_DIR)) + ap.add_argument("--test-csv", default=None, + help="画面の確かめ用. 判定を保存先テーブルの CSV でなくこのファイルに書く (ふだんは付けない)") + a = ap.parse_args() + if not Path(a.runs_dir).is_dir(): + print(f"[ABORT] 記録のフォルダがありません ({a.runs_dir}). KIOXIA をつないでください") + return 2 + csv_path = Path(a.test_csv) if a.test_csv else V.VERDICTS_CSV + if a.test_csv and V.within(a.runs_dir, csv_path): + print(f"[ABORT] --test-csv が記録のフォルダの中です ({csv_path}). 元の記録には書きません") + return 2 + if a.test_csv: + print(f"[TEST ] 確かめ用: 判定は {csv_path} に書きます (保存先テーブルの CSV には書きません)") + app = App(a.runs_dir, csv_path) + srv = ThreadingHTTPServer(("127.0.0.1", a.port), make_handler(app)) + url = f"http://localhost:{a.port}/" + print(f"[VIEW ] {url} (判定は {csv_path} に追記. Ctrl-C で終わる)") + if not a.no_open: + threading.Timer(0.5, lambda: webbrowser.open(url)).start() + try: + srv.serve_forever() + except KeyboardInterrupt: + pass + finally: + srv.server_close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pi05/verdict_viewer/test_verdicts.py b/pi05/verdict_viewer/test_verdicts.py new file mode 100644 index 0000000..07519a9 --- /dev/null +++ b/pi05/verdict_viewer/test_verdicts.py @@ -0,0 +1,378 @@ +#!/usr/bin/env python3 +"""判定ファイルと判定の画面のテスト (依頼 183). KIOXIA にも API にも触らない (一時フォルダだけ). + + python3 pi05/verdict_viewer/test_verdicts.py +""" +import json +import os +import sys +import tempfile +import threading +import unittest +import urllib.error +import urllib.request +from pathlib import Path +from unittest import mock + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) +import verdicts as V # noqa: E402 +import seed_verdicts as S # noqa: E402 +import server # noqa: E402 + + +def row(stem, outcome, labeler, **kw): + r = {"run_stem": stem, "pick": "unknown", "place": "unknown", "outcome": outcome, "labeler": labeler} + r.update(kw) + return r + + +def make_runs(root): + """偽の記録: a = json に success, b = json の outcome null, c = frames と log.jsonl だけ, d = 何もない.""" + root = Path(root) + for stem in ("pi05_20261001_000001", "pi05_20261001_000002", "pi05_20261001_000003", "pi05_20261001_000004"): + (root / stem).mkdir(parents=True) + a = root / "pi05_20261001_000001" + (a / f"{a.name}.json").write_text(json.dumps({"outcome": "success", "ended_at": "2026-10-01T00:00:10", + "judgments": {"pick": {"verdict": "yes"}, "place": {"verdict": "yes"}}})) + b = root / "pi05_20261001_000002" + (b / f"{b.name}.json").write_text(json.dumps({"outcome": None})) + c = root / "pi05_20261001_000003" + (c / "frames").mkdir() + for i in range(3): + (c / "frames" / f"{i:04d}.jpg").write_bytes(b"\xff\xd8\xff\xd9") + (c / "frames" / "._0000.jpg").write_bytes(b"x") + (c / "log.jsonl").write_text("\n".join(json.dumps({"chunk": str(i), "t": str(100 + i), "executed": "True", + "state": json.dumps([i, 1, 2, 3, 4, -30, 0 if i < 2 else 1])}) for i in range(3)) + "\n") + return root + + +class TestAppend(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.csv = Path(self.tmp.name) / "v.csv" + + def tearDown(self): + self.tmp.cleanup() + + def test_append_only_keeps_existing_bytes(self): + V.append_rows([row("pi05_1", "success", "run_input"), row("pi05_2", "failure", "ai")], self.csv) + before = self.csv.read_bytes() + V.append_rows([row("pi05_1", "failure", "human_only")], self.csv) + after = self.csv.read_bytes() + self.assertTrue(after.startswith(before), "既存の行が変わった") + self.assertEqual(after.count(b"labeled_at,"), 1, "見出しが 2 回書かれた") + self.assertEqual(len(V.read_rows(self.csv)), 3) + + def test_header_columns(self): + V.append_rows([row("pi05_1", "success", "ai")], self.csv) + self.assertEqual(self.csv.read_text().splitlines()[0].split(","), V.COLUMNS) + + def test_latest_row_wins(self): + V.append_rows([row("pi05_1", "success", "ai"), row("pi05_2", "failure", "ai")], self.csv) + V.append_rows([row("pi05_1", "failure", "human_fixed_ai")], self.csv) + V.append_rows([row("pi05_1", "unknown", "human_only")], self.csv) + lat = V.latest_by_stem(V.read_rows(self.csv)) + self.assertEqual(lat["pi05_1"]["outcome"], "unknown") + self.assertEqual(lat["pi05_2"]["outcome"], "failure") + self.assertEqual(V.latest_by_stem(V.read_rows(self.csv), labelers=("ai",))["pi05_1"]["outcome"], "success") + + def test_bad_values_write_nothing(self): + V.append_rows([row("pi05_1", "success", "ai")], self.csv) + before = self.csv.read_bytes() + for bad in (row("pi05_1", "great", "ai"), row("pi05_1", "success", "robot"), + row("", "success", "ai"), row("pi05_1", "success", "ai", pick="maybe")): + with self.assertRaises(ValueError): + V.append_rows([row("pi05_9", "success", "ai"), bad], self.csv) + self.assertEqual(self.csv.read_bytes(), before, "まちがった行のときに途中まで書いた") + + def test_comma_and_newline_in_reason(self): + V.append_rows([row("pi05_1", "failure", "ai", ai_reason='箱の外, "落とした"\n2 行目')], self.csv) + self.assertEqual(V.read_rows(self.csv)[0]["ai_reason"], '箱の外, "落とした"\n2 行目') + + def test_broken_file_stops(self): + V.append_rows([row("pi05_1", "success", "ai")], self.csv) + with open(self.csv, "ab") as f: + f.write(b"2026,pi05_2,yes") # 書きかけ (改行なし) + before = self.csv.read_bytes() + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_3", "success", "ai")], self.csv) + self.assertEqual(self.csv.read_bytes(), before) + other = Path(self.tmp.name) / "other.csv" + other.write_text("a,b\n1,2\n") + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_3", "success", "ai")], other) + self.assertEqual(other.read_text(), "a,b\n1,2\n") + + def test_unclosed_quote_stops(self): + V.append_rows([row("pi05_1", "success", "ai")], self.csv) + with open(self.csv, "ab") as f: + f.write(b'2026,pi05_2,yes,yes,success,ai,,,,"unfinished\n') + before = self.csv.read_bytes() + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_3", "success", "ai")], self.csv) + self.assertEqual(self.csv.read_bytes(), before) + + def test_long_note_is_cut(self): + V.append_rows([row("pi05_1", "success", "human_only", note="あ" * 200000, ai_reason="い" * 200000)], self.csv) + r = V.read_rows(self.csv)[0] + self.assertEqual(len(r["note"]), V.MAX_NOTE + 1) + self.assertEqual(len(r["ai_reason"]), V.MAX_REASON + 1) + V.append_rows([row("pi05_2", "success", "ai")], self.csv) + + def test_forbid_dirs(self): + d = Path(self.tmp.name) / "myruns" + d.mkdir() + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_1", "success", "ai")], d / "v.csv", forbid_dirs=(d,)) + self.assertFalse((d / "v.csv").exists()) + + def test_refuses_symlink_and_record_dirs(self): + real = Path(self.tmp.name) / "real.csv" + V.append_rows([row("pi05_1", "success", "ai")], real) + link = Path(self.tmp.name) / "link.csv" + link.symlink_to(real) + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_2", "success", "ai")], link) + runs = Path(self.tmp.name) / "runs" + (runs / "pi05_1").mkdir(parents=True) + with mock.patch.object(V, "RUNS_DIR", runs): + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_2", "success", "ai")], runs / "pi05_1" / "x.csv") + with self.assertRaises(V.BrokenCsv): + V.append_rows([row("pi05_2", "success", "ai")], Path(self.tmp.name) / "x.json") + self.assertEqual(len(V.read_rows(real)), 1) + + def test_skip_if_labeled_by(self): + V.append_rows([row("pi05_1", "failure", "human_only")], self.csv) + out = V.append_rows([row("pi05_1", "success", "ai"), row("pi05_2", "success", "ai")], self.csv, + skip_if_labeled_by=V.HUMAN_LABELERS) + self.assertEqual([r["run_stem"] for r in out], ["pi05_2"]) + self.assertEqual(V.latest_by_stem(V.read_rows(self.csv))["pi05_1"]["labeler"], "human_only") + + +class TestMount(unittest.TestCase): + def test_not_mounted_stops_and_writes_nothing(self): + target = V.KIOXIA_MOUNT / "r2r2r-data" / "runs" / "pi05_policy_verdicts_v1_TEST_never.csv" + with mock.patch.object(V.os.path, "ismount", lambda p: False), \ + mock.patch("builtins.open", side_effect=AssertionError("open してはいけない")): + with self.assertRaises(V.NotMounted): + V.append_rows([row("pi05_1", "success", "ai")], target) + + def test_mounted_check_only_for_kioxia(self): + with mock.patch.object(V.os.path, "ismount", lambda p: False): + V.ensure_ssd_mounted("/tmp/whatever.csv") # KIOXIA の外は確かめない (テスト用) + + def test_seed_aborts_without_mount(self): + with mock.patch.object(V.os.path, "ismount", lambda p: False), \ + mock.patch.object(sys, "argv", ["seed", "--run-input"]): + self.assertEqual(S.main(), 2) + + def test_server_returns_503_without_mount(self): + with tempfile.TemporaryDirectory() as d: + runs = make_runs(Path(d) / "runs") + app = server.App(runs, V.KIOXIA_MOUNT / "r2r2r-data" / "runs" / "never_TEST.csv") + with mock.patch.object(V.os.path, "ismount", lambda p: False): + with self.assertRaises(V.NotMounted): + app.add_verdict({"run_stem": "pi05_20261001_000003", "outcome": "success"}) + + +class TestLabeler(unittest.TestCase): + def test_labeler(self): + self.assertEqual(V.human_labeler_for("success", None), "human_only") + self.assertEqual(V.human_labeler_for("success", {"outcome": "success"}), "human") + self.assertEqual(V.human_labeler_for("failure", {"outcome": "success"}), "human_fixed_ai") + + +class TestSeed(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.runs = make_runs(Path(self.tmp.name) / "runs") + self.csv = Path(self.tmp.name) / "v.csv" + + def tearDown(self): + self.tmp.cleanup() + + def test_run_input_copy_once(self): + c = S.do_run_input(self.csv, self.runs, dry_run=False) + self.assertEqual(c["success"], 1) + rows = V.read_rows(self.csv) + self.assertEqual([(r["run_stem"], r["outcome"], r["labeler"], r["pick"]) for r in rows], + [("pi05_20261001_000001", "success", "run_input", "yes")]) + S.do_run_input(self.csv, self.runs, dry_run=False) + self.assertEqual(len(V.read_rows(self.csv)), 1, "2 回流したら行が増えた") + + def test_ai_targets_and_not_judgeable(self): + calls = [] + + def fake_api(key, imgs, evidence): + calls.append((len(imgs), evidence)) + return {"pick": "yes", "place": "no", "outcome": "failure", "reason": "箱の外"}, "" + S.do_run_input(self.csv, self.runs, dry_run=False) + with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", fake_api): + c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) + self.assertEqual(calls, [(3, "frames")]) # c だけ API を呼ぶ (frames 3 枚・._ は除く) + self.assertEqual(c["failure"], 1) + self.assertEqual(c["not_judgeable"], 2) # b (null・素材なし) と d (何もない) + ai = V.latest_by_stem(V.read_rows(self.csv), labelers=("ai",)) + self.assertNotIn("pi05_20261001_000001", ai) + self.assertEqual(ai["pi05_20261001_000003"]["ai_reason"], "[送ったチャンク 3] 箱の外") + n = len(V.read_rows(self.csv)) + with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", fake_api): + S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) + self.assertEqual(len(V.read_rows(self.csv)), n, "AI の行が 2 回書かれた") + + def test_ai_skips_when_api_unavailable(self): + def boom(*a): + raise S.ApiUnavailable("http_401") + with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", boom): + c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) + self.assertEqual(c["skipped"], 1) + self.assertNotIn("pi05_20261001_000003", V.latest_by_stem(V.read_rows(self.csv))) + + def test_seed_never_adds_after_human(self): + V.append_rows([row("pi05_20261001_000003", "success", "human_only"), + row("pi05_20261001_000001", "failure", "human_only")], self.csv) + with mock.patch.object(S, "api_key", lambda: "k"), \ + mock.patch.object(S, "call_api", lambda *a: ({"outcome": "failure"}, "")): + S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) + S.do_run_input(self.csv, self.runs, dry_run=False) + lat = V.latest_by_stem(V.read_rows(self.csv)) + self.assertEqual(lat["pi05_20261001_000003"]["labeler"], "human_only") + self.assertEqual(lat["pi05_20261001_000001"]["labeler"], "human_only") + + def test_seed_rechecks_inside_lock(self): + # API を待つ間に人が確定した → AI の行は足さない + def api_then_human(*a): + V.append_rows([row("pi05_20261001_000003", "failure", "human_only")], self.csv) + return {"outcome": "success"}, "" + with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", api_then_human): + c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only={"pi05_20261001_000003"}) + self.assertEqual(c["skipped_human"], 1) + self.assertEqual(V.latest_by_stem(V.read_rows(self.csv))["pi05_20261001_000003"]["labeler"], "human_only") + + def test_symlinked_side_files_are_ignored(self): + outside = Path(self.tmp.name) / "outside.jsonl" + outside.write_text(json.dumps({"chunk": "0", "executed": "True"}) + "\n") + rd = self.runs / "pi05_20261001_000004" + (rd / "log.jsonl").symlink_to(outside) + (rd / f"{rd.name}.json").symlink_to(outside) + self.assertEqual(V.read_log(rd), []) + self.assertIsNone(V.read_run_json(rd)) + + def test_server_refuses_csv_inside_runs_dir(self): + app = server.App(self.runs, self.runs / "pi05_20261001_000003" / "v.csv") + with self.assertRaises(V.BrokenCsv): + app.add_verdict({"run_stem": "pi05_20261001_000003", "outcome": "success"}) + self.assertFalse((self.runs / "pi05_20261001_000003" / "v.csv").exists()) + + def test_symlinked_run_and_frames_are_ignored(self): + outside = Path(self.tmp.name) / "outside" + (outside / "frames").mkdir(parents=True) + (outside / "frames" / "0000.jpg").write_bytes(b"secret") + (self.runs / "pi05_20261001_000099").symlink_to(outside) + (self.runs / "pi05_20261001_000003" / "frames" / "0009.jpg").symlink_to(outside / "frames" / "0000.jpg") + self.assertNotIn("pi05_20261001_000099", V.list_run_stems(self.runs)) + self.assertNotIn("0009.jpg", V.frame_files(self.runs / "pi05_20261001_000003")) + + def test_api_key_export_line(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / ".anthropic_key" + p.write_text("export ANTHROPIC_API_KEY='sk-test'\n") + with mock.patch.dict(os.environ, {"HOME": d}, clear=False), \ + mock.patch.dict(os.environ, {}, clear=False) as env: + env.pop("ANTHROPIC_API_KEY", None) + self.assertEqual(S.api_key(), "sk-test") + + +class TestServer(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.runs = make_runs(Path(self.tmp.name) / "runs") + self.csv = Path(self.tmp.name) / "v.csv" + V.append_rows([row("pi05_20261001_000003", "success", "ai", ai_reason="入っている", evidence="frames")], self.csv) + self.app = server.App(self.runs, self.csv) + self.srv = server.ThreadingHTTPServer(("127.0.0.1", 0), server.make_handler(self.app)) + threading.Thread(target=self.srv.serve_forever, daemon=True).start() + self.base = f"http://127.0.0.1:{self.srv.server_address[1]}" + + def tearDown(self): + self.srv.shutdown() + self.srv.server_close() + self.tmp.cleanup() + + def get(self, path): + with urllib.request.urlopen(self.base + path) as r: + return json.loads(r.read()) + + def post(self, body): + req = urllib.request.Request(self.base + "/api/verdict", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, method="POST") + try: + with urllib.request.urlopen(req) as r: + return r.status, json.loads(r.read()) + except urllib.error.HTTPError as e: + return e.code, json.loads(e.read()) + + def test_list_and_detail(self): + runs = {r["run_stem"]: r for r in self.get("/api/runs")["runs"]} + self.assertEqual(len(runs), 4) + self.assertEqual(runs["pi05_20261001_000003"]["ai_outcome"], "success") + self.assertFalse(runs["pi05_20261001_000003"]["confirmed"]) + d = self.get("/api/run/pi05_20261001_000003") + self.assertEqual(d["frames"], ["0000.jpg", "0001.jpg", "0002.jpg"]) + self.assertEqual(len(d["states"]["joints"]), 3) + self.assertEqual(d["states"]["grip"][-1][1], 1.0) + self.assertEqual(d["executed_chunks"], 3) + + def test_post_appends_and_labeler(self): + before = self.csv.read_bytes() + st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "place": "no", "note": "箱の縁"}) + self.assertEqual(st, 200) + self.assertTrue(self.csv.read_bytes().startswith(before)) + last = V.read_rows(self.csv)[-1] + self.assertEqual((last["labeler"], last["outcome"], last["pick"], last["place"], last["ai_outcome"], last["ai_reason"], + last["evidence"], last["note"]), + ("human_fixed_ai", "failure", "unknown", "no", "success", "入っている", "frames", "箱の縁")) + st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "success"}) + self.assertEqual((j["row"]["labeler"], j["row"]["pick"], j["row"]["place"]), ("human", "yes", "yes")) + st, j = self.post({"run_stem": "pi05_20261001_000004", "outcome": "not_judgeable"}) + self.assertEqual(j["row"]["labeler"], "human_only") + runs = {r["run_stem"]: r for r in self.get("/api/runs")["runs"]} + self.assertTrue(runs["pi05_20261001_000003"]["confirmed"]) + self.assertEqual(runs["pi05_20261001_000003"]["outcome"], "success") + + def test_same_op_id_writes_once(self): + n = len(V.read_rows(self.csv)) + b = {"op_id": "abc", "run_stem": "pi05_20261001_000003", "outcome": "failure"} + self.assertEqual(self.post(b)[0], 200) + self.assertEqual(self.post(b)[0], 200) + self.assertEqual(len(V.read_rows(self.csv)), n + 1) + + def test_media_link_to_other_run_is_refused(self): + other = self.runs / "pi05_20261001_000004" / "frames" + other.mkdir() + (other / "0000.jpg").write_bytes(b"other") + (self.runs / "pi05_20261001_000003" / "frames" / "0008.jpg").symlink_to(other / "0000.jpg") + with self.assertRaises(urllib.error.HTTPError) as cm: + urllib.request.urlopen(self.base + "/media/pi05_20261001_000003/frames/0008.jpg") + self.assertEqual(cm.exception.code, 404) + + def test_post_rejects_bad(self): + n = len(V.read_rows(self.csv)) + self.assertEqual(self.post({"run_stem": "pi05_20261001_000003", "outcome": "great"})[0], 400) + self.assertEqual(self.post({"run_stem": "../etc", "outcome": "success"})[0], 400) + self.assertEqual(self.post({"run_stem": "pi05_20990101_000000", "outcome": "success"})[0], 400) + self.assertEqual(len(V.read_rows(self.csv)), n) + + def test_media_and_traversal(self): + with urllib.request.urlopen(self.base + "/media/pi05_20261001_000003/frames/0001.jpg") as r: + self.assertEqual(r.read(), b"\xff\xd8\xff\xd9") + for p in ("/media/pi05_20261001_000003/frames/..%2Flog.jsonl", "/media/..%2F..%2Fetc/top.mp4"): + with self.assertRaises(urllib.error.HTTPError) as cm: + urllib.request.urlopen(self.base + p) + self.assertEqual(cm.exception.code, 404) + + +if __name__ == "__main__": + unittest.main(verbosity=1) diff --git a/pi05/verdict_viewer/verdicts.py b/pi05/verdict_viewer/verdicts.py new file mode 100644 index 0000000..652fce9 --- /dev/null +++ b/pi05/verdict_viewer/verdicts.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""推論の記録 (runs/pi05_policy/) の判定ファイルを読む・足す共通の部品 (依頼 183). + +判定ファイル: /Volumes/KIOXIA/r2r2r-data/runs/pi05_policy_verdicts_v1.csv (MASTER の保存先テーブル) + - 追記だけ. 既存の行は書き換えない. 同じ run_stem はいちばん新しい行 (ファイルの後ろの行) が正 + - labeler=ai の行は未確定. 確定は人の行 (human / human_fixed_ai / human_only) + - 書く前に KIOXIA のマウントを確かめる. なければ止める (ローカルに逃がさない) +元の記録 (runs/pi05_policy/ の中身) は読むだけ. +""" +import csv +import fcntl +import io +import json +import os +import sys +from datetime import datetime +from pathlib import Path + +KIOXIA_MOUNT = Path("/Volumes/KIOXIA") +DATA_ROOT = KIOXIA_MOUNT / "r2r2r-data" +RUNS_DIR = DATA_ROOT / "runs" / "pi05_policy" +VERDICTS_CSV = DATA_ROOT / "runs" / "pi05_policy_verdicts_v1.csv" + +COLUMNS = ["labeled_at", "run_stem", "pick", "place", "outcome", "labeler", + "ai_outcome", "ai_reason", "evidence", "note"] +PICK_PLACE = ("yes", "no", "unknown") +OUTCOMES = ("success", "failure", "unknown", "not_judgeable") +LABELERS = ("run_input", "ai", "human", "human_fixed_ai", "human_only") +HUMAN_LABELERS = ("human", "human_fixed_ai", "human_only") +EVIDENCES = ("top.mp4", "frames", "both", "") +MAX_NOTE = 500 +MAX_REASON = 2000 + + +class NotMounted(RuntimeError): + pass + + +def ensure_ssd_mounted(csv_path=VERDICTS_CSV): + """KIOXIA の下に書くときはマウントを確かめる. なければ NotMounted (ローカルへは書かない). + テストで KIOXIA の外の一時ファイルに書くときは確かめない.""" + p = Path(csv_path) + if str(p).startswith(str(KIOXIA_MOUNT) + os.sep) or p == KIOXIA_MOUNT: + if not os.path.ismount(str(KIOXIA_MOUNT)): + raise NotMounted(f"KIOXIA がマウントされていません ({KIOXIA_MOUNT}). 書かずに止めます") + if not p.parent.is_dir(): + raise NotMounted(f"書き込み先のフォルダがありません ({p.parent}). 書かずに止めます") + + +def now_iso(): + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def validate_row(row): + r = {c: ("" if row.get(c) is None else str(row.get(c))) for c in COLUMNS} + # 長すぎる値は切る (csv の読み取りの上限を超えると以後読めなくなる. codex 3 回目 R3) + for c, n in (("note", MAX_NOTE), ("ai_reason", MAX_REASON)): + if len(r[c]) > n: + r[c] = r[c][:n] + "…" + if not r["run_stem"]: + raise ValueError("run_stem が空です") + if r["pick"] not in PICK_PLACE or r["place"] not in PICK_PLACE: + raise ValueError(f"pick / place は {PICK_PLACE} のどれか: {r['pick']!r} {r['place']!r}") + if r["outcome"] not in OUTCOMES: + raise ValueError(f"outcome は {OUTCOMES} のどれか: {r['outcome']!r}") + if r["labeler"] not in LABELERS: + raise ValueError(f"labeler は {LABELERS} のどれか: {r['labeler']!r}") + if r["ai_outcome"] not in OUTCOMES + ("",): + raise ValueError(f"ai_outcome は {OUTCOMES} のどれかか空: {r['ai_outcome']!r}") + if r["evidence"] not in EVIDENCES: + raise ValueError(f"evidence は {EVIDENCES} のどれか: {r['evidence']!r}") + if not r["labeled_at"]: + r["labeled_at"] = now_iso() + return r + + +class BrokenCsv(RuntimeError): + pass + + +def _check_existing(data): + """追記の前に今のファイルを確かめる. 見出しがちがう・最後が改行で終わらない・列の数がちがう行があれば止める + (そのまま足すと既存の最後の行に続けて書いてしまう. codex R4).""" + if not data: + return [] + if not data.endswith(b"\n"): + raise BrokenCsv("判定ファイルの最後が改行で終わっていません (書きかけの行がある). 書かずに止めます") + try: # strict: 閉じていない引用符も止める (codex 2 回目 R1) + rows = list(csv.reader(io.StringIO(data.decode("utf-8"), newline=""), strict=True)) + except (csv.Error, UnicodeDecodeError) as e: + raise BrokenCsv(f"判定ファイルを読めません ({e}). 書かずに止めます") + if not rows or rows[0] != COLUMNS: + raise BrokenCsv("判定ファイルの見出しがちがいます. 書かずに止めます") + for i, r in enumerate(rows[1:], 2): + if len(r) != len(COLUMNS): + raise BrokenCsv(f"判定ファイルの {i} 行目の列の数が {len(r)} です ({len(COLUMNS)} のはず). 書かずに止めます") + return [dict(zip(COLUMNS, r)) for r in rows[1:]] + + +def append_rows(rows, csv_path=VERDICTS_CSV, skip_if_labeled_by=None, forbid_dirs=()): + """行を末尾に足すだけ. 既存のバイトは触らない (追記モードで開く). 見出しはファイルが空のときだけ書く. + skip_if_labeled_by: この labeler の行がすでにある run_stem の行は足さない. 確かめはロックの中で行う + (seed が人の確定のあとに AI の行を足して未確定へ戻さないため. codex R2). + forbid_dirs: この中には書かない (使っている記録のフォルダ. codex 2 回目 R3). 書いた行を返す.""" + rows = [validate_row(r) for r in rows] + ensure_ssd_mounted(csv_path) + if not rows: + return [] + p = Path(csv_path) + if (p.suffix != ".csv" or within(RUNS_DIR, p) or within(DATA_ROOT / "datasets", p) + or any(within(d, p) for d in forbid_dirs)): + raise BrokenCsv(f"判定ファイルとして書けない場所です ({p}). 元の記録・datasets の中や .csv でないファイルには書きません") + if p.is_symlink(): + raise BrokenCsv(f"判定ファイルがシンボリックリンクです ({p}). 書かずに止めます") + fd = os.open(str(p), os.O_RDWR | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o644) + with os.fdopen(fd, "r+b") as f: + fcntl.flock(f, fcntl.LOCK_EX) + try: + f.seek(0) + existing = _check_existing(f.read()) + if skip_if_labeled_by: + done = {r["run_stem"] for r in existing if r.get("labeler") in skip_if_labeled_by} + rows = [r for r in rows if r["run_stem"] not in done] + if not rows: + return [] + buf = io.StringIO() + w = csv.DictWriter(buf, fieldnames=COLUMNS, lineterminator="\n") + if not existing and f.seek(0, os.SEEK_END) == 0: + w.writeheader() + for r in rows: + w.writerow(r) + f.write(buf.getvalue().encode("utf-8")) + f.flush() + os.fsync(f.fileno()) + finally: + fcntl.flock(f, fcntl.LOCK_UN) + return rows + + +def within(root, path): + """path がリンクをたどっても root の中にあるか (codex R6).""" + try: + Path(path).resolve().relative_to(Path(root).resolve()) + return True + except (ValueError, OSError): + return False + + +def read_rows(csv_path=VERDICTS_CSV): + p = Path(csv_path) + if not p.exists(): + return [] + with open(p, newline="", encoding="utf-8") as f: + return list(csv.DictReader(f)) + + +def latest_by_stem(rows, labelers=None): + """run_stem ごとにいちばん新しい行 (ファイルの後ろの行). labelers を渡すとその種類の行だけから選ぶ.""" + out = {} + for r in rows: + if labelers is not None and r.get("labeler") not in labelers: + continue + out[r["run_stem"]] = r + return out + + +def human_labeler_for(outcome, ai_row): + """人が付けたときの labeler. AI の行がなければ human_only, 同じなら human, 違えば human_fixed_ai.""" + if not ai_row: + return "human_only" + return "human" if ai_row.get("outcome") == outcome else "human_fixed_ai" + + +# ---- 元の記録を読む (読むだけ) ---- + +def list_run_stems(runs_dir=RUNS_DIR): + return sorted(d.name for d in Path(runs_dir).iterdir() + if d.is_dir() and d.name.startswith("pi05_") and within(runs_dir, d)) + + +def safe_file(run_dir, *rel): + """記録の中のファイル. リンクをたどって記録の外へ出るもの・ないものは None (codex 2 回目 R4).""" + p = Path(run_dir).joinpath(*rel) + return p if p.is_file() and within(run_dir, p) else None + + +def read_run_json(run_dir): + p = safe_file(run_dir, f"{Path(run_dir).name}.json") + if p is None: + return None + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception: + return None + + +def frame_files(run_dir): + d = Path(run_dir) / "frames" + if not d.is_dir(): + return [] + return sorted(f.name for f in d.iterdir() + if f.suffix.lower() == ".jpg" and not f.name.startswith("._") and within(run_dir, f)) + + +def read_log(run_dir): + p = safe_file(run_dir, "log.jsonl") + out = [] + if p is None: + return out + for line in p.read_text(encoding="utf-8", errors="replace").splitlines(): + line = line.strip() + if not line: + continue + try: + out.append(json.loads(line)) + except Exception: + pass + return out + + +def executed_chunks(log): + return sum(1 for x in log if "chunk" in x and str(x.get("executed")) == "True") + + +def has_top_mp4(run_dir): + return safe_file(run_dir, "recording", "top.mp4") is not None + + +def run_input_row(run_stem, run_json): + """{run_stem}.json の outcome を写した行 (labeler=run_input). outcome がない / null なら None.""" + if not run_json or run_json.get("outcome") in (None, ""): + return None + oc = run_json["outcome"] + if oc not in OUTCOMES: + oc = "unknown" + jd = run_json.get("judgments") or {} + + def pv(k): + v = (jd.get(k) or {}).get("verdict") + return v if v in PICK_PLACE else "unknown" + return {"labeled_at": run_json.get("ended_at") or now_iso(), "run_stem": run_stem, + "pick": pv("pick"), "place": pv("place"), "outcome": oc, "labeler": "run_input", + "ai_outcome": "", "ai_reason": "", "evidence": "", + "note": "{run_stem}.json の outcome を写した (記録したときの入力)"} + + +if __name__ == "__main__": + rows = read_rows() + lat = latest_by_stem(rows) + print(f"{VERDICTS_CSV}: {len(rows)} 行 / {len(lat)} 回") + sys.exit(0)