mechArm 270 Pi 受け渡し動作の成立を探る Log#2 | アームが差し出し、人が受け取る。設計が4回の外部レビューで固まるまで
ロボットアームが持っているブロックを、人が手で受け取る。この一歩目の設計書を書いた。コードを書く前の設計だけで、外部レビューが4ラウンド入った。人の手がアームの可動域に入るタスクは、置くタスクと安全の桁が違う。その差がレビューの指摘として全部出てきたので、指摘と反映の過程を省略せずに残す。
この回の成果物は3つ。設計書（r2 → r3 → r3.1 → r3.2 → r3.3 と4回改訂）、管理文書の更新2件、そして実装の最初の3点（候補抽出スクリプト・Unity 手順書・新規台帳の初期化）。制御コード本体はまだ書いていない。提示姿勢が Unity での検証を経て確定するまで、アームを動かすコードには着手しない順序にしてある。
一歩目のタスク
作るのはこの流れ。
PICK → 保持 → HOME（既存のまま）
→ 提示姿勢へ差し出して静止
→ 人の合図で N 秒待つ
→ グリッパーを開く
→ 人が手で受ける（掴まれていなければ落ちる）人が掴んだかどうかを、アームはリアルタイムに判断しない。判断する手段が現状ないからだ。画像で成否を見る VLM は API の往復が秒単位で遅く、グリッパーの力や電流は読めていない。だから最初は判断しない。N 秒で決め打ちして開き、取られなければ落とす。落ちた回数が「何秒待てば人は受け取れるか」の測定値になる。
ただし人の安全に関わる箇所は決め打ちにしない。待機の開始と、アームが再び動き出す瞬間は、人の確認入力を経てから進む。測定は開ループ（カメラやセンサーで確かめず、決めた動きをそのまま出すこと）、安全は人の確認。この2つを混ぜないことが設計の背骨になった。
設計の初期判断
最初の版（r2）で決めた骨格は次のとおり。
新しいスクリプトを立てる。既存の Pick & Place 本線 run_pick_place.py には1文字も触らず、受け渡しは run_pick_handover.py として別に作る。共通部品（FK テーブル参照・視覚サーボ・カメラ・通信）は import で再利用する。シリーズとデータの対応も揃えた。Pick & Place = run_pick_place.py + pick_place_ledger_v3.csv、受け渡し = run_pick_handover.py + handover_ledger_v1.csv。台帳を分けたことで、既存台帳への列追加作業そのものが不要になった。
保存先はこう確定した。
/Volumes/KIOXIA/r2r2r-data/runs/
├── pick_place_ledger_v3.csv # 既存・不変
├── run_pick_place/ # 既存・不変
├── handover_ledger_v1.csv # 新規・受け渡し台帳
└── run_handover/
└── {run_stem}/ # 1試行1フォルダ提示姿勢は勘で作らない。FK テーブル（Unity のデジタルツインで実測した、関節角と爪先位置の対応表）から、マット手前側・人が受け取れる高さの実在姿勢を抽出して選ぶ。姿勢を固定1つにすれば、テーブルの穴に当たらない。
この骨格を、コードレビュー依頼テンプレートに載せて外部レビュー（codex）へ出した。以降、4ラウンドの指摘と反映を全量で載せる。設計がどう変わったかの記録そのものが、このタスクの安全設計の中身だからだ。
レビュー第1ラウンド（r2 → r3）
第1ラウンドの結論は「現行 r2 のまま実装へ進む: ❌」。必須修正7点が付いた。全文を引用する。
設計書 r2 のまま Claude Code へ実装発注するのは見送りです。主な必須修正は、①既存 PICK の再利用方式、②提示姿勢・移動軌道の検証、③人が退避したことを確認してから RETREAT するインターロック、④台帳列、⑤「閉じる指令なし」の適用範囲です。
A1. 開ループ測定 ⚠️ — N 秒後に無条件で開き、taken/drop を測る方針は探索的一歩目として妥当です。ただし、現在の適応的な順序「5秒成功→3→2、失敗→8」では、参加者の慣れと N の効果が交絡します。wait_sec_requested と wait_sec_actual（monotonic 実測）、WAIT 開始・RELEASE 送信・RELEASE 応答の各時刻、release_command_ok、試行順・条件順、taken/drop/stuck/unknown を表す排他的な outcome の記録、N の順番のランダム化またはカウンターバランス化を追加推奨します。
A2. 提示姿勢 ❌ — 位置条件だけの候補抽出では、爪の向きもリンク全体の位置も判定できません。ee_x/y/z と関節角だけでは「爪の正面が Z+」という条件は直接表現されていません。Unity/Brep の正しい FK から、各候補の tool-forward / tool-up ベクトルを別の読み取り専用 sidecar CSV に出し、tool_forward · [0,0,1] >= cos(許容角) を条件にしてください。爪先だけでなく、肘・手首・全リンクの人側への張り出し、床/ベースとのクリアランスも候補条件に。
A3. PRESENT_APPROACH ❌ — j0 だけ提示値、他軸 HOME は、妥当な中間姿勢である根拠がありません。HOME から j0 を先に回すことで、保持物・肘・手首が大きな弧を描く可能性があります。PRESENT に近い関節姿勢を FK テーブルから中間点として選び、各軸差分上限を決め、send_angles の関節補間をサンプリングして全リンクの swept volume を Unity/Brep で確認してください。
A4. 台帳25列 ❌ — N と結果の集計には不足があります。trial_status / abort_stage / wait_sec_requested / wait_sec_actual / release_command_ok / PRESENT・RETREAT・HOME の到達結果 / 排他的 outcome（taken/drop/stuck/unknown/not_applicable）/ カメラ別 capture status / hand_clear_confirmed などが最低限必要です。追記専用なので、taken/drop は後から既存行を書き換えず、対話入力後に1回だけ append してください。中断・入力不能時は unknown を許容しないと「必ず追記」と「未入力で進めない」が衝突します。
B1. 既存モジュール再利用 ❌ — 現行の run_pick_place.py には公開された「PICK→保持→HOME だけ」の入口がありません。run() は PICK から PLACE、既存台帳処理までを含む一体型です。「コピーしない」「既存ファイル無変更」「既存 PICK をそのまま呼ぶ」の3条件は、現状では同時に満たせません。(1) PICK フェーズを共有モジュールへ抽出して両本線を回帰確認する、(2) 無変更を絶対条件にするなら新スクリプト側で下位関数から orchestration を再構築し、重複・将来の乖離を正式に受容する、のどちらかを設計で明記してください。
C1. 人が手を出す区間 ❌ — 画面表示2種だけでは不足です。PRESENT 到達・停止確認後、明示的な「受け取り開始」入力で WAIT を開始。RELEASE 後、自動 RETREAT しない。「手を離し、可動域外へ退避した」確認入力後にのみ RETREAT。実機 E-stop を即操作できる位置に。初回人試験は補助者同席、単独で行わない。SPEED=20 を「人体に安全な速度」とは扱わない。
C3. RETREAT ❌ — 終点が Z- でも、関節補間中の爪・肘・手首が一時的に Z+ へ膨らまない保証はありません。hand_clear_confirmed が true になるまで RETREAT 禁止。PRESENT→小さな検証済み RETRACT→APPROACH→HOME の段階化。各区間の補間軌道と全リンク形状を検証。
D7. close 指令 ❌ — PICK には必ず閉じる指令が必要なので、「コード全体に state=1 が存在しない」grep は成立しません。「PICK 完了後の handover phase に閉指令が存在しない」に限定し、open() だけを持つ制限インターフェース、状態機械での拒否、モックのコマンド履歴検証（open 1回・close 0回）で固めてください。
7点すべてが設計の穴だった。特に効いたのは C3 だ。関節を補間して動くアームは、始点と終点が安全でも、途中の体が人側へ膨らむことがある。終点だけ見て安全と言っていた。
反映（r3）: PICK は既存本線を無変更のまま、新スクリプト側で組み直す案を採用（重複を正式受容・参照元コミットをコメントに明記）。爪の向きを Unity から sidecar CSV に出す工程を新設。中間姿勢を FK テーブルから選ぶ方式へ変更。台帳を38列に拡充し outcome を排他1列に。インターロック2段（待機開始と退避開始の両方に人の確認）を導入。close 禁止の範囲を handover phase に限定して制限インターフェースで封じた。
レビュー第2ラウンド（r3 → r3.1）
第2ラウンドで前回の7点は「概ね閉じた」判定。代わりに新規必須4点が出た。
実測到達判定が未定義 ❌ — 静止確認が is_moving 相当 + SENDWAIT だけです。is_moving=0 は「目標に到達した」と同義ではありません。指令黙殺・通信異常でも停止状態になり得ます。PRESENT、PRESENT_MID、RETRACT_SMALL、HOME のすべてで、send_angles → 停止待ち → get_angles 実測 → max(abs(actual-target)) <= tolerance を到達条件にしてください。None/-1/timeout/is_moving=-1 は未到達です。PRESENT 未到達時は「手を出してください」を表示せず中断。
物理電源断は E-stop と確定できない ❌ — 電源断を E-stop 相当とする記述は撤回してください。電源断は関節保持力を失わせ、姿勢によってはアームや保持物が重力で落ちる可能性があります。設計自身も release_all_servos の脱力を危険と判断しており、ここが矛盾します。「緊急時の最終的な電源遮断手段。安全認証された E-stop とは呼ばない」に改め、非常停止方法が確定するまで人を可動域に入れる試験は不可です。
マウントガードの時点分岐 ❌ — 未マウントでも物理動作を継続、は起動時から未マウントの場合にも読めます。試行開始前に未マウント: 物理動作を一切開始せず abort。移動前に mount lost: 新規シーケンスを開始せず abort。人が既に手を出した後: 記録より物理安全を優先し、インターロックを維持して安全シーケンスを完了。/tmp は正本 fallback ではなく emergency journal と明記。「通し試行は必ず1行」は SSD 物理消失時には同期的に保証できません。
RELEASE 応答と物理開放を分ける ⚠️ — result=-1 は物理成否を表しません。release_command_ok は「通信上、指令送信処理を完了した」の意味に限定し、release_observed（open/not_open/unknown）と release_observed_source を別に持ってください。
2点目は自分の設計の内部矛盾を突かれた。release_all_servos の脱力落下を危険と書いた同じ文書で、電源断を安全装置として扱っていた。電源を切れば脱力する。同じことだ。
反映（r3.1）: 全移動に実測角ベースの到達判定。電源断は「最終手段」に格下げし、非常停止方法の確定を別タスク化。マウントガードを3時点で分岐。release の通信成功と物理開放を別列に分離。台帳は44列になった。
レビュー第3ラウンド（r3.1 → r3.2）
r3.1 は前回指摘を正しく反映しており、設計はほぼ実装可能です。ただし、到達待ちの「有界化」と「再送禁止」を明記してから発注してください。
到達待ちを有界化する ❌ — 現状は待機上限が未定義。送信前実測から移動量を算出し、移動ごとに deadline を決める（monotonic 基準）。deadline 超過は motion_failed / arrival_timeout。is_moving 待ちにも上限を。人が退避した後とはいえ、停止確認が永久待ちになると手動復帰判断ができません。
handover phase は再送禁止 ❌ — 既存の _send_and_wait_arrival() は、送信無視と判断すると最大1回再送します。これをそのまま handover phase へ流用してはいけません。特に RETRACT の遅延再送は、操作者が「失敗して止まった」と認識した後に動く危険があります。各 waypoint につき send_angles 1回のみ。新しい send_once_and_wait_arrival 相当を handover 側に置く。PICK phase は parity 維持のため既存挙動のままで構いません。
2点目の指摘が一番怖い種類のものだった。「止まった」と人が思った数秒後に、遅れて届いた再送でアームが動く。既存部品をそのまま持ってくると、置くタスクでは便利だった再送が、人のいるタスクでは凶器になる。
反映（r3.2）: 移動ごとの deadline・is_moving ポーリング上限・再送なしの送角関数を新設。Ctrl-C のモック注入点を8箇所（待機中・移動中・確認入力中・RELEASE 前後・撮影中・台帳追記直前・緊急ログ書き込み中）に拡張。到達許容値は姿勢別に確定（PRESENT 系 3.0°・HOME は既存 2.5° 維持・設定ファイルのみが正で CLI から緩和不可）。
レビュー第4ラウンド（r3.2 → r3.3・設計確定）
r3.2 は前回の3条件を正しく満たしています。ただし、実装発注前に「VLM 実行タイミング」を1点だけ修正してください。人の手が可動域内にある間に秒単位の API 待ちを入れないためです。
シーケンスは「RELEASE → 撮影 → インターロック2」ですが、撮影後いつ API を呼ぶかがありません。このままだと、手が可動域内にある状態で秒単位の VLM API 待ちを行う実装になり得ます。RELEASE・撮影（証拠のため現在位置で各1枚・カメラごと短い deadline）→ インターロック2 → 退避 → HOME → VLM 事後判定 → 対話入力 → 台帳追記、の順序へ固定してください。VLM は参考値なので、物理安全シーケンスより後で問題ありません。
実装レビューでは、deadline を確認するだけでなくブロッキング中の各 RPC に min(remaining, per_call_cap) のタイムアウトが効くこと、算出 deadline にコード側のハード上限があることを確認してください。
最終判断: 設計の安全・制約面 ✅。VLM を退避後へ移す一文と RPC ハード上限を追記後、実装発注 ✅。r3.3 全面改稿は不要。
反映（r3.3）: VLM の実行を HOME 復帰後に固定。撮影と API 呼び出しを時間的に分離。RPC 単位のタイムアウトと deadline のハード上限を追記。緊急ログは原子的書き込み（tmp 作成 → fsync → os.replace）に。ここで設計レビューが完了した。
4ラウンドを通して設計書に積まれた安全装置を並べると、こうなる。
層 | 中身 |
|---|---|
姿勢 | FK テーブル実在姿勢のみ・爪の向きと全リンク張り出しを Unity で事前検証 |
軌道 | swept volume を Unity で確認 → 人なし実機で実軌道確認 → ダミー手 → 補助者同席の順 |
移動 | 実測角ベースの到達判定・deadline で有界化・再送禁止・未到達なら人を呼ばない |
グリッパー | handover phase は open 専用インターフェース・close 要求は状態機械破損として中断 |
人の確認 | 待機開始と退避開始の2段インターロック・補助者が可動域外のキーボードで押す |
中断 | Ctrl-C は新規指令を出さないだけ（安全停止ではないと明記）・自動でどこへも動かない |
記録 | 44列台帳・マウント喪失は3時点で分岐・人がいる間は記録より物理安全を優先 |
一番大きい学びは、レビューの視点の違いそのものだった。設計者は「正しく動く流れ」を書く。レビューは「止まったとき・遅れたとき・嘘の停止のとき」を見る。人がそばにいるタスクでは後者が本体で、4ラウンドの指摘はほぼ全部そこに集中していた。
管理文書の更新2件
実装より先にルールを更新した（保存先を新設するなら、まずテーブルを更新してから実装する、という既存原則に従う）。
MASTER の保存先テーブルへ2行追加。
| 受け渡しの通し試行（PICK→提示→開放の1試行） | runs/run_handover/{run_stem}/ |
| 受け渡しの通し台帳（横断索引） | runs/handover_ledger_v1.csv |data_manifest_rules.md へ handover_ledger_v1 の節を新設。台帳の実体・44列の定義・追記ガード・マウント喪失時の緊急ログと復旧手順・状態語（trial_status / outcome / outcome_source）を規定した。列定義は設計由来のドラフトと明記し、実装後に実体ヘッダを実測して確定転記する。数字を文書に焼いて実体と乖離した過去の反省を、新しい台帳には最初から織り込んである。
実装の最初の3点
設計確定後、実装に着手した。順序はゲート制で、提示姿勢が Unity 検証で確定するまで、アームを動かすコードには入らない。
実装物A: 候補抽出スクリプト（コミット a1236b3）。FK テーブルを読み取り専用で開き、マット手前側・左右中央・高さ3帯・関節安全範囲で一次絞り込みする。読み取り専用の担保（テーブルへの書き込み経路が存在しないこと）と、マウント確認が「パスがあるか」ではなく「本当にマウントされて書けるか」を見ていることを確認できるよう、コードを全量載せる。
#!/usr/bin/env python3
"""受け渡し提示姿勢の候補抽出 (DAY288 実装物 A・読み取り専用).
設計書 DAY288_handover_step1_design_r3.3 の 3節「提示姿勢の決め方」に沿って,
FK テーブル `fk_scan_motion_based_v2.csv` から一次絞り込みを行う.
**読み取り専用**: FK テーブルへの書き込み・追加スキャン・上書きは一切しない
(fk_scan_motion_based_v2.csv は恒久固定名の読み取り専用資産).
一次絞り込み条件:
- ee_z > +0.15 (マット手前側 150mm 以深)
- |ee_x| < 0.05 (左右中央 ±50mm)
- ee_y 3 帯 (0.100 / 0.150 / 0.200m 各 ±0.015)
- J4_SAFE [-90, +60] (j4 の安全帯)
- j1 <= 88 (実機上限・88 超は黙殺される)
出力: /Volumes/KIOXIA/r2r2r-data/diag/present_candidates_DAY{NNN}.csv
列: band_label, selected_idx, fk_table_basename, fk_table_md5,
ee_x, ee_y, ee_z, j0, j1, j2, j3, j4, j5
stdout に帯ごとの候補件数を表示.
Usage:
python3 extract_present_candidates.py --day 288
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import os
import sys
from pathlib import Path
# ---- 定数 --------------------------------------------------------------------
FK_TABLE = Path("/Users/shimazueiichi/r2r2r-scripts/fk_scan_motion_based_v2.csv")
DIAG_ROOT = Path("/Volumes/KIOXIA/r2r2r-data/diag")
KIOXIA_MOUNT = Path("/Volumes/KIOXIA")
EE_Z_MIN = 0.15
EE_X_ABS_MAX = 0.05
J4_SAFE_MIN = -90.0
J4_SAFE_MAX = 60.0
J1_MAX = 88.0
# ee_y 3 帯 (中心値 ± 0.015m).
BANDS = [
("y100", 0.100, 0.015),
("y150", 0.150, 0.015),
("y200", 0.200, 0.015),
]
OUT_COLS = ["band_label", "selected_idx", "fk_table_basename", "fk_table_md5",
"ee_x", "ee_y", "ee_z", "j0", "j1", "j2", "j3", "j4", "j5"]
def ensure_ssd_mounted() -> bool:
"""KIOXIA が真マウントされて書き込み可能か.
- os.path.ismount で shim (単なるパス存在) を弾く
- diag ルートへの一時書き込みテストで write 可否を確認
"""
try:
if not os.path.ismount(str(KIOXIA_MOUNT)):
return False
except Exception:
return False
try:
DIAG_ROOT.mkdir(parents=True, exist_ok=True)
test = DIAG_ROOT / ".present_candidates_write_test"
test.write_text("x")
test.unlink()
return True
except Exception:
return False
def md5_file(path: Path) -> str:
h = hashlib.md5()
with path.open("rb") as f:
for chunk in iter(lambda: f.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
def _in_band(ee_y: float, center: float, half: float) -> bool:
return (center - half) <= ee_y <= (center + half)
def extract(fk_path: Path, band_label: str, band_center: float, band_half: float):
"""band を1つ受け取り, 通過した行を (row_idx, dict) の generator で返す."""
with fk_path.open("r", newline="") as f:
r = csv.DictReader(f)
for idx, row in enumerate(r):
try:
ee_x = float(row["ee_x"])
ee_y = float(row["ee_y"])
ee_z = float(row["ee_z"])
j1 = float(row["j1"])
j4 = float(row["j4"])
except (KeyError, ValueError):
continue
if ee_z <= EE_Z_MIN:
continue
if abs(ee_x) >= EE_X_ABS_MAX:
continue
if not _in_band(ee_y, band_center, band_half):
continue
if not (J4_SAFE_MIN <= j4 <= J4_SAFE_MAX):
continue
if j1 > J1_MAX:
continue
yield idx, row
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--day", required=True, type=int,
help="DAY 番号 (出力ファイル present_candidates_DAY{NNN}.csv に反映)")
args = ap.parse_args()
if not FK_TABLE.exists():
print(f"[ABORT] FK テーブルが見つからない: {FK_TABLE}", file=sys.stderr)
return 2
if not ensure_ssd_mounted():
print(f"[ABORT] KIOXIA 未マウント (ismount + write test 失敗): {KIOXIA_MOUNT}",
file=sys.stderr)
return 2
print(f"[INIT] FK テーブル: {FK_TABLE}")
fk_md5 = md5_file(FK_TABLE)
fk_basename = FK_TABLE.name
print(f"[INIT] FK md5: {fk_md5}")
out_path = DIAG_ROOT / f"present_candidates_DAY{args.day:03d}.csv"
print(f"[INIT] 出力先: {out_path}")
band_counts = {}
total_written = 0
with out_path.open("w", newline="") as f:
w = csv.writer(f)
w.writerow(OUT_COLS)
for band_label, center, half in BANDS:
n = 0
for idx, row in extract(FK_TABLE, band_label, center, half):
w.writerow([
band_label,
idx,
fk_basename,
fk_md5,
row["ee_x"], row["ee_y"], row["ee_z"],
row["j0"], row["j1"], row["j2"],
row["j3"], row["j4"], row["j5"],
])
n += 1
band_counts[band_label] = n
total_written += n
print(f"[BAND {band_label}] center={center:.3f}m ±{half:.3f}m → {n} 件")
print()
print("=" * 60)
print(f"合計: {total_written} 件を出力")
print(f"内訳: " + " / ".join(f"{k}={v}" for k, v in band_counts.items()))
print(f"出力: {out_path}")
print("=" * 60)
return 0
if __name__ == "__main__":
sys.exit(main())実行結果の全量。
FKテーブル: fk_scan_motion_based_v2.csv
md5: 6f069cd5aad5d30fa71c5a2b5cbb0167
帯 中心(m) ±半幅(m) 件数
y100 0.100 0.015 4,992
y150 0.150 0.015 3,465
y200 0.200 0.015 1,751
合計 10,208
出力: /Volumes/KIOXIA/r2r2r-data/diag/present_candidates_DAY288.csv
（10,209行 = ヘッダ + 10,208データ・列13）
禁止動詞 grep 全0: send_coords / release_all_servos / send_angles / set_gripper3帯とも候補は十分にある。テーブルの手前側領域が疎で追加スキャンが要る事態は回避できた。
実装物B: Unity 派生スキャナ手順書（コミット ee2ea18）。候補 CSV を Unity で再生して爪の向きベクトルと全リンクの張り出しを sidecar CSV に出す手順、swept volume の確認手順、到達許容 3.0° が爪先位置で何 mm になるかの換算確認までを、初見で実行できる粒度で文書化した。Unity 操作は人間側の作業として残る。
実装物C: 台帳初期化（コミット d4e2984）。44列の定義を文書ではなくコードの定数1箇所に持たせ、後続モジュールが import で共用する構造にした。その定数の実体と、既存台帳を絶対に上書きしない初期化の作りを確認できるよう、コードを全量載せる。
#!/usr/bin/env python3
"""受け渡し台帳 handover_ledger_v1.csv の 44 列定義 + 初期化スクリプト.
r3.3 §6「台帳 handover_ledger_v1.csv（44列）」を実装として正とする.
実装後に data_manifest_rules.md 9節の型・細目を実測から確定転記する.
このモジュールの役割:
1. HANDOVER_LEDGER_COLS を唯一の正として提供
(D handover_release.py / E run_pick_handover.py が import する)
2. --init CLI で 44 列ヘッダの空 CSV を新規作成 (既存あれば ABORT・上書き禁止)
追記ガード / mount lost 時の emergency journal / recovery の実装は E に持たせる
(本モジュールは定数と初期化のみ・append 経路は E から HANDOVER_LEDGER_COLS を
import して csv.DictWriter で書く).
Usage:
python3 handover_ledger.py --init
"""
from __future__ import annotations
import argparse
import csv
import os
import sys
from pathlib import Path
# ---- 定数 --------------------------------------------------------------------
LEDGER_PATH = Path("/Volumes/KIOXIA/r2r2r-data/runs/handover_ledger_v1.csv")
KIOXIA_MOUNT = Path("/Volumes/KIOXIA")
# r3.3 §6 の 44 列. 列名・順序を実装の唯一の正とする.
# 型・語彙のコメントは r3.3 の該当行から転記 (「実装後に data_manifest_rules.md
# へ確定転記」の材料).
HANDOVER_LEDGER_COLS = [
"timestamp", # 1 ISO8601
"run_stem", # 2 run_handover/ フォルダ名と一致・recovery キー
"trial_no", # 3 int
"trial_status", # 4 completed / pick_failed / motion_failed /
# release_failed / interrupted / mount_lost /
# unexpected_error
"manual_recovery_required", # 5 bool
"abort_stage", # 6 中断段階 (完走時は空)
"abort_reason", # 7 forbidden_gripper_close / arrival_timeout /
# ... (完走時は空)
"pick_verdict", # 8 v3 と同語彙
"pick_reason", # 9
"pick_confidence", # 10
"block_pick_x", # 11 float
"block_pick_z", # 12 float
"align_j0_final", # 13 float
"align_converged", # 14 bool
"pick_pose_id", # 15
"corr_cell_key", # 16
"corr_applied_dx", # 17 float
"corr_applied_dz", # 18 float
"handover_pose_id", # 19 handover_present_pose.json の pose_id
"object_profile_id", # 20 run JSON に内包
"wait_plan_id", # 21 N 計画識別子
"wait_plan_index", # 22 int (計画内の提示順)
"shuffle_seed", # 23 int
"wait_sec_requested", # 24 float 要求 N
"wait_sec_actual", # 25 float 実測 N (monotonic)
"wait_started_at", # 26 WAIT 開始時刻
"release_sent_at", # 27
"release_responded_at", # 28
"release_command_ok", # 29 bool 通信上の送信完了 (物理開放と分離)
"release_observed", # 30 open / not_open / unknown
"release_observed_source", # 31 operator / top_camera / side_camera
"present_reached_bool", # 32 実測角ベース到達判定
"retreat_completed_bool", # 33
"home_reached_bool", # 34
"hand_clear_confirmed", # 35 インターロック2 の確認入力
"outcome", # 36 taken / drop / stuck / unknown /
# not_applicable (排他)
"outcome_source", # 37 operator_visual /
# operator_with_vlm_reference / unknown
# (r3.3 で vlm_post は廃止)
"vlm_taken_verdict", # 38 参考値
"vlm_status", # 39 ok / http_400_no_credit / error
"vlm_reason", # 40
"top_capture_status", # 41 ok / fail:{reason}
"side_capture_status", # 42
"top_capture_path", # 43 fail 時は空
"side_capture_path", # 44 fail 時は空
]
assert len(HANDOVER_LEDGER_COLS) == 44, (
f"HANDOVER_LEDGER_COLS len={len(HANDOVER_LEDGER_COLS)} != 44 "
f"(r3.3 §6 と不一致・実装の唯一の正が崩れる)")
def ensure_ssd_mounted() -> bool:
"""KIOXIA が真マウントされて書き込み可能か.
os.path.ismount で shim を弾き, LEDGER_PATH の親ディレクトリへの
一時書き込みテストで write 可否を確認する.
"""
try:
if not os.path.ismount(str(KIOXIA_MOUNT)):
return False
except Exception:
return False
try:
LEDGER_PATH.parent.mkdir(parents=True, exist_ok=True)
test = LEDGER_PATH.parent / ".handover_ledger_write_test"
test.write_text("x")
test.unlink()
return True
except Exception:
return False
def init_ledger() -> int:
"""44 列ヘッダのみの空 CSV を新規作成. 既存があれば ABORT."""
if not ensure_ssd_mounted():
print(
f"[ABORT] KIOXIA 未マウント (ismount + write test 失敗): {KIOXIA_MOUNT}",
file=sys.stderr,
)
return 2
if LEDGER_PATH.exists():
# 既存の全貌を短く報告してから ABORT (人間の判断材料).
try:
with LEDGER_PATH.open("r", newline="") as f:
existing = list(csv.reader(f))
n_rows = len(existing)
n_cols = len(existing[0]) if existing else 0
print(
f"[ABORT] 既存台帳あり (上書き禁止): {LEDGER_PATH}",
file=sys.stderr,
)
print(
f" 現在: {n_rows} 行 / ヘッダ {n_cols} 列",
file=sys.stderr,
)
except Exception as e:
print(
f"[ABORT] 既存台帳あり (読取失敗: {e}): {LEDGER_PATH}",
file=sys.stderr,
)
return 2
# ヘッダを書いて即検証.
with LEDGER_PATH.open("w", newline="") as f:
w = csv.writer(f)
w.writerow(HANDOVER_LEDGER_COLS)
with LEDGER_PATH.open("r", newline="") as f:
rows = list(csv.reader(f))
if len(rows) != 1 or rows[0] != HANDOVER_LEDGER_COLS:
print(
f"[ABORT] 初期化検証失敗: rows={len(rows)} "
f"header_match={rows[0] == HANDOVER_LEDGER_COLS if rows else False}",
file=sys.stderr,
)
return 3
print(f"[OK] 初期化完了: {LEDGER_PATH}")
print(f" 44 列ヘッダのみ (0 データ行)")
print(f" 先頭 5 列: {HANDOVER_LEDGER_COLS[:5]}")
print(f" 末尾 5 列: {HANDOVER_LEDGER_COLS[-5:]}")
return 0
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
ap.add_argument(
"--init",
action="store_true",
help="handover_ledger_v1.csv を新規作成 (既存があれば ABORT)",
)
args = ap.parse_args()
if args.init:
return init_ledger()
ap.print_help()
return 1
if __name__ == "__main__":
sys.exit(main())実行結果の全量。
実体: /Volumes/KIOXIA/r2r2r-data/runs/handover_ledger_v1.csv
行数: 1（ヘッダのみ・0データ行）
列数: 44
md5: 5f356b6806f26b6ec2fd4db9b9acde98
二重init検証: [ABORT] 既存台帳あり（上書き禁止）+ exit=2
44列は定数 HANDOVER_LEDGER_COLS（assert len==44）として1箇所に持ち、
後続モジュールは import で共用する。
既存台帳 pick_place_ledger_v3.csv: md5 91dac34289c6701e3b530c15315f2f78 不変
run_pick_place.py / PLACE系 / fk_lookup.py: git diff ゼロ列定義をコードの定数1箇所に集約して文書側をドラフト扱いにしたのは、文書とコードの列数がずれて気づけなかった過去への対策で、新台帳では初日から適用した。
ここからの順序
Unity 手順書の実行（爪の向き軸の確定 → sidecar CSV → 3候補 → 1確定）
→ 制御モジュール実装（開く専用・再送なし・有界待ち）
→ 通しスクリプト実装（PICK組み直し・インターロック・台帳追記）
→ モックテスト（PICK一致検証・Ctrl-C 8注入点・close 0回）
→ 実装コードの外部レビュー
→ 人なし実機 → ダミー手 → 補助者同席の限定試験人を可動域に入れる試験は、非常停止方法が別タスクで確定するまで行わない。ここはレビューで固定された順序であり、崩さない。
