mechArm 270 Pi 受け渡し動作の成立を探る Log#3 | 人に手渡す姿勢の決め方と、検証コードが検証になっていなかった話
ロボットアームが掴んだブロックを、人の手に差し出す。差し出したところで止まる、その一つの姿勢を決める回。やることはそれだけのはずだった。姿勢というのは 6 つの数字（各関節の角度）でしかない。だが、その 6 つを勘で書くと、実機が爪先をどこに持ってくるか分からない。だから過去に Unity で測った実測値の対応表（FK テーブル）から選ぶ、という前提で進めていた。
その前提が、最初のコマンドで崩れた。
差し出す姿勢を FK テーブルから引こうとして、1 行も無かった
受け渡しの基準姿勢は HOME、つまり全関節が [0, 0, 0, 0, 0, -30] の姿勢から始まる。まず HOME の爪先が空間のどこに来るかを FK テーブルから引こうとした。FK テーブルは 170 万行ある。その中に HOME があるはずだと思っていた。
$ awk -F, '$4==0 && $5==0 && $6==0 && $7==0 && $8==0 && $9==-30' \
~/r2r2r-scripts/fk_scan_motion_based_v2.csv | head
$何も返らない。列の順番を疑って確認した。
$ head -1 ~/r2r2r-scripts/fk_scan_motion_based_v2.csv
ee_x,ee_y,ee_z,j0,j1,j2,j3,j4,j5
$ head -3 ~/r2r2r-scripts/fk_scan_motion_based_v2.csv
ee_x,ee_y,ee_z,j0,j1,j2,j3,j4,j5
0.231374,0.106889,0.22751,-37,95,-49,18.5,-70,-30.5
0.232657,0.093842,0.232371,-37,95,-49,18.5,-65,-30.5列順は想定どおり（ee_x, ee_y, ee_z, j0, j1, j2, j3, j4, j5）。データを見ると j1 が 95、j3 が 18.5 のような半端な値が並んでいる。基準姿勢から ±15° の格子でスキャンした結果で、格子の目は基準姿勢の値に依存している。HOME の [0,0,0,0,0,-30] がその目の上に乗っている保証はどこにもなかった。
近傍を探しても空だった。
$ awk -F, 'NR>1 && ($4>-2&&$4<2) && ($5>-2&&$5<2) && ($6>-2&&$6<2) && ($7>-2&&$7<2) && ($8>-2&&$8<2) && ($9>-33&&$9<-27)' ~/r2r2r-scripts/fk_scan_motion_based_v2.csv | head
$j1 の分布を見て、理由が確定した。
$ awk -F, 'NR>1{print $5}' ~/r2r2r-scripts/fk_scan_motion_based_v2.csv | sort -n | head -1
20.17
$ awk -F, 'NR>1{print $5}' ~/r2r2r-scripts/fk_scan_motion_based_v2.csv | sort -n | tail -1
125j1（肩の関節）の範囲は 20.17〜125。HOME の j1=0 は範囲外。この FK テーブルは PICK と PLACE のために作ったもので、その領域は j1 が 87〜95 あたりに集中している。腕が前に伸びて爪が下を向く姿勢の集まりだ。HOME 付近（腕が起き上がった姿勢）は、測る理由がなかったのでスキャンしていない。
FK テーブルから HOME 系の姿勢は引けない。ここで、LOG288 に用意していた候補 CSV（present_candidates_DAY288.csv・FK テーブル由来の 10,208 件）も使えないことが確定する。あの候補はすべて j1=87〜95 の PICK 領域の延長で、人に差し出す姿勢としてそもそも向いていなかった可能性が高い。
HOME を基準に ±15° 格子を 16,807 点スキャンして 4 姿勢を選んだ
FK テーブルに無いなら作る。ただし FK テーブル本体は読み取り専用（3 か所で md5 一致を管理している恒久固定ファイル）なので触らない。HOME を基準姿勢とする ±15° の格子スキャンを、Unity の FKScanner で新規に走らせた。
入力は 1 行だけの CSV。
timestamp,label,J1,J2,J3,J4,J5,J6
0,HOME,0,0,0,0,0,-30Unity で MotionLog モードを回すと、Console にこう出た。
[11:59:37] [FKScanner_MotionBased] 完了。16807点 → /Users/shimazueiichi/r2r2r-scripts/home_probe_DAY289_ee.csv1 行を入れたのに 16,807 点。これは MotionLog（1 姿勢再生）ではなく格子スキャンモードで動いた数字で、HOME を基準に各関節を ±15°（刻み 5°）振った全組み合わせ（7^5 に近い展開）がそのまま出た。意図とは違ったが、害はない。FK テーブル本体には書いていないし、新規ファイルに出ている。むしろ HOME 付近の姿勢が 16,807 点得られたことになる。
FK テーブルとの md5 一致も確認した。スキャンで読んだ FK テーブルが正しいものであることの記録として残す。
$ md5 ~/r2r2r-scripts/fk_scan_motion_based_v2.csv
MD5 (/Users/shimazueiichi/r2r2r-scripts/fk_scan_motion_based_v2.csv) = 6f069cd5aad5d30fa71c5a2b5cbb0167
$ md5 /Volumes/KIOXIA/r2r2r-data/fk_tables/fk_scan_motion_based_v2.csv
MD5 (/Volumes/KIOXIA/r2r2r-data/fk_tables/fk_scan_motion_based_v2.csv) = 6f069cd5aad5d30fa71c5a2b5cbb0167
$ md5 /Users/shimazueiichi/Robotics_UX/Robotics_UX_New/Assets/Data/fk_scan_motion_based_v2.csv
MD5 (/Users/shimazueiichi/Robotics_UX/Robotics_UX_New/Assets/Data/fk_scan_motion_based_v2.csv) = 6f069cd5aad5d30fa71c5a2b5cbb01673 か所とも一致。この 16,807 点から HOME を引く。
$ awk -F, 'NR>1 && $4==0 && $5==0 && $6==0 && $7==0 && $8==0 && $9==-30' ~/r2r2r-scripts/home_probe_DAY289_ee.csv
-0.002874,0.232396,0.267074,0.000,0.000,0.000,0.000,0.000,-30.000HOME の爪先が実測で出た。単位を mm に直すと、ee_x = -2.9mm（左右ほぼ中央）、ee_y = 232.4mm（床からの高さ）、ee_z = +267.1mm（Z+ は人が立つ側）。HOME は既に人の方を向いていた。
ここで一つ、思い込みを訂正することになった。作業の途中で「HOME は腕が真上に伸びた姿勢で、爪先はベースの真上・高さ 270mm」と口にしていた。だが実測は高さ 232mm、人側へ 267mm。真上でもベースの上でもなかった。数字は測れば出る。測る前に言った数字は、当たっていなかった。
差し出す姿勢（PRESENT）、その手前の中間姿勢（PRESENT_MID）、開放後に少し引く姿勢（RETRACT_SMALL）を、この 16,807 点から選ぶ。条件は j0=0・j3=0・j5=-30 で固定（左右にぶれず、同じ垂直面の中で動く）、ee_x が HOME と揃う（-2.9mm 付近）。絞り込みの awk を順に振って、4 姿勢が揃った。
j0 j1 j2 j3 j4 j5 ee_mm (x, y, z)
HOME [ 0, 0, 0, 0, 0, -30] -2.9, 232.4, 267.1
PRESENT_MID [ 0, -15, 10, 0, -15, -30] -2.9, 293.8, 234.0
PRESENT [ 0, 15, -15, 0, -5, -30] -2.9, 242.7, 293.2
RETRACT_SMALL [ 0, 5, -5, 0, -5, -30] -2.9, 245.9, 276.0動きの筋はこうなる。HOME から一度上に持ち上げて（MID・高さ 294mm）、そこから下ろしながら人側へ差し出し（PRESENT・人側 293mm）、開放したら少し引く（RETRACT・人側 276mm）、HOME へ戻る。各区間の関節差分は最大 30°、すべて手首の安全域 J4_SAFE [-90, +60] の内側に収まっている。
選んだ姿勢を「仮」と記録して、設計書を r3.4 に改訂した
4 姿勢を JSON に固めた。ただしこの姿勢は仮だ。設計書 r3.3 は、姿勢を選ぶのに二段階の絞り込みを要求していた。一次（高さで絞る）は済んでいたが、二次（爪が人の方を向いているかの内積判定・爪の開き方向・腕が人側へ張り出さないか・床とのクリアランス）をやっていない。動作中に腕が人の側へ膨らまないかを確認する swept volume の検証もしていない。
handover_present_pose.json の validation_status に、その事実をそのまま書いた。
"validation_status": {
"sidecar_secondary_filter": "not_applied",
"swept_volume_visual_check": "pending",
"real_machine_no_human": "pending",
"note": "設計書 r3.4 3-1 の二次絞り込み(tool_forward・[0,0,1] >= cos30° / 開口方向の許容 / 全リンク Z+ 張り出しが爪先を超えない / 床・ベースクリアランス > 30mm)は未実施。本 JSON は仮の提示姿勢。最終合否は swept volume 目視(3-3 計画段階)と人なし実機の実軌道確認(3-3 実測段階)で決める。実軌道確認が済むまでダミー手工程へ進まない。"
}姿勢の出典も設計書の想定と違った。r3.3 は FK テーブル由来の候補集合から選ぶ前提だったが、実際は HOME 基準の新規スキャンから選んだ。この経路変更を設計書に残さないと、後で読んだ人（次の自分、次の AI）が「候補 CSV から選んだ」と誤解する。設計書を r3.4 に改訂し、3-5 節を新設して、実測で分かったこと・実際にやったこと・やっていないことを表にした。
そのうえで判断が一つあった。姿勢の検証（二次絞り込み・swept volume）を先に済ませてから実装に入るか、仮の姿勢のまま実装を進めるか。実装を進める方を選んだ。理由は、実装するコード（グリッパー制御・到達判定・インターロック）は姿勢がどこにあっても同じで、姿勢を差し替えても pose JSON を更新すればコードは変わらないから。ただし「人が手を出す試験の前に、二次絞り込みと swept volume を消化する」義務は設計書に固定した。仮のまま実機に人を近づけることはしない。
parity mock が run() を走らせていなかった
ここから実装の話になる。受け渡しのコードは、既にある PICK（掴む）の処理を再利用する。ただし既存の run_pick_place.py（1300 行超の巨大関数）をコピーするのではなく、その中の下位関数を呼び出して PICK を新しく組み直す方式を取った。コピーと違って、既存本線と新本線がズレる危険がある。そのズレを防ぐのが parity mock ――「既存本線と新本線が、同じ入力に対して同じ下位関数を同じ順序・同じ引数で呼ぶ」ことを検証するテストだ。これは設計書が実装条件として必須指定していた。
最初に上がってきた parity mock は、テストとして通っていた。6 ケース全部パス。だが中身を確認すると、既存本線 run() を一度も実行していなかった。新本線を走らせて、その結果を「こうなるはず」と手で書いた期待値と比べているだけだった。
これは parity ではない。「既存本線と一致」を確かめるはずが、「新本線が自分の想定どおり動く」を確かめていた。比べる相手（run()）がそこにいなかった。
書き直しの方針を決めるために、run() を PICK だけ実行して止める手段があるか調べた。run() は PLACE まで実機 I/O 込みで走る関数で、途中で止める注入点が無い。だが調査の結果、dry_run=True で走らせれば送角の値は計算されてから送信だけスキップされる（値そのものは記録に出る）こと、PLACE に入る直前に観測用の hook があることが分かった。既存コードを一切変えずに、run() を実際に走らせて PICK 区間だけ trace を取れる。
走らせたら写し違いが 3 件出た
parity mock を「run() を dry_run で実行して、新本線と下位関数の呼び出し列を突き合わせる」形に書き直した。走らせた結果、静的な期待値比較では見えなかった写し違いが 3 件出た。
一件目が一番効いた。run() は PICK 開始時に、まずグリッパーを開いてから検出に入る。新本線はこの「開く」を飛ばしていた。前の試行が掴んだまま失敗して、次の試行を始めると、開かれないまま腕が降りて、掴んだブロックを持ったまま次の降下に入る。実機で事故になり得た。常駐ループには「前試行が把持中に失敗した場合、次試行の gripper open で空中落下し得る」という警告を既に表示していた。その警告は、この open があって初めて意味を持つ。open が無ければ、警告と実装が食い違っていた。
二件目は検出のリトライ。run() は検出を複数回試す。新本線は 1 回で諦めて中断していた。実機ではノイズで 1 回目が外れることは普通にあり、成功率が変わる。
三件目は補正記憶の記録位置のずれで、これは記録だけの問題（実機の動きには影響しない）。
3 件とも、新本線を run() に合わせる方向で直した。テストの期待値を新本線に寄せて通す（＝ズレたまま緑にする）ことはしなかった。書き直した mock は、run() 側と新本線側の送角列を実値で並べて、全行が一致することを確認している。
run() 側 send_angles 列:
approach: (-40.0, 87.7, ...) pick: (-40.0, 92.3, ...) lift: (-40.0, 87.7, ...)
E 側 send_angles 列:
approach: (-40.0, 87.7, ...) pick: (-40.0, 92.3, ...) lift: (-40.0, 87.7, ...)この 3 件は、静的 mock のままなら実機で発覚するまで潜んでいた。しかも一件目は、人が近くにいる工程ではないとはいえ、ブロックを持ったまま予期しない降下をする類の事故だ。テストが緑だったこと自体は、何も保証していなかった。
この回で確定した実装コード（全量）
D-1（visual_servo_pick への per-call timeout 追加）は共有モジュールへの差分のため本文では省き、下の 2 ファイルを全量で残す。再現性のための記録であり、読み物としての体裁より完全性を優先する。それぞれ codex の外部レビューを通過してコミット済み（D-2 = 995aec9、E-1 = fff21c3、E-2 = 287a0c5、E-3 = 520ec37）。
handover_release.py（D-2・RELEASE 系の道具・452 行）
グリッパーを開くこと以外できない制限インターフェース（GripperOpenOnly）、close 要求が来たら中断するゲート、再送しない到達待ち、手順的インターロック。受け渡しフェーズの安全側の部品をここに閉じ込めている。
"""
handover_release.py — DAY289 D-2
r3.4 実装物リストの 4。受け渡しフェーズ (PICK_COMPLETE 以降) の RELEASE 系
一連の道具を提供する。呼び出し元 (E: run_pick_handover.py) が sequence を
組む前提であり、本モジュール自体は sequence を回さない。
提供物:
- GripperOpenOnly : open() のみの制限インターフェース (5節・132行)
- reject_close_request : close 系指令が要求された時の中断ヘルパ (5節・133行)
- send_once_and_wait_arrival: 再送なし・deadline 有界の到達待ち (85行)
- wait_interlock : 手順的保護インターロック (2節・8節)
- load_pose_json : pose JSON 読み取り
- compute_deadline_sec : pose JSON 係数 + コード側ハード上限で二重 clamp
- ArrivalResult / InterlockConfirmation / 例外群
絶対ルール (r3.4 5節・8節):
- handover phase へ公開する gripper 能力を GripperOpenOnly.open() に絞る。
open() 内は daemon_set_gripper(0) を state=0 リテラル固定で呼ぶ。
daemon_set_gripper は open() 内で遅延 import する運用（本モジュール内の
他スコープからは名前解決不能）。完全な構造保証（コード編集者を含む）は
E のモックで gripper 送信履歴が state=0 のみであることを検査して担保する。
- 未到達で自動再送・自動 HOME を行わない (r3.4 85行)。
handover phase は再送禁止。呼び出し元 (E) が trial_status=motion_failed
等を判断して手動復帰へ回す。
- send_angles を 1 回でも呼んだ waypoint は「送信済み」として扱う。
socket.timeout / RuntimeError / OSError のいずれでも再送しない
(codex D-1 レビュー D2 反映)。
- tolerance / deadline 係数は pose JSON のみが正。CLI から緩和不可 (87行)。
- コード側ハード上限 DEADLINE_HARD_CAP_SEC で pose JSON 改ざん耐性を持たせる
(83行)。
安全に関する留意 (呼び出し元がバナー表示する想定・STARTUP_BANNER 参照):
- Ctrl-C は安全停止ではない (daemon 受理済み send_angles は継続しうる)。
- SPEED=20 は人体安全速度ではない。
- インターロック 1/2 の Enter は safety-rated interlock ではない (手順的保護)。
"""
from __future__ import annotations
import json
import math
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional
SCRIPTS_ROOT = Path(__file__).resolve().parents[3]
MODULE_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPTS_ROOT))
sys.path.insert(0, str(MODULE_DIR))
# 汎用 gripper 関数 (daemon_set_gripper / daemon_set_gripper_value) は
# 本モジュールの top-level で import しない。GripperOpenOnly.open の内側で
# 遅延 import することで、他の関数からの参照経路を構造上作らない。
from visual_servo_pick import ( # noqa: E402
SPEED,
daemon_get_angles,
daemon_send_angles,
)
# ---- 定数（コードに焼くのはここだけ・pose 値はコードに焼かない） ----
POSE_JSON_PATH = Path.home() / "r2r2r-scripts" / "handover_present_pose.json"
# DAY266 で確定。send 直後の Pi 側 sleep をカバーする最小 wait。
SEND_WAIT_SEC = 1.0
# 到達判定ポーリング周期。参照実装 j0_sweep_scan L108 の POLL_SLEEP と同値。
POLL_SLEEP_SEC = 0.3
# 1 回の daemon_get_angles RPC の上限秒。Pi 側 3-retry 込みで返る現実的な
# 上限として設定 (実測 200-500ms/回 × 3 ≈ 1.5s の 2 倍マージン)。deadline 全体
# (8-30s) に対し 1 RPC がその大半を食わない値にする狙い。
PER_CALL_CAP_SEC = 3.0
# コード側ハード上限 (r3.4 83行)。pose JSON の max_sec を書き換えても超えられ
# ない絶対上限。現行 JSON max_sec=30.0 と同値。目的は JSON 改ざん耐性 (JSON を
# 1000 にしても 30 で clamp)。人なし実機の実測で伸ばしたい正当な理由が出た
# 場合は、この定数と JSON を両方更新する (二重更新のフリクションが意図)。
DEADLINE_HARD_CAP_SEC = 30.0
STARTUP_BANNER = (
"[handover_release] safety notes:\n"
" - Ctrl-C は安全停止ではない (daemon 受理済み send_angles は継続しうる)\n"
" - SPEED=20 は人体安全速度ではない\n"
" - インターロック 1/2 の Enter は手順的保護 (safety-rated ではない)\n"
" - handover phase では close 系指令を送信しない (state 機械で拒否)\n"
)
# ---- 例外 (呼び出し元が種別で trial_status / abort_reason を決める) ----
class ForbiddenGripperClose(RuntimeError):
"""PICK_COMPLETE 以降で close 系指令が要求された時 (r3.4 5節・133行)。
呼び出し元は
trial_status=unexpected_error / abort_reason=forbidden_gripper_close
として handover sequence を中断する。自動移動なし。手動復帰へ。
"""
class PoseUnreachable(RuntimeError):
"""送信前姿勢が取得できず ABORT (fallback しない)。r3.4 89行。
呼び出し元は trial_status=motion_failed 相当で扱う。
"""
class SendFailed(RuntimeError):
"""daemon_send_angles が例外を投げた。socket.timeout / RuntimeError /
OSError のいずれでも同じ waypoint を再送しない (r3.4 85行 / codex D-1
レビュー D2 反映)。
呼び出し元は trial_status=motion_failed 等で扱う。
"""
# ---- 返り値の型 (呼び出し元 E が run_dir JSON / 台帳に書く) ----
@dataclass
class InterlockConfirmation:
"""Enter 入力の記録。呼び出し元 (E) が run_dir JSON に保存する。"""
label: str # "interlock1_wait_start" / "interlock2_hand_clear"
role: str # "operator" / "assistant"
ts_monotonic: float
ts_wall: float
@dataclass
class ArrivalResult:
"""send_once_and_wait_arrival の返り値。E は arrived を見て次工程を決める。"""
arrived: bool
max_axis_diff_deg: float
last_actual: Optional[list]
ts_send_wall: float
ts_arrival_wall: Optional[float]
deadline_sec: float
reason: str = "" # 未到達時: "arrival_timeout" のみ (ABORT/SEND FAILED は例外)
# ---- pose JSON ローダ ----
def load_pose_json(path: Path = POSE_JSON_PATH) -> dict:
"""pose JSON を読む。md5 検証は行わない (外部の運用手順で担保)。
ファイル自体の md5 は selected_day / selected_date / source.file_md5 と
ともに pose_id 文字列に反映されており、呼び出し元 (E) が台帳へ書く。
"""
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def get_tolerance_deg(pose_data: dict, pose_label: str) -> float:
"""pose JSON の arrival_tolerance_deg から姿勢別 tolerance を取り出す。
CLI 引数からは緩和しない (r3.4 87行・pose JSON のみが正)。
"""
tol = pose_data["arrival_tolerance_deg"][pose_label]
return float(tol)
def compute_deadline_sec(max_delta_deg: float, pose_data: dict) -> float:
"""pose JSON の deadline 係数で clamp、その後コード側ハード上限で二重 clamp。
formula = clamp(base_sec + (max_delta_deg / speed) * factor, min_sec, max_sec)
その上で min(_, DEADLINE_HARD_CAP_SEC) を掛ける。
D-2a ⚠️B1 反映: 全係数の有限性・speed>0・0<min_sec<=max_sec を入口で検証。
最終結果も有限かつ 0<result<=DEADLINE_HARD_CAP_SEC を確認する。違反は
ValueError (pose JSON の不正・プログラミングエラー扱いで daemon_get_angles
の except RuntimeError では握らない)。
"""
d = pose_data["deadline"]
base_sec = float(d["base_sec"])
factor = float(d["factor"])
min_sec = float(d["min_sec"])
max_sec = float(d["max_sec"])
speed = float(d["speed"])
delta = float(max_delta_deg)
if not all(math.isfinite(x) for x in (base_sec, factor, min_sec, max_sec, speed, delta)):
raise ValueError(
"compute_deadline_sec: non-finite input "
f"(base_sec={base_sec}, factor={factor}, min_sec={min_sec}, "
f"max_sec={max_sec}, speed={speed}, max_delta_deg={delta})"
)
if speed <= 0:
raise ValueError(f"compute_deadline_sec: speed must be > 0 (got {speed})")
if not (0 < min_sec <= max_sec):
raise ValueError(
f"compute_deadline_sec: require 0 < min_sec <= max_sec "
f"(min_sec={min_sec}, max_sec={max_sec})"
)
t = base_sec + (delta / speed) * factor
if t < min_sec:
t = min_sec
if t > max_sec:
t = max_sec
# コード側ハード上限で二重 clamp (JSON 改ざん耐性・r3.4 83行)
if t > DEADLINE_HARD_CAP_SEC:
t = DEADLINE_HARD_CAP_SEC
if not (math.isfinite(t) and 0 < t <= DEADLINE_HARD_CAP_SEC):
raise ValueError(
f"compute_deadline_sec: invalid result t={t} "
f"(HARD_CAP={DEADLINE_HARD_CAP_SEC})"
)
return t
# ---- 幾何ヘルパ ----
def _max_axis_diff(a: list, b: list) -> float:
"""max_i |a[i] - b[i]| over 6 joints。長さ != 6 は呼び出し元の責任。"""
return max(abs(float(x) - float(y)) for x, y in zip(a, b))
# ---- GripperOpenOnly (open() のみの制限インターフェース) ----
class GripperOpenOnly:
"""RELEASE 用の制限インターフェース。open() のみ提供、close 系 API 無し
(r3.4 5節・132行)。
handover phase へ公開する gripper 能力を本クラスの open() に絞る。open()
内は daemon_set_gripper(0) を state=0 リテラル固定で呼ぶ。daemon_set_gripper
の import は open() の内側で遅延実行することで、本モジュールの他スコープ
からは名前解決不能。ただし「コード編集者を含む完全な構造保証」ではない
(誰でも import 行を書き加えれば close を呼べる)。完全な保証は E のモック
で gripper 送信履歴が state=0 のみであることを検査して担保する
(D-2a D1 反映・codex 指摘の docstring 表現修正)。
r3.4 55行の「送信後 sleep >= 2秒」は daemon_set_gripper の既定 wait=2 で
満たす (Pi 側の固定 2s 事前 sleep 対策で入っている)。
"""
def open(self) -> None:
"""set_gripper(0) を単発送信。timeout_override は渡さない。
gripper 系 RPC への timeout_override 配線は D-1 で明示的にスコープ外と
した (daemon_set_gripper / daemon_set_gripper_value に引数追加なし)。
wait=2 込みで daemon 側 socket timeout は約 5s、実運用の RELEASE は
1-2 秒で返る想定。
"""
from visual_servo_pick import daemon_set_gripper # noqa: E402 # 遅延 import で他スコープからの名前解決を遮断
daemon_set_gripper(0, speed=100, wait=2)
# ---- close 系拒否ゲート (呼び出し元 E から明示的に呼ばれる想定) ----
def reject_close_request(context: str = "") -> None:
"""close 系指令が要求されたことを検知した時に呼ぶ。ForbiddenGripperClose
を上げ、呼び出し元は trial_status=unexpected_error /
abort_reason=forbidden_gripper_close で handover を中断する (r3.4 5節・
133行)。自動移動・自動 HOME なし。「ログして継続」は禁止。
引数 context は例外メッセージへ添付するだけで挙動には影響しない。
"""
msg = "gripper close is forbidden in handover phase"
if context:
msg = f"{msg} (context={context})"
raise ForbiddenGripperClose(msg)
# ---- 到達待ち (再送なし) ----
def send_once_and_wait_arrival(
target: list,
pose_label: str,
pose_data: dict,
*,
observer: Optional[Callable[[Optional[float], Optional[float], int, bool], None]] = None,
) -> ArrivalResult:
"""target を単発 daemon_send_angles 1 回で送信し、pose JSON deadline 内で
max|Δ| <= tolerance の到達を確認する。再送しない (r3.4 85行)。
D-2a F1 反映: tolerance は公開引数から受けない (r3.4 87行「pose JSON のみが
正・CLI から緩和不可」)。pose_label で pose JSON から get_tolerance_deg
経由で引く。pose_label は log 表示にも使う (JSON キーと log 表示を分ける
必要がないため統一)。呼び出し例:
send_once_and_wait_arrival([0,-15,10,0,-15,-30], "PRESENT_MID", pose_data)
D-2a B2/B3 反映: deadline_at は send の前に作る。send にも per-call cap を
渡す (min(remaining, PER_CALL_CAP_SEC))。send + settle + N ポーリングの
合計上限が deadline_sec を超えないことを保証する (r3.4 83-84行「ブロッキング
する各 RPC に per-call cap」に完全準拠)。
未到達 (deadline 超過) は ArrivalResult(arrived=False, reason="arrival_timeout")
として返す。SendFailed / PoseUnreachable は例外で上げる (呼び出し元が別扱い
にできるよう区別する)。
handover phase の再送禁止 (r3.4 85行) を構造で保証:
- daemon_send_angles は 1 回だけ呼ぶ (ループ内で呼ばない)。
- socket.timeout / RuntimeError / OSError のいずれで例外が出ても
SendFailed に包んで上げる (呼び出し元は再送しないことが仕様)。
observer は既定 None。シグネチャは j0_sweep_scan._send_and_wait_arrival と
同じ 4 引数 (ts_send, ts_arrival, attempt, arrived) だが、attempt は本関数
では常に 1 (再送しないため)。
"""
# tolerance は JSON からのみ引く (D-2a F1)。存在しないキーなら KeyError で
# 呼び出し元へ (プログラミングエラー扱い)。
tolerance_deg = get_tolerance_deg(pose_data, pose_label)
# 1. 送信前姿勢 → deadline 算出 (取れない場合は ABORT・fallback しない)
pre_actual = daemon_get_angles(timeout_override=PER_CALL_CAP_SEC)
if pre_actual is None or len(pre_actual) != 6:
raise PoseUnreachable(
f"{pose_label}: pre-actual angles unavailable (no fallback per r3.4 89行)"
)
pre_max = _max_axis_diff(list(target), pre_actual)
deadline_sec = compute_deadline_sec(pre_max, pose_data)
# 2. deadline 絶対期限を send の前に作る (D-2a B2/B3)。以降の send + N
# ポーリングは同じ deadline_at を共有する。monotonic を使うのは壁時計
# が NTP 調整で飛ぶため。
deadline_at = time.monotonic() + deadline_sec
# 3. 単発送信 (1 回のみ・再送しない)。send も per-call cap の対象
# (D-2a B2/B3・r3.4 83-84行)。
remaining = deadline_at - time.monotonic()
if remaining <= 0:
raise SendFailed(f"{pose_label}: deadline exhausted before send")
ts_send_wall = time.time() # 無条件で取る (D-2a ⚠️ ts_send_wall)
# D-2b A2 反映: wait は SEND_WAIT_SEC 定数固定 (DAY266 固定事項)。
# 公開引数から send_wait_sec を削除したため CLI から緩和できない
# (tolerance F1 と同じ扱い・固定事項のみが正)。
try:
daemon_send_angles(
target,
speed=SPEED,
wait=SEND_WAIT_SEC,
timeout_override=min(remaining, PER_CALL_CAP_SEC),
)
except Exception as e:
if observer is not None:
_safe_notify(observer, ts_send_wall, None, 1, False)
raise SendFailed(f"{pose_label}: daemon_send_angles failed: {e}") from e
# 4. ポーリング (deadline_at を継続使用)。get_angles の各呼び出しに
# min(remaining, PER_CALL_CAP_SEC) を timeout_override として渡す。
# daemon_get_angles は内部で RuntimeError (D-1 deadline 切れ含む) を
# 握って None を返す。
last_max_diff = float("inf")
last_actual: Optional[list] = None
while True:
remaining = deadline_at - time.monotonic()
if remaining <= 0:
# deadline 超過 = 未到達 (r3.4 89行)。自動再送・自動 HOME なし。
if observer is not None:
_safe_notify(observer, ts_send_wall, None, 1, False)
return ArrivalResult(
arrived=False,
max_axis_diff_deg=(last_max_diff if last_max_diff != float("inf") else -1.0),
last_actual=last_actual,
ts_send_wall=ts_send_wall,
ts_arrival_wall=None,
deadline_sec=deadline_sec,
reason="arrival_timeout",
)
override = min(remaining, PER_CALL_CAP_SEC)
actual = daemon_get_angles(timeout_override=override)
# D-2b B1-1 反映: RPC 後に deadline を再確認。get_angles が期限直前に
# 返ったケースで arrived=True を返さないため。remaining <= 0 なら
# continue で次周回先頭へ (先頭の deadline チェックで arrival_timeout に
# 落とす)。宣言した deadline を超えて「到達」判定しない。
remaining = deadline_at - time.monotonic()
if remaining <= 0:
continue
if actual is not None and len(actual) == 6:
max_d = _max_axis_diff(actual, list(target))
last_max_diff = max_d
last_actual = list(actual)
if max_d <= tolerance_deg:
ts_arrival_wall = time.time()
if observer is not None:
_safe_notify(observer, ts_send_wall, ts_arrival_wall, 1, True)
return ArrivalResult(
arrived=True,
max_axis_diff_deg=max_d,
last_actual=list(actual),
ts_send_wall=ts_send_wall,
ts_arrival_wall=ts_arrival_wall,
deadline_sec=deadline_sec,
)
# actual が None / 長さ != 6 のときは「未到達」扱い (r3.4 89行) で
# 次周回へ。deadline が生きているうちはポーリングを続ける。
# D-2b B1-2 反映: sleep は remaining を超えない。残り 0.01s でも
# 0.3s 眠ると arrival_timeout の返却が最大 0.3s 遅れるため。
remaining = deadline_at - time.monotonic()
if remaining > 0:
time.sleep(min(POLL_SLEEP_SEC, remaining))
def _safe_notify(
observer: Callable[[Optional[float], Optional[float], int, bool], None],
ts_send: Optional[float],
ts_arrival: Optional[float],
attempt: int,
arrived: bool,
) -> None:
"""observer 内例外を握る (bad observer が制御を止めないための保険)。
j0_sweep_scan の _notify と同じ手口。
"""
try:
observer(ts_send, ts_arrival, attempt, arrived)
except Exception:
pass
# ---- インターロック ----
def wait_interlock(
label: str,
prompt: str,
role: str = "operator",
) -> InterlockConfirmation:
"""Enter があるまで送信なし (r3.4 2節 4・8 / 8節 211-215行)。
label: "interlock1_wait_start" / "interlock2_hand_clear" 等
prompt: 表示するメッセージ (「静止・到達を確認。準備ができたら Enter」等)
role: 誰が押したか。人試験時は補助者運用 (role="assistant") を r3.4 が推奨
(参加者本人が押すと退避と入力の順序が曖昧になるため)。呼び出し元 (E)
が context から決めて渡す。
この Enter は safety-rated interlock ではない (手順的保護)。r3.4 8節に明記。
Ctrl-C 時は KeyboardInterrupt がそのまま呼び出し元へ伝播 (自動移動なし)。
返り値の InterlockConfirmation を呼び出し元 (E) が run_dir JSON に保存する。
"""
print(f"[INTERLOCK {label}] {prompt}")
print(
"[INTERLOCK {label}] Enter で次工程 (safety-rated ではなく手順的保護)".format(
label=label
)
)
input()
return InterlockConfirmation(
label=label,
role=role,
ts_monotonic=time.monotonic(),
ts_wall=time.time(),
)run_pick_handover.py（E-1〜E-3・受け渡し本体・1542 行）
骨格（pose ローダ・N 計画・常駐ループ）、PICK を下位関数で組み直した orchestration、そして PRESENT → 待機 → 開放 → 退避 → HOME の受け渡しシーケンス。close 封じの状態機械と、開放中に Ctrl-C が入っても記録と物理が食い違わないための SIGINT 遅延もここに入っている。
"""
run_pick_handover.py — DAY289 E-1 (骨格・sequence 本体は E-2 以降)
r3.4 実装物リスト 5 の分割発注 (E-1)。E-1 の範囲:
- pose JSON ローダ (schema 検証つき・起動時に fail-fast)
- N 計画 (balanced block・r3.4 4節)
- 常駐ループ (n / Enter / h / q / EOF・r3.4 94行)
- CLI 引数
E-1 の時点では「次試行」の中身はスタブ (print して即戻る)。実際の PICK
orchestration は E-2、handover sequence 本体は E-3、台帳追記 / emergency
journal は E-4、VLM / 撮影は E-5、モックテストは E-6 で足す。
===============================================================================
恒久ルール (r3.4 1節・PICK 再利用方式・案2)
===============================================================================
- PICK orchestration の参照元コミット: **cbfb7b3 (DAY283)**
- PICK ロジックは現物 be6fde1 と同一である必要がある (差分は recorder_notify
のみ・E 初版は recorder を写さない・r3.4 37行 録画は初版なし)
- **PICK ロジック変更時は run_pick_place.py と本ファイルの両方を直す** —
Case-in-point の parity 破りが台帳の意味を壊すため恒久ルール。E-2 で PICK
orchestration を写す際に parity mock を同時に整備する
===============================================================================
安全に関する留意 (STARTUP_BANNER で表示):
- Ctrl-C は安全停止ではない (daemon 受理済み send_angles は継続しうる)
- SPEED=20 は人体安全速度ではない
- インターロック 1/2 の Enter は safety-rated interlock ではない (手順的保護)
- handover phase では close 系指令を送信しない (state 機械で拒否)
"""
from __future__ import annotations
import argparse
import math
import random
import signal
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Optional
SCRIPTS_ROOT = Path(__file__).resolve().parents[3]
MODULE_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPTS_ROOT))
sys.path.insert(0, str(MODULE_DIR))
# handover_release から pose JSON ローダ・sequence 関数・例外・定数を再利用
# する (二重定義しない)。pose JSON のパス既定値 POSE_JSON_PATH もそこから引く。
from handover_release import ( # noqa: E402
POSE_JSON_PATH,
STARTUP_BANNER,
ArrivalResult,
ForbiddenGripperClose,
GripperOpenOnly,
InterlockConfirmation,
PoseUnreachable,
SendFailed,
load_pose_json,
reject_close_request,
send_once_and_wait_arrival,
)
# 注: D-2 の wait_interlock は Q2 反映で E-3 側 wrapper (_wait_interlock_or_abort)
# に差し替え済み (q 中断サポートのため)。D-2 本体は不変。詳細は
# _wait_interlock_or_abort docstring 参照。
# E-2a: PICK orchestration は run_pick_place の下位関数を import して呼ぶ (方式2)。
# コピーせず・値をコードに焼かず・定数もモジュールから引く。写し違いのリスクは
# parity mock (E-2b・別発注) で検証する。
# 恒久ルール: run_pick_place の import 対象を触ったら本ファイルの動作も同時に
# 見直すこと (r3.4 1節・PICK ロジック変更時は両ファイル同期)。
from run_pick_place import ( # noqa: E402
HOME,
SPEED,
SEND_WAIT_SEC as RPP_SEND_WAIT_SEC,
TOL_HOME,
TOL_APPROACH,
TOL_PICK_PLACE,
MAX_PICK_ATTEMPTS,
SWEPT_J0_FOR_PICK_RETREAT,
GRIPPER_SLEEP,
GRIP_VALUE,
USE_CORRECTION,
CORR_MEM_PATH,
DETECT_REPEAT,
_OBSERVE_STATE,
_abort,
_apply_correction,
_do_gripper,
_do_move,
_observe_phase,
stage1_home_and_open,
stage2_detect_block_and_box,
stage_lookup_and_move,
)
from visual_servo_pick import ( # noqa: E402
daemon_set_fresh_mode,
get_daemon_socket,
)
from handover_ledger import ensure_ssd_mounted as ensure_handover_ssd_mounted # noqa: E402
from handover_recording import ( # noqa: E402
AppendAborted,
build_ledger_row,
build_run_payload,
collect_operator_annotations,
make_run_stem,
persist_or_journal,
recover_journals,
)
# 必須姿勢ラベル (r3.4 3-5節の handover_present_pose.json スキーマに準拠)
REQUIRED_POSE_LABELS = ("HOME", "PRESENT_MID", "PRESENT", "RETRACT_SMALL")
# deadline ブロックの必須キー (handover_release.compute_deadline_sec が読む値)
REQUIRED_DEADLINE_KEYS = ("base_sec", "factor", "min_sec", "max_sec", "speed")
# ---------------------------------------------------------------------------
# pose JSON ローダ (schema 検証つき)
# ---------------------------------------------------------------------------
def _is_finite_number(v) -> bool:
"""int/float (bool 除く) かつ有限であること。NaN / Inf / bool / str は False。"""
if isinstance(v, bool):
return False
if not isinstance(v, (int, float)):
return False
return math.isfinite(v)
def load_and_validate_pose_json(path: Path) -> dict:
"""pose JSON を読んで必須キー・型・shape・値域を検証する。sequence 開始前に
落とすため、欠損・型不正・非有限値・値域違反は即 ValueError。
handover_release.load_pose_json を内包 (dict 読取までは再利用)。
E-1a A3 反映: handover_release が runtime に読む全キーを起点に、
handover_release.compute_deadline_sec が内部で持つ検証 (全 finite・speed>0・
0<min_sec<=max_sec) と同じ条件をここで先に走らせる。これにより sequence の
途中で ValueError が飛ぶ事態を fail-fast で防ぐ。
handover_release が実測で読むキー (grep 確認 L164/L179-184):
- pose_data["arrival_tolerance_deg"][label] (get_tolerance_deg)
- pose_data["deadline"]["base_sec"|"factor"|"min_sec"|"max_sec"|"speed"]
(compute_deadline_sec)
E が読むキー:
- pose_data["poses"][label]["angles"] (target 展開・_manual_home 含む)
joint_delta_limits_deg は runtime consumers に居ないため、存在チェックのみ
(E-4 の台帳追記で使う想定・shape は将来チェック追加)。
"""
d = load_pose_json(path)
# poses.<label>.angles: 4 label すべてが長さ 6 の有限数リスト
if "poses" not in d or not isinstance(d["poses"], dict):
raise ValueError(f"pose JSON missing dict 'poses' ({path})")
for label in REQUIRED_POSE_LABELS:
if label not in d["poses"]:
raise ValueError(f"pose JSON missing poses.{label} ({path})")
angles = d["poses"][label].get("angles")
if not isinstance(angles, list) or len(angles) != 6:
raise ValueError(
f"pose JSON poses.{label}.angles must be a list of length 6 (got {angles!r})"
)
for i, v in enumerate(angles):
if not _is_finite_number(v):
raise ValueError(
f"pose JSON poses.{label}.angles[{i}] must be a finite number "
f"(got {type(v).__name__}: {v!r})"
)
# arrival_tolerance_deg.<label>: 4 label すべてが有限かつ >0
if "arrival_tolerance_deg" not in d or not isinstance(d["arrival_tolerance_deg"], dict):
raise ValueError(f"pose JSON missing dict 'arrival_tolerance_deg' ({path})")
for label in REQUIRED_POSE_LABELS:
if label not in d["arrival_tolerance_deg"]:
raise ValueError(f"pose JSON missing arrival_tolerance_deg.{label}")
v = d["arrival_tolerance_deg"][label]
if not _is_finite_number(v):
raise ValueError(
f"pose JSON arrival_tolerance_deg.{label} must be a finite number "
f"(got {type(v).__name__}: {v!r})"
)
if v <= 0:
raise ValueError(
f"pose JSON arrival_tolerance_deg.{label} must be > 0 (got {v})"
)
# deadline.<key>: 5 key すべてが有限、加えて speed>0 / 0<min_sec<=max_sec
# (handover_release.compute_deadline_sec の内部検証と同一条件)
if "deadline" not in d or not isinstance(d["deadline"], dict):
raise ValueError(f"pose JSON missing dict 'deadline' ({path})")
for key in REQUIRED_DEADLINE_KEYS:
if key not in d["deadline"]:
raise ValueError(f"pose JSON missing deadline.{key}")
v = d["deadline"][key]
if not _is_finite_number(v):
raise ValueError(
f"pose JSON deadline.{key} must be a finite number "
f"(got {type(v).__name__}: {v!r})"
)
speed = d["deadline"]["speed"]
min_sec = d["deadline"]["min_sec"]
max_sec = d["deadline"]["max_sec"]
if speed <= 0:
raise ValueError(f"pose JSON deadline.speed must be > 0 (got {speed})")
if not (0 < min_sec <= max_sec):
raise ValueError(
f"pose JSON deadline must satisfy 0 < min_sec <= max_sec "
f"(got min_sec={min_sec}, max_sec={max_sec})"
)
# joint_delta_limits_deg: 存在チェックのみ (E-4 台帳追記で使う想定)
if "joint_delta_limits_deg" not in d or not isinstance(d["joint_delta_limits_deg"], dict):
raise ValueError(f"pose JSON missing dict 'joint_delta_limits_deg' ({path})")
return d
# ---------------------------------------------------------------------------
# N 計画 (balanced block・r3.4 4節)
# ---------------------------------------------------------------------------
def parse_wait_plan_arg(s: str) -> list:
""""2,3,5,8" のような CSV 文字列を int リストに。空要素・非整数は ValueError。"""
parts = [p.strip() for p in s.split(",") if p.strip()]
if not parts:
raise ValueError(f"--wait-plan must contain at least one integer (got {s!r})")
out = []
for p in parts:
try:
v = int(p)
except ValueError:
raise ValueError(f"--wait-plan element must be integer (got {p!r})")
if v <= 0:
raise ValueError(f"--wait-plan element must be > 0 (got {v})")
out.append(v)
return out
def generate_wait_plan(
wait_set: list,
block_repeat: int,
seed: Optional[int] = None,
) -> tuple:
"""balanced block の N 計画を作る (r3.4 4節)。
- wait_set: N 集合 (例: [2, 3, 5, 8])
- block_repeat: ブロック数 (各ブロックは wait_set の順列 1 つ)
- seed: shuffle_seed。None なら生成する (0..2^31-1)
返り値: (trials, wait_plan_id, shuffle_seed)
trials: list of dict [{"wait_sec": int, "wait_plan_index": int, "block_index": int}, ...]
wait_plan_id: 記録用 ID (再現に必要な情報を含む文字列)
shuffle_seed: 実際に使った seed (呼び出し元が run_dir JSON に保存する)
保証:
- **各 N が同数含まれる**: 最終系列で各 N はちょうど block_repeat 回登場
- **完全ランダムではない**: 各ブロック内でだけシャッフル (ブロック間は
独立にシャッフル・balanced block 設計)
- **seed 固定で再現可能**: 同じ (wait_set, block_repeat, seed) は同じ
提示順を返す (random.Random(seed) の deterministic 性を利用)
"""
if not isinstance(block_repeat, int) or block_repeat <= 0:
raise ValueError(f"block_repeat must be a positive integer (got {block_repeat!r})")
if not wait_set:
raise ValueError("wait_set must not be empty")
if seed is None:
seed = random.randint(0, 2**31 - 1)
rng = random.Random(seed)
trials = []
for block_i in range(block_repeat):
block = list(wait_set)
rng.shuffle(block)
for wait_sec in block:
trials.append(
{
"wait_sec": int(wait_sec),
"wait_plan_index": len(trials),
"block_index": block_i,
}
)
wait_plan_id = "wp:{}_x{}:seed{}".format(
",".join(str(n) for n in wait_set),
block_repeat,
seed,
)
return trials, wait_plan_id, seed
# ---------------------------------------------------------------------------
# HOME 手動復帰 (h コマンド)
# ---------------------------------------------------------------------------
def _manual_home(pose_data: dict, dry_run: bool = False) -> None:
"""`h` コマンドで HOME 復帰 (試行外・台帳追記なし)。
参照元 run_pick_place.py L5071 の _manual_home と同型 (short-lived socket・
gripper には触らない・例外を握って戻る)。ただし送角は handover_release
の send_once_and_wait_arrival を使う (D-2 の deadline/per_call_cap 経路を
共有・E 内で socket 予算を独自定義しない)。
"""
home_angles = pose_data["poses"]["HOME"]["angles"]
if dry_run:
print("[HOME] --dry-run のため送角スキップ (planned: HOME={})".format(home_angles))
return
print("[HOME] 手動 HOME 復帰開始 (試行外・台帳追記なし)")
try:
res: ArrivalResult = send_once_and_wait_arrival(home_angles, "HOME", pose_data)
except PoseUnreachable as e:
print("[HOME] pre-actual 取得失敗 → 停止・報告のみ: {}".format(e))
return
except SendFailed as e:
print("[HOME] send_angles 失敗 → 停止・報告のみ: {}".format(e))
return
except Exception as e:
# 常駐ループを止めないよう握って戻る (L5077 と同型)
print("[HOME] 予期しない例外 → 停止・報告のみ: {}: {}".format(type(e).__name__, e))
return
if res.arrived:
print("[HOME] arrived max|Δ|={:.2f}° (deadline_sec={:.1f})".format(
res.max_axis_diff_deg, res.deadline_sec))
else:
print("[HOME] 未到達 (reason={} max|Δ|={:.2f}° deadline_sec={:.1f})".format(
res.reason, res.max_axis_diff_deg, res.deadline_sec))
print("[HOME] 目視確認のこと・必要なら再度 h を試すか手動介入")
# ---------------------------------------------------------------------------
# PICK orchestration (E-2a・方式2・run_pick_place の下位関数を import して呼ぶ)
# ---------------------------------------------------------------------------
# 参照元コミット: cbfb7b3 (DAY283)。PICK ロジックは HEAD で不変 (差分は
# recorder_notify のみ・E は写さない)。DAY277 で PICK 視覚サーボ (_run_pick_
# visual_servo / _visual_servo_loop_pick) は本線から切り離されており、
# stage4 (descend) は j0 lock による真下降下のみ (pick_j0_ref=approach j0)。
# 本 E-2a も同方針: 視覚サーボは呼ばない。
#
# E-2a のスコープ (r3.4 実装物 5・分割発注):
# - PICK 区間 (STAGE1 HOME+open → STAGE2 detect → STAGE3 approach_pick →
# STAGE4 descend_pick → STAGE5 grip close → STAGE6 lift → HOME 復帰) のみ
# - handover sequence (PRESENT 系) は E-3。tip 測定・VLM verdict・regrasp
# ループは E-5 (VLM 系まとめ)。台帳追記は E-4
#
# 保持する状態 (E-2b parity mock で run_pick_place.run() と一致検証する対象):
# - log_record["moves"] : _do_move が append する送角履歴 (target/tol/arrived/post_actual)
# - log_record["gripper"] : _do_gripper が append する open/close 履歴
# - log_record["stages"] : stage_lookup_and_move / _do_gripper が append する label
# - log_record["lookups"] : stage_lookup_and_move が append する fk_lookup 結果
# - log_record["detect"] : stage2_detect_block_and_box が append する検出結果
# - log_record["pick_correction"]: E-2a 独自 (block_xz + target_xz + corr_meta)
# - log_record["stage_failures"] : run() 互換 (stage5/6/home 失敗を記録して継続)
# - log_record["abort"] : run_pick_place._abort が set (stage / reason)。
# E-2a も同じキーを使う (parity 対象)
# - log_record["handover_pick_to_home"]: E-2a 独自 (send_once_and_wait_arrival 経由の
# HOME 復帰。parity 対象外 = run() に相当経路なし)
# 補正記憶更新は E-2a では行わない。理由:
# run() は判定① (VLM verdict + tip_dist override) が "yes" 確定時のみ
# correction_memory.json を更新する (L4171-4208)。E-2a は VLM を持たない
# (E-5) ため update トリガが存在しない。読取 (_apply_correction) は行うが
# 書込 (_corr_update / _corr_save) は E-5 で VLM 判定と同時に配線する。
def _reset_observe_state_for_trial(trial_index: int, dry_run: bool) -> None:
"""run() L3849-3868 と同型の _OBSERVE_STATE リセット。E-2a は E の常駐
ループから毎試行呼ばれるので、前試行の残り状態が UDP notify に混ざらない
ようにする。"""
_OBSERVE_STATE["phase"] = None
_OBSERVE_STATE["gripper"] = None
_OBSERVE_STATE["verdict"] = {"pick": None, "place": None, "side_place": None}
_OBSERVE_STATE["block_xz"] = None
_OBSERVE_STATE["box_xz"] = None
_OBSERVE_STATE["block_place_xz"] = None
_OBSERVE_STATE["run_stem"] = "handover_trial_{:03d}".format(trial_index)
_OBSERVE_STATE["cap_tip"] = None
_OBSERVE_STATE["cap_pick"] = None
_OBSERVE_STATE["cap_place"] = None
_OBSERVE_STATE["cap_side_place"] = None
_OBSERVE_STATE["lookup"] = None
_OBSERVE_STATE["corr"] = None
_OBSERVE_STATE["captures"] = None
_OBSERVE_STATE["dry_run"] = bool(dry_run)
def _init_log_record(trial_index: int, trial: Optional[dict], dry_run: bool) -> dict:
"""PICK helpers が append する log_record の下地。run() L3869-3896 と
互換の構造を持つ (parity mock で列を揃えるため)。"""
return {
"started_at": datetime.now().isoformat(),
"run_stem": make_run_stem(trial_index),
"dry_run": bool(dry_run),
"trial_index": trial_index,
"trial": trial,
"constants": {
# E-2b parity mock はここを見て「両本線が同じ定数で動いた」ことを確認する
"HOME": list(HOME),
"SPEED": SPEED,
"SEND_WAIT_SEC": RPP_SEND_WAIT_SEC,
"TOL_HOME": TOL_HOME,
"TOL_APPROACH": TOL_APPROACH,
"TOL_PICK_PLACE": TOL_PICK_PLACE,
"GRIPPER_SLEEP": GRIPPER_SLEEP,
"GRIP_VALUE": GRIP_VALUE,
"MAX_PICK_ATTEMPTS": MAX_PICK_ATTEMPTS,
"SWEPT_J0_FOR_PICK_RETREAT": SWEPT_J0_FOR_PICK_RETREAT,
"DETECT_REPEAT": DETECT_REPEAT,
"USE_CORRECTION": bool(USE_CORRECTION),
"CORR_MEM_PATH": str(CORR_MEM_PATH),
},
"stages": [],
"detect": [],
"lookups": [],
"moves": [],
"gripper": [],
"vlm_captures": [], # E-2a では常に空 (VLM は E-5)
"judgments": {}, # E-2a では常に空
"pick_attempts": [], # E-2a では regrasp なし・常に空
"stage_failures": [],
# abort は run_pick_place._abort() が入れる ({"stage":..., "reason":...})。
# 事前に None は入れない (run() 側も抽選前は未 set = key 存在せず)。
}
def _guard_mount_before_sequence(log_record: dict, stage: str, dry_run: bool) -> bool:
"""Stop before a new physical sequence while the human has not been invited in.
No guard is called after interlock1 succeeds: from that point through retreat,
physical safety has priority and a later write failure goes to the journal.
"""
if dry_run or ensure_handover_ssd_mounted():
return True
log_record["trial_status"] = "mount_lost"
has_closed_gripper = any(
x.get("action") == "close" for x in log_record.get("gripper", []))
log_record["manual_recovery_required"] = has_closed_gripper
_abort(log_record, stage, "KIOXIA mount lost before next sequence")
print("[MOUNT-GUARD] {} 開始前に mount lost → 新規指令なし".format(stage))
return False
# ---------------------------------------------------------------------------
# E-3: handover sequence 部品 (r3.4 2節 sequence 2-11・5節 close 封じ・8節 Ctrl-C)
# ---------------------------------------------------------------------------
class _HandoverPhaseGate:
"""PICK_COMPLETE 以降の handover 区間で close 系呼出しを構造的に拒否する
ゲート (r3.4 5節・E-3 中核安全要件)。
実装方針 3 層:
(1) 構造的保護: 公開する gripper API は release_open() のみ。内部で
GripperOpenOnly.open() を呼ぶ (D-2)。close 系メソッドを本クラスに
**定義しない**ため、E-3 のコードから close を呼ぶ経路は構造上ゼロ。
(2) runtime 保護: assert_no_close_attempted(action) を通じて、万一
外部コードが本クラス経由で gripper 操作を試みた際に action != "open"
なら ForbiddenGripperClose を上げる。E-3 の release 直前で明示的に
呼び「この phase では open 以外拒否」の宣言を残す (防衛線 + 監査痕)。
(3) phase 追跡: enter_handover / exit_handover で内部 flag を切替、
log_record に phase 遷移を記録 (E-4 台帳・監査用)。
handover phase = PICK_COMPLETE (E-2 の HOME 復帰完了) 以降、sequence 11
(HOME 到達) までの区間。この区間で gripper は open (RELEASE) を 1 回のみ。
"""
def __init__(self):
self._gripper_open_only = GripperOpenOnly()
self._active = False
def enter_handover(self) -> None:
self._active = True
def exit_handover(self) -> None:
self._active = False
@property
def is_active(self) -> bool:
return self._active
def release_open(self) -> None:
"""RELEASE 用: gripper open のみ。close 系メソッドは本クラスに未定義。"""
if not self._active:
raise RuntimeError(
"release_open() called outside handover phase "
"(call enter_handover first)"
)
self._gripper_open_only.open()
def assert_no_close_attempted(self, action: str) -> None:
"""runtime gate: gripper 操作意図がある呼び出し側から呼ぶ。
active 区間で action != 'open' なら reject_close_request 経由で
ForbiddenGripperClose (r3.4 5節・133行)。"""
if self._active and action != "open":
reject_close_request(
"handover phase active, requested action={!r}".format(action))
class _DeferredSigint:
"""SIGINT (Ctrl-C) を一時的に遅延させる context manager (E-3a C3 反映)。
目的: RELEASE 送信〜応答確定〜log_record 記録の区間を Ctrl-C から保護し、
「爪は物理的に開いた (daemon が受理・Pi 側 wait 2s 中)」のに
「release_command_ok が確定しないまま interrupted になる」記録ズレを防ぐ。
使い方:
with _DeferredSigint() as deferred:
# SIGINT が来ても KeyboardInterrupt は発生せず、フラグだけ立つ
gate.release_open() # daemon 送信 + Pi 側 wait 2s
release_command_ok = True # 記録確定
# ここで元のハンドラに戻る
if deferred.was_interrupted:
raise KeyboardInterrupt() # 外側 except が interrupted 記録
設計:
- handler は「フラグを立てるだけ」の最小処理 (signal 内で長い処理禁止)
- try/finally で元のハンドラを必ず復元 (例外経路含む)
- 非メインスレッド (signal.signal が ValueError) では「遅延なし」に劣化
(現状挙動と同型)、install_error を log に残す
- 遅延区間は最小: 送信直前〜記録確定まで。WAIT やインターロックは含めない
"""
def __init__(self):
self.was_interrupted = False
self._orig_handler = None
self._installed = False
self._install_error: Optional[str] = None
def __enter__(self):
def _defer(_signum, _frame):
self.was_interrupted = True
try:
self._orig_handler = signal.signal(signal.SIGINT, _defer)
self._installed = True
except ValueError as e:
# signal.signal は main thread からしか呼べない
# → 劣化動作: 遅延なし・現状挙動 (Ctrl-C 即応)
self._install_error = "signal.signal failed: {}".format(e)
return self
def __exit__(self, _exc_type, _exc_val, _exc_tb):
if self._installed:
try:
signal.signal(signal.SIGINT, self._orig_handler)
except Exception:
pass
return False # 例外を握らない (伝播させる)
def _do_handover_move(
target_angles: list,
pose_label: str,
pose_data: dict,
log_record: dict,
dry_run: bool,
) -> dict:
"""handover sequence 内の 1 移動 (send_once_and_wait_arrival 経由・D-2)。
tolerance / deadline は pose JSON からのみ (D-2 が構造保証)。返り値 dict:
{arrived: bool, reason: Optional[str], max_axis_diff_deg: float,
deadline_sec: float, dry_run: bool, exception: Optional[str]}
log_record["handover"]["sequence"] に追記。dry_run は skip して stub 返す。"""
entry = {
"label": pose_label,
"target_angles": list(target_angles),
}
if dry_run:
entry.update({
"arrived": True, "reason": "dry_run", "dry_run": True,
"max_axis_diff_deg": None, "deadline_sec": None,
"ts_send_wall": None, "ts_arrival_wall": None,
"last_actual": None,
})
log_record["handover"].setdefault("sequence", []).append(entry)
print("[HANDOVER] {} dry_run（送信なし）".format(pose_label))
return entry
try:
res: ArrivalResult = send_once_and_wait_arrival(
target_angles, pose_label, pose_data)
except PoseUnreachable as e:
entry.update({
"arrived": False, "reason": "pre_actual_unreachable",
"exception": str(e), "dry_run": False,
"max_axis_diff_deg": None, "deadline_sec": None,
})
log_record["handover"].setdefault("sequence", []).append(entry)
return entry
except SendFailed as e:
entry.update({
"arrived": False, "reason": "send_failed",
"exception": str(e), "dry_run": False,
"max_axis_diff_deg": None, "deadline_sec": None,
})
log_record["handover"].setdefault("sequence", []).append(entry)
return entry
entry.update({
"arrived": bool(res.arrived),
"reason": res.reason or ("arrived" if res.arrived else "arrival_timeout"),
"dry_run": False,
"max_axis_diff_deg": float(res.max_axis_diff_deg),
"deadline_sec": float(res.deadline_sec),
"last_actual": list(res.last_actual) if res.last_actual is not None else None,
"ts_send_wall": float(res.ts_send_wall),
"ts_arrival_wall":
float(res.ts_arrival_wall) if res.ts_arrival_wall is not None else None,
})
log_record["handover"].setdefault("sequence", []).append(entry)
return entry
def _capture_after_release_stub(dry_run: bool) -> dict:
"""r3.4 2節 7 (RELEASE 直後の撮影) の E-3 スタブ。E-5 で本実装。
「[E-3 STUB] capture (E-5)」を表示して即継続。
返り値の shape は E-4 台帳の 41〜44 列 (top_capture_status /
side_capture_status / top_capture_path / side_capture_path) と直接
対応する dict (Q3 反映)。status 語彙は r3.4 6節「ok / fail:{reason}」に
「skipped:e3_stub」を暫定追加 — E-5 で ok/fail に置換される。E-4 発注時に
data_manifest_rules.md 9節と突合して「skipped:」を正式採用するか確認する。
"""
print("[E-3 STUB] capture (E-5)")
return {
"top": {"status": "skipped:e3_stub", "path": ""},
"side": {"status": "skipped:e3_stub", "path": ""},
}
def _wait_interlock_or_abort(label: str, prompt: str, role: str = "operator"):
"""Q2 反映・E-3 wrapper: D-2 の wait_interlock は Enter しか受けないため
q(中断) サポートを E-3 側で実装する (D-2 は不変)。
入力受付:
- Enter (空文字) or それ以外 = 次工程へ (InterlockConfirmation を返す)
- "q" = 中断 (None を返す・呼び出し側が abort_stage 記録 + return False)
Ctrl-C と同じ終着 (「その場保持・記録して return・メニューへ戻る」・r3.4 8節)
にするため、q は KeyboardInterrupt に変換せず戻り値 None で表現する
(Ctrl-C は Python 実行系が例外として上げるが、q は明示的意思のため区別可能
な signal にしておくと監査記録で区別できる)。呼び出し側で None 判定して
abort_stage=<label> + trial_status=interrupted を記録する。
"""
print("[INTERLOCK {}] {}".format(label, prompt))
print("[INTERLOCK {}] Enter で次工程 / q で中断 (safety-rated ではなく手順的保護)".format(label))
try:
resp = input().strip().lower()
except EOFError:
# EOF は q 同等 (stdin 終了)
return None
if resp == "q":
return None
# Enter (空文字) or 他 → 次工程 (D-2 の InterlockConfirmation と同型を返す)
return InterlockConfirmation(
label=label, role=role,
ts_monotonic=time.monotonic(), ts_wall=time.time(),
)
def _record_interlock_abort(log_record: dict, interlock_label: str) -> None:
"""Q2 反映: interlock で q 受領 → Ctrl-C と同じ終着処理 (r3.4 8節)。
新規指令なし・自動 HOME なし・グリッパー自動開放なし・ラッチ表示・
trial_status=interrupted / abort_stage=<interlock_label> を記録。
常駐ループはメニューへ戻る (セッション終了しない)。"""
_abort(log_record, "handover_interlock_abort",
"user pressed 'q' at {} (Ctrl-C 相当・その場保持・手動復帰へ)"
.format(interlock_label))
log_record["trial_status"] = "interrupted"
log_record["abort_stage"] = interlock_label
log_record.setdefault("handover", {})["phase"] = "interrupted"
print("[HANDOVER] {} で q 受領・handover 中断 (r3.4 8節)".format(interlock_label))
print("[HANDOVER] アームに触らない。復帰は退避確認後に手動で。")
print("[HANDOVER] メニューへ戻ります (セッション継続)。")
def _record_handover_interrupt(log_record: dict) -> None:
"""r3.4 8節: handover 中 Ctrl-C 受領時の記録。新規指令なし・自動 HOME
なし・グリッパー自動開放なし・中断表示ラッチ。trial_status=interrupted
と abort_stage を log_record に残し、常駐ループはメニューへ戻る
(セッション終了しない・E-1 の暫定 break を廃止)。
E-3a C3 反映: RELEASE 送信中に Ctrl-C を受けた場合、_DeferredSigint が
フラグを立て log_record["handover"]["release_interrupted_after_send"] = True
を残してから KeyboardInterrupt を再送出する。このケースでは release_command_ok
が既に確定しており abort_stage を "release_after_send" として区別する
(「爪は物理的に開いた・その後に Ctrl-C」を明示)。
"""
hv = log_record.get("handover") or {}
if hv.get("release_interrupted_after_send"):
# C3: RELEASE 完了後 (爪 open 確定後) の Ctrl-C
abort_stage = "release_after_send"
else:
seq = hv.get("sequence") or []
if seq:
abort_stage = seq[-1].get("label", "handover_unknown")
else:
abort_stage = "handover_unknown"
_abort(log_record, "handover_interrupt",
"handover 中 Ctrl-C ({}): アームに触らない。復帰は退避確認後に手動で。"
.format(abort_stage))
log_record["trial_status"] = "interrupted"
log_record["abort_stage"] = abort_stage
log_record.setdefault("handover", {})["phase"] = "interrupted"
print("[HANDOVER] Ctrl-C 受領・handover 中断 (r3.4 8節)")
print("[HANDOVER] アームに触らない。復帰は退避確認後に手動で。")
print("[HANDOVER] メニューへ戻ります (セッション継続)。")
def _run_handover_sequence(
trial: Optional[dict],
pose_data: dict,
log_record: dict,
dry_run: bool,
) -> bool:
"""r3.4 2節 sequence 2-11 を実行。呼び出し前提: E-2 の PICK 完了 + HOME
復帰が成功していること。返り値: True (sequence 11 まで到達) / False (中断・
未到達・失敗のいずれか)。log_record["handover"] に全結果を残す (E-4 台帳
が拾える形)。close 送信ゼロは _HandoverPhaseGate が構造保証。
r3.4 2節 の中断条件対応:
- PRESENT 未到達 (3): インターロック1 表示を出さない (r3.4 2節3・
「手を出してください」を出さない = 人に手を出させない安全要件)
- 9-11 未到達: 完了表示を出さず manual_recovery_required を記録
"""
trial_wait_sec = int(trial["wait_sec"]) if trial else 5
log_record["handover"] = {
"phase": "start",
"trial_wait_sec_requested": trial_wait_sec,
"sequence": [],
}
gate = _HandoverPhaseGate()
gate.enter_handover()
try:
# ---- 2. PRESENT_MID へ移動 + 到達判定 ----
if not _guard_mount_before_sequence(log_record, "handover_seq2", dry_run):
log_record["handover"]["phase"] = "mount_lost_before_present_mid"
return False
pmid_angles = pose_data["poses"]["PRESENT_MID"]["angles"]
r2 = _do_handover_move(pmid_angles, "PRESENT_MID", pose_data,
log_record, dry_run)
if not r2["arrived"]:
log_record["handover"]["phase"] = "aborted"
log_record["trial_status"] = "motion_failed"
_abort(log_record, "handover_seq2",
"PRESENT_MID 未到達: {}".format(r2.get("reason")))
return False
# ---- 3. PRESENT へ移動 + 到達判定 ----
if not _guard_mount_before_sequence(log_record, "handover_seq3", dry_run):
log_record["handover"]["phase"] = "mount_lost_before_present"
return False
pres_angles = pose_data["poses"]["PRESENT"]["angles"]
r3 = _do_handover_move(pres_angles, "PRESENT", pose_data,
log_record, dry_run)
if not r3["arrived"]:
# r3.4 2節3: PRESENT 未到達なら「手を出してください」を表示せず
# 中断。**インターロック1 の入力プロンプトを出さない** (下記の 4
# 番へ進まないことで実現)。
log_record["handover"]["phase"] = "aborted_before_interlock1"
log_record["trial_status"] = "motion_failed"
log_record["handover"]["interlock1_skipped_reason"] = \
"PRESENT unreached (r3.4 2節3・手を出させない)"
_abort(log_record, "handover_seq3",
"PRESENT 未到達 (interlock1 表示なし・手動復帰へ): {}"
.format(r3.get("reason")))
print("[HANDOVER] PRESENT 未到達。インターロック1 は表示しません "
"(r3.4 2節3・手動復帰へ)。")
return False
# ---- 4. インターロック1 (WAIT 開始前・Enter まで送信なし・Q2: q で中断可) ----
# PRESENT 移動中に mount lost した場合も、人へ手を出すよう促す前に止める。
# この確認を最後の mount guard とし、Enter 通過後は退避完了まで確認で
# 止めない (記録より物理安全を優先し、最後に journal へ退避)。
if not _guard_mount_before_sequence(log_record, "interlock1", dry_run):
log_record["handover"]["phase"] = "mount_lost_before_interlock1"
return False
if dry_run:
interlock1 = InterlockConfirmation(
label="interlock1_wait_start", role="operator",
ts_monotonic=time.monotonic(), ts_wall=time.time())
print("[HANDOVER dry_run] interlock1 skip (input なし)")
else:
# Q2 反映: D-2 の wait_interlock ではなく E-3 wrapper 経由 (q 対応)。
# D-2 は不変。q → None で受けて Ctrl-C と同じ終着へ。
interlock1 = _wait_interlock_or_abort(
"interlock1_wait_start",
"静止・到達を確認。準備ができたら Enter (受け取り開始)",
role="operator")
if interlock1 is None:
log_record["handover"]["phase"] = "aborted_at_interlock1"
_record_interlock_abort(log_record, "interlock1_wait_start")
return False
log_record["handover"]["interlock1"] = {
"label": interlock1.label, "role": interlock1.role,
"ts_monotonic": interlock1.ts_monotonic,
"ts_wall": interlock1.ts_wall,
}
# ---- 5. WAIT: trial["wait_sec"] 秒・monotonic 実測・送信ゼロ ----
# E-3a B2 反映: 一発 time.sleep(wait_sec) から分割 sleep + monotonic
# 経過判定へ。Ctrl-C は従来どおり即応 (WAIT 中の中断は安全: RELEASE 前)、
# 中断時も wait_sec_actual が monotonic 実測で正確に残る。
# 送信ゼロは不変 (time.sleep と時刻取得のみ)。
_WAIT_SLICE_SEC = 0.1
wait_started_mono = time.monotonic()
wait_started_wall = time.time()
wait_interrupted = False
try:
if not dry_run:
while True:
elapsed = time.monotonic() - wait_started_mono
remaining = trial_wait_sec - elapsed
if remaining <= 0:
break
time.sleep(min(_WAIT_SLICE_SEC, remaining))
except KeyboardInterrupt:
wait_interrupted = True
finally:
wait_sec_actual = time.monotonic() - wait_started_mono
log_record["handover"]["wait"] = {
"wait_sec_requested": trial_wait_sec,
"wait_sec_actual": wait_sec_actual,
"wait_started_wall": wait_started_wall,
"dry_run": dry_run,
"interrupted": wait_interrupted,
}
if wait_interrupted:
# WAIT 中の Ctrl-C は RELEASE 前で安全 → 外側 except に伝播させ
# _record_handover_interrupt が interrupted 記録
raise KeyboardInterrupt()
# ---- 6. RELEASE: GripperOpenOnly.open() (再送・再接続・移動なし) ----
# 防衛線: この直前で「open 以外の action は禁止」を宣言的に確認。
gate.assert_no_close_attempted("open")
release_command_ok = False
release_error: Optional[str] = None
release_sent_at: Optional[float] = None
release_responded_at: Optional[float] = None
sigint_install_error: Optional[str] = None
if dry_run:
# dry_run: signal を触らない (テスト環境で SIGINT ハンドラを差替
# ない・現状 pattern と同じ)。送信も skip。
release_sent_at = time.time()
release_command_ok = True
release_responded_at = time.time()
print("[HANDOVER dry_run] RELEASE skip (GripperOpenOnly.open 未実行)")
deferred_was_interrupted = False
else:
# E-3a C3: RELEASE 送信〜応答〜記録確定を SIGINT から保護。
# Ctrl-C が届いてもフラグだけ立ち、release_command_ok が確定した
# 後に KeyboardInterrupt を再送出する。「爪 open 済み・記録未確定」
# で interrupted になる record ズレを防ぐ。
deferred = _DeferredSigint()
with deferred:
release_sent_at = time.time()
try:
gate.release_open() # daemon 送信 + Pi 側 wait 2s
release_command_ok = True
except ForbiddenGripperClose as e:
release_error = "ForbiddenGripperClose: {}".format(e)
release_command_ok = False
print("[HANDOVER] RELEASE で close が要求された (構造破損): {}"
.format(e))
except Exception as e:
release_error = "{}: {}".format(type(e).__name__, e)
release_command_ok = False
print("[HANDOVER] RELEASE 失敗 (再送・再接続・移動なし): {}"
.format(e))
print("[HANDOVER] アームに触らない。復帰は退避確認後に手動で。")
release_responded_at = time.time()
deferred_was_interrupted = deferred.was_interrupted
sigint_install_error = deferred._install_error
log_record["handover"]["release"] = {
"release_command_ok": release_command_ok,
"release_error": release_error,
"release_sent_at": release_sent_at,
"release_responded_at": release_responded_at,
"dry_run": dry_run,
"sigint_deferred_during_send": (not dry_run),
"sigint_install_error": sigint_install_error,
}
# E-3a C3: 遅延中に SIGINT が来ていた場合、記録が確定した今こそ
# KeyboardInterrupt を再送出する。外側 except handler
# (_record_handover_interrupt) が interrupted 記録を追加する
# (release_command_ok は既に上で確定済み・両方 log_record に残る)。
if not dry_run and deferred_was_interrupted:
log_record["handover"]["release_interrupted_after_send"] = True
raise KeyboardInterrupt()
if not release_command_ok:
log_record["handover"]["phase"] = "aborted_at_release"
log_record["trial_status"] = "release_failed"
_abort(log_record, "handover_seq6",
"release 失敗 (自動移動なし・手動復帰へ)")
return False
# ---- 7. 撮影スタブ (E-5 で実装) ----
cap_status = _capture_after_release_stub(dry_run)
log_record["handover"]["capture"] = cap_status
# ---- 8. インターロック2 (退避開始前・hand_clear_confirmed・Q2: q で中断可) ----
if dry_run:
interlock2 = InterlockConfirmation(
label="interlock2_hand_clear", role="assistant",
ts_monotonic=time.monotonic(), ts_wall=time.time())
print("[HANDOVER dry_run] interlock2 skip (input なし)")
else:
# Q2 反映: interlock2 も同 wrapper 経由 (q 対応)。D-2 不変。
interlock2 = _wait_interlock_or_abort(
"interlock2_hand_clear",
"手を離し、アームの届く範囲の外へ出たことを確認したら Enter (退避開始)",
role="assistant")
if interlock2 is None:
log_record["handover"]["phase"] = "aborted_at_interlock2"
_record_interlock_abort(log_record, "interlock2_hand_clear")
return False
log_record["handover"]["interlock2"] = {
"label": interlock2.label, "role": interlock2.role,
"ts_monotonic": interlock2.ts_monotonic,
"ts_wall": interlock2.ts_wall,
}
log_record["handover"]["hand_clear_confirmed"] = True
# ---- 9. RETRACT_SMALL ----
rs_angles = pose_data["poses"]["RETRACT_SMALL"]["angles"]
r9 = _do_handover_move(rs_angles, "RETRACT_SMALL", pose_data,
log_record, dry_run)
if not r9["arrived"]:
log_record["handover"]["phase"] = "aborted_at_retract"
log_record["trial_status"] = "motion_failed"
log_record["manual_recovery_required"] = True
_abort(log_record, "handover_seq9",
"RETRACT_SMALL 未到達 (完了表示なし・手動復帰へ): {}"
.format(r9.get("reason")))
return False
# ---- 10. PRESENT_MID (退避) ----
r10 = _do_handover_move(pmid_angles, "PRESENT_MID", pose_data,
log_record, dry_run)
if not r10["arrived"]:
log_record["handover"]["phase"] = "aborted_at_pmid_retreat"
log_record["trial_status"] = "motion_failed"
log_record["manual_recovery_required"] = True
_abort(log_record, "handover_seq10",
"PRESENT_MID (退避) 未到達 (完了表示なし・手動復帰へ): {}"
.format(r10.get("reason")))
return False
# ---- 11. HOME ----
home_angles = pose_data["poses"]["HOME"]["angles"]
r11 = _do_handover_move(home_angles, "HOME", pose_data,
log_record, dry_run)
if not r11["arrived"]:
log_record["handover"]["phase"] = "aborted_at_home"
log_record["trial_status"] = "motion_failed"
log_record["manual_recovery_required"] = True
_abort(log_record, "handover_seq11",
"HOME 未到達 (完了表示なし・手動復帰へ): {}"
.format(r11.get("reason")))
return False
log_record["handover"]["phase"] = "complete"
log_record["trial_status"] = "completed"
return True
finally:
gate.exit_handover()
def _run_one_trial(
trial_index: int,
trial: Optional[dict],
pose_data: dict,
dry_run: bool,
log_record: Optional[dict] = None,
) -> tuple:
"""E-2a: PICK 区間 (STAGE1〜STAGE6 + HOME 復帰) を run_pick_place の下位
関数を呼んで実行する。handover sequence (PRESENT 系) は E-3・台帳追記は
E-4・VLM/撮影/regrasp は E-5 で追加する。
戻り値: (success: bool, log_record: dict)
success=True : PICK 完走 (stage1-4 到達・grip close 発火・lift → HOME)
success=False : abort_reason に理由 (stage1/2/3/4/home のいずれかの失敗)
r3.4 2節: PICK 失敗なら 2 以降 (提示工程) へ進まない。E-2a は成功時に
「PICK 完了・handover は E-3」を表示するだけで PRESENT へは進まない。
"""
_reset_observe_state_for_trial(trial_index, dry_run)
if log_record is None:
log_record = _init_log_record(trial_index, trial, dry_run)
# 実機モードのみ SSD マウント確認 + daemon 接続 + fresh_mode=1
# (run() L3836-3930 と同型・handover 側も同じ前提条件を確認)
# E-2b 準備で確認: _ensure_ssd_mounted は bool 返しで raise しない
# (L1086・fire-and-forget)。したがって if not _ensure_ssd_mounted() で受ける。
if not dry_run:
if not ensure_handover_ssd_mounted():
log_record["trial_status"] = "mount_lost"
_abort(log_record, 0,
"KIOXIA SSD 未マウント (run データの出力先が無い)")
return False, log_record
try:
get_daemon_socket()
except Exception as e:
# run() L3924 と同型 (daemon 接続失敗は stage=0 の abort)
_abort(log_record, 0, "daemon 接続失敗: {}".format(e))
return False, log_record
try:
daemon_set_fresh_mode(1)
except Exception as e:
# run() L3930 と同じく WARN で継続 (致命的ではない)
print("[E-2a] fresh_mode=1 失敗 (継続): {}".format(e))
# STAGE1: HOME + gripper open (run() L3933-3937 と同型)
# DAY265 で stage1_home_and_open は常に True を返す (失敗は stage_failures
# へ記録して継続する仕様・L3579-3600 参照)。ここの if は run() L3934 と
# 完全同一の形にしておく (parity・stage1 の I/F が将来変わっても対応)。
if not _guard_mount_before_sequence(log_record, "stage1", dry_run):
return False, log_record
_observe_phase("stage1")
if not stage1_home_and_open(log_record, dry_run):
# helpers 内で abort は set 済み想定 (現状は unreachable path)
return False, log_record
# STAGE2: block 検出 (require_box=False・handover は PLACE を経由しない)
# 失敗時は stage2_detect_block_and_box が内部で _abort(log_record, 2, ...) を
# 呼ぶ (L3646/3669/3673/3676 参照)。ここでは log_record["abort"] を上書き
# しない (helpers の記録がそのまま有効)。
if not _guard_mount_before_sequence(log_record, "stage2", dry_run):
return False, log_record
_observe_phase("stage2")
block_xz, _box_xz = stage2_detect_block_and_box(
log_record, dry_run,
dummy_block_xz=None, dummy_box_xz=None,
require_box=False,
)
if block_xz is None:
# stage2 内で _abort 済み → そのまま return
return False, log_record
# STAGE3: 補正記憶適用 + approach_pick (run() L3963-3991 と同型)
# phase="pick" は run() と一致 (correction_memory の名前空間分離)。
# 補正記憶の書込み (update) は VLM verdict "yes" 確定時のみ発火する仕様
# (run() L4171-4208)。E-2a は VLM を持たない (E-5) ため読取のみ実施し、
# 書込みは行わない (現状の correction_memory.json を汚さない)。
if not _guard_mount_before_sequence(log_record, "stage3", dry_run):
return False, log_record
target_xz_pick, corr_meta = _apply_correction("pick", block_xz)
log_record["pick_correction"] = {
"phase": "pick",
"block_xz": list(block_xz),
"target_xz_pick": list(target_xz_pick) if target_xz_pick is not None else None,
"corr_meta": corr_meta,
}
_observe_phase("stage3_approach_pick_handover")
ok, r_app_pick = stage_lookup_and_move(
"approach_pick",
target_xz_pick,
"handover_approach_pick", # label のみ handover 用に変更 (run() は "stage3_approach_pick")
TOL_APPROACH, # run() と同一 tol (定数 import)
log_record, dry_run, 3,
)
if not ok:
# stage_lookup_and_move 内で _abort 済み (L3789 or L3806) → そのまま return
return False, log_record
# stage6 lift で再利用 (run() L3992 と同型・run() ローカル変数の thread through)
approach_pick_target = list(r_app_pick["target_angles"])
# STAGE4: descend_pick (真下降下・j0 lock は run() L3998-4001 と同型)
if not _guard_mount_before_sequence(log_record, "stage4", dry_run):
return False, log_record
_observe_phase("stage4_pick_handover")
ok, _r_pick = stage_lookup_and_move(
"pick",
target_xz_pick,
"handover_pick", # label のみ handover 用
TOL_PICK_PLACE, # run() と同一 tol
log_record, dry_run, 4,
pick_j0_ref=float(approach_pick_target[0]), # run() と同型の j0 lock
)
if not ok:
# stage_lookup_and_move 内で _abort 済み → そのまま return
return False, log_record
# STAGE5: gripper close (run() L4050-4063 と同型・失敗は記録して継続)
# E-2b-redo bug#2 反映: run() L4051 が log_record["stages"].append("stage5"+suf)
# を明示的に呼んでいるので E-2a も同型で append する (parity mock 検出済)。
if not _guard_mount_before_sequence(log_record, "stage5", dry_run):
return False, log_record
_observe_phase("stage5_grip_close_handover")
log_record["stages"].append("stage5")
grip_ok = _do_gripper("close", "handover_grip_close", log_record, dry_run)
if not grip_ok:
# run() L4054-4059 と同型: 記録のみ・中断しない (次 stage で判断)
log_record["stage_failures"].append(
{"stage": 5, "label": "handover_grip_close", "reason": "gripper_close_failed"}
)
print("[E-2a] STAGE5 gripper close 失敗 (記録して継続・run() と同型)")
# STAGE6: lift back to approach_pick (run() L4066-4086 と同型・失敗は記録して継続)
# E-2b-redo bug#2 反映: run() L4067 が log_record["stages"].append("stage6"+suf)
# を明示的に呼んでいるので E-2a も同型で append する (parity mock 検出済)。
if not _guard_mount_before_sequence(log_record, "stage6", dry_run):
return False, log_record
_observe_phase("stage6_lift_handover")
log_record["stages"].append("stage6")
ok, _ = _do_move(
approach_pick_target,
"handover_lift",
TOL_APPROACH, # run() と同一 tol
log_record, dry_run=dry_run,
)
if not ok:
log_record["stage_failures"].append(
{"stage": 6, "label": "handover_lift", "reason": "lift_failed"}
)
print("[E-2a] STAGE6 lift 失敗 (記録して継続・run() と同型)")
# E-2a 独自: PICK 完了後の HOME 復帰 (parity 対象外・run() には相当経路なし)
#
# r3.4 3-5 節の handover_present_pose.json は HOME を基準姿勢とする ±15°
# 格子スキャンから選定しており、handover sequence は HOME から始まる。
# よって PICK → HOME への明示的な復帰を E-2a の最後で行う (E-3 が HOME
# 起点で PRESENT_MID へ進める前提)。run() 本線は PICK → PLACE (stage7) と
# 進むため run() 側にこの HOME 復帰は存在しない。
#
# 【重要・Q3 対応】この HOME 移動は _do_move (PICK 側の到達判定) ではなく
# handover_release.send_once_and_wait_arrival (handover 側の到達判定・
# deadline・per_call_cap) を使う。人へ向かう手前の HOME を handover の
# 安全規約下に置くため (r3.4 83-85行 + D-1 の per-call cap 経路)。
# pose JSON の HOME angles を使い、tolerance/deadline も pose JSON の値
# (arrival_tolerance_deg["HOME"] = 2.5° / deadline は clamp 済み) を使う。
# SPEED は SEND_WAIT_SEC=1.0 と MASTER 固定 (handover_release 内部)。
if not _guard_mount_before_sequence(log_record, "handover_pick_to_home", dry_run):
return False, log_record
_observe_phase("handover_pick_to_home")
home_angles = pose_data["poses"]["HOME"]["angles"]
if dry_run:
log_record["handover_pick_to_home"] = {
"dry_run": True,
"target": list(home_angles),
"path": "handover_release.send_once_and_wait_arrival (skipped in dry_run)",
}
else:
try:
res: ArrivalResult = send_once_and_wait_arrival(
home_angles, "HOME", pose_data
)
except PoseUnreachable as e:
_abort(log_record, "handover_home",
"handover_pick_to_home pre-actual unreachable: {}".format(e))
log_record["stage_failures"].append(
{"stage": "home", "label": "handover_pick_to_home",
"reason": "pre_actual_unreachable"}
)
return False, log_record
except SendFailed as e:
_abort(log_record, "handover_home",
"handover_pick_to_home send failed: {}".format(e))
log_record["stage_failures"].append(
{"stage": "home", "label": "handover_pick_to_home",
"reason": "send_failed"}
)
return False, log_record
log_record["handover_pick_to_home"] = {
"arrived": bool(res.arrived),
"max_axis_diff_deg": float(res.max_axis_diff_deg),
"deadline_sec": float(res.deadline_sec),
"ts_send_wall": float(res.ts_send_wall),
"ts_arrival_wall":
float(res.ts_arrival_wall) if res.ts_arrival_wall is not None else None,
"reason": res.reason,
"path": "handover_release.send_once_and_wait_arrival",
}
if not res.arrived:
log_record["stage_failures"].append(
{"stage": "home", "label": "handover_pick_to_home",
"reason": res.reason or "arrival_timeout"}
)
_abort(log_record, "handover_home",
"handover_pick_to_home {}".format(res.reason or "arrival_timeout"))
print("[E-2a] HOME 復帰未到達 → 手動復帰が必要 (r3.4 8節・自動移動なし)")
return False, log_record
print("[E-2a] PICK 完了・handover へ移行 (E-3)")
# ------------------------------------------------------------------
# E-3: handover sequence (r3.4 2節 sequence 2-11)
# Ctrl-C の分離:
# - PICK 区間 (上記まで): 内部 try/except KeyboardInterrupt を **持たない**。
# Ctrl-C は上位 (run_resident_loop) の except KeyboardInterrupt に届き、
# run_pick_place.run() と同じ「メニューに戻らずセッション終了」挙動
# (parity 対象・E-1/E-2 互換・変えない)。
# - handover 区間 (下記): 内部で try/except KeyboardInterrupt を張り、
# _record_handover_interrupt で trial_status=interrupted を記録し
# return False, log_record (通常戻り)。上位の except KeyboardInterrupt
# には届かないため常駐ループはメニューへ戻る (セッション継続・r3.4 8節)。
# ------------------------------------------------------------------
try:
handover_ok = _run_handover_sequence(trial, pose_data, log_record, dry_run)
except KeyboardInterrupt:
_record_handover_interrupt(log_record)
return False, log_record
if not handover_ok:
return False, log_record
print("[E-3] handover 完了")
return True, log_record
def _finalize_and_record_trial(
log_record: dict,
success: bool,
pose_data: dict,
wait_plan_id: str,
shuffle_seed: int,
dry_run: bool,
) -> str:
"""Finalize annotations, run JSON and one ledger append (or journal)."""
if not log_record.get("trial_status"):
if success:
log_record["trial_status"] = "completed"
elif log_record.get("handover"):
log_record["trial_status"] = "unexpected_error"
else:
log_record["trial_status"] = "pick_failed"
has_closed_gripper = any(
x.get("action") == "close" for x in log_record.get("gripper", []))
status_needs_manual = log_record["trial_status"] in {
"motion_failed", "release_failed", "interrupted", "unexpected_error"
}
log_record.setdefault(
"manual_recovery_required", bool(has_closed_gripper or status_needs_manual))
log_record["ended_at"] = datetime.now().astimezone().isoformat()
if dry_run:
log_record["operator_annotations"] = {
"outcome": ("not_applicable" if log_record["trial_status"] == "pick_failed"
else "unknown"),
"outcome_source": "unknown",
"release_observed": "unknown",
"release_observed_source": "operator",
}
else:
collect_operator_annotations(log_record)
trial = log_record.get("trial") or {}
plan = {
"wait_plan_id": wait_plan_id,
"wait_plan_index": trial.get("wait_plan_index", ""),
"shuffle_seed": shuffle_seed,
"wait_sec_requested": trial.get("wait_sec", ""),
"block_index": trial.get("block_index", ""),
}
row = build_ledger_row(log_record, pose_data, plan)
payload = build_run_payload(log_record, pose_data, plan, row)
if dry_run:
print("[E-4 dry_run] 台帳/run_dir/journal 書込みなし: {}".format(
log_record["run_stem"]))
return "dry_run"
deferred = _DeferredSigint()
with deferred:
result = persist_or_journal(payload)
log_record["recording_result"] = result
log_record["recording_sigint_install_error"] = deferred._install_error
if deferred.was_interrupted:
print("[E-4] Ctrl-C は記帳確定後に処理しました。")
raise KeyboardInterrupt()
return result
def _format_trial_summary(log_data) -> str:
"""常駐ループの表示用の短い要約 (log_record 全文は verbose なため)。"""
if not isinstance(log_data, dict):
return str(log_data)
parts = []
abort = log_data.get("abort")
if isinstance(abort, dict):
parts.append("abort=stage{} reason={!r}".format(
abort.get("stage"), abort.get("reason")))
if log_data.get("stage_failures"):
parts.append("stage_failures={}".format(len(log_data["stage_failures"])))
n_moves = len(log_data.get("moves", []))
n_grip = len(log_data.get("gripper", []))
parts.append("moves={} gripper={}".format(n_moves, n_grip))
return " ".join(parts)
# ---------------------------------------------------------------------------
# 常駐ループ (r3.4 94行・参照元 run_pick_place.py L5181-5241)
# ---------------------------------------------------------------------------
def run_resident_loop(
pose_data: dict,
trials: list,
wait_plan_id: str,
shuffle_seed: int,
dry_run: bool,
) -> None:
"""DAY283 §3-A の常駐ループを handover 側に写した版。
n または Enter = 次試行、h = HOME、q または EOF(Ctrl-D) = 終了。
プロンプト冒頭に「掴んだまま失敗した場合は h で HOME 確認」を明示するのは
参照元 L5189-5191 と同じ (前試行が把持中に失敗した場合、次試行の gripper
open で空中落下しうるため)。
trials 使い切り後の semantics (E-1a codex 意見反映で確定):
**循環しない・使い切ったら常駐ループを終了**。balanced block は各 N が
同数ちょうどの測定設計であり、循環は偏りを生む。E-2 で「追加ブロック
生成」が要るなら、そのとき別途 CLI オプションを足す。
h / q は plan 消化に関わらず通常どおり効く。
"""
print("[LOOP] 常駐モード開始. コマンド: n または Enter=次試行 / h=HOME 復帰 / q=終了")
print(
"[LOOP] 注意: 前試行が把持中に失敗 (HOME 未到達で握ったまま等) した場合,"
" 次試行の gripper open で空中落下し得る."
" その場合は h で HOME を試行してから n で次試行へ進むこと."
)
print("[LOOP] wait_plan_id={} shuffle_seed={} trials_total={}".format(
wait_plan_id, shuffle_seed, len(trials)))
trial_counter = 0
try:
while True:
try:
cmd = input(
"\n[LOOP] 次の操作 [n=次試行 (Enter可) / h=HOME / q=終了]: "
).strip().lower()
except EOFError:
print("[LOOP] EOF 検出 → 常駐モード終了")
break
if cmd in ("q", "quit", "exit"):
print("[LOOP] q 受領 → 常駐モード終了")
break
if cmd == "h":
_manual_home(pose_data, dry_run=dry_run)
continue
if cmd in ("", "n", "next"):
# E-1a 保留1 確定: 使い切ったら常駐ループを終了 (循環なし)
if trial_counter >= len(trials):
print(
"[LOOP] 全 trial 消化済み ({}/{}) → 常駐モード終了".format(
trial_counter, len(trials)
)
)
break
trial_counter += 1
trial = trials[trial_counter - 1]
print("[LOOP] === 試行 #{}/{} 開始 ===".format(trial_counter, len(trials)))
log_data = _init_log_record(trial_counter, trial, dry_run)
success = False
try:
success, log_data = _run_one_trial(
trial_counter, trial, pose_data, dry_run, log_record=log_data
)
except Exception as e:
log_data["trial_status"] = "unexpected_error"
log_data["manual_recovery_required"] = True
_abort(log_data, "unexpected_error",
"{}: {}".format(type(e).__name__, e))
print("[LOOP] unexpected error; 新規指令・自動 HOME なし: {}: {}"
.format(type(e).__name__, e))
_finalize_and_record_trial(
log_data, False, pose_data, wait_plan_id, shuffle_seed, dry_run)
print(
"[LOOP] === 試行 #{}/{} 終了: success=False {} ===".format(
trial_counter, len(trials), _format_trial_summary(log_data)))
continue
except KeyboardInterrupt:
log_data["trial_status"] = "interrupted"
log_data["abort_stage"] = "pick"
log_data["manual_recovery_required"] = True
_abort(log_data, "pick_interrupt",
"PICK 中 Ctrl-C: 新規指令なし・自動 HOME なし")
print(
"[LOOP] PICK 中 Ctrl-C 受領. 新規指令・自動 HOME なし。"
" unknown 対話値で記帳してセッション終了します。"
)
log_data["operator_annotations"] = {
"outcome": "unknown", "outcome_source": "unknown",
"release_observed": "unknown",
"release_observed_source": "operator",
}
log_data["ended_at"] = datetime.now().astimezone().isoformat()
trial_plan = {
"wait_plan_id": wait_plan_id,
"wait_plan_index": trial.get("wait_plan_index", ""),
"shuffle_seed": shuffle_seed,
"wait_sec_requested": trial.get("wait_sec", ""),
"block_index": trial.get("block_index", ""),
}
if not dry_run:
row = build_ledger_row(log_data, pose_data, trial_plan)
persist_or_journal(build_run_payload(
log_data, pose_data, trial_plan, row))
break
_finalize_and_record_trial(
log_data, success, pose_data, wait_plan_id, shuffle_seed, dry_run)
print(
"[LOOP] === 試行 #{}/{} 終了: success={} {} ===".format(
trial_counter, len(trials), success,
_format_trial_summary(log_data),
)
)
continue
print("[LOOP] 未知のコマンド: {!r}. n / Enter / h / q のいずれか".format(cmd))
except KeyboardInterrupt:
print("\n[LOOP] プロンプト中 Ctrl-C 受領.")
print(
"[LOOP] 前試行が未完走または HOME 未到達の場合, 現在姿勢と把持状態を"
" 目視確認のこと. 必要なら h で HOME を試行してから次試行へ."
)
print("[LOOP] 常駐モード終了. 実行試行数 = {}".format(trial_counter))
# ---------------------------------------------------------------------------
# CLI エントリポイント
# ---------------------------------------------------------------------------
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
description="run_pick_handover.py: r3.4 実装物 5 (E-1 骨格・E-2 以降で sequence 本体)"
)
p.add_argument(
"--wait-plan",
type=str,
default="2,3,5,8",
help="N 集合 (CSV・整数)。既定 '2,3,5,8' (r3.4 §12 codex 承認済み)。",
)
p.add_argument(
"--block-repeat",
type=int,
default=2,
help="balanced block のブロック数 (各ブロックが wait_plan の順列 1 つ)。既定 2。",
)
p.add_argument(
"--seed",
type=int,
default=None,
help="shuffle_seed。省略時は 0..2^31-1 の乱数を生成し記録する。",
)
p.add_argument(
"--dry-run",
action="store_true",
help="実機送角なし (スタブに渡す・_manual_home も skip)。",
)
p.add_argument(
"--pose-json",
type=str,
default=str(POSE_JSON_PATH),
help="handover_present_pose.json のパス。既定は handover_release.POSE_JSON_PATH。",
)
p.add_argument(
"--print-plan",
action="store_true",
help="N 計画を生成して表示するだけで終了 (sequence を回さない・balanced block 再現性の検証用)。",
)
p.add_argument(
"--recover",
nargs="?",
const="",
metavar="RUN_STEM",
help=("SSD 復旧後に emergency journal を回収。値なしで全件、"
"RUN_STEM 指定で1件。物理動作は一切行わない。"),
)
return p
def main(argv: Optional[list] = None) -> int:
parser = build_arg_parser()
args = parser.parse_args(argv)
if args.recover is not None:
try:
recovered, skipped, failed = recover_journals(args.recover or None)
except AppendAborted as e:
print(str(e), file=sys.stderr)
return 2
print("[RECOVER] appended={} already_recorded={} failed={}".format(
recovered, skipped, failed))
return 0 if failed == 0 else 3
# 起動時バナー (handover_release.STARTUP_BANNER を先に表示)
print(STARTUP_BANNER)
print("[INIT] pose JSON: {}".format(args.pose_json))
# pose JSON を先に読んで validate (sequence 開始前に落とす)
try:
pose_data = load_and_validate_pose_json(Path(args.pose_json))
except (ValueError, FileNotFoundError, PermissionError) as e:
print("[INIT] pose JSON 検証失敗: {}: {}".format(type(e).__name__, e), file=sys.stderr)
return 2
# N 計画を作る (seed 未指定なら生成・記録用に seed を保持)
try:
wait_set = parse_wait_plan_arg(args.wait_plan)
trials, wait_plan_id, shuffle_seed = generate_wait_plan(
wait_set, args.block_repeat, seed=args.seed
)
except ValueError as e:
print("[INIT] wait-plan 検証失敗: {}".format(e), file=sys.stderr)
return 2
print(
"[INIT] wait_plan={} block_repeat={} shuffle_seed={} trials_total={}".format(
wait_set, args.block_repeat, shuffle_seed, len(trials)
)
)
print("[INIT] wait_plan_id={}".format(wait_plan_id))
if args.print_plan:
# sequence を回さず、生成した提示順を出して終わる (再現性検証用)
for i, t in enumerate(trials):
print(
"[PLAN] trial#{i:03d} wait={wait}s wait_plan_index={idx} block={blk}".format(
i=i + 1, wait=t["wait_sec"], idx=t["wait_plan_index"], blk=t["block_index"]
)
)
return 0
# 常駐ループへ (n/h/q・E-1 の trial は _run_one_trial_stub がスタブ)
run_resident_loop(
pose_data=pose_data,
trials=trials,
wait_plan_id=wait_plan_id,
shuffle_seed=shuffle_seed,
dry_run=args.dry_run,
)
return 0
if __name__ == "__main__":
sys.exit(main())二つに共通していたこと
FK テーブルの件も、parity mock の件も、同じ形をしていた。
FK テーブルには HOME があるはずだと思っていた。無かった。parity mock は既存本線と比べているつもりだった。比べていなかった。どちらも「もう測ってある」「もう検証してある」と思い込んでいた場所が、実は測っても検証してもいなかった。そして、どちらも実際にコマンドを一つ打てば――awk を一回、run() を一回走らせれば――即座に分かることだった。
「測っていない」を「測った」と思い込む。それが一番静かに、一番深く潜む。
