"""Step 8-14 感度分析: blendbsdf weight × alpha の 3×3 グリッドスイープ. 内壁の非 Lambertian 性（光沢成分混合）が輝度分布「形」と数値指標に 与える影響を観察する感度分析スクリプト. スイープグリッド: weight ∈ {0.05, 0.15, 0.30} (specular 混合比) alpha ∈ {0.1, 0.3, 0.6 } (GGX 粗さ) → 9 組合せ × (sm-render + sm-evaluate) を順次実行 実行方法: uv run python scripts/sweep_blend_bsdf.py 前提条件: - configs/scene_baseline_blend.yaml が存在すること - configs/eval_baseline_dng_target.yaml が存在すること - sm-render / sm-evaluate が PATH に通っていること 出力: - output/blend_sweep/w{w}_a{a}/luminance.npy (各ジョブのレンダリング結果) - output/blend_sweep/w{w}_a{a}/metrics.csv (各ジョブの評価指標) - output/blend_sweep/summary.csv (全 9 ジョブの集約結果) - configs/_tmp_blend/scene_w{w}_a{a}.yaml (一時 YAML，実行後も残す) TECH_10 参照: 残差主因候補 §A1「内壁の非 Lambertian 性（光沢成分混合 BSDF）」 """ from __future__ import annotations import csv import subprocess import sys import time from pathlib import Path import yaml # --------------------------------------------------------------------------- # 定数 # --------------------------------------------------------------------------- ROOT = Path(__file__).resolve().parents[1] TEMPLATE_YAML = ROOT / "configs" / "scene_baseline_blend.yaml" EVAL_CONFIG_YAML = ROOT / "configs" / "eval_baseline_dng_target.yaml" TMP_DIR = ROOT / "configs" / "_tmp_blend" OUTPUT_BASE = ROOT / "output" / "blend_sweep" # スイープグリッド WEIGHTS = [0.05, 0.15, 0.30] ALPHAS = [0.1, 0.3, 0.6] # 集約 CSV のカラム定義（metrics.csv に存在するキー） SUMMARY_EXTRA_COLS = [ "cov", "std", "max_min_ratio", "mean", "max", "min", "center_periphery_ratio", "gradient_magnitude", "radial_min_max_ratio", "radial_min_distance", "radial_min", "radial_max", ] # --------------------------------------------------------------------------- # ヘルパ関数 # --------------------------------------------------------------------------- def _format_param(value: float) -> str: """パラメータ値をファイル名に使える文字列に変換する. 例: 0.05 → "0.05", 0.3 → "0.3" Args: value: float のパラメータ値. Returns: ファイル名に使える文字列. """ return str(value) def _make_job_name(weight: float, alpha: float) -> str: """ジョブ識別子を生成する. Args: weight: blendbsdf の weight 値. alpha: roughconductor の alpha 値. Returns: ジョブ識別子文字列（例: "w0.05_a0.1"）. """ return f"w{_format_param(weight)}_a{_format_param(alpha)}" def _generate_tmp_yaml(weight: float, alpha: float) -> Path: """一時 YAML ファイルを生成して返す. scene_baseline_blend.yaml のテンプレートを読み込み， materials.printer_white.weight と components[1].alpha を書き換えて 一時 YAML として保存する. Note: UTF-8 BOM なしで書き込む（Mitsuba YAML パーサは BOM 付きで失敗するため）. Args: weight: blendbsdf の weight 値（specular 混合比）. alpha: roughconductor の alpha 値（GGX 粗さ）. Returns: 生成した一時 YAML ファイルのパス. """ TMP_DIR.mkdir(parents=True, exist_ok=True) with TEMPLATE_YAML.open(encoding="utf-8") as f: data = yaml.safe_load(f) # geometry.meshes の STL 相対パスを絶対化（一時 YAML の配置場所が変わっても解決可能にする） template_base = TEMPLATE_YAML.parent for item in data.get("geometry", {}).get("meshes", []): rel_path = item.get("path") if rel_path and not Path(rel_path).is_absolute(): item["path"] = str((template_base / rel_path).resolve()) # printer_white の blendbsdf パラメータを書き換える printer_white = data["materials"]["printer_white"] printer_white["weight"] = weight # components[1] が roughconductor（スキーマ順: [0]=diffuse, [1]=roughconductor） printer_white["components"][1]["alpha"] = alpha # name を識別しやすい値に更新する job_name = _make_job_name(weight, alpha) data["name"] = f"baseline_minitias_blend_{job_name}" tmp_path = TMP_DIR / f"scene_{job_name}.yaml" with tmp_path.open("w", encoding="utf-8") as f: yaml.dump(data, f, allow_unicode=True, default_flow_style=False, sort_keys=False) return tmp_path def _run_command(cmd: list[str], label: str) -> int: """外部コマンドを実行して終了コードを返す. Args: cmd: 実行するコマンドリスト. label: ログ出力用のラベル. Returns: 終了コード（0 = 成功）. """ print(f" [run] {' '.join(cmd)}") result = subprocess.run(cmd, check=False) # noqa: S603 if result.returncode != 0: print(f" [ERROR] {label} failed (returncode={result.returncode})", file=sys.stderr) return result.returncode def _read_metrics_csv(csv_path: Path) -> dict: """metrics.csv を読み込んで辞書を返す. Args: csv_path: 読み込む metrics.csv のパス. Returns: 指標名 → 値の辞書.1 行目のヘッダ + 2 行目の値を読む. Raises: FileNotFoundError: ファイルが存在しない場合. ValueError: ファイルが空または形式不正の場合. """ if not csv_path.exists(): msg = f"metrics.csv が見つかりません: {csv_path}" raise FileNotFoundError(msg) with csv_path.open(encoding="utf-8", newline="") as f: reader = csv.DictReader(f) rows = list(reader) if not rows: msg = f"metrics.csv が空です: {csv_path}" raise ValueError(msg) return rows[0] # --------------------------------------------------------------------------- # メイン処理 # --------------------------------------------------------------------------- def main() -> int: """スイープのメインエントリポイント. Returns: 終了コード（全ジョブ成功なら 0，失敗があれば 1）. """ # 前提ファイルの確認 if not TEMPLATE_YAML.exists(): print(f"エラー: テンプレート YAML が見つかりません: {TEMPLATE_YAML}", file=sys.stderr) return 1 if not EVAL_CONFIG_YAML.exists(): print(f"エラー: 評価 config YAML が見つかりません: {EVAL_CONFIG_YAML}", file=sys.stderr) return 1 OUTPUT_BASE.mkdir(parents=True, exist_ok=True) # ジョブリスト生成 jobs = [(w, a) for w in WEIGHTS for a in ALPHAS] total = len(jobs) print(f"=== blend BSDF sweep: {total} jobs ({len(WEIGHTS)} weights × {len(ALPHAS)} alphas) ===") print(f" weights: {WEIGHTS}") print(f" alphas: {ALPHAS}") print(f" output: {OUTPUT_BASE}") print() failed_jobs: list[str] = [] summary_rows: list[dict] = [] total_start = time.time() for job_idx, (weight, alpha) in enumerate(jobs, start=1): job_name = _make_job_name(weight, alpha) output_dir = OUTPUT_BASE / job_name metrics_csv = output_dir / "metrics.csv" print(f"--- Job {job_idx}/{total}: {job_name} ---") job_start = time.time() # 一時 YAML 生成 tmp_yaml = _generate_tmp_yaml(weight, alpha) print(f" tmp YAML: {tmp_yaml}") # sm-render 実行 render_ret = _run_command( ["sm-render", "--scene", str(tmp_yaml), "--output", str(output_dir)], label=f"sm-render ({job_name})", ) if render_ret != 0: failed_jobs.append(f"{job_name} [render]") print(" -> スキップ: render 失敗\n") continue # sm-evaluate 実行 npy_path = output_dir / "luminance.npy" evaluate_ret = _run_command( [ "sm-evaluate", "--image", str(npy_path), "--config", str(EVAL_CONFIG_YAML), "--output", str(output_dir), ], label=f"sm-evaluate ({job_name})", ) if evaluate_ret != 0: failed_jobs.append(f"{job_name} [evaluate]") print(" -> スキップ: evaluate 失敗\n") continue # metrics 読み込み try: metrics = _read_metrics_csv(metrics_csv) except (FileNotFoundError, ValueError) as exc: print(f" [ERROR] metrics.csv 読み込み失敗: {exc}", file=sys.stderr) failed_jobs.append(f"{job_name} [metrics_read]") print() continue elapsed = time.time() - job_start print(f" -> 完了 ({elapsed:.1f}s)") cov_val = metrics.get("cov", "N/A") grad_val = metrics.get("gradient_magnitude", "N/A") print(f" CoV={cov_val}, gradient={grad_val}%") row: dict = { "job": job_name, "weight": weight, "alpha": alpha, "elapsed_s": f"{elapsed:.1f}", } for col in SUMMARY_EXTRA_COLS: row[col] = metrics.get(col, "") summary_rows.append(row) print() # 集約 CSV 保存 summary_csv = OUTPUT_BASE / "summary.csv" if summary_rows: fieldnames = ["job", "weight", "alpha", "elapsed_s"] + SUMMARY_EXTRA_COLS with summary_csv.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(summary_rows) print(f"集約 CSV 保存: {summary_csv}") else: print( "警告: 成功したジョブが 0 件のため summary.csv を保存しませんでした.", file=sys.stderr, ) total_elapsed = time.time() - total_start print(f"\n=== スイープ完了: {len(summary_rows)}/{total} 成功, 総時間 {total_elapsed:.1f}s ===") if failed_jobs: print(f"失敗ジョブ ({len(failed_jobs)}件):", file=sys.stderr) for j in failed_jobs: print(f" - {j}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())