"""照明配置スイープ: LED の位置・向き・個数への依存を球と立方体で比較する（TEST_18）.
TEST_16 で「軸上配置なら同内面積の立方体が球より均一」，TEST_17 で「側壁 1 点への
オフ軸配置では球のほうが頑健」と分かった．本スイープは「積分球が使われる理由は
照明配置への依存が小さいからではないか」という仮説を，配置を系統的に振って検証する．
条件の軸（同内面積の球・立方体ペアごと）:
A. 位置 : LED をキャビティ中心から見た極角 θ（0°=底の中心＝軸上，90°=中心と
同じ高さの側壁，120°=板のすぐ下）で壁面上を動かす．板中心向き＋バッフル
B. 向き : θ=60/90/120° で「板中心向き」「壁法線向き（キャビティ中心向き）」
「真下向き（板が発光面の背後＝バッフル不要）」を比べる
C. 個数 : θ の輪に N 灯を等方位に置き総光束を保存する（radiance / N）
D. 再現性 : 同一条件の再レンダでノイズフロアを測る
E. 方位角 : 立方体の面中央（φ=0）→ 稜寄り（φ=40°）で位置依存が変わるか（追加実験）
F. 真下向き多灯 : バッフル無しの真下向き N=2/4/8 灯（最も簡単な実機構成，追加実験）
θ の定義（両形状共通）:
キャビティ中心 c（球: 球中心，立方体: 立方体中心）から方向
u = (sinθ cosφ, sinθ sinφ, -cosθ) に伸ばした半直線が内壁と交わる点に LED を置く．
球ではこれが「LED から見た板の立体角」を一意に決める自然なパラメータで，
立方体には同じ角度をそのまま適用する（θ<45° は底面，45°<θ<135° は側壁）．
立方体の稜（θ=45°）は LED を置けないので θ グリッドから外す．
バッフルは TEST_17 と同じ「LED から BAFFLE_GAP 先に置く傾いた円板」を，発光面の前方に
ある板の全点を隠す最小半径 × 安全率で数値的に決める．TEST_17 では外殻の貫通を
エラーにしたが，板の近く（θ が大きい）では円板が壁を大きく貫かざるを得ないので
本スイープでは許容し，食い込み量を記録するだけにする（外側に出た部分はキャビティ
内から見えないため光輸送には影響しない）．
固定条件は TEST_15/16/17 と同一（材質・レンダ・評価・評価板・LED 面積・総光束）．
実行方法:
uv run python scripts/sweep_led_placement.py --prepare-only # 幾何 + YAML のみ
uv run python scripts/sweep_led_placement.py --shapes sphere # 球だけレンダ
uv run python scripts/sweep_led_placement.py --groups position count # グループ指定
uv run python scripts/sweep_led_placement.py --workers 3 # 3 条件同時に実行
"""
from __future__ import annotations
import argparse
import copy
import csv
import io
import json
import multiprocessing
import sys
import time
from collections.abc import Callable
from concurrent.futures import ProcessPoolExecutor, as_completed
from contextlib import redirect_stdout
from dataclasses import dataclass
from pathlib import Path
import numpy as np
import trimesh
import yaml
sys.path.insert(0, str(Path(__file__).resolve().parent))
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from generate_cube_housing import build_cube_housing # noqa: E402
from sweep_cube_size import cavity_geometry as cube_cavity_geometry # noqa: E402
from sweep_offaxis_led import STANDOFF_MARGIN # noqa: E402
from sweep_sphere_radius import ( # noqa: E402
BAFFLE_GAP,
BAFFLE_MARGIN,
BAFFLE_THICKNESS,
BOARD_HALF,
BOARD_SURFACE_Z,
EVAL_CONFIG_YAML,
LED_RADIUS,
RIM_INSET,
RIM_Z,
TEMPLATE_YAML,
THICKNESS,
raw_stats,
read_metrics_csv,
)
from sweep_sphere_radius import (
build_mesh as build_sphere_mesh,
)
from sweep_sphere_radius import (
cavity_geometry as sphere_cavity_geometry,
)
from sweep_sphere_radius import (
center_z as sphere_center_z,
)
from sweep_sphere_radius import (
cut_z as sphere_cut_z,
)
from sm_tias_lightsim.evaluation.pipeline import evaluate_image # noqa: E402
from sm_tias_lightsim.render.pipeline import render_yaml # noqa: E402
from sm_tias_lightsim.runtime.overrides import resolve_render_variant # noqa: E402
from sm_tias_lightsim.scene.paths import relative_mesh_path # noqa: E402
ROOT = Path(__file__).resolve().parents[1]
TMP_DIR = ROOT / "configs" / "_tmp_led_placement"
MESH_DIR = TMP_DIR / "meshes"
OUTPUT_BASE = ROOT / "output" / "led_placement_sweep"
SPHERE_CAVITY = "similar_radial" # TEST_15 最良・TEST_17 と同じ
# 比較する 2 形状（SIZE_PAIRS の要素順と対応させる）
SHAPES = ("sphere", "cube")
# 同内面積ペア（球の内半径, 立方体の内寸）．立方体は 6L^2+594 = 球の内面積 から丸めた
SIZE_PAIRS: list[tuple[float, float]] = [
(100.0, 141.0),
(135.0, 187.0),
(170.0, 234.0),
(250.0, 340.0),
]
REPEAT_PAIR_INDEX = 2 # 再現性チェックに使うペア（r=170 / L=234）
# A. 位置: 極角グリッド（立方体の稜 45° を避ける．120° は球の loft 接続（約 121°）直下）
THETA_GRID = [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0, 110.0, 120.0]
# B. 向き
DIRECTION_THETAS = [60.0, 90.0, 120.0]
AIMS = ("board", "normal", "down")
# C. 個数
COUNT_THETA = 90.0
COUNT_N = [2, 3, 4]
COUNT_N4_THETAS = [60.0, 120.0]
# D. 再現性（同一条件の追加レンダ回数）
REPEATS = 3
# E. 方位角（追加実験）: 立方体の面中央（φ=0）→ 稜寄り（φ=40°）．球は φ=45° が対照（軸対称）．
# φ=90° は m=1 の投影の符号較正用（+Y 面＝面中央なので φ=0 と同じ値になるはず）
AZIMUTH_THETA = 90.0
AZIMUTH_CUBE = [15.0, 30.0, 40.0]
AZIMUTH_SPHERE = [45.0]
AZIMUTH_CALIBRATION = ("cube", 234.0, 90.0)
# F. 真下向き多灯（追加実験）: バッフル無しで最も簡単な実機構成．(向き, θ, N)
# 立方体の N=8 は稜（φ=45°）を避けるため方位を 22.5° ずらす
RING_SPECS = [("down", 90.0, 2), ("down", 90.0, 4), ("down", 90.0, 8), ("down", 60.0, 4),
("board", 90.0, 8)]
CUBE_N8_AZIMUTH = 22.5
GROUPS = ("position", "direction", "count", "repeat", "azimuth", "ring")
# G. 校正材質での引き直し（追加実験）: TEST_14 で確定した純 diffuse ρ=0.81 に内壁を置き換え，
# 主要な結論（位置依存の幅・多灯・真下向き）が材質に依らないことを確かめる．
# 出力・一時 YAML は別ディレクトリ（サフィックス）に分ける
MATERIALS = {
"legacy": None, # テンプレートのまま（blendbsdf ρ=0.85・ω=0.05・α=0.1）
"calibrated": {"type": "diffuse", "reflectance": [0.81, 0.81, 0.81]},
}
MATERIAL_SUFFIX = {"legacy": "", "calibrated": "_rho081"}
_material = "legacy"
# 「主要条件だけ」のサブセット: 位置スイープをペア 1・3 で，多灯（N=2/4）と真下向きをペア 3 で
KEY_SUBSET_PAIRS = (0, 2)
KEY_SUBSET_PAIR_MULTI = 2
def set_material(material: str) -> None:
"""内壁材質の種類を切り替え，出力・一時ディレクトリを材質ごとに分ける."""
global _material, TMP_DIR, MESH_DIR, OUTPUT_BASE # noqa: PLW0603
if material not in MATERIALS:
raise ValueError(f"未知の材質指定: {material}（{', '.join(MATERIALS)}）")
_material = material
suffix = MATERIAL_SUFFIX[material]
TMP_DIR = ROOT / "configs" / f"_tmp_led_placement{suffix}"
MESH_DIR = TMP_DIR / "meshes"
OUTPUT_BASE = ROOT / "output" / f"led_placement_sweep{suffix}"
def current_material() -> str:
"""現在の材質指定を返す."""
return _material
def key_subset(conds: list[Condition]) -> list[Condition]:
"""材質の引き直しに使う主要条件だけを残す."""
keep = []
for c in conds:
pair = next((i for i, p in enumerate(SIZE_PAIRS)
if p[0 if c.shape == "sphere" else 1] == c.size), -1)
if c.group == "position" and pair in KEY_SUBSET_PAIRS:
keep.append(c)
elif (c.group == "count" and pair == KEY_SUBSET_PAIR_MULTI
and c.theta_deg == COUNT_THETA and c.n_led in (2, 4)):
keep.append(c)
elif (c.group == "direction" and pair == KEY_SUBSET_PAIR_MULTI
and c.theta_deg == COUNT_THETA and c.aim == "down"):
keep.append(c)
elif (c.group == "ring" and pair == KEY_SUBSET_PAIR_MULTI
and c.theta_deg == COUNT_THETA and c.aim == "down" and c.n_led == 4):
keep.append(c)
return keep
# 評価板の中心（キャビティ側の面）＝ 板向き LED の照準点
BOARD_CENTER = np.array([0.0, 0.0, BOARD_SURFACE_Z])
# LED 円板を立方体の稜に近づけすぎない余裕（mm）
EDGE_MARGIN = 3.0
# バッフル半径の上限（キャビティ寸法比）．超える条件は物理的に無理があるので実行しない
BAFFLE_RADIUS_CAP_RATIO = 0.6
SUMMARY_COLUMNS = [
"name", "group", "shape", "size_mm", "theta_deg", "aim", "n_led", "rep", "azimuth_deg",
"material",
"cavity_area_mm2", "led_x_mm", "led_y_mm", "led_z_mm", "led_standoff_mm", "tilt_deg",
"baffled", "baffle_radius_mm", "baffle_embed_mm", "board_front_frac",
"occlusion_blocked_frac", "occlusion_total",
"elapsed_s", "render_variant", "raw_mean_roi", "raw_mean_full", "raw_rel_std_roi",
"cov", "center_periphery_ratio", "gradient_magnitude", "radial_min_max_ratio",
"angular_ratio", "pca_eccentricity", "intensity_centroid_offset_norm",
"intensity_centroid_x_norm", "intensity_centroid_y_norm", "max_min_ratio",
]
@dataclass(frozen=True)
class Condition:
"""1 レンダ条件（配置の指定）."""
group: str
shape: str
size: float
theta_deg: float
aim: str
n_led: int
rep: int = 1
azimuth_deg: float = 0.0
@property
def base_name(self) -> str:
"""幾何（メッシュ・バッフル）を共有する名前（再レンダ条件は seed だけが違う）.
方位角は 0 以外のときだけ付ける（追加実験より前の条件名を変えないため）．
"""
prefix = "r" if self.shape == "sphere" else "L"
azimuth = f"_az{self.azimuth_deg:03.0f}" if self.azimuth_deg else ""
return (f"{self.shape}_{prefix}{self.size:g}_th{self.theta_deg:03.0f}"
f"_{self.aim}_n{self.n_led}{azimuth}")
@property
def name(self) -> str:
"""出力ディレクトリ名."""
return self.base_name + (f"_rep{self.rep}" if self.rep > 1 else "")
@property
def azimuths_deg(self) -> list[float]:
"""各 LED の方位角（1 灯目は azimuth_deg，以降等間隔）."""
return [self.azimuth_deg + 360.0 * k / self.n_led for k in range(self.n_led)]
def shape_sizes() -> list[tuple[str, float]]:
"""(形状, サイズ) を同内面積ペアの順に返す（形状が外側・サイズが内側）."""
return [(shape, pair[i]) for i, shape in enumerate(SHAPES) for pair in SIZE_PAIRS]
def conditions() -> list[Condition]:
"""全条件を「重要な順」（位置 → 個数 → 向き → 再現性）に並べて返す.
途中で止まっても主実験（位置）が先に揃うようにする．同じ条件が複数グループから
生まれる場合（θ=90 板向き 1 灯は位置・向き・個数の共通アンカー）は最初の 1 つだけ残す．
"""
result: list[Condition] = []
seen: set[str] = set()
def add(cond: Condition) -> None:
if cond.name not in seen:
seen.add(cond.name)
result.append(cond)
for shape, size in shape_sizes():
for theta in THETA_GRID:
add(Condition("position", shape, size, theta, "board", 1))
for shape, size in shape_sizes():
for n in COUNT_N:
add(Condition("count", shape, size, COUNT_THETA, "board", n))
for theta in COUNT_N4_THETAS:
add(Condition("count", shape, size, theta, "board", 4))
for shape, size in shape_sizes():
for theta in DIRECTION_THETAS:
for aim in AIMS:
add(Condition("direction", shape, size, theta, aim, 1))
for shape_index, shape in enumerate(SHAPES):
size = SIZE_PAIRS[REPEAT_PAIR_INDEX][shape_index]
for rep in range(2, REPEATS + 2):
add(Condition("repeat", shape, size, COUNT_THETA, "board", 1, rep))
for shape, size in shape_sizes():
for azimuth in AZIMUTH_CUBE if shape == "cube" else AZIMUTH_SPHERE:
add(Condition("azimuth", shape, size, AZIMUTH_THETA, "board", 1,
azimuth_deg=azimuth))
add(Condition("azimuth", *AZIMUTH_CALIBRATION[:2], AZIMUTH_THETA, "board", 1,
azimuth_deg=AZIMUTH_CALIBRATION[2]))
for shape, size in shape_sizes():
for aim, theta, n in RING_SPECS:
azimuth = CUBE_N8_AZIMUTH if (shape == "cube" and n == 8) else 0.0
add(Condition("ring", shape, size, theta, aim, n, azimuth_deg=azimuth))
return result
# ---------------------------------------------------------------------------
# 幾何
# ---------------------------------------------------------------------------
def cavity_center(shape: str, size: float) -> np.ndarray:
"""θ の基準になるキャビティ中心（mm）を返す."""
if shape == "cube":
return np.array([0.0, 0.0, RIM_Z - size / 2.0])
return np.array([0.0, 0.0, sphere_center_z(size, SPHERE_CAVITY)])
def cavity_area(shape: str, size: float) -> float:
"""比較の横軸に使う内面積（mm²）を返す（各スイープと同じ定義）."""
if shape == "cube":
return cube_cavity_geometry(size)["cavity_area_mm2"]
return sphere_cavity_geometry(size, SPHERE_CAVITY)["cavity_area_mm2"]
def polar_direction(theta_deg: float, azimuth_deg: float) -> np.ndarray:
"""キャビティ中心から見た単位方向 u（θ=0 が -Z＝底）を返す."""
theta, phi = np.radians(theta_deg), np.radians(azimuth_deg)
return np.array([np.sin(theta) * np.cos(phi), np.sin(theta) * np.sin(phi), -np.cos(theta)])
def wall_anchor(
shape: str, size: float, theta_deg: float, azimuth_deg: float
) -> tuple[np.ndarray, np.ndarray]:
"""LED を貼り付ける内壁上の点と，そこでの内向き法線を返す（mm）.
Raises:
ValueError: 立方体の稜・球の loft 接続に近すぎて LED 円板を置けない場合．
"""
center = cavity_center(shape, size)
u = polar_direction(theta_deg, azimuth_deg)
if shape == "sphere":
point = center + size * u
# 球の壁は loft 接続平面（cut_z）より下だけ．接続の上では法線の定義が変わる
if point[2] > sphere_cut_z(size, SPHERE_CAVITY):
raise ValueError(f"θ={theta_deg:g}: 球の loft 接続より上に出ます")
return point, -u
half = size / 2.0
with np.errstate(divide="ignore"):
t_axes = np.where(np.abs(u) > 1e-12, half / np.abs(u), np.inf)
axis = int(np.argmin(t_axes))
point = center + t_axes[axis] * u
normal = np.zeros(3)
normal[axis] = -np.sign(u[axis])
others = [i for i in range(3) if i != axis]
margin = float(min(half - abs(point[i] - center[i]) for i in others))
if margin < LED_RADIUS + EDGE_MARGIN:
raise ValueError(f"θ={theta_deg:g}: 立方体の稜に近すぎます（余裕 {margin:.1f}mm）")
return point, normal
def _board_points(grid: int = 41) -> np.ndarray:
"""評価板のキャビティ側の面の格子点（四隅を含む）を返す."""
gx = np.linspace(-BOARD_HALF[0], BOARD_HALF[0], grid)
gy = np.linspace(-BOARD_HALF[1], BOARD_HALF[1], grid)
mesh_x, mesh_y = np.meshgrid(gx, gy)
z = np.full(mesh_x.size, BOARD_SURFACE_Z)
return np.stack([mesh_x.ravel(), mesh_y.ravel(), z], axis=1)
def _disk_points(center: np.ndarray, direction: np.ndarray, radius: float, n: int) -> np.ndarray:
"""direction を法線とする円板の外周点 n 個 + 中心を返す."""
axis = np.array([0.0, 1.0, 0.0]) if abs(direction[1]) < 0.9 else np.array([0.0, 0.0, 1.0])
u = np.cross(direction, axis)
u /= np.linalg.norm(u)
v = np.cross(direction, u)
angles = 0.1234 + np.linspace(0.0, 2.0 * np.pi, n, endpoint=False)
ring = center + radius * (np.outer(np.cos(angles), u) + np.outer(np.sin(angles), v))
return np.vstack([center[None, :], ring])
def led_geometry(
shape: str, size: float, theta_deg: float, azimuth_deg: float, aim: str
) -> dict:
"""LED 円板の中心・発光方向・浮かせ量・傾き角と，板への直接光の有無を返す.
浮かせ量は「傾いた円板の縁が壁面へ食い込む量（LED_RADIUS×sinτ）+ マージン」
（τ は発光方向と壁法線のなす角）．板向きは浮かせた後の中心から向け直す．
"""
wall, normal = wall_anchor(shape, size, theta_deg, azimuth_deg)
center = wall.copy()
for _ in range(2):
if aim == "board":
direction = BOARD_CENTER - center
elif aim == "normal":
direction = normal.copy()
elif aim == "down":
direction = np.array([0.0, 0.0, -1.0])
else:
raise ValueError(f"未知の向き指定: {aim}")
direction = direction / np.linalg.norm(direction)
sin_tilt = float(np.linalg.norm(np.cross(direction, normal)))
standoff = LED_RADIUS * sin_tilt + STANDOFF_MARGIN
center = wall + standoff * normal
tilt_deg = float(np.degrees(np.arctan2(sin_tilt, float(direction @ normal))))
front = ((_board_points() - center) @ direction) > 0.0
return {
"center": center, "direction": direction, "normal": normal,
"standoff": standoff, "tilt_deg": tilt_deg,
"board_front_frac": float(front.mean()),
}
def baffle_radius(led: dict) -> float:
"""発光面の前方にある板の全点から LED 全点を隠すバッフル半径（mm）を返す.
LED 円板上の点 Q × 板上の点 P の直線がバッフル平面（LED から BAFFLE_GAP 先・
法線 = 発光方向）を横切る位置の，バッフル中心からの面内距離の最大値 × 安全率．
発光面の背後にある P（直接光が届かない）は除外する．
"""
center, direction = led["center"], led["direction"]
b_center = center + BAFFLE_GAP * direction
q = _disk_points(center, direction, LED_RADIUS, 64)
p = _board_points()
delta = p[None, :, :] - q[:, None, :] # (M, N, 3)
denom = delta @ direction
front = denom > 1e-9
if not front.any():
return 0.0
t = np.where(front, ((b_center - q) @ direction)[:, None] / np.where(front, denom, 1.0), 0.0)
x = q[:, None, :] + t[..., None] * delta
offset = x - b_center
radial = offset - (offset @ direction)[..., None] * direction[None, None, :]
return float(BAFFLE_MARGIN * np.linalg.norm(radial, axis=-1)[front].max())
def _wall_distance(shape: str, size: float, points: np.ndarray) -> float:
"""点群から解析的なキャビティ内壁までの最小距離（mm）を返す（負なら壁の外）."""
if shape == "cube":
margins = np.concatenate([
(size / 2.0 - np.abs(points[:, 0:2])).ravel(),
(points[:, 2] - (RIM_Z - size)),
(RIM_Z - points[:, 2]),
])
return float(margins.min())
return float((size - np.linalg.norm(points - cavity_center(shape, size), axis=1)).min())
def build_baffle_mesh(
shape: str, size: float, led: dict, dest: Path
) -> tuple[trimesh.Trimesh, float, float]:
"""LED の視線上に置く傾いたバッフル円板を生成して書き出す.
Returns:
(メッシュ, 半径 mm, 内壁への食い込み深さ mm)．食い込みは外殻の貫通も含めて許容する
（壁の外側に出た部分はキャビティ内から見えず，光輸送に影響しない）．
Raises:
ValueError: 半径がキャビティ寸法に対して大きすぎる（物理的に成立しない）場合．
"""
sections = 128
radius = baffle_radius(led)
if radius > BAFFLE_RADIUS_CAP_RATIO * size:
raise ValueError(
f"バッフル半径 {radius:.1f}mm がキャビティ寸法 {size:g}mm の "
f"{BAFFLE_RADIUS_CAP_RATIO:.0%} を超えます（板が発光面の前方に掠めて入る配置）"
)
mesh = trimesh.creation.cylinder(radius=radius, height=BAFFLE_THICKNESS, sections=sections)
mesh.apply_transform(trimesh.transformations.rotation_matrix(np.pi / sections, [0, 0, 1]))
direction = led["direction"]
mesh.apply_transform(trimesh.geometry.align_vectors([0.0, 0.0, 1.0], direction))
mesh.apply_translation(led["center"] + BAFFLE_GAP * direction)
embed = max(0.0, -_wall_distance(shape, size, mesh.vertices))
dest.parent.mkdir(parents=True, exist_ok=True)
mesh.export(dest)
return mesh, radius, embed
def build_housing_mesh(shape: str, size: float, dest: Path) -> trimesh.Trimesh:
"""筐体メッシュを生成して書き出す（軸上スイープと同一のビルダ）."""
if dest.exists():
return trimesh.load(dest, force="mesh")
if shape == "sphere":
return build_sphere_mesh(size, dest, SPHERE_CAVITY)
mesh = build_cube_housing(
size=size, thickness=THICKNESS,
board_half_x=BOARD_HALF[0], board_half_y=BOARD_HALF[1],
rim_inset=RIM_INSET, rim_z=RIM_Z,
)
dest.parent.mkdir(parents=True, exist_ok=True)
mesh.export(dest)
return mesh
def housing_mesh_path(shape: str, size: float) -> Path:
"""筐体 STL の置き場所（同じ形状・サイズの条件で共有）."""
prefix = "r" if shape == "sphere" else "L"
return MESH_DIR / f"筐体_{shape}_{prefix}{size:g}.stl"
def build_scene_yaml(
cond: Condition, mesh_path: Path, leds: list[dict], baffles: list[Path | None], dest: Path
) -> None:
"""テンプレートから当該条件のシーン YAML を書き出す.
材質・センサ・レンダ設定・LED の半径は変更しない．LED の位置・向き・個数と
バッフルだけを変え，多灯では radiance を 1/N にして総光束を保存する．
"""
config = copy.deepcopy(yaml.safe_load(TEMPLATE_YAML.read_text(encoding="utf-8")))
config["name"] = f"led_placement_{cond.name}{MATERIAL_SUFFIX[_material]}"
config["render"]["seed"] = cond.rep - 1
if MATERIALS[_material] is not None:
config["materials"]["printer_white"] = dict(MATERIALS[_material])
config["description"] = (
f"照明配置スイープ [{cond.shape} / {cond.size:g}mm]: θ={cond.theta_deg:g}° の壁面に "
f"{cond.n_led} 灯（向き={cond.aim}）．"
+ ("バッフルで直接光を遮断．" if any(b is not None for b in baffles)
else "板は発光面の背後（直接光なし・バッフル無し）．")
)
yaml_dir = dest.parent
for mesh in config["geometry"]["meshes"]:
if mesh["id"] == "sphere_housing":
if cond.shape == "cube":
mesh["id"] = "cube_housing"
mesh["path"] = relative_mesh_path(mesh_path, yaml_dir)
else:
mesh["path"] = relative_mesh_path(TEMPLATE_YAML.parent / mesh["path"], yaml_dir)
for k, baffle_path in enumerate(baffles):
if baffle_path is None:
continue
config["geometry"]["meshes"].append({
"id": f"baffle_{k}",
"path": relative_mesh_path(baffle_path, yaml_dir),
"unit": "mm",
"material": "printer_white",
})
template_light = config["lights"][0]
config["lights"] = []
for k, led in enumerate(leds):
light = copy.deepcopy(template_light)
light["id"] = f"led_{k}"
light["center"] = [round(float(c), 4) for c in led["center"]]
light["direction"] = [round(float(d), 6) for d in led["direction"]]
light["radiance"] = [float(r) / cond.n_led for r in template_light["radiance"]]
config["lights"].append(light)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(yaml.safe_dump(config, allow_unicode=True, sort_keys=False), encoding="utf-8")
def prepare_condition(cond: Condition) -> tuple[Path, dict]:
"""1 条件の筐体・LED・バッフルを生成し，シーン YAML のパスと幾何の記録を返す.
Raises:
ValueError: LED を置けない（稜・接続に近い），円板が壁を貫く，バッフルが大きすぎる．
"""
mesh_path = housing_mesh_path(cond.shape, cond.size)
housing = build_housing_mesh(cond.shape, cond.size, mesh_path)
if not housing.is_watertight:
raise ValueError(f"{cond.name}: watertight でないメッシュが生成されました")
leds: list[dict] = []
baffles: list[Path | None] = []
record: dict = {
"cavity_area_mm2": cavity_area(cond.shape, cond.size),
"baffled": 0, "baffle_radius_mm": 0.0, "baffle_embed_mm": 0.0,
"board_front_frac": 0.0,
}
for k, azimuth in enumerate(cond.azimuths_deg):
led = led_geometry(cond.shape, cond.size, cond.theta_deg, azimuth, cond.aim)
rim = _disk_points(led["center"], led["direction"], LED_RADIUS, 64)
# 球は解析的な球面に対して測る．θ=120° でも円板は loft 接続平面より下に収まり，
# 接続直上 12mm では loft が球の延長より外側（r=100）か 1.7mm 以内の内側なので
# 解析球面での判定が実質そのまま成り立つ（生成メッシュで確認済み）
clearance = _wall_distance(cond.shape, cond.size, rim)
if clearance < 0.0:
raise ValueError(f"{cond.name}: LED 円板が壁を貫いています（余裕 {clearance:.3f}mm）")
leds.append(led)
if led["board_front_frac"] > 0.0:
baffle_path = MESH_DIR / f"バッフル_{cond.base_name}_{k}.stl"
_, radius, embed = build_baffle_mesh(cond.shape, cond.size, led, baffle_path)
baffles.append(baffle_path)
record["baffled"] = 1
record["baffle_radius_mm"] = round(radius, 3)
record["baffle_embed_mm"] = round(max(record["baffle_embed_mm"], embed), 3)
else:
baffles.append(None)
if k == 0:
record.update({
"led_x_mm": round(float(led["center"][0]), 3),
"led_y_mm": round(float(led["center"][1]), 3),
"led_z_mm": round(float(led["center"][2]), 3),
"led_standoff_mm": round(float(led["standoff"]), 3),
"tilt_deg": round(led["tilt_deg"], 2),
"board_front_frac": round(led["board_front_frac"], 4),
"led_clearance_mm": round(clearance, 3),
})
# 再レンダ条件（rep>1）は幾何が同じでも seed を変えた別 YAML にする．seed が同じだと
# シーンを読み直しても指標が 4 桁まで一致し，ノイズフロアにならない（サーバで実測）
scene_yaml = TMP_DIR / f"scene_{cond.name}.yaml"
build_scene_yaml(cond, mesh_path, leds, baffles, scene_yaml)
return scene_yaml, record
def prepare(
conds: list[Condition], log: Callable[[str], None], report_suffix: str = ""
) -> dict[str, tuple[Path, dict]]:
"""条件群の幾何を生成し，置けない条件はスキップして記録する.
report_suffix は形状別に並列で走らせるときにレポートを分けるための識別子．
"""
prepared: dict[str, tuple[Path, dict]] = {}
skipped: list[tuple[str, str]] = []
log(f"=== 幾何生成: {len(conds)} 条件 ===")
for cond in conds:
try:
scene_yaml, record = prepare_condition(cond)
except ValueError as exc:
skipped.append((cond.name, str(exc)))
log(f" SKIP {cond.name}: {exc}")
continue
prepared[cond.name] = (scene_yaml, record)
baffle_note = (f"バッフル R={record['baffle_radius_mm']:6.2f} "
f"食い込み={record['baffle_embed_mm']:5.2f}" if record["baffled"]
else "バッフル無し（板は背後）")
log(f" {cond.name:>34} LED=({record['led_x_mm']:7.2f},{record['led_y_mm']:7.2f},"
f"{record['led_z_mm']:8.2f}) 傾き={record['tilt_deg']:5.1f}° "
f"余裕={record['led_clearance_mm']:4.2f} 板前方={record['board_front_frac']:.2f} "
f"{baffle_note}")
report = TMP_DIR / f"prepare_report{report_suffix}.csv"
keys = ["name", "status", "note", "cavity_area_mm2", "led_x_mm", "led_y_mm", "led_z_mm",
"led_standoff_mm", "tilt_deg", "led_clearance_mm", "board_front_frac",
"baffled", "baffle_radius_mm", "baffle_embed_mm"]
with report.open("w", newline="", encoding="utf-8") as fh:
writer = csv.DictWriter(fh, fieldnames=keys)
writer.writeheader()
for cond in conds:
if cond.name in prepared:
row = {"name": cond.name, "status": "ok", "note": ""}
row.update(prepared[cond.name][1])
else:
note = next(msg for name, msg in skipped if name == cond.name)
row = {"name": cond.name, "status": "skip", "note": note}
writer.writerow({k: row.get(k, "") for k in keys})
log(f" 生成 {len(prepared)} / スキップ {len(skipped)}（{report}）")
return prepared
# ---------------------------------------------------------------------------
# 実行
# ---------------------------------------------------------------------------
# 解決済みの Mitsuba variant（プロセス内で 1 回だけ決める．ワーカープロセスでも各 1 回）
_render_variant: str | None = None
def render_variant() -> str:
"""このスイープが使う variant を返す（環境変数 SM_RENDER_VARIANT > テンプレート YAML）.
条件ごとにシーン YAML を読み直すと 1 条件あたり 1 秒近く掛かるので，全条件で共通の
テンプレート（TEMPLATE_YAML）の render.variant を 1 回だけ解決して使い回す．
"""
global _render_variant # noqa: PLW0603
if _render_variant is None:
template = yaml.safe_load(TEMPLATE_YAML.read_text(encoding="utf-8"))
_render_variant = resolve_render_variant(str(template["render"]["variant"]))
return _render_variant
def ensure_render_variant() -> str:
"""レンダ前にプロセスの variant を確定させる（遮蔽検証も同じ variant で走らせるため）.
check_occlusion は mi.variant() が設定済みならそれを使うので，先に呼んでおくと
YAML の llvm_ad_rgb ではなくサーバの cuda_ad_rgb でレイキャストが走る
（球 1 条件で 25 秒 → 1 秒未満）．import mitsuba を伴うので，
既存結果を再利用するだけの条件では呼ばない．
"""
# mitsuba を読み込む import なので，モジュールレベルではなくここで行う
from sm_tias_lightsim.render.runner import ensure_variant
variant = render_variant()
ensure_variant(variant)
return variant
def verify_occlusion(scene_yaml: Path, messages: list[str]) -> dict[str, float]:
"""バッフル付き条件について板→LED の遮蔽率をレイキャストで測る（失敗しても続行）.
Dr.Jit のスレッド数は ensure_render_variant 内の apply_drjit_thread_count が
SM_DRJIT_THREADS を見て制限するので，ここでは触らない．
"""
try:
from verify_baffle_occlusion import check_occlusion
result = check_occlusion(scene_yaml, board_grid=41, led_count=32)
return {
"occlusion_blocked_frac": round(result["blocked"] / result["total"], 6),
"occlusion_total": int(result["total"]),
}
except Exception as exc: # noqa: BLE001 - 検証の失敗でスイープを止めない
messages.append(f" !! 遮蔽検証に失敗（続行）: {exc}")
return {"occlusion_blocked_frac": float("nan"), "occlusion_total": 0}
def condition_row(cond: Condition, record: dict, variant: str) -> dict:
"""レンダ前に決まる列（条件の指定・幾何の記録・variant）だけの集約行を作る."""
row: dict = {
"name": cond.name, "group": cond.group, "shape": cond.shape, "size_mm": cond.size,
"theta_deg": cond.theta_deg, "aim": cond.aim, "n_led": cond.n_led, "rep": cond.rep,
"azimuth_deg": cond.azimuth_deg,
}
row.update(record)
row["render_variant"] = variant
row["material"] = _material
return row
def render_and_evaluate(
scene_yaml: Path, out_dir: Path, variant: str, messages: list[str]
) -> bool:
"""1 条件をプロセス内でレンダ → 評価する（失敗しても例外を投げず False を返す）.
sm-render / sm-evaluate をサブプロセスで呼ぶ代わりにプロセス内で実行する
（起動 + import mitsuba + 図の描画で 1 条件あたり 4 秒前後を使っていたため）．
標準出力は条件ごとの評価サマリーで埋まるので取り込んでおき，失敗時だけ末尾を残す．
Returns:
成功したら True．失敗したら messages に理由を足して False．
"""
captured = io.StringIO()
try:
with redirect_stdout(captured):
render_yaml(scene_yaml, out_dir, variant)
evaluate_image(
out_dir / "luminance.npy", EVAL_CONFIG_YAML, out_dir, save_figures=False,
)
except Exception as exc: # noqa: BLE001 - 1 条件の失敗でスイープ全体を止めない
messages.append(f" !! 失敗: {exc}")
tail = captured.getvalue().strip()[-800:]
if tail:
messages.append(f" 出力: {tail}")
return False
return True
def result_columns(out_dir: Path) -> dict:
"""レンダ結果（metrics.csv / luminance.npy）から集約行の指標列を読む."""
columns: dict = dict(read_metrics_csv(out_dir / "metrics.csv"))
with (out_dir / "metrics.csv").open(encoding="utf-8") as fh:
metrics = next(csv.DictReader(fh))
for key in ("intensity_centroid_x_norm", "intensity_centroid_y_norm"):
columns[key] = float(metrics[key]) if metrics.get(key) else float("nan")
columns.update(raw_stats(out_dir / "luminance.npy"))
return columns
def run_job(
cond: Condition, scene_yaml: Path, record: dict, material: str = "legacy"
) -> tuple[dict | None, list[str]]:
"""1 条件をレンダ→評価まで実行し row.json を書いて集約行とログ行を返す.
ワーカープロセスからも呼べるよう，引数・戻り値は pickle 可能なものだけにする
（主プロセスの log クロージャは渡せないので，ログはメッセージとして返して
主プロセスが書き出す）．材質はワーカー（spawn）に主プロセスの設定が伝わらないので
引数で受け取り，ここで出力ディレクトリを切り替える．
Returns:
(集約行, ログ行のリスト)．失敗した条件の集約行は None．
"""
if material != _material: # 主プロセスや同一材質のワーカーでは出力先を触らない
set_material(material)
messages: list[str] = []
out_dir = OUTPUT_BASE / cond.name
row_json = out_dir / "row.json"
if row_json.exists():
messages.append(f"--- {cond.name}: 既存結果を再利用")
return json.loads(row_json.read_text(encoding="utf-8")), messages
variant = ensure_render_variant()
row = condition_row(cond, record, variant)
if record["baffled"]:
row.update(verify_occlusion(scene_yaml, messages))
messages.append(f"--- {cond.name}: レンダ開始")
out_dir.mkdir(parents=True, exist_ok=True)
start = time.time()
if not render_and_evaluate(scene_yaml, out_dir, variant, messages):
return None, messages
row["elapsed_s"] = round(time.time() - start, 1)
row.update(result_columns(out_dir))
row_json.write_text(json.dumps(row, ensure_ascii=False, indent=1), encoding="utf-8")
occl = row.get("occlusion_blocked_frac", float("nan"))
cx, cy = row["intensity_centroid_x_norm"], row["intensity_centroid_y_norm"]
messages.append(
f" -> {row['elapsed_s']:.0f}s CoV={row['cov']:.4f} centroid=({cx:+.4f},{cy:+.4f}) "
f"angular_ratio={row['angular_ratio']:.4f} 生mean={row['raw_mean_roi']:.4f} "
f"遮蔽={occl:.4f}")
return row, messages
def collect_rows() -> list[dict]:
"""出力ディレクトリ配下の row.json をすべて読む（並列ジョブの結果を合流させる）."""
rows = []
for path in sorted(OUTPUT_BASE.glob("*/row.json")):
rows.append(json.loads(path.read_text(encoding="utf-8")))
return rows
def write_summary() -> Path:
"""全 row.json から集約 CSV を書き直す（並列に走る形状別ジョブでも壊れない）."""
order = {name: i for i, name in enumerate(c.name for c in conditions())}
rows = sorted(collect_rows(), key=lambda r: order.get(r["name"], len(order)))
summary_csv = OUTPUT_BASE / "summary.csv"
summary_csv.parent.mkdir(parents=True, exist_ok=True)
with summary_csv.open("w", newline="", encoding="utf-8") as fh:
writer = csv.DictWriter(fh, fieldnames=SUMMARY_COLUMNS)
writer.writeheader()
for row in rows:
writer.writerow({k: row.get(k, "") for k in SUMMARY_COLUMNS})
return summary_csv
def main() -> int:
"""照明配置スイープを実行するメインエントリポイント."""
parser = argparse.ArgumentParser(description="照明配置スイープ（球 + 立方体）")
parser.add_argument("--prepare-only", action="store_true",
help="メッシュとシーン YAML の生成だけ行う（3D 配置確認用）")
parser.add_argument("--shapes", choices=("sphere", "cube"), nargs="*", default=None,
help="実行する形状（省略時は両方）")
parser.add_argument("--groups", choices=GROUPS, nargs="*", default=None,
help="実行する条件グループ（省略時は全部）")
parser.add_argument("--limit", type=int, default=None,
help="先頭 N 条件だけ実行する（動作確認用）")
parser.add_argument("--material", choices=tuple(MATERIALS), default="legacy",
help="内壁材質（calibrated=TEST_14 の純 diffuse ρ=0.81．出力は別 dir）")
parser.add_argument("--key-subset", action="store_true",
help="材質の引き直し用の主要条件だけに絞る")
parser.add_argument("--workers", type=int, default=1,
help="同時に実行する条件数（既定 1＝逐次．GPU 1 枚なら 2〜3 が目安）")
parser.add_argument("--list", action="store_true", help="条件一覧を表示して終了")
args = parser.parse_args()
set_material(args.material)
conds = [c for c in conditions()
if (not args.shapes or c.shape in args.shapes)
and (not args.groups or c.group in args.groups)]
if args.key_subset:
conds = key_subset(conds)
if args.limit:
conds = conds[: args.limit]
if args.list:
for c in conds:
print(c.name)
print(f"{len(conds)} 条件")
return 0
log_path = ROOT / "output" / "_sweep_led_placement.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
log_file = log_path.open("a", encoding="utf-8")
def log(message: str) -> None:
print(message, flush=True)
log_file.write(message + "\n")
log_file.flush()
log(f"=== 照明配置スイープ（{len(conds)} 条件）===")
suffix = "_" + "_".join(args.shapes) if args.shapes else ""
prepared = prepare(conds, log, suffix)
if args.prepare_only:
log(f"--prepare-only のため終了．シーン YAML: {TMP_DIR}")
log_file.close()
return 0
jobs = [(cond, *prepared[cond.name]) for cond in conds if cond.name in prepared]
done = 0
def log_result(row: dict | None, messages: list[str]) -> int:
"""1 条件の結果をログに流し，成功していれば集約 CSV を書き直して 1 を返す."""
for message in messages:
log(message)
if row is None:
return 0
write_summary()
return 1
if args.workers > 1:
# ワーカーは spawn（Windows 互換）で立ち上げ，各プロセスが自前で variant を
# 1 回設定してから条件を処理する．集約 CSV の書き出しは主プロセスだけが行う．
log(f"=== {args.workers} 並列で実行（1 ワーカー 1 プロセス）===")
context = multiprocessing.get_context("spawn")
with ProcessPoolExecutor(max_workers=args.workers, mp_context=context) as pool:
futures = {
pool.submit(run_job, cond, scene_yaml, geometry_record, args.material): cond
for cond, scene_yaml, geometry_record in jobs
}
for future in as_completed(futures):
cond = futures[future]
try:
row, messages = future.result()
except Exception as exc: # noqa: BLE001 - 1 条件の異常で全体を止めない
log(f" !! {cond.name}: ワーカーが異常終了（続行）: {exc}")
continue
done += log_result(row, messages)
else:
for cond, scene_yaml, geometry_record in jobs:
done += log_result(*run_job(cond, scene_yaml, geometry_record, args.material))
log(f"\n=== 完了: {done} / {len(prepared)} 条件 ===")
log(f" 集約 CSV: {write_summary()}")
log_file.close()
return 0
if __name__ == "__main__":
sys.exit(main())
