ajustado scripts de calibragem

This commit is contained in:
Diego Freitas 2026-09-09 15:52:08 -03:00
parent f35553ec08
commit 0319fd204d
8 changed files with 4297 additions and 2597 deletions

File diff suppressed because it is too large Load Diff

View File

@ -15,7 +15,9 @@ Topologia oficial do produto
Objetivos
---------
- Detectar e validar o hardware antes de permitir a calibração.
- Carregar o module profile criado pelo Script 0 antes da calibração.
- Conectar automaticamente ao MX ID registrado no profile.
- Validar o hardware conectado contra o contrato do profile.
- Suportar OV9782 e AR0234 na CAM_A.
- Mostrar RGB + RE + NIR simultaneamente.
- Deixar AE estabilizar e depois congelar exposição/ISO.
@ -24,6 +26,7 @@ Objetivos
- Exigir estabilidade próxima ao pico antes do ACCEPT.
- Medir centro + 4 regiões periféricas para detectar tilt/decentering.
- Salvar evidência auditável por sessão.
- Nunca depender de module_params (ele ainda não existe nesta etapa).
- Nunca alterar module_params nem qualquer calibração de runtime.
- Nunca sobrescrever um resultado ativo aprovado com uma sessão FAIL/CANCELLED.
@ -34,7 +37,12 @@ Foco é um ajuste FÍSICO da lente. Portanto:
module_params_fragment = null
O JSON produzido por esta ferramenta é um relatório de QC/rastreabilidade.
O assembler final do module_params pode ignorá-lo.
O assembler final pode validar seu PASS e sua assinatura de hardware, mas não
há fragmento de configuração de runtime para mesclar.
O Bayer NÃO participa da medição de foco, pois a RGB é medida na saída ISP.
Mesmo assim, a identidade de decode definida no Script 0 é copiada para o
relatório como proveniência, com used_by_focus_measurement=false.
Target recomendado
------------------
@ -105,9 +113,17 @@ Há um override de engenharia:
que permite aceitar sem bracket/field QA, mas isso fica registrado no JSON
e o resultado não é considerado homologação normal.
Saídas padrão
-------------
calibration/focus_candidates/<session_id>/
Entrada obrigatória
-------------------
Por padrão, o script lê:
calibration/active_module_profile.json
Também é possível selecionar explicitamente:
--module-profile calibration/mp_ar0234/module_profile_v1.json
Saídas padrão (herdadas do module profile)
------------------------------------------
calibration/<profile>/focus_candidates/<session_id>/
report.json
PASS.txt / FAIL.txt / CANCELLED.txt
snapshots/
@ -118,7 +134,7 @@ calibration/focus_candidates/<session_id>/
nir_...
Se PASS:
calibration/focus_qc_active.json
calibration/<profile>/focus_qc_active.json
Nenhum NPZ é gerado.
Nenhuma chave do module_params é alterada.
@ -207,6 +223,9 @@ SENSOR_ORIENTATION = {
}
SCHEMA = "multispec_focus_qc_v3"
MODULE_PROFILE_SCHEMA = "multispec_module_profile_v1"
ACTIVE_SELECTOR_SCHEMA = "multispec_active_module_profile_v1"
BAYER_PATTERNS = ("RGGB", "BGGR", "GRBG", "GBRG")
# ============================================================
@ -302,6 +321,161 @@ def sha256_file(path: str | Path) -> str:
return h.hexdigest()
def load_json_file(path: str | Path) -> dict:
p = Path(path)
if not p.is_file():
raise RuntimeError(f"JSON obrigatório não encontrado: {p}")
try:
with p.open("r", encoding="utf-8") as f:
data = json.load(f)
except Exception as exc:
raise RuntimeError(f"Falha ao ler JSON {p}: {exc}") from exc
if not isinstance(data, dict):
raise RuntimeError(f"JSON deve conter um objeto na raiz: {p}")
return data
def resolve_existing_path(value: str | Path, anchor: Optional[Path] = None) -> Path:
"""Resolve paths do contrato sem depender do diretório do próprio script."""
raw = Path(value).expanduser()
candidates = [raw]
if not raw.is_absolute() and anchor is not None:
candidates.append(anchor.parent / raw)
for candidate in candidates:
if candidate.is_file():
return candidate.resolve()
tried = ", ".join(str(x) for x in candidates)
raise RuntimeError(f"Arquivo de contrato não encontrado. Tentativas: {tried}")
def validate_module_profile(profile: dict, profile_path: Path):
if profile.get("schema") != MODULE_PROFILE_SCHEMA:
raise RuntimeError(
f"Schema de module profile inesperado em {profile_path}: "
f"{profile.get('schema')!r}"
)
if str(profile.get("status", "")).lower() != "active":
raise RuntimeError(
f"Module profile não está ativo: status={profile.get('status')!r}"
)
profile_name = str(profile.get("profile_name") or "").strip()
profile_dir = str(profile.get("profile_dir") or "").strip()
device_mx_id = str(profile.get("device_mx_id") or "").strip()
if not profile_name or not profile_dir or not device_mx_id:
raise RuntimeError(
"Module profile incompleto: profile_name, profile_dir e "
"device_mx_id são obrigatórios."
)
signature = profile.get("hardware_signature")
if not isinstance(signature, dict):
raise RuntimeError("Module profile sem hardware_signature válido.")
for role in ROLES:
item = signature.get(role)
if not isinstance(item, dict):
raise RuntimeError(f"hardware_signature sem role {role!r}.")
if not item.get("socket") or not item.get("sensor"):
raise RuntimeError(f"hardware_signature/{role} sem socket/sensor.")
size = item.get("size")
if not isinstance(size, (list, tuple)) or len(size) != 2:
raise RuntimeError(f"hardware_signature/{role}/size inválido: {size!r}")
rgb_decode = profile.get("rgb_decode")
if not isinstance(rgb_decode, dict):
raise RuntimeError("Module profile sem rgb_decode válido.")
bayer = str(rgb_decode.get("bayer_pattern") or "").upper()
if bayer not in BAYER_PATTERNS:
raise RuntimeError(f"rgb_decode.bayer_pattern inválido: {bayer!r}")
orientations = profile.get("preview_orientation_defaults")
if not isinstance(orientations, dict):
raise RuntimeError("Module profile sem preview_orientation_defaults.")
artifacts = profile.get("artifact_paths")
candidate_roots = artifacts.get("candidate_roots") if isinstance(artifacts, dict) else None
if not isinstance(candidate_roots, dict) or not candidate_roots.get("focus"):
raise RuntimeError("Module profile sem artifact_paths.candidate_roots.focus.")
if not artifacts.get("focus_qc_json"):
raise RuntimeError("Module profile sem artifact_paths.focus_qc_json.")
def load_module_profile_contract(
module_profile_arg: Optional[str],
active_selector_arg: str,
):
selector_path = None
selector = None
if module_profile_arg:
profile_path = resolve_existing_path(module_profile_arg)
else:
try:
selector_path = resolve_existing_path(active_selector_arg)
except RuntimeError as exc:
raise RuntimeError(
"Active module profile não encontrado. Execute primeiro o "
"Script 0 ou informe --module-profile explicitamente."
) from exc
selector = load_json_file(selector_path)
if selector.get("schema") != ACTIVE_SELECTOR_SCHEMA:
raise RuntimeError(
f"Schema do seletor ativo inesperado: {selector.get('schema')!r}"
)
declared = selector.get("profile_path")
if not declared:
raise RuntimeError("Seletor ativo sem profile_path.")
profile_path = resolve_existing_path(declared, anchor=selector_path)
profile = load_json_file(profile_path)
validate_module_profile(profile, profile_path)
profile_sha256 = sha256_file(profile_path)
if selector is not None:
declared_sha = str(selector.get("profile_sha256") or "").lower()
if declared_sha and declared_sha != profile_sha256.lower():
raise RuntimeError(
"Hash do module profile diverge do active selector. "
"Execute novamente o Script 0 para oficializar o profile."
)
selector_mx = str(selector.get("device_mx_id") or "")
profile_mx = str(profile.get("device_mx_id") or "")
if selector_mx and selector_mx != profile_mx:
raise RuntimeError("MX ID do selector diverge do module profile.")
return {
"profile": profile,
"profile_path": profile_path,
"profile_sha256": profile_sha256,
"selector": selector,
"selector_path": selector_path,
}
def resolve_output_path(value: str | Path) -> Path:
"""Paths de artifact_paths são deliberadamente relativos à raiz do projeto."""
return Path(value).expanduser()
def safe_int(v, default=None):
try:
if v is None:
@ -599,9 +773,52 @@ def rect_from_norm(shape_hw, x0, y0, x1, y1):
)
def normalize_preview_orientation(cfg: Optional[dict], label: str = "preview") -> dict:
cfg = cfg or {}
try:
rotate_deg = int(cfg.get("rotate_deg", 0)) % 360
except Exception as exc:
raise RuntimeError(f"{label}: rotate_deg inválido.") from exc
if rotate_deg not in (0, 90, 180, 270):
raise RuntimeError(
f"{label}: rotate_deg={rotate_deg}; permitidos=0,90,180,270."
)
flip_horizontal = cfg.get("flip_horizontal", False)
flip_vertical = cfg.get("flip_vertical", False)
if not isinstance(flip_horizontal, bool) or not isinstance(flip_vertical, bool):
raise RuntimeError(f"{label}: flips devem ser booleanos.")
return {
"rotate_deg": rotate_deg,
"flip_horizontal": flip_horizontal,
"flip_vertical": flip_vertical,
}
def resolve_preview_orientation(profile: dict, specs: Dict[str, "SensorSpec"]) -> dict:
declared = profile.get("preview_orientation_defaults", {})
result = {}
for role in ROLES:
spec = specs[role]
fallback = SENSOR_ORIENTATION.get(spec.sensor_name, {})
cfg = declared.get(role, fallback)
result[role] = normalize_preview_orientation(
cfg,
label=f"preview_orientation_defaults/{role}",
)
return result
def apply_sensor_orientation(
img: np.ndarray,
sensor_name: str,
orientation_cfg: Optional[dict] = None,
) -> np.ndarray:
if img is None:
return img
@ -611,12 +828,11 @@ def apply_sensor_orientation(
or ""
).strip().upper()
cfg = (
SENSOR_ORIENTATION.get(
sensor,
{}
)
or {}
cfg = normalize_preview_orientation(
orientation_cfg
if orientation_cfg is not None
else SENSOR_ORIENTATION.get(sensor, {}),
label=f"orientation/{sensor or 'unknown'}",
)
rotate_deg = int(
@ -860,6 +1076,51 @@ def build_specs(camera_rows) -> Dict[str, SensorSpec]:
return specs
def validate_profile_against_device(
profile: dict,
specs: Dict[str, SensorSpec],
actual_mx: Optional[str],
):
expected_mx = str(profile.get("device_mx_id") or "")
actual_mx_text = str(actual_mx or "")
if expected_mx and actual_mx_text != expected_mx:
raise RuntimeError(
f"Câmera conectada não pertence ao profile ativo: "
f"MX esperado={expected_mx}, detectado={actual_mx_text or None}."
)
signature = profile["hardware_signature"]
errors = []
for role in ROLES:
expected = signature[role]
actual = specs[role]
expected_size = tuple(int(x) for x in expected["size"])
actual_size = (actual.configured_width, actual.configured_height)
if str(expected["socket"]).upper() != actual.socket_name.upper():
errors.append(
f"{role}: socket profile={expected['socket']} device={actual.socket_name}"
)
if str(expected["sensor"]).upper() != actual.sensor_name.upper():
errors.append(
f"{role}: sensor profile={expected['sensor']} device={actual.sensor_name}"
)
if expected_size != actual_size:
errors.append(
f"{role}: size profile={expected_size} device={actual_size}"
)
if errors:
raise RuntimeError(
"Hardware conectado diverge do module profile ativo:\n - "
+ "\n - ".join(errors)
)
# ============================================================
# Pipeline e controles
# ============================================================
@ -1851,11 +2112,19 @@ class FocusRuntime:
specs: Dict[str, SensorSpec],
args,
qa,
preview_orientation: Dict[str, dict],
):
self.device = device
self.specs = specs
self.args = args
self.qa = qa
self.preview_orientation = {
role: normalize_preview_orientation(
preview_orientation.get(role),
label=f"runtime_preview/{role}",
)
for role in ROLES
}
self.output_queues = {
role: device.getOutputQueue(
@ -1947,6 +2216,7 @@ class FocusRuntime:
img = apply_sensor_orientation(
img,
sensor_name=spec.sensor_name,
orientation_cfg=self.preview_orientation[role],
)
if img.ndim == 2:
@ -3308,7 +3578,14 @@ def build_base_report(
isp_settings,
qa,
candidate_dir,
profile_contract,
preview_orientation,
):
profile = profile_contract["profile"]
profile_path = profile_contract["profile_path"]
selector_path = profile_contract["selector_path"]
rgb_decode = dict(profile["rgb_decode"])
return {
"schema": SCHEMA,
"created_at": now_str(),
@ -3326,7 +3603,41 @@ def build_base_report(
),
"device_mx_id": actual_mx,
"usb_speed": usb_speed,
"module_profile_contract": {
"schema": profile.get("schema"),
"profile_name": profile.get("profile_name"),
"profile_path": str(profile_path),
"profile_sha256": profile_contract["profile_sha256"],
"active_selector_path": (
str(selector_path) if selector_path is not None else None
),
"device_match": True,
"artifact_routing": {
"candidate_root": str(args.candidate_root),
"candidate_root_source": (
"cli_override"
if args.candidate_root_overridden
else "module_profile.artifact_paths.candidate_roots.focus"
),
"active_json": str(args.active_json),
"active_json_source": (
"cli_override"
if args.active_json_overridden
else "module_profile.artifact_paths.focus_qc_json"
),
},
},
"rgb_decode_provenance": {
**rgb_decode,
"used_by_focus_measurement": False,
"focus_measurement_source": "ISP",
"notes": (
"Bayer/decode pertence ao module profile. O Focus apenas "
"registra a identidade; sua métrica usa o stream ISP."
),
},
"topology": topology_report,
"hardware_signature": profile["hardware_signature"],
"camera_inventory": [
{
k: v
@ -3386,6 +3697,7 @@ def build_base_report(
},
"notes": args.notes or "",
"candidate_dir": str(candidate_dir),
"active_json": str(args.active_json),
"preflight": None,
"locked_controls": None,
"calibration_roi_by_role": None,
@ -3395,14 +3707,8 @@ def build_base_report(
"orientation_used": {
role: {
"sensor": specs[role].sensor_name,
**SENSOR_ORIENTATION.get(
specs[role].sensor_name,
{
"rotate_deg": 0,
"flip_horizontal": False,
"flip_vertical": False,
},
),
**preview_orientation[role],
"source": "module_profile.preview_orientation_defaults",
}
for role in ROLES
},
@ -3426,6 +3732,23 @@ def main():
parser.add_argument(
"--mx-id",
default=None,
help=(
"Confirmação explícita opcional. Deve coincidir com o module profile ativo."
),
)
parser.add_argument(
"--module-profile",
default=None,
help=(
"Module profile explícito. Se omitido, resolve pelo active selector."
),
)
parser.add_argument(
"--active-selector",
default="calibration/active_module_profile.json",
help="Seletor criado pelo Script 0.",
)
parser.add_argument(
@ -3559,12 +3882,19 @@ def main():
# Output
parser.add_argument(
"--candidate-root",
default="calibration/focus_candidates",
default=None,
help=(
"Override de saída. Por padrão usa o module profile "
"artifact_paths.candidate_roots.focus."
),
)
parser.add_argument(
"--active-json",
default="calibration/focus_qc_active.json",
default=None,
help=(
"Override. Por padrão usa artifact_paths.focus_qc_json do profile."
),
)
parser.add_argument(
@ -3611,6 +3941,36 @@ def main():
args = parser.parse_args()
# O Script 0 é a autoridade de identidade e roteamento. Focus não lê
# module_params, pois ele ainda não existe nesta etapa do fluxo.
profile_contract = load_module_profile_contract(
module_profile_arg=args.module_profile,
active_selector_arg=args.active_selector,
)
module_profile = profile_contract["profile"]
artifacts = module_profile["artifact_paths"]
args.candidate_root_overridden = args.candidate_root is not None
args.active_json_overridden = args.active_json is not None
expected_mx = str(module_profile.get("device_mx_id") or "")
if args.mx_id and str(args.mx_id) != expected_mx:
raise RuntimeError(
f"--mx-id={args.mx_id} diverge do module profile ativo ({expected_mx})."
)
args.mx_id = expected_mx
args.candidate_root = str(
resolve_output_path(
args.candidate_root or artifacts["candidate_roots"]["focus"]
)
)
args.active_json = str(
resolve_output_path(
args.active_json or artifacts["focus_qc_json"]
)
)
# Campo mutável apenas em runtime para E/M.
args.equalize_runtime = bool(args.equalize)
@ -3667,11 +4027,24 @@ def main():
camera_rows
)
validate_profile_against_device(
module_profile,
specs,
actual_mx,
)
preview_orientation = resolve_preview_orientation(
module_profile,
specs,
)
print("=" * 82)
print("FOCUS CALIBRATION - PRODUCTION")
print(f"DepthAI : {getattr(dai, '__version__', 'unknown')}")
print(f"MX ID : {actual_mx}")
print(f"USB : {usb_speed}")
print(f"Profile : {module_profile['profile_name']}")
print(f"Profile JSON: {profile_contract['profile_path']}")
print("-" * 82)
for role in ROLES:
@ -3686,6 +4059,13 @@ def main():
f"{spec.source}"
)
orient = preview_orientation[role]
print(
f" preview rotate={orient['rotate_deg']} "
f"flip_h={orient['flip_horizontal']} "
f"flip_v={orient['flip_vertical']} | source=module_profile"
)
print("=" * 82)
# ----------------------------------------------------
@ -3722,6 +4102,8 @@ def main():
isp_settings,
qa,
candidate_dir,
profile_contract,
preview_orientation,
)
with device_ctx as device:
@ -3730,6 +4112,7 @@ def main():
specs,
args,
qa,
preview_orientation,
)
# ------------------------------------------------

View File

@ -16,7 +16,7 @@ Domínio:
sem undistort externo
Gera:
calibration/intrinsics_calibration_v1.json
calibration/<profile>/intrinsics_calibration_v1.json
O artefato contém:
module_params_fragment.intrinsics_config
@ -31,15 +31,24 @@ IMPORTANTE:
espaço nativo do sensor.
Fluxo:
1) Mostre o ChArUco nas 3 câmeras.
2) ENTER trava EXP/ISO.
3) R seleciona a ROI operacional da câmera escolhida (opcional).
4) Enquadre o ChArUco no quadrilátero-guia.
5) G captura e avança; F força captura fora da tolerância do guia.
6) O roteiro varia X/Y, distância, yaw, pitch e roll.
7) Ideal: completar as 25 poses guiadas.
8) A calibra e executa acceptance.
9) PASS promove o JSON ativo.
1) Carrega o module profile criado pelo Script 0.
2) Exige Focus QC ativo, homologado e pertencente ao mesmo profile.
3) Mostre o ChArUco nas 3 câmeras.
4) ENTER trava EXP/ISO.
5) A ROI operacional nasce da CALIBRATION ROI do Focus.
6) R pode selecionar outra ROI para esta sessão; C usa frame completo.
7) Enquadre o ChArUco no quadrilátero-guia.
8) G captura e avança; F força captura fora da tolerância do guia.
9) O roteiro varia X/Y, distância, yaw, pitch e roll.
10) Ideal: completar as 25 poses guiadas.
11) A calibra e executa acceptance.
12) PASS promove o JSON ativo dentro da pasta do profile.
Contrato de bootstrap:
- NÃO lê module_params.json; ele ainda não existe nesta etapa.
- O MX ID, hardware, Bayer/decode, orientação e destinos vêm do Script 0.
- Bayer é registrado como proveniência, mas não participa da medição:
RGB usa ISP; RE/NIR usam MONO_OUT.
Teclas:
G captura view quando o guia estiver READY
@ -56,6 +65,7 @@ Teclas:
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
@ -74,6 +84,10 @@ import numpy as np
ROLES = ("rgb", "re", "nir")
SCHEMA = "multispec_intrinsics_calibration_v1"
CALIBRATION_SPACE = "native_stream_no_external_undistort"
MODULE_PROFILE_SCHEMA = "multispec_module_profile_v1"
ACTIVE_SELECTOR_SCHEMA = "multispec_active_module_profile_v1"
FOCUS_SCHEMA = "multispec_focus_qc_v3"
BAYER_PATTERNS = ("RGGB", "BGGR", "GRBG", "GBRG")
TOPOLOGY = {
"rgb": ("CAM_A", ("OV9782", "AR0234")),
@ -212,6 +226,209 @@ def save_json_atomic(path, data):
os.replace(tmp, p)
def sha256_file(path):
h = hashlib.sha256()
with Path(path).open("rb") as f:
while True:
chunk = f.read(1024 * 1024)
if not chunk:
break
h.update(chunk)
return h.hexdigest()
def load_json_file(path):
p = Path(path)
if not p.is_file():
raise RuntimeError(f"JSON obrigatório não encontrado: {p}")
try:
with p.open("r", encoding="utf-8") as f:
data = json.load(f)
except Exception as exc:
raise RuntimeError(f"Falha ao ler JSON {p}: {exc}") from exc
if not isinstance(data, dict):
raise RuntimeError(f"JSON deve conter um objeto na raiz: {p}")
return data
def resolve_existing_path(value, anchor=None):
raw = Path(value).expanduser()
candidates = [raw]
if not raw.is_absolute() and anchor is not None:
candidates.append(Path(anchor).parent / raw)
for candidate in candidates:
if candidate.is_file():
return candidate.resolve()
tried = ", ".join(str(x) for x in candidates)
raise RuntimeError(f"Arquivo de contrato não encontrado. Tentativas: {tried}")
def resolve_output_path(value):
"""artifact_paths do Script 0 são relativos à raiz do projeto."""
return Path(value).expanduser()
def validate_module_profile(profile, profile_path):
if profile.get("schema") != MODULE_PROFILE_SCHEMA:
raise RuntimeError(
f"Schema de module profile inesperado em {profile_path}: "
f"{profile.get('schema')!r}"
)
if str(profile.get("status", "")).lower() != "active":
raise RuntimeError(
f"Module profile não está ativo: status={profile.get('status')!r}"
)
required = ("profile_name", "profile_dir", "device_mx_id")
missing = [key for key in required if not str(profile.get(key) or "").strip()]
if missing:
raise RuntimeError(f"Module profile incompleto: campos ausentes={missing}")
signature = profile.get("hardware_signature")
if not isinstance(signature, dict):
raise RuntimeError("Module profile sem hardware_signature válido.")
for role in ROLES:
item = signature.get(role)
if not isinstance(item, dict):
raise RuntimeError(f"hardware_signature sem role {role!r}.")
size = item.get("size")
if not item.get("socket") or not item.get("sensor"):
raise RuntimeError(f"hardware_signature/{role} sem socket/sensor.")
if not isinstance(size, (list, tuple)) or len(size) != 2:
raise RuntimeError(f"hardware_signature/{role}/size inválido: {size!r}")
rgb_decode = profile.get("rgb_decode")
if not isinstance(rgb_decode, dict):
raise RuntimeError("Module profile sem rgb_decode válido.")
bayer = str(rgb_decode.get("bayer_pattern") or "").upper()
if bayer not in BAYER_PATTERNS:
raise RuntimeError(f"rgb_decode.bayer_pattern inválido: {bayer!r}")
orientations = profile.get("preview_orientation_defaults")
if not isinstance(orientations, dict):
raise RuntimeError("Module profile sem preview_orientation_defaults.")
for role in ROLES:
normalize_preview_orientation(orientations.get(role))
artifacts = profile.get("artifact_paths")
roots = artifacts.get("candidate_roots") if isinstance(artifacts, dict) else None
if not isinstance(roots, dict) or not roots.get("intrinsics"):
raise RuntimeError(
"Module profile sem artifact_paths.candidate_roots.intrinsics."
)
for key in ("focus_qc_json", "intrinsics_json"):
if not artifacts.get(key):
raise RuntimeError(f"Module profile sem artifact_paths.{key}.")
def load_module_profile_contract(module_profile_arg, active_selector_arg):
selector = None
selector_path = None
if module_profile_arg:
profile_path = resolve_existing_path(module_profile_arg)
else:
try:
selector_path = resolve_existing_path(active_selector_arg)
except RuntimeError as exc:
raise RuntimeError(
"Active module profile não encontrado. Execute primeiro o "
"Script 0 ou informe --module-profile explicitamente."
) from exc
selector = load_json_file(selector_path)
if selector.get("schema") != ACTIVE_SELECTOR_SCHEMA:
raise RuntimeError(
f"Schema do seletor ativo inesperado: {selector.get('schema')!r}"
)
declared = selector.get("profile_path")
if not declared:
raise RuntimeError("Seletor ativo sem profile_path.")
profile_path = resolve_existing_path(declared, anchor=selector_path)
profile = load_json_file(profile_path)
validate_module_profile(profile, profile_path)
profile_sha256 = sha256_file(profile_path)
if selector is not None:
declared_sha = str(selector.get("profile_sha256") or "").lower()
if declared_sha and declared_sha != profile_sha256.lower():
raise RuntimeError(
"Hash do module profile diverge do active selector. "
"Execute novamente o Script 0."
)
if str(selector.get("device_mx_id") or "") != str(
profile.get("device_mx_id") or ""
):
raise RuntimeError("MX ID do selector diverge do module profile.")
return {
"profile": profile,
"profile_path": profile_path,
"profile_sha256": profile_sha256,
"selector": selector,
"selector_path": selector_path,
}
def load_focus_prerequisite(profile_contract):
profile = profile_contract["profile"]
declared = profile["artifact_paths"]["focus_qc_json"]
try:
path = resolve_existing_path(declared)
except RuntimeError as exc:
raise RuntimeError(
"Focus QC ativo não encontrado para este profile. Execute e aprove "
"o Script 1 antes dos intrínsecos."
) from exc
focus = load_json_file(path)
if focus.get("schema") != FOCUS_SCHEMA:
raise RuntimeError(f"Schema do Focus QC inesperado: {focus.get('schema')!r}")
if str(focus.get("status", "")).lower() != "pass":
raise RuntimeError(f"Focus QC não está PASS: status={focus.get('status')!r}")
if focus.get("promoted") is not True:
raise RuntimeError("Focus QC não foi promovido.")
if focus.get("normal_production_homologation") is not True:
raise RuntimeError("Focus QC não representa homologação normal de produção.")
if str(focus.get("device_mx_id") or "") != str(profile["device_mx_id"]):
raise RuntimeError("Focus QC pertence a outro MX ID.")
provenance = focus.get("module_profile_contract")
if not isinstance(provenance, dict):
raise RuntimeError(
"Focus QC não possui o novo module_profile_contract. "
"Rode novamente o Script 1 revisado."
)
if provenance.get("profile_sha256") != profile_contract["profile_sha256"]:
raise RuntimeError("Focus QC foi gerado com outra versão do module profile.")
if provenance.get("profile_name") != profile.get("profile_name"):
raise RuntimeError("Focus QC pertence a outro profile.")
results = focus.get("results_by_role")
if not isinstance(results, dict) or not all(
isinstance(results.get(role), dict) and results[role].get("accepted") is True
for role in ROLES
):
raise RuntimeError("Focus QC não contém acceptance das três câmeras.")
focus_rois = focus.get("calibration_roi_by_role")
if not isinstance(focus_rois, dict) or not all(
isinstance(focus_rois.get(role), (list, tuple))
and len(focus_rois[role]) == 4
for role in ROLES
):
raise RuntimeError("Focus QC não contém CALIBRATION ROI das três câmeras.")
if focus.get("hardware_signature") != profile.get("hardware_signature"):
raise RuntimeError("Assinatura de hardware do Focus diverge do profile.")
return {
"report": focus,
"path": path,
"sha256": sha256_file(path),
}
def status_rank(s):
return {"good": 0, "warning": 1, "bad": 2}.get(str(s), 2)
@ -480,38 +697,20 @@ def build_guide_targets():
]
def resolve_preview_orientation(specs, args):
def resolve_preview_orientation(profile, specs):
"""
Resolve orientação visual por sensor, com override opcional por role.
Resolve a orientação visual exclusivamente pelo module profile.
Isso NÃO muda o domínio científico da calibração.
"""
result = {}
declared = profile.get("preview_orientation_defaults", {})
for role in ROLES:
sensor = str(specs[role].sensor).upper()
base = dict(
PREVIEW_ORIENTATION_BY_SENSOR.get(
sensor,
{
"rotate_deg": 0,
"flip_horizontal": False,
"flip_vertical": False,
},
)
)
rotate_override = getattr(args, f"{role}_preview_rotate_deg")
if rotate_override is not None:
base["rotate_deg"] = int(rotate_override)
if getattr(args, f"{role}_preview_flip_horizontal"):
base["flip_horizontal"] = True
if getattr(args, f"{role}_preview_flip_vertical"):
base["flip_vertical"] = True
base = declared.get(role)
if not isinstance(base, dict):
base = PREVIEW_ORIENTATION_BY_SENSOR.get(sensor, {})
result[role] = normalize_preview_orientation(base)
return result
@ -686,6 +885,64 @@ def hardware_signature(specs):
}
def validate_profile_against_device(profile, specs, actual_mx):
expected_mx = str(profile.get("device_mx_id") or "")
detected_mx = str(actual_mx or "")
if detected_mx != expected_mx:
raise RuntimeError(
"Câmera conectada não pertence ao profile ativo: "
f"MX esperado={expected_mx}, detectado={detected_mx or None}."
)
expected_signature = profile["hardware_signature"]
detected_signature = hardware_signature(specs)
errors = []
for role in ROLES:
expected = expected_signature[role]
detected = detected_signature[role]
for key in ("socket", "sensor", "size"):
if expected.get(key) != detected.get(key):
errors.append(
f"{role}/{key}: profile={expected.get(key)!r} "
f"device={detected.get(key)!r}"
)
if errors:
raise RuntimeError(
"Hardware conectado diverge do module profile ativo:\n - "
+ "\n - ".join(errors)
)
def focus_roi_to_native(rect, spec, orientation):
"""Converte a ROI canônica [x0,y0,x1,y1] do Focus para SENSOR-NATIVE."""
if not isinstance(rect, (list, tuple)) or len(rect) != 4:
raise RuntimeError(f"ROI do Focus inválida para {spec.role}: {rect!r}")
x0, y0, x1, y1 = (int(x) for x in rect)
if x1 <= x0 or y1 <= y0:
raise RuntimeError(f"ROI do Focus vazia para {spec.role}: {rect!r}")
canonical = np.asarray([
[x0, y0], [x1 - 1, y0],
[x1 - 1, y1 - 1], [x0, y1 - 1],
], dtype=np.float32)
native = unorient_points(
canonical,
(spec.height, spec.width),
orientation,
)
nx0, ny0 = np.floor(native.min(axis=0)).astype(int)
nx1, ny1 = np.ceil(native.max(axis=0)).astype(int)
return normalize_roi(
{
"x": int(nx0),
"y": int(ny0),
"width": int(nx1 - nx0 + 1),
"height": int(ny1 - ny0 + 1),
},
(spec.width, spec.height),
)
def build_pipeline(specs, fps, sharpness, luma, chroma):
p = dai.Pipeline()
isp = {"sharpness": None, "luma_denoise": None, "chroma_denoise": None,
@ -2234,7 +2491,19 @@ def main():
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
ap.add_argument("--mx-id", default=None)
ap.add_argument(
"--mx-id", default=None,
help="Confirmação opcional; deve coincidir com o module profile ativo.",
)
ap.add_argument(
"--module-profile", default=None,
help="Profile explícito; se omitido, usa o active selector do Script 0.",
)
ap.add_argument(
"--active-selector",
default="calibration/active_module_profile.json",
help="Seletor criado pelo Script 0.",
)
ap.add_argument("--fps", type=float, default=20.0)
ap.add_argument("--isp-sharpness", type=int, default=0)
@ -2272,17 +2541,20 @@ def main():
metavar="X,Y,W,H",
help=(
"ROI operacional em coordenadas SENSOR-NATIVE. "
"Vazio usa o frame completo; também pode ser selecionada com R."
"Vazio herda a CALIBRATION ROI do Focus; também pode ser "
"selecionada com R."
),
)
ap.add_argument(
"--candidate-root",
default="calibration/intrinsics_candidates",
default=None,
help="Override; por padrão vem de artifact_paths.candidate_roots.intrinsics.",
)
ap.add_argument(
"--active-json",
default="calibration/intrinsics_calibration_v1.json",
default=None,
help="Override; por padrão vem de artifact_paths.intrinsics_json.",
)
ap.add_argument("--module-id", default="")
@ -2292,31 +2564,42 @@ def main():
ap.add_argument("--lens-nir", default="")
ap.add_argument("--notes", default="")
# Orientação SOMENTE de preview.
# Defaults vêm do sensor detectado; as flags abaixo são overrides locais.
for role in ROLES:
ap.add_argument(
f"--{role}-preview-rotate-deg",
type=int,
default=None,
choices=[0, 90, 180, 270],
help=(
"Rotação SOMENTE do preview. "
"Não altera o domínio da calibração intrínseca."
),
)
ap.add_argument(
f"--{role}-preview-flip-horizontal",
action="store_true",
help="Flip horizontal somente para preview.",
)
ap.add_argument(
f"--{role}-preview-flip-vertical",
action="store_true",
help="Flip vertical somente para preview.",
args = ap.parse_args()
if args.charuco_squares_x < 3 or args.charuco_squares_y < 3:
raise ValueError("ChArUco deve ter ao menos 3x3 quadrados.")
if not (
args.charuco_square_length > 0
and args.charuco_marker_length > 0
and args.charuco_marker_length < args.charuco_square_length
):
raise ValueError(
"Medidas ChArUco inválidas: use 0 < marker-length < square-length."
)
args = ap.parse_args()
profile_contract = load_module_profile_contract(
args.module_profile,
args.active_selector,
)
module_profile = profile_contract["profile"]
focus_contract = load_focus_prerequisite(profile_contract)
artifacts = module_profile["artifact_paths"]
args.candidate_root_overridden = args.candidate_root is not None
args.active_json_overridden = args.active_json is not None
expected_mx = str(module_profile.get("device_mx_id") or "")
if args.mx_id and str(args.mx_id) != expected_mx:
raise RuntimeError(
f"--mx-id={args.mx_id} diverge do module profile ativo ({expected_mx})."
)
args.mx_id = expected_mx
args.candidate_root = str(resolve_output_path(
args.candidate_root or artifacts["candidate_roots"]["intrinsics"]
))
args.active_json = str(resolve_output_path(
args.active_json or artifacts["intrinsics_json"]
))
sid = stamp()
candidate_dir = Path(args.candidate_root) / sid
@ -2349,13 +2632,26 @@ def main():
try:
rows, actual_mx, usb = discover(args.mx_id)
specs = validate_specs(rows)
preview_orientation = resolve_preview_orientation(specs, args)
operational_rois = {
role: parse_roi_arg(
getattr(args, f"{role}_operational_roi"), specs[role]
)
for role in ROLES
}
validate_profile_against_device(module_profile, specs, actual_mx)
preview_orientation = resolve_preview_orientation(module_profile, specs)
focus_rois = focus_contract["report"].get("calibration_roi_by_role", {})
operational_rois = {}
operational_roi_sources = {}
for role in ROLES:
explicit_roi = getattr(args, f"{role}_operational_roi")
if explicit_roi:
operational_rois[role] = parse_roi_arg(explicit_roi, specs[role])
operational_roi_sources[role] = "cli_override"
else:
operational_rois[role] = focus_roi_to_native(
focus_rois.get(role),
specs[role],
preview_orientation[role],
)
operational_roi_sources[role] = (
"focus_qc.calibration_roi_by_role_converted_to_sensor_native"
)
guide_targets = [] if args.no_guide else build_guide_targets()
pipeline, isp = build_pipeline(
@ -2370,6 +2666,9 @@ def main():
print("INTRINSICS CALIBRATION - PRODUCTION")
print(f"MX ID: {actual_mx}")
print(f"USB : {usb}")
print(f"Profile: {module_profile['profile_name']}")
print(f"Profile JSON: {profile_contract['profile_path']}")
print(f"Focus QC: {focus_contract['path']}")
for role in ROLES:
s = specs[role]
@ -2387,7 +2686,13 @@ def main():
f"rotate={o['rotate_deg']:3d} | "
f"flip_h={o['flip_horizontal']} | "
f"flip_v={o['flip_vertical']} | "
"runtime_effect=NONE"
"source=module_profile | runtime_effect=NONE"
)
roi = operational_rois[role]
print(
f"[ROI] {role.upper():3s} | "
f"{roi['x']},{roi['y']},{roi['width']},{roi['height']} | "
f"source={operational_roi_sources[role]}"
)
print("=" * 82)
@ -2403,8 +2708,57 @@ def main():
"device_mx_id": actual_mx,
"usb_speed": usb,
"hardware_signature": hardware_signature(specs),
"module_profile_contract": {
"schema": module_profile.get("schema"),
"profile_name": module_profile.get("profile_name"),
"profile_path": str(profile_contract["profile_path"]),
"profile_sha256": profile_contract["profile_sha256"],
"active_selector_path": (
str(profile_contract["selector_path"])
if profile_contract["selector_path"] is not None
else None
),
"device_match": True,
"artifact_routing": {
"candidate_root": str(args.candidate_root),
"candidate_root_source": (
"cli_override"
if args.candidate_root_overridden
else "module_profile.artifact_paths.candidate_roots.intrinsics"
),
"active_json": str(args.active_json),
"active_json_source": (
"cli_override"
if args.active_json_overridden
else "module_profile.artifact_paths.intrinsics_json"
),
},
},
"prerequisites": {
"focus_qc": {
"path": str(focus_contract["path"]),
"sha256": focus_contract["sha256"],
"schema": focus_contract["report"].get("schema"),
"status": focus_contract["report"].get("status"),
"promoted": focus_contract["report"].get("promoted"),
"normal_production_homologation": focus_contract[
"report"
].get("normal_production_homologation"),
"validated": True,
},
},
"rgb_decode_provenance": {
**module_profile["rgb_decode"],
"used_by_intrinsics_measurement": False,
"intrinsics_rgb_source": "ISP",
"notes": (
"Bayer/decode é identidade do profile, mas K/D da RGB são "
"medidos diretamente no stream ISP full-resolution."
),
},
"calibration_space": CALIBRATION_SPACE,
"operational_rois_native": operational_rois,
"operational_roi_sources": operational_roi_sources,
"guided_capture": {
"enabled": not args.no_guide,
"guide_role": args.guide_role,
@ -2662,7 +3016,9 @@ def main():
)
if chosen is not None:
operational_rois[selected_role] = chosen
operational_roi_sources[selected_role] = "interactive_manual"
report["operational_rois_native"] = operational_rois
report["operational_roi_sources"] = operational_roi_sources
save_json_atomic(report_path, report)
message = (
f"ROI {selected_role.upper()}="
@ -2682,7 +3038,9 @@ def main():
operational_rois[selected_role] = full_roi(
specs[selected_role]
)
operational_roi_sources[selected_role] = "interactive_full_frame"
report["operational_rois_native"] = operational_rois
report["operational_roi_sources"] = operational_roi_sources
save_json_atomic(report_path, report)
message = f"ROI {selected_role.upper()} = frame completo."
message_t = time.time()