FishServer/fish_api/app/measure_debug_cli.py

"""CLI：对单条 fish 跑 FishMeasure 子进程（不写 SQLite、不发布 /media）。

使用 ``run_measure_batch_subprocess``（合并点云）。线上 ``measure_watch`` 为逐段 ``run_full_measure``，齐套后再 aggregate final。
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List

from app.services.measure import run_measure_batch_subprocess
from app.settings import get_settings


def _sorted_svo2_in_folder(folder: Path) -> List[Path]:
    """与 measure_watch.iter_svo2_folders 中单目录内 .svo2 列表一致（按文件名排序）。"""
    return sorted(
        [p for p in folder.iterdir() if p.is_file() and p.suffix.lower() == ".svo2"],
        key=lambda p: p.name,
    )


def _print_weight_summary(output_root: Path) -> None:
    combined = output_root / "weight_prediction.json"
    if not combined.is_file():
        print(f"[measure-debug] no combined file: {combined}", file=sys.stderr)
        return
    try:
        with open(combined, encoding="utf-8") as f:
            data: Dict[str, Any] = json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        print(f"[measure-debug] failed to read {combined}: {e}", file=sys.stderr)
        return

    summary = data.get("dgcnn_summary") or data.get("weight_summary") or {}
    pred_g = data.get("pred_weight_g")
    rule = data.get("pred_weight_rule")
    print("[measure-debug] --- weight_prediction.json (summary) ---")
    if pred_g is not None:
        print(f"  pred_weight_g: {pred_g}")
    if rule is not None:
        print(f"  pred_weight_rule: {rule}")
    if summary:
        top_k = summary.get("top_k")
        tbl = summary.get("top_by_length")
        print(f"  dgcnn_summary.top_k: {top_k}")
        print(f"  dgcnn_summary.top_by_length: {tbl}")
    mean_all = data.get("mean_all_pred_g_after_filters")
    if mean_all is not None:
        print(f"  mean_all_pred_g_after_filters: {mean_all}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Run FishMeasure for one fish folder (same subprocess as fish_api)."
    )
    parser.add_argument(
        "--fish-id",
        default="1",
        help='Fish id used for output dir measure_output/fish{N} (default: 1).',
    )
    parser.add_argument(
        "--batch-folder",
        type=Path,
        default=None,
        help="Folder containing .svo2 files (overrides MEASURE_WATCH_DIR/fish{N}).",
    )
    parser.add_argument(
        "--watch-dir",
        type=Path,
        default=None,
        help="Override MEASURE_WATCH_DIR when resolving fish{N}/ (ignored if --batch-folder is set).",
    )
    parser.add_argument(
        "--output-root",
        type=Path,
        default=None,
        help="Override save-output directory (default: MEASURE_OUTPUT_ROOT/fish{N}).",
    )
    args = parser.parse_args()

    settings = get_settings()
    fish_id = str(args.fish_id).strip() or "1"

    if args.batch_folder is not None:
        batch_folder = args.batch_folder.expanduser().resolve()
    else:
        watch = args.watch_dir if args.watch_dir is not None else settings.measure_watch_dir
        if watch is None:
            print(
                "未配置输入目录：请设置 --batch-folder，或在 .env 中设置 MEASURE_WATCH_DIR",
                file=sys.stderr,
            )
            raise SystemExit(2)
        watch = watch.expanduser().resolve()
        batch_folder = watch / f"fish{fish_id}"

    if not batch_folder.is_dir():
        print(f"不是目录或不存在: {batch_folder}", file=sys.stderr)
        raise SystemExit(2)

    svo_paths = _sorted_svo2_in_folder(batch_folder)
    if not svo_paths:
        print(f"目录下无 .svo2 文件: {batch_folder}", file=sys.stderr)
        raise SystemExit(2)

    if args.output_root is not None:
        out_root = args.output_root.expanduser().resolve()
    else:
        out_root = settings.measure_output_root / f"fish{fish_id}"
    out_root.mkdir(parents=True, exist_ok=True)

    print(f"[measure-debug] batch_folder={batch_folder}")
    print(f"[measure-debug] output_root={out_root}")
    print(f"[measure-debug] {len(svo_paths)} SVO(s): {', '.join(p.name for p in svo_paths)}")

    run_measure_batch_subprocess(svo_paths, settings, output_root=out_root)

    _print_weight_summary(out_root)
    print("[measure-debug] done.")


if __name__ == "__main__":
    main()