# フィジカルAI研究室を作る Log#36 | ロボットアームの学習データに、成否を付ける判定の画面を作った - 記録: LOG512（フィジカルAI研究室） - 公開: 2026/11/15 - 記事のページ: https://pixelstomotors.com/record/mecharm-270-demo-data-half-motion-interpolated - サイト: Pixels to Motors（ADAPT DESIGN / 島津 栄一） https://pixelstomotors.com/ - 要約: 小型ロボットアーム mechArm 270 で π0.5 を動かした記録176回に、成否とだれが付けたかを付ける判定の画面を作り、成功率59.8%を基準にした。お手本の動きの半分ほどは、実測でなく直線で補った値だった。 > AI が読むための記事の全文です（コード・実行ログを含む）。引用するときは上の「記事のページ」の住所を使ってください。 --- **小型ロボットアーム mechArm 270 で Physical Intelligence の π0.5 を実機で動かした記録176回に、成功か失敗かと「だれが付けた判定か」を付ける判定の画面を作り、全部に印を付けた。成功率は59.8%（55/92）で、学習し直したあとに比べる基準になった。** AI が先に判定を付け、人が録画と関節のグラフを見て確かめて確定する形です。同じ作業の中で、学習に使っていたお手本（アームを動かした記録）の角度を1コマずつ数え、アームが動いた量の半分ほどが実測ではなく、前後の実測を直線で結んで補った値だったことも分かりました。手で教えたお手本では動いた量の56〜59%が角度を読めない1秒の待ちの中にあり、推論を人が直して作った補正のお手本では、学習に使った本のコマの41〜46%が補った値でした。 この回の作業で、次のことが変わりました。 - 推論の記録176回すべてに、成功・失敗と「だれが付けた判定か」が入った。人が画面で見て確定したのは25回で、画面の直しを挟みながら約40分（22:49〜23:29）。いまの成功率59.8%（55/92）が、学習し直したあとに比べる基準になった - これから作る補正のお手本には、1コマずつ「実測」「補った値」の印が残る。値そのものは1つも変えていない（既存の14データセット・95本を作り直して、すべて一致することを確かめた） - 手で教えたお手本の角度が0.32秒おきにしか読めていない原因が、記録の作りだと分かった。アームの動きを変えずに読み取りを増やす案があり、撮り直せば補った値は減る見込み（実機ではまだ確かめていない） きっかけは、ものづくり太郎チャンネルの動画 [【予言的中？】製造現場のAI活用はどこまで来たか？](https://www.youtube.com/watch?v=GmwphgG7cG4) です。エアリオンの河村さんが、フィジカルAI の学習データは量より質が問われるようになると話し、その例として「ちゃんとラベリングしてデータの質を記録してますか？」と問いかけていました。どんな動きか、いつの記録か、成功したときのデータか失敗したときのデータか。そうした印（ラベル）を付けて仕分けておくことが大事だ、という話です。手元のデータで、この問いに答えられるかを確かめました。 答えは半分だけ「はい」でした。日付・動き・外した理由は、ファイルに残っていました。ところが推論の記録176回のうち108回には成功か失敗かの印がなく、付いている印も人が付けたのか AI が付けたのか分かりませんでした。さらに、学習に入れている角度のかなりの部分が、測った値ではなく、間を直線で埋めた値でした。 | やったこと | なぜ | 結果 | |---|---|---| | 手元のデータのラベルを数えた | 動画の問いに答えられるかを見る | 日付・動き・外した理由はあった。推論の記録は108回に成否がなく、だれが付けたかも分からなかった | | 判定の画面を作り、176回を確定した（依頼183〜195・198） | 大きな仕組みは作らず、成否を付ける画面を1枚だけ作る | AI が先に判定し、人が確かめて確定。176回すべてに印が付き、成功率は59.8% | | 補ったコマに印を付け、既存のお手本を数えた（依頼196） | 補った値がどれだけ入っているかを測る | 学習に使った補正のお手本は、コマの41〜46%が補った値 | | 手教示の読み取りが0.32秒おきの原因を調べた（依頼197） | 手で教えたお手本にも補った値が多かった | 原因は記録の作り。動いた量の56〜59%が、角度を読まない1秒の待ちの中にあった | 試しの記録と、この回に使った材料（依頼文・報告・差分・データ・画面の写し）は KIOXIA の `lab/Log36/` に入れています。 GPU を買うか借りるかの判断は、Runpod で実際に借りて学習させた費用をもとに、別のページにまとめています。この回で測ったお手本も、その借り方で学習させてきたものです。 [ロボットAIの学習用GPU、買うか借りるか](https://pixelstomotors.com/gpu-buy-or-rent) [図解: LOG512_data_quality.html](https://pixelstomotors.com/embeds/LOG512_data_quality.html) ## 作業ログ ― 動画の問いから、推論176回のラベルとお手本の補った値を数えるまで ここから下は作業の記録です。依頼文・報告・コードの差分・データは全量で載せます。効果を書いた所のうち、数字で書いたものは測った値、「見込み」と書いたものはまだ確かめていない値です。 この回で何度も出てくる言葉を先に言い換えておきます。 - π0.5：[Physical Intelligence](https://www.pi.website/blog/pi05) が公開しているロボット用の AI。上カメラの画像と関節の角度を見て、次の動きを出す - お手本：π0.5 に学ばせるための、アームを動かした記録。画像と関節の角度が0.1秒ごとに並ぶ。手で教えた要所を再生して撮ったもの（手教示）と、推論を人がゲームパッドで直して成功させたもの（補正）の2種類がある - ラベル：記録1本ごとに付ける印。この回では、成功したか・失敗したか、つかめたか（pick）・箱に置けたか（place）、だれが付けたか - 推論の記録：π0.5 を実機で動かした1回分の記録。KIOXIA の `runs/pi05_policy/{名前}/` に入る - 補った値：アームの角度が読めなかった間を、前後の実測の角度を直線で結んで埋めた値。0.1秒ごとのコマに並べ直すときに作る - 依頼箱：Claude Code への依頼をファイルで置いておく Mac のフォルダ。置くと見張り係が順に Claude Code に渡し、結果を書き戻す - [Claude Code](https://www.anthropic.com/claude-code)：Anthropic の、コードを書いて動かす AI。この回のコードはすべて依頼箱から Claude Code が書いた ## 動画で挙がったラベルの項目と、手元のデータを突き合わせた 動画の画面の録画（10分55秒）から字幕を切り出して読み、ラベルについて話している所を拾いました。挙がっていたのは、どんな動作をしたか、日付、タスクが成功したときのデータか失敗したときのデータか、そして、AI が自動でラベルを付けて点数を付け、人の判断と AI でブレを確かめて記録すること、の4つです。 そのうえで、KIOXIA の `r2r2r-data/` を Mac 連携の shell で数えました。 | 動画の項目 | 手元にあるか | どこにあるか | |---|---|---| | 日付 | ある | すべての記録の名前に日時が入っている | | どんな動きか | ある | タスクの文、1コマごとの「モデル／人の補正」の区別（`step_modes`）、爪の指令 | | 成功／失敗 | ある（抜けが多い） | 台帳の pick/place の判定（813行）、お手本の `human_verdict`、推論の記録の `outcome` | | 点数 | 一部ある | 把持の深さ（閉じて1秒後のはみ出しの画素数）で浅い本を外している。1コマで一番大きく動いた角度（`max_step_deg`） | | 外した理由 | ある | データセットの `index.json` に、外した本と理由 | | 人＋AI の確認 | ある（区別できない） | AI（VLM）の判定の列と、判定の列 | 弱いところを数で並べると、こうなりました。 | 弱いところ | 数 | |---|---| | 推論の記録のうち成否の印がない回 | 176回のうち108回（結果のファイルがない97回・`outcome` が null の11回） | | 台帳の自信度（`pick_confidence`）が固定に近い | 813行のうち397行が 0.82 | | 台帳の pick の判定と AI の判定の一致 | 464件のうち413件。人が確かめたかどうかを区別する列がない | | 台帳の空欄と unknown | pick は空欄152・unknown 27、place は空欄268・unknown 57。試しの行（`run_0042_test`）と dry-run の行も残っている | | ずれが分かっているラベル | `scripted_v1` の爪を開く印が約2秒（約20コマ）早い。注記だけで未補正 | | 補った値に1コマごとの印がない | `episode.npz` は `t`・`t_orig`・`state`・`actions` だけ | | 同じ元の記録の使い回し | 19データセット・375本の元は82回。1回の記録が最大14のデータセットに入っている | | 種類 | 作業は1つ（赤いブロックを白い箱へ）、物も場所も1つ。画像は 384×216 | ## 大きなプラットフォームは作らず、判定の画面を1枚だけ作る 動画の中では、エアリオンが作っているフィジカルAI の開発基盤（動画では Physical AI Studio、画面には Physical AI Canvas と出ていた）が紹介されていました。センサー・ロボット・記録の部品を線でつないでデータ集めを組み、記録の一覧で自動のラベルと点数を見られるものです。エアリオンは矢崎総業にこの基盤を提供すると[報じられています](https://itbusinesstoday.com/ja/tech/ai/airion-yazaki-team-up-on-autonomous-physical-ai/)。 同じものが要るかを考え、要らないと判断しました。理由は3つです。 - その基盤は、Python やターミナルを使わない現場の人が組むためのもの。こちらは Claude Code で組めるので、そこは足りている - 困っているのは仕組みがないことではなく、印が付いていない回が多いこと。これは画面1枚で片付く - MASTER の設備と作業の決まりでは、実測された詰まりを解く物だけを足す。詰まりは「108回に成否がない」とはっきりしている ただ、画面の作りから1つ取り入れたい考えがありました。記録を独立した部品にして、映像と状態を同じ時計でそろえていたことです。こちらの弱いところ（爪の印が2秒早い・角度が1.6秒抜ける）は、どちらも時刻のそろえ方から来ています。 作る画面は、1回ずつ録画と関節のグラフと AI の判定を並べ、人がキーで成功・失敗を付けるものにしました。AI が先に判定を付け、人が確かめて直す形です。動画で言っていた「人の判断と AI でブレを確かめる」に当たります。判定の置き場所は新しいファイルなので、MASTER の保存先テーブルに次の行を足してから実装しました。 ```text | フィジカルAI研究室の推論の記録の成功/失敗の判定（AI の先回り判定と人の確定。追記のみ・元の記録は読むだけ） | `runs/pi05_policy_verdicts_v1.csv` | ``` - ねらい：成否の印がない108回をなくし、だれが付けた印かをデータから分かるようにする - 変わったこと：判定は新しいファイルに追記だけで残し、1行ごとに `labeler`（記録したときの入力・AI・人・人が AI を直した・決まり）を持つ。元の記録は書き換えない - 見込める効果：成功率を、学習し直した前と後で同じ数え方で比べられる。測った結果は下の「176回の判定を終えた結果」 ## 依頼183：推論の記録に成否を付ける判定の画面を作る 依頼箱は11月14日20:19に始め、20:53に終えました（34分）。AI の先回り判定は、`vlm_judge` と同じ Claude API（claude-sonnet-4-6）で付けています。録画の最後には人がブロックを箱から出している所が写ることがあるので、最後の待ちと HOME へ戻る間のコマは使わないようにしていました。 ### 依頼183 の依頼文 ```markdown --- 作業フォルダ: /Users/shimazueiichi/r2r2r-scripts 上限時間: 120 --- # 依頼183：推論の記録に成功/失敗を付ける判定の画面を作る 実機は動かさない（記録を読むだけ）。フィジカルAI研究室のデータ集めや補正の道具として作る。 ## なぜ runs/pi05_policy/ の推論の記録 176 回のうち、成功/失敗の印があるのは 68 回（成功 41・失敗 27）だけ。残り 108 回には印がない（2026-11-14 に数えた）。 - 94 回は {run_stem}.json も recording/ もなく、frames/ と log.jsonl だけ（10 月の記録） - 3 回は recording/ はあるが {run_stem}.json がない - 11 回は {run_stem}.json の outcome が null もうひとつ、今ある判定は誰が付けたか（人か AI か）がデータから分からない。 エイイチの決め（2026-11-14）：大きなプラットフォームは作らず、判定のための画面を 1 枚だけ作る。AI が先に判定を付け、人が確かめて直す。 ## 守ること - 元の記録（runs/pi05_policy/ の中身・台帳・datasets/）は読むだけ。上書き・移動・削除しない - 判定は新しいファイル 1 本に追記だけで残す（既存の行は書き換えない。同じ run_stem に後から行を足したら、いちばん新しい行が正） - 書き込むときは ensure_ssd_mounted() 相当のマウント確認を入れる。KIOXIA がなければ止めて報告する（ローカルに逃がさない） - 画面で alert / confirm などのダイアログは使わない ## 判定の置き場所（保存先テーブルに行を足してから実装する） /Volumes/KIOXIA/r2r2r-data/runs/pi05_policy_verdicts_v1.csv 列： | 列 | 中身 | |---|---| | labeled_at | 判定した時刻（ISO8601） | | run_stem | 記録の名前 | | pick | yes / no / unknown | | place | yes / no / unknown | | outcome | success / failure / unknown / not_judgeable（映像が足りず判定できない） | | labeler | run_input（記録したときの入力を写したもの） / ai / human（AI と同じ判定を人が確かめた） / human_fixed_ai（人が AI と違う判定にした） / human_only（AI の判定なしで人が付けた） | | ai_outcome | AI の判定（参考。人の行にも写しておく） | | ai_reason | AI の判定の理由 | | evidence | 判定に使った素材（top.mp4 / frames / both） | | note | 人のひとこと（空でよい） | 決まり：AI だけの行（labeler=ai）は「まだ確定していない」扱い。確定は人の行。handover の台帳の outcome_source（VLM 単独で確定しない）と同じ考え方。 ## やること 1. 今ある判定の写し：{run_stem}.json に outcome があり null でない 68 回を、labeler=run_input の行として書く（json は書き換えない） 2. AI の先回り判定：印のない 108 回に、AI（今の vlm_judge と同じ Claude API）で成功/失敗を付けて labeler=ai の行を書く。素材は recording/top.mp4 があれば最後のほうのコマ、なければ frames/ の最後の数枚。素材が足りなければ not_judgeable にする。API が使えないときは飛ばして、何回分を飛ばしたか報告に書く 3. 判定の画面（pi05/verdict_viewer/ に置く。Mac で python3 一発で立ち上がり、ブラウザで localhost を開く形） - 左に記録の一覧。既定は「人の確定がまだの回」だけ。絞り込みで全部・失敗だけ・AI と人が違う回も見られる - 右に 1 回分：top.mp4 の再生（なければ frames/ のコマ送り）、関節 6 本と爪のグラフ（recording/states.csv、なければ log.jsonl の state）、AI の判定と理由、今の確定の判定 - キーで判定：y 成功 / n 失敗 / u 分からない / x 判定できない、←→ で前後の回。押したらその場で 1 行追記して次の回へ進む - pick と place を分けて付けたいときは、画面のボタンで別々に選べるようにする（キーは outcome だけでよい） 4. 確かめ：テスト（追記だけで既存行を変えないこと、マウントがないと止まること、最新行が正になること）。画面を実際に立ち上げ、1 回分の判定が CSV に入るところまで確かめ、画面の写し（png）を報告に添える 5. codex にレビューを頼み、変えたファイルだけを指定して commit（git add . と git clean は使わない） ## 報告に書くこと - 写した判定の数（success / failure / unknown）と、AI が付けた数（success / failure / unknown / not_judgeable / 飛ばした） - 画面の起動コマンド（エイイチが打つ 1 行） - 画面の写し - 決めてほしいことがあれば - 最後の 1 行に「判定：合格／止めた（決めてほしいことあり）／失敗」 ``` ### 依頼183 で Claude Code が返した報告 ```text # 結果：20261114_183_推論の記録に成功と失敗を付ける判定の画面を作る - 状態：完了 - 始めた時刻：2026-11-14 20:19 - 終わった時刻：2026-11-14 20:53（34分） - 作業フォルダ：/Users/shimazueiichi/r2r2r-scripts - この続きを Claude Code で話すとき：作業フォルダで `claude --resume 6f23d901-7e95-4c8d-ba56-63a99a700e3e` ## Claude Code の報告 やったこと - 依頼に従い、実装の前に MASTER の保存先テーブルへ判定ファイル `runs/pi05_policy_verdicts_v1.csv` の行を足しました。更新履歴にも 1 行足しています。直す前の写しは `MASTER_old_20261114.md` です。 - `pi05/verdict_viewer/` を新しく作りました。 - 判定ファイルへは追記だけで書き、同じ回はいちばん新しい行が正です。 - 書く前に KIOXIA のマウントを確かめ、なければ止めます（ローカルには書きません）。 - 元の記録・台帳・datasets は読むだけで、その中には書きません。 - 今ある判定の写し（labeler=run_input）と、AI の先回り判定（labeler=ai）を判定ファイルに書きました。 - AI は vlm_judge と同じ Claude API（claude-sonnet-4-6）を使いました。 - 素材は、top.mp4 があれば方策が動いていた区間の後ろ寄りの 6 コマ、なければ frames/ の最後を含む 6 枚です。 - 録画の最後は、人がブロックを箱から出しているところが写っていることがあります。そのため、最後の待ちと HOME へ戻る間のコマは使っていません。 - 画面では次のことができます。 - 左の一覧の絞り込み（既定は人の確定がまだの回。ほかに全部・失敗だけ・AI と人が違う回） - top.mp4 の再生（なければ frames/ のコマ送り） - 関節 6 本と爪のグラフ - AI の判定と理由、今の判定、その回の行の一覧を見る - y/n/u/x で 1 行追記して次の回へ進む、←→ で前後の回へ移る - pick と place はボタンで別々に選ぶ - ダイアログは使っていません。 - codex のレビューを 7 回受けました。必須の指摘は 1 回目から順に 5・3・1・1・1・0・0 件で、すべて直しました。7 回目は推奨 1 件だけで、それも直しました。 - 主な直しは、読み込み中に違う回へ書かない、キーの押しっぱなしで続けて書かない、応答が届かずに押し直しても 2 回書かない、書きかけの行がある CSV には足さない、人が確定した回に AI の行を足さない、の 5 つです。 - commit は d595d07 です（5 ファイルを個別に add）。 結果 - 写した判定：68 回（success 41 / failure 27 / unknown 0） - AI が付けた判定：108 回（success 14 / failure 17 / unknown 0 / not_judgeable 77 / 飛ばした 0） - 1 回目は、AI の答えを読めなかった 1 回を飛ばしました。流し直して埋めたので、最後に飛ばした回は 0 です。 - not_judgeable 77 回の内訳は、素材がなく API を呼ばなかった回が 35 回、AI が判定できないとした回が 42 回です。 - 判定ファイルは 176 行・176 回です。 - テストは 31 件すべて通りました。追記で既存の行が変わらないこと、マウントがないと止まること、最新の行が正になることを確かめています。 - 画面は実際に立ち上げ、ブラウザで y を押して 1 回分の判定が CSV に入るところまで確かめました。 - 書いた先は /tmp の写しの CSV です。人が付けていない「人の行」を本物のファイルに残さないためです。 - 入った行は pi05_20261114_182327・success・labeler=human・pick yes / place yes です。 - 画面の起動コマンド： `python3 ~/r2r2r-scripts/pi05/verdict_viewer/server.py` ブラウザで http://localhost:8765/ が開きます。 - 画面の写し： - `claude_code_reports/20261114_183_screen_before.png`（判定する前） - `claude_code_reports/20261114_183_screen_after_y.png`（y で追記したあと） 範囲の外で気づいたこと（変えていません）：`~/.anthropic_key` は「export ANTHROPIC_API_KEY=...」の形で書かれています。そのため、環境変数がないときは今の vlm_judge.judge() は鍵を正しく読めず、401 で失敗します。今回のスクリプトはこの形も読めるようにしました。 変えたファイル - `pi05/verdict_viewer/verdicts.py`・`seed_verdicts.py`・`server.py`・`index.html`・`test_verdicts.py`（新しく作成・commit 済み） - MASTER.md（保存先テーブルに 1 行と更新履歴に 1 行。直す前の写しは MASTER_old_20261114.md） - KIOXIA の `runs/pi05_policy_verdicts_v1.csv`（新しく作成・176 行） - 報告：`claude_code_reports/20261114_183.txt` と画面の写し 2 枚 - 作業メモ：`logs/20261114_2052.md` - AI のログ：`logs/verdict_seed_ai_20261114.txt` - 報告・作業メモ・ログと画面の写しは commit していません。 決めてほしいこと 1. 記録したときの入力（run_input）の 68 回を人の確定に数えるか。誰が付けたか分からないので、今は数えていません。そのため既定の一覧には 176 回すべてが出ます。 2. アームを動かしていない回（送ったチャンク 0）が 51 回あります。AI は 44 回を not_judgeable、7 回を failure としました。成功率に入れるか外すか。 3. frames/ だけの回は、最後の画像が最後のチャンクを送る前に撮られています。そのため AI は 42 回を not_judgeable としました。これを人が画面で確かめる形でよいか。 4. 本物の判定ファイルには、まだ人の行がありません。人の判定はエイイチが画面で付けてください。 判定：止めた（決めてほしいことあり） ``` ### 依頼183 のコードの差分（コミット d595d07） ```diff 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) ``` ![最初の判定の画面。左に推論の記録の一覧、真ん中に録画と関節のグラフ、右に判定のボタンと AI の判定が並ぶ](https://pixelstomotors.com/media/LOG512_01_verdict_viewer_first.png) AI が「判定できない」とした77回のうち35回は、素材がなくて API を呼ばなかった回、42回は画像だけの古い回で最後の画像が最後の動きより前に撮られていた回でした。報告の「決めてほしいこと」には、次の3つを決めました。 - 記録したときの入力（推論を終えたときに人が y/n を打った判定）の68回は、人の確定に数える - アームを動かしていない回（送ったチャンク0）は試行になっていないので、成功率から外す - 画像だけの古い回で AI が判定できないとした回は、人が見ずに「判定できない」で確定する ## 依頼184：判定の画面を立ち上げてブラウザで開く 画面の立ち上げも依頼箱で行いました。次からはダブルクリックで立ち上がる `.command` も作っています。 ### 依頼184 の依頼文 ```markdown --- 作業フォルダ: /Users/shimazueiichi/r2r2r-scripts 上限時間: 10 --- # 依頼184：判定の画面（依頼183）を立ち上げてブラウザで開く コードは変えない。立ち上げるだけ。 ## やること 1. すでに 8765 番で動いていれば立ち上げ直さない（lsof -i :8765 で確かめる） 2. 動いていなければ、この依頼が終わっても止まらないように裏で立ち上げる nohup setsid python3 ~/r2r2r-scripts/pi05/verdict_viewer/server.py > ~/r2r2r-scripts/logs/verdict_viewer_server.log 2>&1 & （setsid がなければ nohup と disown で） 3. curl で http://localhost:8765/ が 200 を返すことを確かめる 4. open http://localhost:8765/ で既定のブラウザに開く 5. 次から人がダブルクリックで立ち上げられるように、~/r2r2r-scripts/pi05/verdict_viewer/判定の画面を開く.command を作る（中身は 1 と 2 と 4 と同じ。chmod +x）。commit は不要 6. 報告の最後の 1 行に「判定：合格／止めた（決めてほしいことあり）／失敗」 ``` ### 依頼184 で Claude Code が返した報告 ```text # 結果：20261114_184_判定の画面を立ち上げて開く - 状態：完了 - 始めた時刻：2026-11-14 22:40 - 終わった時刻：2026-11-14 22:41（1分） - 作業フォルダ：/Users/shimazueiichi/r2r2r-scripts - この続きを Claude Code で話すとき：作業フォルダで `claude --resume 95cd7519-d774-42d2-a40f-cdce0f562ed1` ## Claude Code の報告 **やったこと** - `lsof -i :8765` で調べたところ、8765 番では何も動いていなかったので、判定の画面のサーバーを立ち上げました。Mac には `setsid` がないため、`nohup` と `disown` を使っています。この依頼が終わっても止まらずに動き続けます。ログは `~/r2r2r-scripts/logs/verdict_viewer_server.log` に出ます。 - 依頼の手順と 1 点だけ変えました。`server.py` は何も付けないと立ち上げたときに自分でもブラウザを開きます。そのため `--no-open` を付けて、手順 4 の `open` だけで開くようにしました。こうしないとタブが 2 枚開きます。コードは変えていません。 - `open http://localhost:8765/` で既定のブラウザに開きました。 - `pi05/verdict_viewer/判定の画面を開く.command` を作り、`chmod +x` しました。commit はしていません。 - 中身は手順 1・2・4 と同じで、`setsid` がない Mac では `nohup` と `disown` で立ち上げます。 - 立ち上げたあと、最大 10 秒、200 が返るのを待ちます。返らなければログの最後の 20 行を出して止まります。 - 作業の記録を `~/r2r2r-scripts/logs/` に残しました。 **結果** - `curl http://localhost:8765/` は 200 を返しました（プロセス番号 14361 で待ち受けています）。 - `open` は問題なく終わりました。 - `.command` は書き方の確認が通りました。すでに動いているときの流れも実行し、「すでに 8765 番で動いています」と出てブラウザを開き、問題なく終わりました。 - 動いていないときに `.command` から立ち上げる流れは、サーバーがすでに動いていたため試していません。 **変えたファイル** - `pi05/verdict_viewer/判定の画面を開く.command`（新しく作成） - `logs/verdict_viewer_server.log`（サーバーのログ） - `logs/` の作業記録 1 件 **決めてほしいこと** なし 判定：合格 ``` ## 依頼185：決めた3つを判定ファイルと画面に入れる ### 依頼185 の依頼文 ```markdown --- 作業フォルダ: /Users/shimazueiichi/r2r2r-scripts 上限時間: 60 --- # 依頼185：判定の決め 3 つを判定ファイルと画面に入れる 依頼183 の「決めてほしいこと」1〜3 へのエイイチの決め（2026-11-14）。実機は動かさない。元の記録は読むだけ、判定ファイルは追記だけ（依頼183 と同じ決まり）。画面（依頼184 で立ち上げたもの）が動いていたら、直したあとに立ち上げ直す。 ## 決め 1. 記録したときの入力（labeler=run_input）の 68 回は、人の確定に数える。推論を終えたときにエイイチが y/n を打った判定のため。行は書き換えず、確定かどうかを決めるコードの側で run_input を人の確定として扱う 2. アームを動かしていない回（送ったチャンク 0。依頼183 の報告では 51 回）は、試行になっていないので成功率から外す。outcome に not_applicable を足し（受け渡しの台帳と同じ語）、labeler=rule の行を追記する 3. 画像だけ（frames/ のみ）で AI が not_judgeable とした回と、素材がなく API を呼ばなかった回は、人が見ずに not_judgeable で確定する。labeler=rule の行を追記する - 2 と 3 の両方に当たる回は 2（not_applicable）にする - 人の行（human / human_fixed_ai / human_only）がすでにある回には rule の行を足さない - rule の行の note に「エイイチの決め 2026-11-14 依頼185」と、どの決めに当たったか（2 か 3）を書く ## やること 1. verdicts.py の語に not_applicable と labeler=rule を足す。確定の扱いは run_input・human・human_fixed_ai・human_only・rule 2. 2 と 3 の行を判定ファイルに追記する（送ったチャンクの数え方は log.jsonl から。数え方を報告に書く） 3. 画面：既定の一覧は「まだ確定していない回」。成功率（success / (success + failure)）を画面の上に出し、not_applicable・not_judgeable・unknown は分母に入れない 4. テスト（rule の行が人の行のある回に入らないこと・2 と 3 の重なりが 2 になること・成功率の分母）。変えたファイルだけを指定して commit（git add . と git clean は使わない）。codex のレビューは画面の数え方と追記の部分だけ 1 回でよい ## 報告に書くこと - 2 に当たった回・3 に当たった回・重なった回の数 - エイイチが画面で見る回の数（まだ確定していない回） - 今の成功率（確定した回だけ） - 最後の 1 行に「判定：合格／止めた（決めてほしいことあり）／失敗」 ``` ### 依頼185 で Claude Code が返した報告 ```text # 結果：20261114_185_判定の決め3つを判定ファイルと画面に入れる - 状態：完了 - 始めた時刻：2026-11-14 22:41 - 終わった時刻：2026-11-14 22:45（4分） - 作業フォルダ：/Users/shimazueiichi/r2r2r-scripts - この続きを Claude Code で話すとき：作業フォルダで `claude --resume dd730ac4-d733-4a65-8ff1-cd0b7d3b4ad8` ## Claude Code の報告 **やったこと** - `verdicts.py` に outcome の `not_applicable` と、付けた人の種類 `labeler=rule` を足しました。確定の扱いは run_input・human・human_fixed_ai・human_only・rule の 5 つです。既存の行は書き換えず、確定かどうかを判断するコードの側で run_input を人の確定として数えています。確定したあとに AI の行が足されても、未確定には戻りません。 - `seed_verdicts.py` に `--rule` を足し、判定ファイルに rule の行を追記しました。2 と 3 の両方に当たる回は 2 にしています。確定の行(run_input・人・rule)がすでにある回には足しません。この確かめは書き込みのロックの中でも行います。note には「エイイチの決め 2026-11-14 依頼185」と、当たった決め(決め2 か 決め3)を書きました。 - 送ったチャンクの数え方：log.jsonl の chunk がある行のうち、executed が True の行の数です。log.jsonl がない回は 0 と数えます(今は該当なし)。 - AI が not_applicable を付けないようにし、確定した回には AI の行も足さないようにしました。 - 画面の既定の一覧を「まだ確定していない回」にしました。上には成功率(success / (success + failure))と、分母に入れない回の数(試行でない・判定できない・分からない)を出します。 - テストを足し、39 本すべて合格しました。足したのは、人の行がある回と run_input の回に rule の行が入らないこと、2 と 3 の重なりが 2 になること、成功率の分母、2 回流しても増えないことなどです。 - codex のレビューは画面の数え方と追記の部分に 1 回かけ、指摘はありませんでした。 - 画面(8765 番)は動いていたので、止めて立ち上げ直しました。新しい一覧と成功率が出ることを確かめています。 - commit は 31113c0 です。 **結果** - 2 に当たった回(not_applicable)：51 回 - 3 に当たった回(not_judgeable)：33 回 - 両方に当たった回：44 回(すべて 2 として 51 回に入っています) - 追記した rule の行：84 行。既存のバイトは変わっていません。もう一度流しても足す行は 0 でした。 - エイイチが画面で見る回(まだ確定していない回)：24 回(AI が成功とした 14 回、失敗とした 10 回) - 今の成功率(確定した回だけ)：41 / 68 = 60.3%(確定 152 回のうち、not_applicable 51 回と not_judgeable 33 回は分母に入れていません) **変えたファイル** - `pi05/verdict_viewer/verdicts.py` - `pi05/verdict_viewer/seed_verdicts.py` - `pi05/verdict_viewer/server.py` - `pi05/verdict_viewer/index.html` - `pi05/verdict_viewer/test_verdicts.py` - 判定ファイル `/Volumes/KIOXIA/r2r2r-data/runs/pi05_policy_verdicts_v1.csv`(追記だけ) - 作業の記録 `logs/20261114_2245.md`(commit していません) **決めてほしいこと** - pi05_20261031_185246 は送ったチャンクが 0 なのに、記録したときの入力が success になっています。決め 1 で run_input は人の確定に数えるため、rule の行(not_applicable)は足していません。今は成功率の分子と分母に入っています。そのまま success とするか、not_applicable にするかを決めてください。not_applicable にすると、成功率は 40 / 67 = 59.7% になります。 - 依頼183 の報告にあった 51 回と、今回送ったチャンクを数えた 52 回の差は、この 1 回です。 判定：止めた（決めてほしいことあり） ``` ### 依頼185 のコードの差分（コミット 31113c0） ```diff commit 31113c0d27bfb195a1577d5becde7f61f89520f0 Author: Eiichi Shimazu Date: Sat Nov 14 22:45:17 2026 +0900 判定の決め 3 つを判定ファイルと画面に入れる (依頼 185・エイイチの決め 2026-11-14). verdicts.py に outcome=not_applicable と labeler=rule を足し, 確定の扱いを run_input・human・human_fixed_ai・human_only・rule にする (CONFIRMED_LABELERS. 行は書き換えない). current_by_stem (確定の行があればそれ. 後ろの AI の行で未確定へ戻らない)・success_rate (確定した回の success / (success + failure). not_applicable・not_judgeable・unknown は分母に入れない)・rule_row (決め 2: 送ったチャンク 0 → not_applicable. 決め 3: AI が frames/ だけ・素材なしで not_judgeable → not_judgeable. 両方なら 2. note に「エイイチの決め 2026-11-14 依頼185」と決めの番号). seed_verdicts.py に --rule (送ったチャンクは log.jsonl の chunk の行で executed が True の数. 確定の行がある回には足さない・ロックの中でも確かめる). AI は not_applicable を付けない・確定した回に AI の行を足さない. 画面: 既定の一覧を「まだ確定していない回」に, 上に成功率と分母に入れない数. 判定ファイルに rule 84 行 (決め 2 51・決め 3 33・重なり 44) を追記. 未確定 24 回・成功率 41/68 = 60.3%. テスト test_verdicts 39 (依頼 185. codex 1 回・指摘なし) Co-Authored-By: Claude Opus 5.5 --- pi05/verdict_viewer/index.html | 23 +++++++-- pi05/verdict_viewer/seed_verdicts.py | 50 +++++++++++++++++-- pi05/verdict_viewer/server.py | 13 +++-- pi05/verdict_viewer/test_verdicts.py | 97 ++++++++++++++++++++++++++++++++++++ pi05/verdict_viewer/verdicts.py | 50 +++++++++++++++++-- 5 files changed, 216 insertions(+), 17 deletions(-) diff --git a/pi05/verdict_viewer/index.html b/pi05/verdict_viewer/index.html index 461bab8..f9008e1 100644 --- a/pi05/verdict_viewer/index.html +++ b/pi05/verdict_viewer/index.html @@ -18,6 +18,8 @@ #side header { padding: 10px; border-bottom: 1px solid var(--border); } #side h1 { font-size: 14px; margin: 0 0 6px; } #filter { width: 100%; font-size: 13px; padding: 3px; } + #rate { margin-top: 6px; font-size: 13px; } + #rate b { font-size: 16px; } #count { color: var(--text-secondary); margin-top: 4px; font-size: 12px; } #list { overflow-y: auto; flex: 1; } .item { padding: 5px 10px; border-bottom: 1px solid var(--border); cursor: pointer; display: flex; justify-content: space-between; gap: 6px; } @@ -28,7 +30,7 @@ .tag { font-size: 11px; padding: 0 5px; border-radius: 3px; border: 1px solid var(--border); white-space: nowrap; align-self: center; } .tag.success { color: var(--ok); border-color: var(--ok); } .tag.failure { color: var(--bad); border-color: var(--bad); } - .tag.unknown, .tag.not_judgeable { color: var(--warn); border-color: var(--warn); } + .tag.unknown, .tag.not_judgeable, .tag.not_applicable { color: var(--warn); border-color: var(--warn); } .tag.ai { border-style: dashed; } #main { flex: 1; overflow-y: auto; padding: 12px 16px; } #title { font-family: ui-monospace, Menlo, monospace; font-size: 15px; font-weight: 600; } @@ -69,11 +71,12 @@

推論の記録の判定

+
@@ -109,8 +112,8 @@
``` ### test_verdicts.py ```python #!/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_place_not_applicable_only_for_place(self): # 依頼 193: place は not_applicable (そこまで行かなかった) を書ける. pick には書けない V.append_rows([row("pi05_1", "failure", "human_only", pick="no", place="not_applicable")], self.csv) self.assertEqual(V.read_rows(self.csv)[-1]["place"], "not_applicable") before = self.csv.read_bytes() with self.assertRaises(ValueError): V.append_rows([row("pi05_2", "failure", "human_only", pick="not_applicable", place="not_applicable")], 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_post_place_not_applicable(self): # 依頼 193: 画面から place=not_applicable を送れる. pick=not_applicable は 400 で書かない st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "pick": "no", "place": "not_applicable"}) self.assertEqual(st, 200) self.assertEqual((j["row"]["pick"], j["row"]["place"]), ("no", "not_applicable")) before = self.csv.read_bytes() st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "pick": "not_applicable"}) self.assertEqual(st, 400) self.assertEqual(self.csv.read_bytes(), before) 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) class TestConfirm192(unittest.TestCase): """依頼 192: y/n/u/x は選ぶだけ・確定 (ボタンか Enter) で 1 行だけ書く. 画面を Chrome で動かす (playwright がなければ飛ばす).""" @classmethod def setUpClass(cls): try: from playwright.sync_api import sync_playwright cls.pw = sync_playwright().start() cls.browser = cls.pw.chromium.launch(channel="chrome") except Exception as e: # noqa: BLE001 raise unittest.SkipTest(f"playwright / Chrome がない: {e}") @classmethod def tearDownClass(cls): cls.browser.close() cls.pw.stop() 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.srv = server.ThreadingHTTPServer(("127.0.0.1", 0), server.make_handler(server.App(self.runs, self.csv))) threading.Thread(target=self.srv.serve_forever, daemon=True).start() self.pg = self.browser.new_page() self.pg.goto(f"http://127.0.0.1:{self.srv.server_address[1]}/") self.wait_loaded() def tearDown(self): self.pg.close() self.srv.shutdown() self.srv.server_close() self.tmp.cleanup() def wait_loaded(self): self.pg.wait_for_function("() => { const t = document.querySelector('#title').textContent;" " return t.startsWith('pi05_') && !t.includes('読み込み中'); }") def n_rows(self): return len(V.read_rows(self.csv)) def on_keys(self): return self.pg.eval_on_selector_all("button.key.on", "e => e.map(x => x.dataset.oc)") def disabled(self): return self.pg.eval_on_selector("#confirm", "e => e.disabled") def test_choose_only_does_not_write(self): n = self.n_rows() self.assertTrue(self.disabled()) self.pg.keyboard.press("n") self.assertEqual(self.on_keys(), ["failure"]) self.assertFalse(self.disabled()) self.pg.click('button.key[data-oc="success"]') # 選び直し self.assertEqual(self.on_keys(), ["success"]) self.pg.wait_for_timeout(600) self.assertEqual(self.n_rows(), n) def test_confirm_before_choose_does_nothing(self): n = self.n_rows() title = self.pg.text_content("#title") self.pg.keyboard.press("Enter") self.pg.click("#confirm", force=True) self.pg.wait_for_timeout(600) self.assertEqual(self.n_rows(), n) self.assertEqual(self.pg.text_content("#title"), title) def test_enter_writes_one_row(self): n = self.n_rows() stem = self.pg.text_content("#title") self.pg.keyboard.press("n") self.pg.keyboard.press("Enter") self.pg.keyboard.press("Enter") # 0.4 秒の間の 2 回目は書かない self.pg.wait_for_function("() => document.querySelector('#toast').textContent.includes('確定しました')") self.assertIn(stem, self.pg.text_content("#toast")) self.pg.wait_for_timeout(800) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["run_stem"], rows[-1]["outcome"]), (stem, "failure")) def test_button_writes_one_row(self): n = self.n_rows() stem = self.pg.text_content("#title") self.pg.keyboard.press("y") self.pg.click("#confirm") self.pg.click("#confirm", force=True) self.pg.wait_for_timeout(1000) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["run_stem"], rows[-1]["outcome"]), (stem, "success")) def test_enter_in_note_confirms_but_keys_do_not_choose(self): n = self.n_rows() self.pg.keyboard.press("u") self.pg.click("#note") self.pg.keyboard.type("yn") self.assertEqual(self.on_keys(), ["unknown"]) self.pg.keyboard.press("Enter") self.pg.wait_for_timeout(1000) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["outcome"], rows[-1]["note"]), ("unknown", "yn")) def test_moving_discards_choice(self): n = self.n_rows() stem = self.pg.text_content("#title") self.pg.keyboard.press("n") self.pg.keyboard.press("ArrowRight") self.wait_loaded() self.assertNotEqual(self.pg.text_content("#title"), stem) self.assertEqual(self.on_keys(), []) self.assertTrue(self.disabled()) self.pg.keyboard.press("ArrowLeft") self.wait_loaded() self.assertEqual(self.pg.text_content("#title"), stem) self.assertEqual(self.on_keys(), []) self.pg.keyboard.press("Enter") self.pg.wait_for_timeout(600) self.assertEqual(self.n_rows(), n) # ---- 依頼 193: つかむ / 置くを日本語に・つかめなかったで置くは自動でそこまで行かなかった ---- def pp_on(self): return self.pg.eval_on_selector_all("[data-pp].on", "e => e.map(x => x.dataset.pp + '=' + x.dataset.v)") def test_pp_labels_are_japanese(self): got = self.pg.eval_on_selector_all("[data-pp]", "e => e.map(x => x.dataset.pp + '=' + x.dataset.v + ':' + x.textContent)") self.assertEqual(got, ["pick=yes:つかめた", "pick=no:つかめなかった", "pick=unknown:分からない", "place=yes:置けた", "place=no:置けなかった", "place=not_applicable:そこまで行かなかった", "place=unknown:分からない"]) self.assertEqual(self.pg.eval_on_selector_all("#panel .row .lbl", "e => e.map(x => x.textContent)"), ["pick", "place"]) def test_pick_no_selects_place_not_applicable(self): n = self.n_rows() self.pg.click('[data-pp="pick"][data-v="no"]') self.assertEqual(self.pp_on(), ["pick=no", "place=not_applicable"]) self.pg.click('[data-pp="place"][data-v="unknown"]') # 選び直せる self.assertEqual(self.pp_on(), ["pick=no", "place=unknown"]) self.pg.click('[data-pp="pick"][data-v="yes"]') # つかめたに戻すと置くは選んでいない self.assertEqual(self.pp_on(), ["pick=yes"]) self.pg.click('[data-pp="pick"][data-v="no"]') self.pg.wait_for_timeout(300) self.assertEqual(self.n_rows(), n) # 選ぶだけでは書かない stem = self.pg.text_content("#title") self.pg.keyboard.press("n") self.pg.keyboard.press("Enter") self.pg.wait_for_timeout(1000) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["run_stem"], rows[-1]["pick"], rows[-1]["place"]), (stem, "no", "not_applicable")) class TestRule185(unittest.TestCase): """エイイチの決め (2026-11-14 依頼 185): run_input も確定・送ったチャンク 0 → not_applicable・素材不足 → not_judgeable.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.runs = Path(self.tmp.name) / "runs" self.csv = Path(self.tmp.name) / "v.csv" # stem: (送ったチャンク, 先に入れる行) self.case = { "pi05_20261002_000001": (0, [row("pi05_20261002_000001", "not_judgeable", "ai", evidence="frames")]), # 2 と 3 → 2 "pi05_20261002_000002": (2, [row("pi05_20261002_000002", "not_judgeable", "ai", evidence="frames")]), # 3 "pi05_20261002_000003": (2, [row("pi05_20261002_000003", "not_judgeable", "ai", evidence="")]), # 3 (素材なし) "pi05_20261002_000004": (0, [row("pi05_20261002_000004", "failure", "ai", evidence="top.mp4")]), # 2 "pi05_20261002_000005": (0, [row("pi05_20261002_000005", "not_judgeable", "ai", evidence="frames"), row("pi05_20261002_000005", "failure", "human_fixed_ai")]), # 人の行 → 足さない "pi05_20261002_000006": (2, [row("pi05_20261002_000006", "not_judgeable", "ai", evidence=""), row("pi05_20261002_000006", "success", "human_only")]), # 人の行 → 足さない "pi05_20261002_000007": (0, [row("pi05_20261002_000007", "success", "run_input")]), # run_input → 足さない "pi05_20261002_000008": (2, [row("pi05_20261002_000008", "not_judgeable", "ai", evidence="top.mp4")]), # 当たらない "pi05_20261002_000009": (3, [row("pi05_20261002_000009", "failure", "ai", evidence="frames")]), # 当たらない } rows = [] for stem, (n, rs) in self.case.items(): d = self.runs / stem d.mkdir(parents=True) lines = [{"type": "model"}] + [{"chunk": str(i), "executed": "True"} for i in range(n)] + [{"chunk": "9", "executed": "False"}] (d / "log.jsonl").write_text("\n".join(json.dumps(x) for x in lines) + "\n") rows += rs V.append_rows(rows, self.csv) def tearDown(self): self.tmp.cleanup() def rule_rows(self): return {r["run_stem"]: r for r in V.read_rows(self.csv) if r["labeler"] == "rule"} def test_rule_rows(self): c = S.do_rule(self.csv, self.runs, dry_run=False) got = {k[-1]: v["outcome"] for k, v in self.rule_rows().items()} self.assertEqual(got, {"1": "not_applicable", "2": "not_judgeable", "3": "not_judgeable", "4": "not_applicable"}) self.assertEqual((c["rule2"], c["rule3"], c["both"]), (2, 2, 1)) for r in self.rule_rows().values(): self.assertIn("エイイチの決め 2026-11-14 依頼185", r["note"]) self.assertIn("決め2", self.rule_rows()["pi05_20261002_000001"]["note"]) self.assertIn("決め3", self.rule_rows()["pi05_20261002_000002"]["note"]) def test_rule_never_after_human_or_run_input(self): S.do_rule(self.csv, self.runs, dry_run=False) for stem in ("pi05_20261002_000005", "pi05_20261002_000006", "pi05_20261002_000007"): self.assertNotIn(stem, self.rule_rows()) def test_rule_twice_adds_nothing(self): S.do_rule(self.csv, self.runs, dry_run=False) before = self.csv.read_bytes() S.do_rule(self.csv, self.runs, dry_run=False) self.assertEqual(self.csv.read_bytes(), before) def test_rule_rechecks_inside_lock(self): # 一覧を読んだあとに人の行が足されたら, rule の行は足さない real = V.read_rows def late_human(p): rows = real(p) V.append_rows([row("pi05_20261002_000002", "failure", "human")], self.csv) return rows with mock.patch.object(V, "read_rows", side_effect=late_human): S.do_rule(self.csv, self.runs, dry_run=False) self.assertNotIn("pi05_20261002_000002", self.rule_rows()) def test_confirmed_and_success_rate(self): S.do_rule(self.csv, self.runs, dry_run=False) app = server.App(self.runs, self.csv) conf = {r["run_stem"][-1]: r["confirmed"] for r in app.runs_summary()} self.assertEqual(conf, {"1": True, "2": True, "3": True, "4": True, "5": True, "6": True, "7": True, "8": False, "9": False}) st = app.stats() # 確定: success 2 (6・7) / failure 1 (5). not_applicable 2・not_judgeable 2 は分母に入れない. AI の行も入れない self.assertEqual((st["success"], st["failure"], st["denominator"], st["confirmed"]), (2, 1, 3, 7)) self.assertAlmostEqual(st["rate"], 2 / 3) self.assertEqual(st["by_outcome"]["not_applicable"], 2) def test_success_rate_denominator(self): rows = [row("a", "success", "human"), row("b", "failure", "run_input"), row("c", "unknown", "human"), row("d", "not_judgeable", "rule"), row("e", "not_applicable", "rule"), row("f", "success", "ai"), row("b", "success", "human_fixed_ai")] # b は後ろの人の行が正 st = V.success_rate(rows) self.assertEqual((st["success"], st["failure"], st["denominator"]), (2, 0, 2)) self.assertEqual(st["rate"], 1.0) self.assertIsNone(V.success_rate([row("x", "not_applicable", "rule")])["rate"]) def test_ai_after_confirmed_does_not_unconfirm(self): rows = [row("a", "not_applicable", "rule"), row("a", "failure", "ai")] self.assertEqual(V.current_by_stem(rows)["a"]["labeler"], "rule") def test_ai_cannot_write_not_applicable(self): r = S.ai_row("pi05_x", {"outcome": "not_applicable", "pick": "no", "place": "no"}, "frames", 0) self.assertEqual(r["outcome"], "unknown") if __name__ == "__main__": unittest.main(verbosity=1) ``` ### 判定の画面を開く.command ```bash #!/bin/zsh # 判定の画面 (依頼 183) を立ち上げてブラウザで開く. ダブルクリックで動く. # すでに 8765 番で動いていれば立ち上げ直さずに開くだけ. PORT=8765 URL="http://localhost:${PORT}/" LOG="$HOME/r2r2r-scripts/logs/verdict_viewer_server.log" if lsof -i :${PORT} >/dev/null 2>&1; then echo "[VIEWER] すでに ${PORT} 番で動いています" else echo "[VIEWER] 立ち上げます (ログ: ${LOG})" cd "$HOME" if command -v setsid >/dev/null 2>&1; then nohup setsid python3 "$HOME/r2r2r-scripts/pi05/verdict_viewer/server.py" --no-open > "$LOG" 2>&1 & else nohup python3 "$HOME/r2r2r-scripts/pi05/verdict_viewer/server.py" --no-open > "$LOG" 2>&1 & disown fi # 応答が返るまで最大 10 秒待つ for i in {1..20}; do code=$(curl -s -o /dev/null -w '%{http_code}' "$URL") [ "$code" = 200 ] && break sleep 0.5 done if [ "$code" != 200 ]; then echo "[VIEWER] 立ち上がりませんでした. ログを見てください: ${LOG}" tail -n 20 "$LOG" exit 1 fi fi open "$URL" ``` ### test_state_source.py ```python #!/usr/bin/env python3 """test_state_source.py — 1 コマごとの state の出どころ (state_source. 依頼 196) の確かめ. python3 pi05/test_state_source.py -v 前半はファイルを読まない部分だけ. 後半 (TestSameAsBefore) は KIOXIA の runs/pi05_policy を読み, 直す前のコード (git の HEAD) と今のコードで /tmp に作り, state / actions / t がバイト単位で同じことを見る (KIOXIA がなければ飛ばす. データセットにもアームにも触らない). """ import importlib.util import json import subprocess import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace import numpy as np _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) import build_correction_episodes as bce # noqa: E402 def samples(still=False): """0〜1 秒は 0.1 秒おき → 1.0〜2.6 秒は抜け (j1 が 6° 動く. still なら動かない) → 2.6〜3.6 秒は 0.1 秒おき.""" a = np.round(np.arange(0.0, 1.01, 0.1), 3) b = np.round(np.arange(2.6, 3.61, 0.1), 3) at = np.concatenate([a, b]) aq = np.zeros((len(at), 6)) if not still: aq[len(a):, 1] = 6.0 return at, aq def runs_of_nonzero(src): nz = np.r_[0, (np.asarray(src) != 0).astype(int)] return int((np.diff(nz) == 1).sum()) class TestFlags(unittest.TestCase): def test_motion_gap_is_1(self): at, aq = samples() t = np.arange(0.0, 3.6, 0.1) src = bce.gap_fill_flags(t, at, aq) self.assertEqual(src.dtype, np.int8) inside = (t > 1.0) & (t < 2.6) self.assertTrue(np.all(src[inside] == bce.SRC_FILLED_MOVE)) self.assertTrue(np.all(src[~inside] == bce.SRC_MEASURED)) def test_still_gap_is_2(self): at, aq = samples(still=True) t = np.arange(0.0, 3.6, 0.1) src = bce.gap_fill_flags(t, at, aq) self.assertEqual(set(np.unique(src)), {bce.SRC_MEASURED, bce.SRC_FILLED_STILL}) def test_flags_mark_exactly_what_speed20_changes(self): """speed20 が線形補間から変えたコマは, すべて印が 0 でないコマ.""" at, aq = samples() t = np.arange(0.0, 3.6, 0.1) q_lin, _ = bce.fill_motion_gaps(t, at, aq, mode="linear") q_sp, _ = bce.fill_motion_gaps(t, at, aq, mode="speed20") src = bce.gap_fill_flags(t, at, aq) changed = np.any(np.abs(q_lin - q_sp) > 1e-9, axis=1) self.assertTrue(changed.any()) self.assertTrue(np.all(src[changed] != bce.SRC_MEASURED)) def test_no_gap_no_flag(self): at = np.round(np.arange(0.0, 3.01, 0.1), 3) aq = np.cumsum(np.ones((len(at), 6)), axis=0) self.assertEqual(int(bce.gap_fill_flags(np.arange(0.0, 3.0, 0.1), at, aq).sum()), 0) def test_compat_ignores_state_source(self): """お手本にない state_source のキーがあっても check_compat はキーの違いとして数えない.""" with tempfile.TemporaryDirectory() as d: import cv2 for name, extra in (("ours", True), ("ref", False)): ep = Path(d) / name / "e0" (ep / "frames").mkdir(parents=True) cv2.imwrite(str(ep / "frames" / "000000.jpg"), np.zeros((4, 6, 3), np.uint8)) kw = {"state_source": np.zeros(1, np.int8)} if extra else {} np.savez_compressed(ep / "episode.npz", t=np.zeros(1), t_orig=np.zeros(1), state=np.zeros((1, 7), np.float32), actions=np.zeros((1, 7), np.float32), **kw) (ep / "meta.json").write_text(json.dumps({"fps": 10, "image_size": [6, 4], "state_dim": 7, "state_names": [], "task": "x"})) self.assertEqual(bce.check_compat(Path(d) / "ours", Path(d) / "ref"), []) def _old_module(): """直す前の build_correction_episodes.py (git の HEAD) を別の名前で読み込む.""" src = subprocess.run(["git", "show", "HEAD:pi05/build_correction_episodes.py"], cwd=_HERE, capture_output=True, text=True, check=True).stdout tmp = Path(tempfile.mkdtemp()) / "bce_old.py" tmp.write_text(src) spec = importlib.util.spec_from_file_location("bce_old", tmp) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod # 補正の多い本・巻き戻しのある本など. 依頼 196 の時点で KIOXIA にあるもの SAMPLE_RUNS = ("pi05_20261031_175527", "pi05_20261031_180435") @unittest.skipUnless(Path("/Volumes/KIOXIA").is_mount() and bce.RUNS_ROOT.is_dir(), "KIOXIA がない") class TestSameAsBefore(unittest.TestCase): @classmethod def setUpClass(cls): cls.old = _old_module() def _build(self, mod, run, d, gap_fill, trim_hold): args = SimpleNamespace(gap_fill=gap_fill, trim_hold=trim_hold, no_trim_mid_idle=False, max_protrude_px=None, force=True) out = Path(d) / mod.__name__ / f"{gap_fill}_{trim_hold}" / run res = mod.build_one(bce.RUNS_ROOT / run, out, args) self.assertTrue(res.get("used"), res) z = np.load(out / "episode.npz") return {k: z[k] for k in z.files}, json.loads((out / "meta.json").read_text()) def test_bytes_and_counts(self): with tempfile.TemporaryDirectory() as d: for run in SAMPLE_RUNS: if not (bce.RUNS_ROOT / run).is_dir(): self.skipTest(f"{run} がない") for gap_fill, trim_hold in (("linear", False), ("speed20", False), ("speed20", True)): with self.subTest(run=run, gap_fill=gap_fill, trim_hold=trim_hold): a, _ = self._build(self.old, run, d, gap_fill, trim_hold) b, meta = self._build(bce, run, d, gap_fill, trim_hold) self.assertNotIn("state_source", a) for k in ("t", "t_orig", "state", "actions"): self.assertEqual(a[k].dtype, b[k].dtype, k) self.assertEqual(a[k].tobytes(), b[k].tobytes(), k) src = b["state_source"] self.assertEqual(src.dtype, np.int8) self.assertEqual(len(src), len(b["state"])) self.assertTrue(set(np.unique(src)) <= set(bce.STATE_SOURCE_CODES)) gf = meta["gap_fill"] filled = int((src != 0).sum()) self.assertEqual(meta["filled_steps"], filled) self.assertEqual(meta["filled_motion_steps"], int((src == 1).sum())) self.assertAlmostEqual(meta["filled_ratio"], filled / len(src), places=4) # 印のつながりの数は抜けの数を超えない (1 つの抜けが印を 2 つに分けることはない) self.assertLessEqual(runs_of_nonzero(src), gf["gaps"]) self.assertLessEqual(runs_of_nonzero(src == 1), gf["motion_gaps"]) if gf["gaps"] == 0: self.assertEqual(filled, 0) if __name__ == "__main__": unittest.main() ``` ### diag_fill_flags.py ```python #!/usr/bin/env python3 """diag_fill_flags.py — 既存の補正データセットで, 実測角の抜けを埋めたコマの割合を測る (2026-11-14 依頼 196). datasets/ の名前に corr / deep_even10 が入るデータセットの, 補正エピソード (meta.json の source が pi05_policy) ごとに, 元の runs/pi05_policy/{run_stem}/ から build_correction_episodes.build_one を meta.json と同じ条件 (gap_fill の mode・trim_mid_idle・trim_hold) で /tmp に作り直し, state_source (コマごとの印) を数える. 作り直した state / actions / t がデータセットの episode.npz と一致しない本は数えない (理由を CSV に残す). データセット・元の記録は読むだけ (書かない). 書くのは /tmp と diag/fill_flags_20261114/ だけ. python3 pi05/diag_fill_flags.py """ import argparse import csv import json import sys from pathlib import Path from types import SimpleNamespace import numpy as np _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) import build_correction_episodes as bce # noqa: E402 import build_demo_episodes as bde # noqa: E402 OUT_NAME = "fill_flags_20261114" TMP_ROOT = Path("/tmp") / OUT_NAME # 学習に使ったことがあるデータセットを先に並べる (依頼 196) TRAINED_FIRST = ("mecharm_redblock_box_teach_v1_trim2_corr_v5", "mecharm_redblock_box_deep_even10_v1", "mecharm_redblock_box_teach_v1_trim2_corr_v5_deep") def target_datasets(ds_root): names = sorted(p.name for p in ds_root.iterdir() if p.is_dir() and not p.name.startswith(".") and ("corr" in p.name or "deep_even10" in p.name)) return sorted(names, key=lambda n: (n not in TRAINED_FIRST, TRAINED_FIRST.index(n) if n in TRAINED_FIRST else 0, n)) def rebuild(run_stem, meta, cache): """meta.json と同じ条件で /tmp に作り直す. 戻り値: (npz の中身 dict / None, 理由).""" gf = meta.get("gap_fill") or {} key = (run_stem, gf.get("mode"), bool(meta.get("trim_mid_idle")), meta.get("trim_hold") is not None) if key in cache: return cache[key] out = TMP_ROOT / "rebuild" / "_".join(str(k) for k in key) / run_stem args = SimpleNamespace(gap_fill=gf.get("mode") or "linear", trim_hold=key[3], no_trim_mid_idle=not key[2], max_protrude_px=None, force=True) res = bce.build_one(bce.RUNS_ROOT / run_stem, out, args) if not res.get("used"): cache[key] = (None, f"作り直せない: {res.get('reason')}") else: z = np.load(out / "episode.npz") cache[key] = ({k: z[k] for k in z.files}, "ok") return cache[key] def main(): ap = argparse.ArgumentParser() ap.add_argument("--dataset", action="append", help="このデータセットだけ (複数可)") args = ap.parse_args() bde.ensure_ssd() ds_root = bde.DATA_ROOT / "datasets" out_dir = bde.DATA_ROOT / "diag" / OUT_NAME if out_dir.exists(): sys.exit(f"[ABORT] 出力先がもうある: {out_dir}") names = args.dataset or target_datasets(ds_root) rows, cache = [], {} for name in names: for mp in sorted((ds_root / name / "raw_episodes").glob("[!.]*/meta.json")): meta = json.loads(mp.read_text()) if not str(meta.get("source", "")).startswith("pi05_policy"): continue ep = mp.parent stem = meta.get("run_stem", ep.name) z = np.load(ep / "episode.npz") row = {"dataset": name, "episode": ep.name, "run_stem": stem, "gap_fill": (meta.get("gap_fill") or {}).get("mode"), "trim_hold": meta.get("trim_hold") is not None, "steps": len(z["state"]), "filled_steps": "", "filled_motion_steps": "", "filled_ratio": "", "gaps": (meta.get("gap_fill") or {}).get("gaps"), "motion_gaps": (meta.get("gap_fill") or {}).get("motion_gaps"), "status": ""} new, why = rebuild(stem, meta, cache) if new is None: row["status"] = why elif not all(k in new and new[k].shape == z[k].shape and new[k].tobytes() == z[k].tobytes() for k in ("state", "actions", "t")): row["status"] = "state が一致しない (数えない)" else: src = new["state_source"] n_f = int((src != bce.SRC_MEASURED).sum()) row.update(filled_steps=n_f, filled_motion_steps=int((src == bce.SRC_FILLED_MOVE).sum()), filled_ratio=round(n_f / len(src), 4), status="ok") rows.append(row) print(f"{name} {ep.name} {row['status']} {row['filled_steps']}/{row['steps']}") if not rows: # codex P2: 出力先を作る前に止める (空のまま作ると次に走らせられない) sys.exit("[ABORT] 対象の補正エピソードがない") out_dir.mkdir(parents=True) with (out_dir / "episodes.csv").open("w", newline="") as f: w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) w.writeheader(); w.writerows(rows) summ = [] for name in names: rs = [r for r in rows if r["dataset"] == name] ok = [r for r in rs if r["status"] == "ok"] steps = sum(r["steps"] for r in ok); filled = sum(r["filled_steps"] for r in ok) top = max(ok, key=lambda r: r["filled_ratio"]) if ok else None summ.append({"dataset": name, "corr_episodes": len(rs), "counted": len(ok), "not_counted": len(rs) - len(ok), "steps": steps, "filled_steps": filled, "filled_motion_steps": sum(r["filled_motion_steps"] for r in ok), "filled_ratio": round(filled / steps, 4) if steps else "", "max_episode": top["episode"] if top else "", "max_episode_ratio": top["filled_ratio"] if top else ""}) with (out_dir / "summary.csv").open("w", newline="") as f: w = csv.DictWriter(f, fieldnames=list(summ[0].keys())) w.writeheader(); w.writerows(summ) (out_dir / "README.md").write_text( "# fill_flags_20261114 (依頼 196)\n\n" "既存の補正データセットで, 実測角の抜けを埋めたコマ (state_source が 1 または 2) の数と割合。\n" "`python3 pi05/diag_fill_flags.py` で作った。データセットは読むだけ。作り直しは /tmp。\n\n" "- state_source: 0 = 実測角 / 1 = 動いた抜けを埋めた値 / 2 = 動かなかった抜けを埋めた値\n" "- filled_ratio = (1 と 2 のコマ) / 全コマ。filled_motion_steps は 1 のコマの数\n" "- status が ok でない本 (作り直した state / actions / t がデータセットと一致しない・作り直せない) は数えない\n" "- episodes.csv: 本ごと / summary.csv: データセットごと (max_episode = 割合がいちばん多い本)\n") print(f"→ {out_dir}") return 0 if __name__ == "__main__": sys.exit(main()) ``` ## 次にやること：案1で手教示の記録を直し、撮り直したお手本で成功率59.8%と比べる この回で、データの質を測る道具と、比べるための基準の数字がそろいました。次は次の順で進めます。 | 順 | やること | 要るもの | |---|---|---| | 1 | 案1（電流の記録の角度を手教示の記録にも書く）を作る。実装・テスト・レビューまで | 依頼箱 | | 2 | 実機で1本、次に5本で確かめる。その前に Pi の daemon が11月5日の版かを確かめる | 実機・Pi | | 3 | 直した記録でお手本を撮り直す | 実機 | | 4 | 学習し直す | Runpod の GPU | | 5 | 実機で推論し、判定の画面で成否を付けて、59.8%と比べる | 実機・判定の画面 | 既存の補正のお手本は、作り直していません。印はこの回の道具（`diag_fill_flags.py`）でいつでも数えられるので、次にそのデータで学習するときに印付きで作ります。 ## 出典 - ものづくり太郎チャンネル「[【予言的中？】製造現場のAI活用はどこまで来たか？](https://www.youtube.com/watch?v=GmwphgG7cG4)」 - IT Business Today「[Airion と矢崎総業の提携](https://itbusinesstoday.com/ja/tech/ai/airion-yazaki-team-up-on-autonomous-physical-ai/)」 - Physical Intelligence「[π0.5](https://www.pi.website/blog/pi05)」