R2D2を作りたい その14　ついに物体認識！

sudo apt update
sudo apt full-upgrade        # AI Camera はファームウェアが新しい必要がある
sudo apt install python3-picamera2 imx500-all imx500-tools
sudo reboot

rpicam-hello --list-cameras
# → "imx500" が表示されれば OK

# 学習・変換用の venv
python3.11 -m venv .venv-train

# PyTorch は CPU 版を明示（GPU 版を引くと数 GB 級のダウンロードになる）
.venv-train/bin/pip install torch==2.11.0 torchvision==0.26.0 \
  --index-url https://download.pytorch.org/whl/cpu

# Ultralytics と IMX500 変換ツールチェーン（バージョン固定）
.venv-train/bin/pip install ultralytics==8.4.115 \
  model-compression-toolkit==2.4.5 edge-mdt-cl==1.0.0 "imx500-converter[pt]==3.17.3"

# 修正前
p.drawRect(left_top.x(), left_top.y(), rect_width, rect_height)
...
p.drawLine(self.prev_point.x(), 0, self.prev_point.x(), self.pixmap.height())
p.drawLine(0, self.prev_point.y(), self.pixmap.width(), self.prev_point.y())

# 修正後
p.drawRect(int(left_top.x()), int(left_top.y()), int(rect_width), int(rect_height))
...
px, py = int(self.prev_point.x()), int(self.prev_point.y())
p.drawLine(px, 0, px, int(self.pixmap.height()))
p.drawLine(0, py, int(self.pixmap.width()), py)

# scroll_request（ホイールスクロール。966 行付近）
# 修正前
bar.setValue(bar.value() + bar.singleStep() * units)
# 修正後
bar.setValue(int(bar.value() + bar.singleStep() * units))

# set_zoom（ズーム値の設定。972 行付近）
# 修正前
self.zoom_widget.setValue(value)
# 修正後
self.zoom_widget.setValue(int(value))

# zoom_request（ズーム時のスクロール位置調整。1026 行付近）
# 修正前
h_bar.setValue(new_h_bar_value)
v_bar.setValue(new_v_bar_value)
# 修正後
h_bar.setValue(int(new_h_bar_value))
v_bar.setValue(int(new_v_bar_value))

#!/usr/bin/env python3
"""カスタムモデル学習用のデータセットを Zero 2 W の AI Camera で収集する。

一定間隔で JPEG を連番保存する。カメラを手で動かしながら実行し、
角度・距離・照明のバリエーションを稼ぐ。

推論はしないので IMX500 のネットワークファーム転送は発生せず、すぐ撮影が始まる。
本番と同じ光学系・画角で集めることが目的なので、スマホ等ではなくこのカメラで撮る。

例:
    # テレビ電源オフの状態を 100 枚（1 秒間隔）
    python3 capture_dataset.py --out ~/dataset/tv_off --count 100

    # 中断しても連番は続きから振られるので、そのまま再実行してよい
    python3 capture_dataset.py --out ~/dataset/tv_off --count 50
"""

import argparse
import sys
import time
from pathlib import Path

from picamera2 import Picamera2


def parse_size(text):
    try:
        w, h = text.lower().split("x")
        return int(w), int(h)
    except ValueError:
        raise argparse.ArgumentTypeError(f"WIDTHxHEIGHT の形式で指定する: {text}") from None


def parse_args():
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--out", required=True, type=Path, help="保存先ディレクトリ（無ければ作る）")
    p.add_argument("--count", type=int, default=100, help="撮影枚数")
    p.add_argument("--interval", type=float, default=1.0, help="撮影間隔（秒）")
    p.add_argument("--size", type=parse_size, default=(1280, 960),
                   help="解像度。推論時と同じ 4:3 を保つこと（既定 1280x960）")
    p.add_argument("--prefix", help="ファイル名の接頭辞（既定: 出力ディレクトリ名）")
    p.add_argument("--delay", type=float, default=3.0, help="開始までの猶予秒数")
    return p.parse_args()


def next_index(out_dir, prefix):
    """既存の連番の続きを返す。中断して再開しても上書きしないため。"""
    existing = sorted(out_dir.glob(f"{prefix}_*.jpg"))
    for path in reversed(existing):
        tail = path.stem.rsplit("_", 1)[-1]
        if tail.isdigit():
            return int(tail) + 1
    return 1


def main():
    args = parse_args()
    out_dir = args.out.expanduser()
    out_dir.mkdir(parents=True, exist_ok=True)
    prefix = args.prefix or out_dir.name

    picam2 = Picamera2()
    picam2.configure(picam2.create_still_configuration(main={"size": args.size}))
    picam2.start()
    time.sleep(2)  # AE / AWB が落ち着くのを待つ

    index = next_index(out_dir, prefix)
    w, h = args.size
    print(f"# 保存先: {out_dir}  接頭辞: {prefix}  開始番号: {index:04d}", file=sys.stderr)
    print(f"# {w}x{h} を {args.interval}s 間隔で {args.count} 枚", file=sys.stderr)
    for remaining in range(int(args.delay), 0, -1):
        print(f"# {remaining}...", file=sys.stderr, flush=True)
        time.sleep(1)
    print("# 開始（カメラを少しずつ動かしてください / Ctrl-C で中断）", file=sys.stderr, flush=True)

    saved = 0
    try:
        for i in range(args.count):
            started = time.monotonic()
            path = out_dir / f"{prefix}_{index:04d}.jpg"
            picam2.capture_file(str(path))
            saved += 1
            index += 1
            print(f"[{i + 1:4d}/{args.count}] {path.name}", flush=True)
            # 撮影自体にかかった時間を差し引いて間隔を一定に保つ
            time.sleep(max(0.0, args.interval - (time.monotonic() - started)))
    except KeyboardInterrupt:
        print(file=sys.stderr)
    finally:
        picam2.stop()
        total = len(list(out_dir.glob(f"{prefix}_*.jpg")))
        print(f"# {saved} 枚保存（{out_dir} の合計 {total} 枚）", file=sys.stderr)


if __name__ == "__main__":
    main()

# WSL 側: スクリプトを実機へ転送
rsync -az --exclude '__pycache__' pi/ <実機ホスト>:~/r2d2_dev/pi/

# 実機側: セッションごとに撮影（対象の状態を変えながら 6 回）
cd ~/r2d2_dev/pi
python3 capture_dataset.py --out ~/dataset/tv1_on  --count 80   # 大テレビ 電源オン
python3 capture_dataset.py --out ~/dataset/tv1_off --count 80   # 大テレビ 電源オフ
python3 capture_dataset.py --out ~/dataset/tv2_on  --count 80   # 小テレビ 電源オン
python3 capture_dataset.py --out ~/dataset/tv2_off --count 80   # 小テレビ 電源オフ
python3 capture_dataset.py --out ~/dataset/laptop  --count 80   # ノート PC
python3 capture_dataset.py --out ~/dataset/mixed   --count 80   # 複数対象の構図

# WSL 側: 撮影した画像を開発機へ回収
rsync -az <実機ホスト>:~/dataset/ dataset/

<クラス番号> <中心x> <中心y> <幅> <高さ>

printf 'tv1\ntv2\nlaptop\n' > dataset/classes.txt
.venv-annot/bin/labelImg dataset/tv1_on dataset/classes.txt dataset/tv1_on

#!/usr/bin/env python3
"""アノテーションの抜けと古いクラスを洗い出す。WSL 側で実行する。

物体検出の学習では、ラベルが付いていない領域は「背景」として学習される。
そのため写り込んだ対象にラベルを付け忘れると、同じ物体に対して
「あるセッションでは前景、別のセッションでは背景」という矛盾した教師信号を与え、
クラスの区別が壊れる。

ここでは COCO 版 YOLO11n で画面らしきものを検出し、既存のラベルと IoU で
照合して「検出されたのにラベルが無い領域」を持つ画像を報告する。
COCO モデルはテレビを苦手（実測で tv セッションの 3 割が未検出）なので、
これは取りこぼしを減らす補助であって、目視確認の代わりにはならない。

例:
    .venv-train/bin/python tools/audit_labels.py --dataset dataset
    .venv-train/bin/python tools/audit_labels.py --dataset dataset --conf 0.15
"""

import argparse
import sys
from pathlib import Path

# 最終的に学習させたいクラス。これ以外が残っていれば修正漏れ。
WANTED = {"tv1", "tv2", "laptop"}

# COCO 側で画面らしきものとして拾うクラス
COCO_SCREENS = {"tv", "laptop"}


def parse_args():
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--dataset", type=Path, default=Path("dataset"))
    p.add_argument("--model", default="yolo11n.pt")
    p.add_argument("--conf", type=float, default=0.20, help="検出閾値。低くすると取りこぼしが減る")
    p.add_argument("--iou", type=float, default=0.35,
                   help="この IoU 以上で既存ラベルと重なれば「ラベル済み」と見なす")
    return p.parse_args()


def to_corners(box):
    """YOLO 形式 (cx, cy, w, h) を (x1, y1, x2, y2) にする。"""
    cx, cy, w, h = box
    return cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2


def iou(a, b):
    ax1, ay1, ax2, ay2 = to_corners(a)
    bx1, by1, bx2, by2 = to_corners(b)
    ix1, iy1 = max(ax1, bx1), max(ay1, by1)
    ix2, iy2 = min(ax2, bx2), min(ay2, by2)
    if ix2 <= ix1 or iy2 <= iy1:
        return 0.0
    inter = (ix2 - ix1) * (iy2 - iy1)
    area_a = (ax2 - ax1) * (ay2 - ay1)
    area_b = (bx2 - bx1) * (by2 - by1)
    return inter / (area_a + area_b - inter)


def read_labels(txt_path):
    """YOLO 形式のラベルを [(class_id, (cx, cy, w, h)), ...] で返す。"""
    if not txt_path.exists():
        return []
    rows = []
    for line in txt_path.read_text().splitlines():
        parts = line.split()
        if len(parts) >= 5:
            rows.append((int(parts[0]), tuple(float(x) for x in parts[1:5])))
    return rows


def main():
    args = parse_args()
    dataset = args.dataset.expanduser().resolve()
    classes = (dataset / "classes.txt").read_text().splitlines()
    print(f"# classes.txt: {classes}")

    stale = []   # 学習対象でないクラスが残っているもの
    empty = []   # ラベルが空のもの
    missing = {}  # セッション → 未ラベル領域がある画像

    from ultralytics import YOLO

    model = YOLO(args.model)
    coco_ids = [i for i, name in model.names.items() if name in COCO_SCREENS]

    for session_dir in sorted(d for d in dataset.iterdir() if d.is_dir()):
        images = sorted(session_dir.glob("*.jpg"))
        if not images:
            continue

        # 先にラベルの中身だけを検査する（推論より速いので分けておく）
        labels = {}
        for image in images:
            rows = read_labels(image.with_suffix(".txt"))
            labels[image] = rows
            if not rows:
                empty.append(image.stem)
            for class_id, _ in rows:
                name = classes[class_id] if class_id < len(classes) else f"?{class_id}"
                if name not in WANTED:
                    stale.append((image.with_suffix(".txt").relative_to(dataset), name))

        predictions = model.predict(source=images, conf=args.conf, classes=coco_ids,
                                    stream=True, verbose=False)
        found = []
        for image, result in zip(images, predictions):
            existing = [box for _, box in labels[image]]
            uncovered = 0
            for box in result.boxes:
                cand = tuple(box.xywhn[0].tolist())
                if all(iou(cand, ref) < args.iou for ref in existing):
                    uncovered += 1
            if uncovered:
                found.append((image.stem, uncovered))
        if found:
            missing[session_dir.name] = found

    print()
    print("=== 学習対象でないクラスが残っているラベル ===")
    if stale:
        for path, name in stale:
            print(f"  {path}  → {name}")
    else:
        print("  なし")

    print()
    print("=== COCO が検出したのにラベルが無い領域を含む画像 ===")
    print("（COCO はテレビを苦手なので、ここに出ないものも目視確認が必要）")
    if not missing:
        print("  なし")
    for session, found in missing.items():
        names = ", ".join(f"{stem}({n})" for stem, n in found[:8])
        more = f" ほか{len(found) - 8}枚" if len(found) > 8 else ""
        print(f"  {session:<12} {len(found):3d} 枚: {names}{more}")

    print()
    print(f"=== ラベルが空の画像: {len(empty)} 枚 ===")
    print("（対象が写っていないなら背景サンプルとして有効。意図しない空なら要修正）")
    if empty:
        print("  " + ", ".join(empty[:12]) + (f" ほか{len(empty) - 12}枚" if len(empty) > 12 else ""))
    return 1 if stale else 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""dataset/ を YOLO 学習形式へ統合し、train/val に分割する。WSL 側で実行する。

セッションフォルダ（撮影管理の単位）ごとに同じ比率で val を抽出する層化分割。
単純ランダムだと val に特定の撮影条件（例: tv2_off）がほとんど入らない偏りが
起こり得るため、全条件が両方の分割に含まれるようにする。

dataset/ は原本としてそのまま残し、datasets/<名前>/ へコピーする。
ラベルが空の画像も背景サンプルとしてコピーする（YOLO はラベル無しを背景として扱う）。

例:
    .venv-train/bin/python tools/split_dataset.py
    .venv-train/bin/python tools/split_dataset.py --val-ratio 0.2 --seed 42
"""

import argparse
import random
import shutil
import sys
from pathlib import Path

CLASSES = ["tv1", "tv2", "laptop"]


def parse_args():
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--dataset", type=Path, default=Path("dataset"), help="原本（セッションフォルダ群）")
    p.add_argument("--out", type=Path, default=Path("datasets/appliances"), help="出力先")
    p.add_argument("--val-ratio", type=float, default=0.2)
    p.add_argument("--seed", type=int, default=42)
    return p.parse_args()


def main():
    args = parse_args()
    dataset = args.dataset.expanduser().resolve()
    out = args.out.expanduser().resolve()

    if out.exists():
        print(f"出力先 {out} が既に存在する → 中断（消すか --out を変えてから再実行）")
        return 1

    classes = (dataset / "classes.txt").read_text().splitlines()
    if classes != CLASSES:
        print(f"classes.txt が想定と違う: {classes}（想定: {CLASSES}）→ 中断")
        return 1

    # セッションごとに収集し、ファイル名の重複が無いことを確かめる
    # （出力先はフラットに置くため、重複すると上書きが起こる）
    sessions = {}
    seen = {}
    for session_dir in sorted(d for d in dataset.iterdir() if d.is_dir()):
        images = sorted(session_dir.glob("*.jpg"))
        if not images:
            continue
        sessions[session_dir.name] = images
        for image in images:
            if image.name in seen:
                print(f"ファイル名が重複: {seen[image.name]} と {image} → 中断")
                return 1
            seen[image.name] = image

    rng = random.Random(args.seed)
    plan = {"train": [], "val": []}
    print(f"層化分割 (val {args.val_ratio:.0%}, seed {args.seed}):")
    for name, images in sessions.items():
        shuffled = images[:]
        rng.shuffle(shuffled)
        n_val = round(len(images) * args.val_ratio)
        plan["val"] += shuffled[:n_val]
        plan["train"] += shuffled[n_val:]
        print(f"  {name:<10} {len(images):3d} 枚 → train {len(images) - n_val:3d} / val {n_val:2d}")

    for split, images in plan.items():
        img_dir = out / "images" / split
        lbl_dir = out / "labels" / split
        img_dir.mkdir(parents=True)
        lbl_dir.mkdir(parents=True)
        empty = 0
        for image in images:
            shutil.copy2(image, img_dir / image.name)
            label = image.with_suffix(".txt")
            if label.exists() and label.read_text().strip():
                shutil.copy2(label, lbl_dir / label.name)
            else:
                empty += 1  # ラベル無し = 背景サンプル。ファイルは置かなくてよい
        print(f"{split}: 画像 {len(images)} 枚（うち背景 {empty} 枚）")

    yaml_path = out / "data.yaml"
    names = "\n".join(f"  {i}: {name}" for i, name in enumerate(CLASSES))
    yaml_path.write_text(
        f"path: {out}\ntrain: images/train\nval: images/val\nnames:\n{names}\n"
    )
    print(f"\ndata.yaml: {yaml_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

ULTRALYTICS_AUTOINSTALL=false .venv-train/bin/yolo train \
  model=yolo11n.pt data=datasets/appliances/data.yaml \
  imgsz=640 epochs=100 patience=20 batch=8 workers=4 device=cpu

Remapped 1/3 cls head rows from pretrained weights by class name
Transferred 451/499 items from pretrained weights

ULTRALYTICS_AUTOINSTALL=false .venv-train/bin/yolo export \
  model=runs/detect/train/weights/best.pt format=imx \
  data=datasets/appliances/data.yaml imgsz=640

# WSL 側
scp runs/detect/train/weights/best_imx_model/{packerOut.zip,labels.txt} \
  <実機>:~/models/yolo11n_appliances/

# 実機側
cd ~/models/yolo11n_appliances
imx500-package -i packerOut.zip -o .   # → network.rpk (3.2MB)

{
 "Memory Report": {
  "Runtime Memory Physical Size": "4.48MB",
  "Model Memory Physical Size": "2.64MB",
  "Memory Usage": "7.12MB",
  "Total Memory Available On Chip": "8.00MB",
  "Memory Utilization": "90%",
  "Fit In Chip": true
 }
}

"""cv2 が無い環境で picamera2 の IMX500 サポートを import できるようにする。

`picamera2.devices.imx500.__init__` が postprocess モジュール群を無条件に import し、
それらが `import cv2` する。ただし cv2 の実際の呼び出しは
  - 検出結果の描画 (rectangle / putText / line / circle)
  - Pi 側で後処理が必要なモデル (nanodet / yolov5 / yolov8) の resize
に限られる。pp 済みモデル (imx500_network_*_pp.rpk) をヘッドレスで使う本プロジェクトでは
どちらも通らない。

Debian の python3-opencv は libopencv-viz 経由で VTK / GDAL / Qt5 まで引き込み
72 パッケージ増えるため、RAM 415MB の Zero 2 W では割に合わない。
そこで cv2 が無ければスタブを差す。本物が入っていればそちらを使うので、
後から `sudo apt install python3-opencv` しても壊れない。
"""

import sys
import types


class _Cv2Stub(types.ModuleType):
    """属性アクセスされた時点で、何が足りないかを明示して失敗する。

    黙って None を返すと検出結果が静かに壊れるので、必ず例外にする。
    """

    def __getattr__(self, name):
        raise ModuleNotFoundError(
            f"cv2.{name} が必要な処理が呼ばれた。描画または Pi 側後処理"
            "（nanodet / yolo 系モデル）を使うなら実機に opencv が必要:\n"
            "  sudo apt install python3-opencv   # 72 パッケージ増える\n"
            "  もしくは venv に opencv-python-headless"
        )


def install() -> bool:
    """cv2 が無ければスタブを sys.modules に差す。picamera2 の import より前に呼ぶ。

    Returns:
        True なら本物の cv2、False ならスタブを使っている。
    """
    existing = sys.modules.get("cv2")
    if existing is not None:
        return not isinstance(existing, _Cv2Stub)
    try:
        import cv2  # noqa: F401
    except ModuleNotFoundError:
        sys.modules["cv2"] = _Cv2Stub("cv2")
        return False
    return True

#!/usr/bin/env python3
"""M1: AI Camera (IMX500) 単体で物体検出できることをヘッドレスで確認する。

Zero 2 W 上で実行する。公式デモ (imx500_object_detection_demo.py) と違い
プレビュー表示も OpenCV も使わないので SSH 越しにそのまま動く。
推論は IMX500 内で完結し、ここでは検出結果のメタデータを読むだけ。

SSD MobileNetV2 は画角全体を弱いスコアで何かのクラスに当てはめる誤検出が多い。
それを切り分けられるよう、面積比 (bbox が画角に占める割合) を常に出力し、
終了時にラベル別の統計を表示する。

例:
    python3 check_detect.py --list-labels        # モデルの認識クラス一覧
    python3 check_detect.py --summary            # 統計のみ（誤検出の実態調査用）
    python3 check_detect.py --label tv -t 0      # tv だけを Ctrl-C まで監視
    python3 check_detect.py --max-area 0.8 --stable 5   # 誤検出を絞る
    python3 check_detect.py --json > det.jsonl   # 1 行 1 JSON で記録
    python3 check_detect.py --preview -t 0       # デスクトップで映像+検出枠を表示

--preview は実機のデスクトップ（またはそのターミナル）から使う。cv2 は使わず、
picamera2 のプレビューにオーバーレイ（numpy 描画の RGBA 画像）で枠を重ねる。
文字は描けないためクラスは枠の色で区別する（起動時に色の対応を表示）。
"""

import argparse
import json
import statistics
import sys
import time
from collections import deque
from pathlib import Path

import numpy as np

import _cv2_stub

# picamera2 の IMX500 サポートは cv2 を無条件 import する。実機に opencv を
# 入れていない場合はスタブで代替する（詳細は _cv2_stub の docstring）。
HAVE_REAL_CV2 = _cv2_stub.install()

from picamera2 import Picamera2
from picamera2.devices import IMX500
from picamera2.devices.imx500 import NetworkIntrinsics

DEFAULT_MODEL = "/usr/share/imx500-models/imx500_network_ssd_mobilenetv2_fpnlite_320x320_pp.rpk"


def parse_args():
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--model", default=DEFAULT_MODEL, help="rpk モデルのパス")
    p.add_argument("--threshold", type=float, default=0.55, help="検出スコアの閾値")
    p.add_argument("--label", action="append", default=None,
                   help="対象クラス名（複数指定可）。省略時は全クラス")
    p.add_argument("--labels", metavar="PATH",
                   help="ラベルファイル（1 行 1 クラス）。自作 rpk はラベルを持たないので必須")
    p.add_argument("--bbox-order", choices=["yx", "xy"], default=None,
                   help="bbox の座標順。yx=(y0,x0,y1,x1) / xy=(x0,y0,x1,y1)。自作 rpk では明示が必要")
    p.add_argument("--bbox-normalization", action=argparse.BooleanOptionalAction, default=None,
                   help="bbox が入力サイズで正規化されている場合に指定")
    p.add_argument("--dump-outputs", action="store_true",
                   help="最初の推論フレームの出力テンソルの形と値を表示して終了（出力形式の実測用）")
    p.add_argument("-t", "--duration", type=float, default=30.0,
                   help="観測秒数。0 なら Ctrl-C まで無期限")
    p.add_argument("--max-area", type=float, default=1.0, metavar="FRAC",
                   help="bbox が画角に占める割合の上限。画角全体を覆う誤検出を切る（例: 0.8）")
    p.add_argument("--stable", type=int, default=1, metavar="N",
                   help="直近 N フレーム連続で出たラベルだけ採用する。ちらつく誤検出を除去（既定 1 = 無効）")
    p.add_argument("--interval", type=float, default=0.0, metavar="SEC",
                   help="同一ラベルの出力を最短 SEC 間隔に間引く（既定 0 = 全件出力）")
    p.add_argument("--summary", action="store_true",
                   help="毎フレームの行を出さず、終了時の統計だけ表示する")
    p.add_argument("--preview", action="store_true",
                   help="デスクトップにプレビューを表示し、検出枠をオーバーレイで重ねる"
                        "（クラスは枠の色で区別。実機の GUI 環境が必要）")
    p.add_argument("--json", action="store_true", help="1 行 1 JSON で出力する")
    p.add_argument("--snapshot", metavar="PATH",
                   help="画角確認用に JPEG を 1 枚保存して終了（ヘッドレスでの向き合わせ用）")
    p.add_argument("--list-labels", action="store_true", help="クラス一覧を表示して終了")
    p.add_argument("--print-intrinsics", action="store_true",
                   help="モデルの network_intrinsics を表示して終了")
    return p.parse_args()


def load_intrinsics(imx500, args):
    """モデルに埋め込まれた intrinsics を取得し、コマンドライン指定で上書きする。

    自作の rpk にはラベルも bbox の形式も埋め込まれない（imx500-package の -f は
    入力テンソルの正規化設定であってこれらの情報ではない）。そのため
    --labels / --bbox-order / --bbox-normalization での明示が必要になる。
    """
    intrinsics = imx500.network_intrinsics
    if not intrinsics:
        intrinsics = NetworkIntrinsics()
        intrinsics.task = "object detection"
    elif intrinsics.task != "object detection":
        sys.exit(f"物体検出用のモデルではない: task={intrinsics.task}")
    if args.labels:
        intrinsics.labels = Path(args.labels).expanduser().read_text().splitlines()
    if args.bbox_order is not None:
        intrinsics.bbox_order = args.bbox_order
    if args.bbox_normalization is not None:
        intrinsics.bbox_normalization = args.bbox_normalization
    if intrinsics.labels is None:
        sys.exit("モデルにラベルが埋め込まれていない。--labels でラベルファイル（1 行 1 クラス）を渡す")
    intrinsics.update_with_defaults()
    return intrinsics


def get_labels(intrinsics):
    labels = intrinsics.labels
    if intrinsics.ignore_dash_labels:
        labels = [x for x in labels if x and x != "-"]
    return labels


# --preview 用。クラス番号順に割り当てる枠の色 (RGBA)。文字描画は cv2 無しでは
# 難しいので、クラスの区別は色で行う（起動時に対応表を stderr へ出す）。
PREVIEW_COLORS = [
    ("赤", (255, 60, 60, 220)),
    ("青", (60, 120, 255, 220)),
    ("緑", (60, 255, 60, 220)),
    ("黄", (255, 255, 60, 220)),
]


def start_qt_preview(picam2):
    """デスクトップに Qt プレビューウィンドウを開く。

    show_preview=True の自動選択に任せると、DISPLAY の無い環境（SSH など）では
    DRM プレビューが選ばれ、デスクトップが画面を握っている実機では
    "Failed to reserve DRM plane" で（別スレッド内で）落ちる。
    そのため Qt を明示して開き、開けなければ対処法を出して終了する。
    """
    from picamera2 import Preview
    errors = []
    for kind in (Preview.QTGL, Preview.QT):
        try:
            picam2.start_preview(kind)
            return
        except Exception as e:  # ImportError / Qt のディスプレイ接続失敗など
            errors.append(f"{kind.name}: {e}")
    sys.exit(
        "プレビューを開けなかった:\n  " + "\n  ".join(errors) + "\n"
        "実機デスクトップのターミナルから実行する。SSH からは DISPLAY=:0 を付けて試す。\n"
        "それでも駄目なら: sudo apt install python3-pyqt5 python3-opengl"
    )


def draw_box(overlay, x, y, w, h, color, thickness=4):
    """RGBA オーバーレイ配列に枠線だけの矩形を描く（塗りつぶさない）。"""
    frame_h, frame_w = overlay.shape[:2]
    x0, y0 = max(0, x), max(0, y)
    x1, y1 = min(frame_w, x + w), min(frame_h, y + h)
    if x1 <= x0 or y1 <= y0:
        return
    t = thickness
    overlay[y0:min(y0 + t, y1), x0:x1] = color  # 上辺
    overlay[max(y1 - t, y0):y1, x0:x1] = color  # 下辺
    overlay[y0:y1, x0:min(x0 + t, x1)] = color  # 左辺
    overlay[y0:y1, max(x1 - t, x0):x1] = color  # 右辺


def parse_detections(imx500, intrinsics, picam2, metadata, labels, threshold):
    """メタデータの出力テンソルを検出リストへ変換する。

    推論結果がまだ乗っていないフレームでは None を返す。
    座標は convert_inference_coords で ISP 出力（main ストリーム）の
    ピクセル座標 (x, y, w, h) に変換される。
    """
    np_outputs = imx500.get_outputs(metadata, add_batch=True)
    if np_outputs is None:
        return None

    if intrinsics.postprocess == "nanodet":
        # nanodet 系は Pi 側で NMS が必要。既定の SSD MobileNetV2 系のみ対象。
        sys.exit("nanodet 系モデルは未対応（postprocess_nanodet_detection が必要）")

    boxes, scores, classes = np_outputs[0][0], np_outputs[1][0], np_outputs[2][0]
    _, input_h = imx500.get_input_size()
    if intrinsics.bbox_normalization:
        boxes = boxes / input_h
    if intrinsics.bbox_order == "xy":
        boxes = boxes[:, [1, 0, 3, 2]]

    frame_w, frame_h = picam2.camera_configuration()["main"]["size"]
    frame_area = frame_w * frame_h

    detections = []
    for box, score, category in zip(boxes, scores, classes):
        if score <= threshold:
            continue
        x, y, w, h = imx500.convert_inference_coords(box, metadata, picam2)
        detections.append({
            "label": labels[int(category)],
            "score": round(float(score), 3),
            "box": [int(x), int(y), int(w), int(h)],
            # 画角中心を (0.5, 0.5) とする正規化中心座標。M5 の正対制御で使う。
            "center": [round((x + w / 2) / frame_w, 4), round((y + h / 2) / frame_h, 4)],
            # bbox が画角に占める割合。1.0 に近いものは「画角全体を何かに当てはめた」誤検出の疑い。
            "area": round(w * h / frame_area, 3),
        })
    return detections


def print_summary(stats, frames, elapsed):
    """ラベル別の統計を stderr に出す。テレビ有無など条件を変えて比較するのに使う。

    1 フレームに同じラベルの bbox が複数出ることがあるため、
    「件数」（bbox の総数）と「出現F」（そのラベルが 1 件以上出たフレーム数）を分けて示す。
    """
    print(f"# --- 統計: {elapsed:.1f}s / 推論 {frames} フレーム ---", file=sys.stderr)
    if not stats:
        print("# 検出なし", file=sys.stderr)
        return
    print(f"# {'label':<14}{'件数':>7}{'出現F':>7}{'出現率':>8}   "
          f"{'score min/med/max':<22}{'area med':>9}", file=sys.stderr)
    for label, rec in sorted(stats.items(), key=lambda kv: -kv[1]["frame_count"]):
        scores, areas = rec["scores"], rec["areas"]
        rate = rec["frame_count"] / frames * 100 if frames else 0.0
        span = f"{min(scores):.2f}/{statistics.median(scores):.2f}/{max(scores):.2f}"
        print(f"# {label:<14}{len(scores):>7}{rec['frame_count']:>7}{rate:>7.1f}%   "
              f"{span:<22}{statistics.median(areas):>9.2f}", file=sys.stderr)


def main():
    args = parse_args()

    # Picamera2 のインスタンス化より前に呼ぶ必要がある
    imx500 = IMX500(args.model)
    intrinsics = load_intrinsics(imx500, args)
    labels = get_labels(intrinsics)

    if args.print_intrinsics:
        print(intrinsics)
        return
    if args.list_labels:
        for i, name in enumerate(labels):
            print(f"{i:3d}  {name}")
        return

    picam2 = Picamera2(imx500.camera_num)
    config = picam2.create_preview_configuration(
        controls={"FrameRate": intrinsics.inference_rate}, buffer_count=12
    )
    # 初回はカメラへのネットワークファーム転送に時間がかかるので進捗を出す
    imx500.show_network_fw_progress_bar()
    picam2.configure(config)
    if args.preview:
        start_qt_preview(picam2)
    picam2.start()
    if intrinsics.preserve_aspect_ratio:
        imx500.set_auto_aspect_ratio()

    if args.snapshot:
        time.sleep(2)  # AE / AWB が落ち着くのを待つ
        picam2.capture_file(args.snapshot)
        picam2.stop()
        print(f"# snapshot 保存: {args.snapshot}", file=sys.stderr)
        return

    if args.dump_outputs:
        # 出力テンソルの並びと値域を実測する。自作 rpk では bbox の座標順も
        # 正規化の有無も rpk からは分からないため、ここで確かめてから指定する。
        outputs = None
        for _ in range(60):  # 推論結果がメタデータに乗るまで数フレームかかる
            outputs = imx500.get_outputs(picam2.capture_metadata(), add_batch=True)
            if outputs is not None:
                break
        picam2.stop()
        if outputs is None:
            sys.exit("60 フレーム待っても推論結果が得られなかった")
        print(f"# input_size={imx500.get_input_size()} 出力テンソル数={len(outputs)}")
        for i, tensor in enumerate(outputs):
            values = tensor[0] if tensor.ndim > 1 else tensor
            print(f"# output_{i}: shape={tensor.shape} dtype={tensor.dtype}")
            print(f"#   先頭3件: {values[:3].tolist()}")
        return

    wanted = set(args.label) if args.label else None
    frame_w, frame_h = picam2.camera_configuration()["main"]["size"]

    overlay = None
    if args.preview:
        overlay = np.zeros((frame_h, frame_w, 4), dtype=np.uint8)
        legend = "  ".join(
            f"{name}={PREVIEW_COLORS[i % len(PREVIEW_COLORS)][0]}"
            for i, name in enumerate(labels)
        )
        print(f"# preview 枠の色: {legend}", file=sys.stderr)

    print(f"# model={args.model}", file=sys.stderr)
    print(f"# cv2={'real' if HAVE_REAL_CV2 else 'stub'} frame={frame_w}x{frame_h} "
          f"inference_rate={intrinsics.inference_rate}", file=sys.stderr)
    print(f"# threshold={args.threshold} max_area={args.max_area} stable={args.stable} "
          f"interval={args.interval}", file=sys.stderr)

    history = deque(maxlen=max(1, args.stable))
    last_printed = {}
    stats = {}
    start = time.monotonic()
    frames = 0
    try:
        while True:
            elapsed = time.monotonic() - start
            if args.duration and elapsed >= args.duration:
                break

            detections = parse_detections(
                imx500, intrinsics, picam2, picam2.capture_metadata(), labels, args.threshold
            )
            if detections is None:
                continue
            frames += 1

            # 画角全体を覆う bbox は「何もないシーンを丸ごと分類した」誤検出の典型
            detections = [d for d in detections if d["area"] <= args.max_area]
            if wanted:
                detections = [d for d in detections if d["label"] in wanted]

            # プレビューには --stable の確定を待たず、フィルタ通過分をそのまま描く
            # （ちらつきも含めて生の検出挙動を見るための表示なので）
            if overlay is not None:
                overlay.fill(0)
                for d in detections:
                    color = PREVIEW_COLORS[labels.index(d["label"]) % len(PREVIEW_COLORS)][1]
                    draw_box(overlay, *d["box"], color)
                picam2.set_overlay(overlay)

            # 直近 N フレームすべてに出ているラベルだけを確定扱いにする
            history.append({d["label"] for d in detections})
            if len(history) < history.maxlen:
                continue
            confirmed = set.intersection(*history)

            seen_this_frame = set()
            for d in detections:
                if d["label"] not in confirmed:
                    continue
                rec = stats.setdefault(d["label"], {"scores": [], "areas": [], "frame_count": 0})
                rec["scores"].append(d["score"])
                rec["areas"].append(d["area"])
                if d["label"] not in seen_this_frame:
                    seen_this_frame.add(d["label"])
                    rec["frame_count"] += 1

                if args.summary:
                    continue
                if args.interval and elapsed - last_printed.get(d["label"], -1e9) < args.interval:
                    continue
                last_printed[d["label"]] = elapsed

                if args.json:
                    print(json.dumps({"t": round(elapsed, 2), **d}), flush=True)
                else:
                    x, y, w, h = d["box"]
                    cx, cy = d["center"]
                    print(f"[{elapsed:6.1f}s] {d['label']:<16} {d['score']:.2f} "
                          f"box=({x},{y},{w},{h}) center=({cx:.3f},{cy:.3f}) area={d['area']:.2f}",
                          flush=True)
    except KeyboardInterrupt:
        pass
    finally:
        picam2.stop()
        print_summary(stats, frames, time.monotonic() - start)


if __name__ == "__main__":
    main()

python3 check_detect.py --model ~/models/yolo11n_appliances/network.rpk \
  --labels ~/models/yolo11n_appliances/labels.txt \
  --bbox-order xy --bbox-normalization \
  --threshold 0.55 --stable 3 --preview -t 0

DISPLAY=:0 python3 check_detect.py --model ~/models/yolo11n_appliances/network.rpk \
  --labels ~/models/yolo11n_appliances/labels.txt \
  --bbox-order xy --bbox-normalization \
  --threshold 0.55 --stable 3 --preview -t 0

# label      件数   出現F   出現率   score min/med/max
# tv2         156    156    98.1%   0.87/0.96/0.98
# tv1         156    156    98.1%   0.88/0.93/0.96

# label      件数   出現F   出現率   score min/med/max
# tv2         186    186    98.9%   0.90/0.95/0.97
# tv1         155    155    82.4%   0.56/0.85/0.95
# laptop       23     22    11.7%   0.56/0.68/0.82

# label      件数   出現F   出現率   score min/med/max
# tv1         185    182    97.3%   0.56/0.90/0.95
# laptop      181    175    93.6%   0.56/0.93/0.97
# tv2         165    165    88.2%   0.90/0.95/0.97