1148 lines
45 KiB
Python
1148 lines
45 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
_capture_dataset.py
|
|
===================
|
|
|
|
Captura de dataset multiespectral em campo.
|
|
|
|
Contrato de configuração
|
|
------------------------
|
|
O hardware não é descrito por argumentos duplicados nem por config.json.
|
|
O script carrega o Module Profile ativo criado pelo Script 0 e, por ele,
|
|
resolve o module_params final correspondente ao módulo físico.
|
|
|
|
Obrigatórios para o operador:
|
|
--cana
|
|
--horario
|
|
|
|
Opcionais de seleção:
|
|
--module-profile usa diretamente um profile específico
|
|
--active-profile seletor produzido pelo Script 0
|
|
|
|
Aquisição científica
|
|
--------------------
|
|
O payload oficial é sempre RAW_BRUTO nativo e empacotado, uma câmera por
|
|
arquivo. Intrinsics, Flat-Field, orientação e Homography não são aplicados aos
|
|
bytes salvos. O module_params é carregado pelo OakFcc3Client para que startup,
|
|
exposição e diagnóstico sigam o mesmo contrato da aplicação de produção.
|
|
|
|
O PNG de anotação é sempre reconstruído pelo RawProcessorPreview a partir do
|
|
mesmo RAW10 da CAM_A: auto-level, gamma, demosaico EA, gray-world e contraste.
|
|
Depois do mini-ISP visual, o script aplica somente camera_orientation da role
|
|
RGB, vinda do module_params. Assim o RAW salvo continua nativo, enquanto o PNG
|
|
e a máscara desenhada sobre ele pertencem ao mesmo espaço canônico usado pelo
|
|
tensor final. A tecla M apenas alterna essa versão também na tela.
|
|
|
|
Isso permite reprocessar o dataset no futuro com calibrações novas. Não permite
|
|
recriar calibrações a partir de imagens agrícolas comuns: foco, intrínsecos,
|
|
flat-field, radiometria e homografia continuam exigindo seus alvos e protocolos.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
import json
|
|
import argparse
|
|
import hashlib
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
from core.oak_fcc3_client import OakFcc3Client as MultiSpectralClient
|
|
|
|
MODULE_PROFILE_SCHEMA = "multispec_module_profile_v1"
|
|
ACTIVE_SELECTOR_SCHEMA = "multispec_active_module_profile_v1"
|
|
MODULE_PARAMS_SCHEMA = "multispec_module_params_v3"
|
|
ROLES = ("rgb", "re", "nir")
|
|
BAYER_PATTERNS = ("RGGB", "BGGR", "GRBG", "GBRG")
|
|
|
|
ANNOTATION_PREVIEW_CONFIG = {
|
|
"schema": "rgb_raw_annotation_preview_v2",
|
|
"version": "raw_processor_preview_v2_plus_canonical_orientation_2026_09_10",
|
|
"purpose": "human_annotation_only",
|
|
"implementation": "core.raw_processor_preview.RawProcessorPreview",
|
|
"levels": {
|
|
"mode": "robust_percentile",
|
|
"black_percentile": 0.20,
|
|
"white_percentile": 99.80,
|
|
},
|
|
"gamma": 2.2,
|
|
"demosaic": {
|
|
"algorithm": "opencv_edge_aware",
|
|
"output_color_order": "BGR",
|
|
},
|
|
"white_balance": {
|
|
"method": "neutral_bright_with_gray_world_fallback",
|
|
"strength": 0.65,
|
|
"gain_range": [0.60, 1.70],
|
|
},
|
|
"clahe": {"clip_limit": 1.55, "tile_grid_size": 8},
|
|
"saturation_factor": 1.04,
|
|
"contrast": {"alpha": 1.04, "beta": 0.0},
|
|
"sharpen": {"sigma": 0.85, "amount": 0.55, "threshold": 2.0},
|
|
"output_format": "PNG_BGR_U8",
|
|
}
|
|
|
|
|
|
# =========================
|
|
# Helpers gerais
|
|
# =========================
|
|
|
|
def ts_name() -> str:
|
|
return datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
|
|
|
|
|
|
def sha256_file(path: str | Path) -> str:
|
|
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 sha256_array(arr: np.ndarray) -> str:
|
|
value = np.ascontiguousarray(arr)
|
|
return hashlib.sha256(value.view(np.uint8)).hexdigest()
|
|
|
|
|
|
def load_json(path: str | Path) -> dict:
|
|
p = Path(path)
|
|
if not p.is_file():
|
|
raise FileNotFoundError(f"JSON obrigatório não encontrado: {p}")
|
|
with p.open("r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError(f"JSON root deve ser objeto: {p}")
|
|
return data
|
|
|
|
|
|
def resolve_rgb_annotation_orientation(
|
|
module_params: dict,
|
|
expected_native_size: tuple[int, int],
|
|
) -> dict:
|
|
"""Resolve exatamente a mesma orientação RGB que o RawProcessorCore usa."""
|
|
root = module_params.get("camera_orientation")
|
|
if not isinstance(root, dict):
|
|
raise RuntimeError("module_params sem camera_orientation.")
|
|
|
|
by_role = root.get("by_role")
|
|
if not isinstance(by_role, dict) or not isinstance(by_role.get("rgb"), dict):
|
|
raise RuntimeError("camera_orientation.by_role.rgb ausente.")
|
|
|
|
role_cfg = by_role["rgb"]
|
|
try:
|
|
rotate_deg = int(role_cfg.get("rotate_deg", 0)) % 360
|
|
except Exception as exc:
|
|
raise RuntimeError("camera_orientation RGB rotate_deg inválido.") from exc
|
|
if rotate_deg not in (0, 90, 180, 270):
|
|
raise RuntimeError(f"camera_orientation RGB rotate_deg inválido: {rotate_deg}")
|
|
|
|
flip_h = role_cfg.get("flip_horizontal", False)
|
|
flip_v = role_cfg.get("flip_vertical", False)
|
|
if not isinstance(flip_h, bool) or not isinstance(flip_v, bool):
|
|
raise RuntimeError("camera_orientation RGB flips devem ser bool.")
|
|
|
|
enabled = root.get("enabled", False)
|
|
if not isinstance(enabled, bool):
|
|
raise RuntimeError("camera_orientation.enabled deve ser bool.")
|
|
|
|
transform_declared = rotate_deg != 0 or flip_h or flip_v
|
|
if transform_declared and not enabled:
|
|
raise RuntimeError(
|
|
"camera_orientation RGB declara transformação, mas enabled=false."
|
|
)
|
|
|
|
native_w, native_h = map(int, expected_native_size)
|
|
declared_native = role_cfg.get("native_size")
|
|
if declared_native is not None:
|
|
if list(map(int, declared_native)) != [native_w, native_h]:
|
|
raise RuntimeError(
|
|
"camera_orientation RGB native_size diverge do hardware: "
|
|
f"{declared_native} != {[native_w, native_h]}"
|
|
)
|
|
|
|
if enabled and rotate_deg in (90, 270):
|
|
oriented_size = [native_h, native_w]
|
|
else:
|
|
oriented_size = [native_w, native_h]
|
|
|
|
declared_oriented = role_cfg.get("oriented_size")
|
|
if declared_oriented is not None:
|
|
if list(map(int, declared_oriented)) != oriented_size:
|
|
raise RuntimeError(
|
|
"camera_orientation RGB oriented_size inconsistente: "
|
|
f"{declared_oriented} != {oriented_size}"
|
|
)
|
|
|
|
return {
|
|
"enabled": enabled,
|
|
"rotate_deg": rotate_deg if enabled else 0,
|
|
"flip_horizontal": flip_h if enabled else False,
|
|
"flip_vertical": flip_v if enabled else False,
|
|
"native_size": [native_w, native_h],
|
|
"oriented_size": oriented_size,
|
|
"input_space": str(root.get("input_space") or "native_stream_no_external_undistort"),
|
|
"output_space": str(root.get("output_space") or "canonical_oriented_stream_no_external_undistort"),
|
|
"apply_stage": str(root.get("apply_stage") or "after_native_flat_before_fusion"),
|
|
}
|
|
|
|
|
|
def apply_rgb_annotation_orientation(
|
|
image: np.ndarray,
|
|
orientation: dict,
|
|
) -> np.ndarray:
|
|
"""Aplica rotate e flips na mesma ordem usada pelo RawProcessorCore."""
|
|
out = np.asarray(image)
|
|
rotate_deg = int(orientation.get("rotate_deg", 0)) % 360
|
|
|
|
if rotate_deg == 90:
|
|
out = cv2.rotate(out, cv2.ROTATE_90_CLOCKWISE)
|
|
elif rotate_deg == 180:
|
|
out = cv2.rotate(out, cv2.ROTATE_180)
|
|
elif rotate_deg == 270:
|
|
out = cv2.rotate(out, cv2.ROTATE_90_COUNTERCLOCKWISE)
|
|
elif rotate_deg != 0:
|
|
raise RuntimeError(f"rotate_deg RGB inválido: {rotate_deg}")
|
|
|
|
if bool(orientation.get("flip_horizontal", False)):
|
|
out = cv2.flip(out, 1)
|
|
if bool(orientation.get("flip_vertical", False)):
|
|
out = cv2.flip(out, 0)
|
|
|
|
return np.ascontiguousarray(out)
|
|
|
|
|
|
def resolve_existing_path(
|
|
value: str | Path,
|
|
*,
|
|
anchor: Optional[Path] = None,
|
|
) -> Path:
|
|
raw = Path(value).expanduser()
|
|
candidates = [raw]
|
|
if not raw.is_absolute() and anchor is not None:
|
|
candidates.extend(parent / raw for parent in anchor.resolve().parents)
|
|
for candidate in candidates:
|
|
if candidate.is_file():
|
|
return candidate.resolve()
|
|
tried = ", ".join(str(x) for x in candidates)
|
|
raise FileNotFoundError(f"Arquivo de contrato não encontrado. Tentativas: {tried}")
|
|
|
|
|
|
def canonical_hardware_signature(value: dict) -> dict:
|
|
if not isinstance(value, dict):
|
|
raise RuntimeError("hardware_signature inválida no Module Profile.")
|
|
|
|
result = {}
|
|
for role in ROLES:
|
|
item = value.get(role)
|
|
if not isinstance(item, dict):
|
|
raise RuntimeError(f"hardware_signature sem role {role}.")
|
|
socket = item.get("socket", item.get("socket_name"))
|
|
sensor = item.get("sensor", item.get("sensor_name"))
|
|
size = item.get("size")
|
|
if size is None:
|
|
width = item.get("width", item.get("configured_width"))
|
|
height = item.get("height", item.get("configured_height"))
|
|
if width is not None and height is not None:
|
|
size = [width, height]
|
|
if not socket or not sensor or not isinstance(size, (list, tuple)) or len(size) != 2:
|
|
raise RuntimeError(f"hardware_signature/{role} incompleta: {item}")
|
|
result[role] = {
|
|
"socket": str(socket),
|
|
"sensor": str(sensor).upper(),
|
|
"size": [int(size[0]), int(size[1])],
|
|
}
|
|
return result
|
|
|
|
|
|
def load_module_profile_contract(
|
|
module_profile_arg: Optional[str],
|
|
active_profile_arg: str,
|
|
) -> dict:
|
|
selector = None
|
|
selector_path = None
|
|
|
|
if module_profile_arg:
|
|
profile_path = resolve_existing_path(module_profile_arg)
|
|
else:
|
|
selector_path = resolve_existing_path(active_profile_arg)
|
|
selector = load_json(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("Active Module Profile sem profile_path.")
|
|
profile_path = resolve_existing_path(declared, anchor=selector_path)
|
|
|
|
profile = load_json(profile_path)
|
|
if profile.get("schema") != MODULE_PROFILE_SCHEMA:
|
|
raise RuntimeError(
|
|
f"Schema do Module Profile inesperado: {profile.get('schema')!r}"
|
|
)
|
|
if str(profile.get("status") or "").lower() != "active":
|
|
raise RuntimeError(f"Module Profile não está ativo: {profile.get('status')!r}")
|
|
for key in ("profile_name", "device_mx_id", "artifact_paths", "rgb_decode"):
|
|
if not profile.get(key):
|
|
raise RuntimeError(f"Module Profile sem {key}.")
|
|
|
|
signature = canonical_hardware_signature(profile.get("hardware_signature"))
|
|
decode = profile["rgb_decode"]
|
|
bayer = str(decode.get("bayer_pattern") or "").upper()
|
|
if bayer not in BAYER_PATTERNS:
|
|
raise RuntimeError(f"rgb_decode.bayer_pattern inválido: {bayer!r}")
|
|
|
|
module_params_declared = profile["artifact_paths"].get("module_params_json")
|
|
if not module_params_declared:
|
|
raise RuntimeError("Module Profile sem artifact_paths.module_params_json.")
|
|
|
|
profile_sha = sha256_file(profile_path)
|
|
if selector is not None:
|
|
selector_sha = str(selector.get("profile_sha256") or "").lower()
|
|
if selector_sha and selector_sha != profile_sha.lower():
|
|
raise RuntimeError(
|
|
"Hash do Module Profile diverge do seletor ativo. Rode novamente o Script 0."
|
|
)
|
|
if str(selector.get("device_mx_id") or "") != str(profile["device_mx_id"]):
|
|
raise RuntimeError("MX ID do seletor diverge do Module Profile.")
|
|
|
|
module_params_path = resolve_existing_path(
|
|
module_params_declared,
|
|
anchor=profile_path,
|
|
)
|
|
module_params = load_json(module_params_path)
|
|
if module_params.get("schema") != MODULE_PARAMS_SCHEMA:
|
|
raise RuntimeError(
|
|
f"Schema do module_params inesperado: {module_params.get('schema')!r}"
|
|
)
|
|
|
|
provenance = module_params.get("module_profile_contract")
|
|
if not isinstance(provenance, dict):
|
|
raise RuntimeError(
|
|
"module_params sem module_profile_contract. Gere-o novamente com o assembler novo."
|
|
)
|
|
if str(provenance.get("profile_sha256") or "").lower() != profile_sha.lower():
|
|
raise RuntimeError(
|
|
"module_params foi montado a partir de outra versão do Module Profile."
|
|
)
|
|
if str(provenance.get("profile_name") or "") != str(profile["profile_name"]):
|
|
raise RuntimeError("profile_name do module_params diverge do Module Profile.")
|
|
if str(provenance.get("device_mx_id") or "") != str(profile["device_mx_id"]):
|
|
raise RuntimeError("MX ID do module_params diverge do Module Profile.")
|
|
|
|
mp_signature = canonical_hardware_signature(module_params.get("camera_hardware"))
|
|
if mp_signature != signature:
|
|
raise RuntimeError("camera_hardware do module_params diverge do Module Profile.")
|
|
if str(module_params.get("bayer_pattern") or "").upper() != bayer:
|
|
raise RuntimeError("Bayer do module_params diverge do Module Profile.")
|
|
mp_rgb = module_params.get("rgb_processing")
|
|
if not isinstance(mp_rgb, dict):
|
|
raise RuntimeError("module_params sem rgb_processing.")
|
|
for key in ("mode", "demosaic_algorithm"):
|
|
expected = str(decode.get(key) or "").lower()
|
|
actual = str(mp_rgb.get(key) or "").lower()
|
|
if actual != expected:
|
|
raise RuntimeError(
|
|
f"rgb_processing.{key}={actual!r} diverge do Module Profile ({expected!r})."
|
|
)
|
|
|
|
return {
|
|
"profile": profile,
|
|
"profile_path": profile_path,
|
|
"profile_sha256": profile_sha,
|
|
"selector_path": selector_path,
|
|
"hardware_signature": signature,
|
|
"bayer_pattern": bayer,
|
|
"module_params": module_params,
|
|
"module_params_path": module_params_path,
|
|
"module_params_sha256": sha256_file(module_params_path),
|
|
}
|
|
|
|
|
|
def validate_raw_triplet(
|
|
packed_by_camera: dict,
|
|
meta: dict,
|
|
hardware_signature: dict,
|
|
) -> dict:
|
|
"""Valida que o bundle contém exatamente uma fonte útil por role esperada."""
|
|
if not isinstance(packed_by_camera, dict):
|
|
raise RuntimeError("RAW_BRUTO multiespectral não veio como dict por câmera.")
|
|
|
|
camera_info = meta.get("camera_info", {}) or {}
|
|
if not isinstance(camera_info, dict):
|
|
raise RuntimeError("stream_meta.camera_info inválido.")
|
|
|
|
resolved = {}
|
|
for cam_id, arr in packed_by_camera.items():
|
|
info = camera_info.get(cam_id, {}) or {}
|
|
role = str(info.get("role") or "").lower()
|
|
if role not in ROLES:
|
|
for candidate, expected in hardware_signature.items():
|
|
if str(cam_id) == expected["socket"]:
|
|
role = candidate
|
|
break
|
|
if role not in ROLES:
|
|
continue
|
|
if role in resolved:
|
|
raise RuntimeError(f"Bundle RAW possui duas fontes para a role {role}.")
|
|
|
|
expected = hardware_signature[role]
|
|
expected_w, expected_h = expected["size"]
|
|
declared_w = info.get("width")
|
|
declared_h = info.get("height")
|
|
declared_sensor = info.get("sensor", info.get("sensor_name"))
|
|
if declared_w is not None and int(declared_w) != expected_w:
|
|
raise RuntimeError(f"{role}: width={declared_w}, esperado={expected_w}.")
|
|
if declared_h is not None and int(declared_h) != expected_h:
|
|
raise RuntimeError(f"{role}: height={declared_h}, esperado={expected_h}.")
|
|
if declared_sensor and str(declared_sensor).upper() != expected["sensor"]:
|
|
raise RuntimeError(
|
|
f"{role}: sensor={declared_sensor}, esperado={expected['sensor']}."
|
|
)
|
|
|
|
value = np.asarray(arr)
|
|
if value.dtype != np.uint8:
|
|
raise RuntimeError(
|
|
f"{role}: payload RAW empacotado deve ser uint8; recebido={value.dtype}."
|
|
)
|
|
minimum_bytes = expected_w * expected_h * 10 // 8
|
|
if value.nbytes < minimum_bytes:
|
|
raise RuntimeError(
|
|
f"{role}: RAW10 curto ({value.nbytes} bytes; mínimo={minimum_bytes})."
|
|
)
|
|
resolved[role] = {
|
|
"camera_id": str(cam_id),
|
|
"socket": expected["socket"],
|
|
"sensor": expected["sensor"],
|
|
"size": [expected_w, expected_h],
|
|
"bytes": int(value.nbytes),
|
|
}
|
|
|
|
missing = [role for role in ROLES if role not in resolved]
|
|
if missing:
|
|
raise RuntimeError(f"Bundle RAW incompleto; roles ausentes: {missing}")
|
|
return resolved
|
|
|
|
|
|
def overlay_hud(
|
|
img_bgr: np.ndarray,
|
|
lines: list[str],
|
|
base_h: int = 720,
|
|
base_font_scale: float = 0.75,
|
|
base_line_step: int = 28,
|
|
):
|
|
h, w = img_bgr.shape[:2]
|
|
|
|
scale = h / float(base_h)
|
|
scale = max(scale, 0.4)
|
|
|
|
font_scale = base_font_scale * scale
|
|
line_step = int(base_line_step * scale)
|
|
|
|
thick_outline = max(1, int(3 * scale))
|
|
thick_text = max(1, int(2 * scale))
|
|
|
|
y = int(24 * scale)
|
|
x = int(12 * scale)
|
|
|
|
for s in lines:
|
|
cv2.putText(img_bgr, s, (x, y), cv2.FONT_HERSHEY_SIMPLEX, font_scale, (0, 0, 0), thick_outline, cv2.LINE_AA)
|
|
cv2.putText(img_bgr, s, (x, y), cv2.FONT_HERSHEY_SIMPLEX, font_scale, (255, 255, 255), thick_text, cv2.LINE_AA)
|
|
y += line_step
|
|
|
|
|
|
def save_sample(
|
|
base_dir: str,
|
|
frame_type: str,
|
|
preview_bgr: np.ndarray,
|
|
meta: dict,
|
|
raw_payload: np.ndarray | None = None,
|
|
packed_raw: np.ndarray | None = None,
|
|
packed_raw_by_camera: dict | None = None,
|
|
):
|
|
os.makedirs(base_dir, exist_ok=True)
|
|
name = ts_name()
|
|
|
|
png_path = os.path.join(base_dir, f"{name}.png")
|
|
json_path = os.path.join(base_dir, f"{name}.json")
|
|
|
|
if frame_type in ("RGB", "MULTISPEC"):
|
|
if raw_payload is None:
|
|
raise ValueError(f"raw_payload não pode ser None quando frame_type='{frame_type}'")
|
|
|
|
payload_path = os.path.join(base_dir, f"{name}.raw")
|
|
payload_value = raw_payload.astype(np.float32)
|
|
payload_tmp = payload_path + ".tmp"
|
|
payload_value.tofile(payload_tmp)
|
|
os.replace(payload_tmp, payload_path)
|
|
|
|
meta["saved_payload_type"] = frame_type.lower()
|
|
meta["saved_payload_path"] = os.path.basename(payload_path)
|
|
meta["saved_payload_dtype"] = "float32"
|
|
meta["saved_payload_shape"] = list(raw_payload.shape)
|
|
meta["saved_payload_sha256"] = sha256_array(payload_value)
|
|
|
|
elif frame_type == "RAW_BRUTO":
|
|
if packed_raw_by_camera is not None:
|
|
payload_files = {}
|
|
payload_shapes = {}
|
|
payload_dtypes = {}
|
|
payload_hashes = {}
|
|
|
|
for cam_id, arr in packed_raw_by_camera.items():
|
|
path = os.path.join(base_dir, f"{name}_{cam_id}.bin")
|
|
value = np.ascontiguousarray(arr)
|
|
tmp = path + ".tmp"
|
|
value.tofile(tmp)
|
|
os.replace(tmp, path)
|
|
payload_files[cam_id] = os.path.basename(path)
|
|
payload_shapes[cam_id] = list(value.shape)
|
|
payload_dtypes[cam_id] = str(value.dtype)
|
|
payload_hashes[cam_id] = sha256_array(value)
|
|
|
|
meta["saved_payload_type"] = "raw_native_multi"
|
|
meta["saved_payload_paths"] = payload_files
|
|
meta["saved_payload_shapes"] = payload_shapes
|
|
meta["saved_payload_dtypes"] = payload_dtypes
|
|
meta["saved_payload_sha256"] = payload_hashes
|
|
|
|
else:
|
|
if packed_raw is None:
|
|
raise ValueError("packed_raw não pode ser None quando frame_type='RAW_BRUTO'")
|
|
|
|
payload_path = os.path.join(base_dir, f"{name}.bin")
|
|
packed_value = np.ascontiguousarray(packed_raw)
|
|
payload_tmp = payload_path + ".tmp"
|
|
packed_value.tofile(payload_tmp)
|
|
os.replace(payload_tmp, payload_path)
|
|
|
|
meta["saved_payload_type"] = "raw_native_single"
|
|
meta["saved_payload_path"] = os.path.basename(payload_path)
|
|
meta["saved_payload_dtype"] = str(packed_value.dtype)
|
|
meta["saved_payload_shape"] = list(packed_value.shape)
|
|
meta["saved_payload_sha256"] = sha256_array(packed_value)
|
|
|
|
else:
|
|
raise ValueError(f"frame_type não suportado para save: {frame_type}")
|
|
|
|
png_tmp = os.path.join(base_dir, f"{name}.tmp.png")
|
|
if not cv2.imwrite(png_tmp, preview_bgr):
|
|
raise RuntimeError(f"Falha ao salvar preview: {png_tmp}")
|
|
os.replace(png_tmp, png_path)
|
|
|
|
json_tmp = json_path + ".tmp"
|
|
with open(json_tmp, "w", encoding="utf-8") as f:
|
|
json.dump(meta, f, ensure_ascii=False, indent=2)
|
|
f.write("\n")
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(json_tmp, json_path)
|
|
|
|
return png_path, json_path
|
|
|
|
|
|
def get_camera_map_from_status(status: dict) -> dict:
|
|
result = {}
|
|
for cam in status.get("cameras", []):
|
|
result[cam.get("id")] = cam
|
|
return result
|
|
|
|
|
|
def build_preview_to_save(
|
|
cam,
|
|
packed_raw_by_camera: dict,
|
|
meta_stream: dict,
|
|
raw_integrity: dict,
|
|
expected_size: tuple[int, int],
|
|
bayer_pattern: str,
|
|
rgb_orientation: dict,
|
|
) -> tuple[np.ndarray, str, dict]:
|
|
"""
|
|
Gera o PNG beauty somente a partir do RAW RGB/CAM_A e o leva ao
|
|
espaço canônico de anotação.
|
|
|
|
O RawProcessorPreview pertence ao sensor RGB e, portanto, nunca deve
|
|
receber os RAWs mono RE/NIR (que possuem outro raster).
|
|
"""
|
|
rgb_info = raw_integrity.get("rgb") or {}
|
|
cam_id = rgb_info.get("camera_id")
|
|
if not cam_id or cam_id not in packed_raw_by_camera:
|
|
raise RuntimeError("Não foi possível localizar o RAW RGB para gerar o preview.")
|
|
|
|
preview_processor = getattr(cam, "preview", None)
|
|
if preview_processor is None:
|
|
raise RuntimeError("Cliente OAK não expõe o RawProcessorPreview RGB.")
|
|
|
|
build_rgb_preview = getattr(
|
|
preview_processor,
|
|
"packed_raw10_to_preview_bgr",
|
|
None,
|
|
)
|
|
if not callable(build_rgb_preview):
|
|
raise RuntimeError(
|
|
"RawProcessorPreview não expõe packed_raw10_to_preview_bgr()."
|
|
)
|
|
|
|
# Fundamental: processar apenas CAM_A. Não passe o dicionário completo,
|
|
# pois CAM_B/C são OV9282 1280x800 e este processor foi configurado para
|
|
# o raster Bayer do RGB (por exemplo, AR0234 1920x1200).
|
|
preview = build_rgb_preview(packed_raw_by_camera[cam_id])
|
|
|
|
preview = np.asarray(preview)
|
|
expected_w, expected_h = map(int, expected_size)
|
|
if preview.ndim != 3 or preview.shape[2] != 3 or preview.dtype != np.uint8:
|
|
raise RuntimeError(
|
|
f"Preview RGB deve ser HWC/BGR/uint8; shape={preview.shape}, dtype={preview.dtype}."
|
|
)
|
|
if preview.shape[:2] != (expected_h, expected_w):
|
|
raise RuntimeError(
|
|
"RawProcessorPreview alterou o raster RGB: "
|
|
f"{preview.shape[:2]} != {(expected_h, expected_w)}. "
|
|
"O mini-ISP visual deve preservar o raster nativo."
|
|
)
|
|
|
|
preview = apply_rgb_annotation_orientation(preview, rgb_orientation)
|
|
|
|
oriented_w, oriented_h = map(int, rgb_orientation["oriented_size"])
|
|
if preview.shape != (oriented_h, oriented_w, 3):
|
|
raise RuntimeError(
|
|
"Orientação RGB produziu raster inesperado: "
|
|
f"{preview.shape} != {(oriented_h, oriented_w, 3)}"
|
|
)
|
|
|
|
orientation_applied = {
|
|
"rotate_deg": int(rgb_orientation["rotate_deg"]),
|
|
"flip_horizontal": bool(rgb_orientation["flip_horizontal"]),
|
|
"flip_vertical": bool(rgb_orientation["flip_vertical"]),
|
|
}
|
|
transformed = any((
|
|
orientation_applied["rotate_deg"] != 0,
|
|
orientation_applied["flip_horizontal"],
|
|
orientation_applied["flip_vertical"],
|
|
))
|
|
|
|
report = {
|
|
**ANNOTATION_PREVIEW_CONFIG,
|
|
"camera_id": str(cam_id),
|
|
"bayer_pattern": str(bayer_pattern).upper(),
|
|
"geometry": "rgb_canonical_oriented_no_external_undistort",
|
|
"source_space": rgb_orientation["input_space"],
|
|
"coordinate_space": rgb_orientation["output_space"],
|
|
"source_image_size": [expected_w, expected_h],
|
|
"image_size": [oriented_w, oriented_h],
|
|
"orientation_applied": orientation_applied,
|
|
"orientation_apply_stage": rgb_orientation["apply_stage"],
|
|
"pixel_correspondence": "deterministic_orientation_of_rgb_sensor_raster",
|
|
"calibrations_applied": ["camera_orientation"] if transformed else [],
|
|
"mask_contract": {
|
|
"coordinate_space": rgb_orientation["output_space"],
|
|
"already_oriented": True,
|
|
"normalize_must_not_reapply_camera_orientation": True,
|
|
"downstream_geometry": ["common_crop", "final_resize_nearest"],
|
|
},
|
|
}
|
|
return (
|
|
np.ascontiguousarray(preview),
|
|
"raw_processor_preview_rgb_beauty_canonical_v2",
|
|
report,
|
|
)
|
|
|
|
|
|
def build_capture_metadata(
|
|
*,
|
|
args,
|
|
contract: dict,
|
|
stream_meta: dict,
|
|
cam,
|
|
effective_capture_mode: str,
|
|
raw_policy: str,
|
|
raw_integrity: dict,
|
|
preview_source_id: str,
|
|
annotation_preview: dict,
|
|
note: str,
|
|
) -> dict:
|
|
profile = contract["profile"]
|
|
signature = contract["hardware_signature"]
|
|
rgb_w, rgb_h = signature["rgb"]["size"]
|
|
return {
|
|
"schema": "multispec_raw_dataset_sample_v2",
|
|
"ts": datetime.now().isoformat(timespec="milliseconds"),
|
|
"cana": args.cana,
|
|
"horario": args.horario,
|
|
"sensor_width": rgb_w,
|
|
"sensor_height": rgb_h,
|
|
"sensor_size_by_role": {
|
|
role: list(signature[role]["size"])
|
|
for role in ROLES
|
|
},
|
|
"camera_hardware": signature,
|
|
"bayer_pattern": contract["bayer_pattern"],
|
|
"fps_target": args.fps,
|
|
"frame_type": "RAW_BRUTO",
|
|
"capture_mode_requested": effective_capture_mode,
|
|
"capture_mode_effective": effective_capture_mode,
|
|
"raw_policy": raw_policy,
|
|
"raw_integrity": raw_integrity,
|
|
"stream_meta": stream_meta,
|
|
"startup_camera_controls": cam.applied_camera_controls,
|
|
"actual_camera_controls": cam.get_current_camera_controls(),
|
|
"radiometric_last_result": cam.get_radiometric_last_result(),
|
|
"module_profile_contract": {
|
|
"profile_name": profile["profile_name"],
|
|
"profile_path": contract["profile_path"].as_posix(),
|
|
"profile_sha256": contract["profile_sha256"],
|
|
"device_mx_id": profile["device_mx_id"],
|
|
},
|
|
"module_params_contract": {
|
|
"path": contract["module_params_path"].as_posix(),
|
|
"sha256": contract["module_params_sha256"],
|
|
"schema": contract["module_params"].get("schema"),
|
|
},
|
|
"rgb_decode_contract": dict(profile["rgb_decode"]),
|
|
"calibration_application": {
|
|
"payload": "none_raw_native_preserved",
|
|
"preview_only": True,
|
|
"note": (
|
|
"Module params governa controles de aquisição. O PNG de anotação "
|
|
"usa RawProcessorPreview no RAW RGB e depois aplica somente a "
|
|
"camera_orientation canônica. Calibrações radiométricas, Flat-Field, "
|
|
"Intrinsics e Homography não alteram o PNG nem os bytes RAW_BRUTO."
|
|
),
|
|
},
|
|
"annotation_preview": annotation_preview,
|
|
"note": note,
|
|
"raw_preview_reference_camera": preview_source_id,
|
|
}
|
|
|
|
|
|
# =========================
|
|
# MAIN
|
|
# =========================
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Captura RAW_BRUTO de dataset usando o Module Profile ativo e "
|
|
"o module_params final correspondente."
|
|
),
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
|
|
parser.add_argument("--cana", required=True, choices=["baixa", "media", "alta"], help="Estado da cana no momento da coleta.")
|
|
parser.add_argument("--horario", required=True, choices=["cedo", "meio_dia", "entardecer", "nublado"], help="Janela de iluminação / horário da coleta.")
|
|
parser.add_argument(
|
|
"--module-profile",
|
|
default=None,
|
|
help="Override opcional do Module Profile. Sem ele, usa o seletor ativo.",
|
|
)
|
|
parser.add_argument(
|
|
"--active-profile",
|
|
default="calibration/active_module_profile.json",
|
|
help="Seletor ativo criado pelo Script 0.",
|
|
)
|
|
parser.add_argument("--out-root", default="dataset", help="Pasta raiz do dataset.")
|
|
parser.add_argument("--fps", type=float, default=20.0, help="FPS de aquisição; política operacional, não identidade do sensor.")
|
|
parser.add_argument("--interval", type=float, default=1.0, help="Intervalo em segundos para auto-save quando ligado.")
|
|
parser.add_argument("--preview-upscale", type=int, default=2, help="Fator de upscale exclusivamente visual.")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.fps <= 0:
|
|
raise ValueError("--fps deve ser > 0.")
|
|
if args.interval <= 0:
|
|
raise ValueError("--interval deve ser > 0.")
|
|
if args.preview_upscale < 1:
|
|
raise ValueError("--preview-upscale deve ser >= 1.")
|
|
|
|
contract = load_module_profile_contract(
|
|
args.module_profile,
|
|
args.active_profile,
|
|
)
|
|
profile = contract["profile"]
|
|
module_params = contract["module_params"]
|
|
hardware_signature = contract["hardware_signature"]
|
|
|
|
# Dataset científico: não salvamos frames processados nem bundles parciais.
|
|
frame_type_requested = "RAW_BRUTO"
|
|
output_dtype = "float32" # Não altera o payload quando frame_type=RAW_BRUTO.
|
|
effective_capture_mode = str(
|
|
module_params.get("capture_mode_effective")
|
|
or module_params.get("capture_mode_requested")
|
|
or "AUTO"
|
|
).upper()
|
|
if effective_capture_mode not in {"AUTO", "SINGLE", "DOUBLE", "TRIPLE"}:
|
|
raise RuntimeError(
|
|
f"capture_mode_effective inválido no module_params: {effective_capture_mode!r}"
|
|
)
|
|
raw_policy = "require_triple"
|
|
|
|
raw_w, raw_h = hardware_signature["rgb"]["size"]
|
|
bayer_pattern = contract["bayer_pattern"]
|
|
module_params_path = contract["module_params_path"]
|
|
rgb_orientation = resolve_rgb_annotation_orientation(
|
|
module_params,
|
|
expected_native_size=(raw_w, raw_h),
|
|
)
|
|
|
|
session_dir = os.path.join(
|
|
args.out_root,
|
|
"brutas",
|
|
f"cana_{args.cana}",
|
|
args.horario,
|
|
datetime.now().strftime("%Y%m%d"),
|
|
)
|
|
os.makedirs(session_dir, exist_ok=True)
|
|
|
|
print("============================================")
|
|
print("Coleta de dataset - Módulo Multiespectral")
|
|
print(f"Cana : {args.cana}")
|
|
print(f"Horário : {args.horario}")
|
|
print(f"Saída : {session_dir}")
|
|
print(f"Profile : {profile['profile_name']}")
|
|
print(f"MX ID : {profile['device_mx_id']}")
|
|
print(f"Module Params: {module_params_path}")
|
|
for role in ROLES:
|
|
hw = hardware_signature[role]
|
|
print(
|
|
f"{role.upper():3s} : {hw['socket']} | {hw['sensor']} | "
|
|
f"{hw['size'][0]}x{hw['size'][1]}"
|
|
)
|
|
print(f"RGB Bayer : {bayer_pattern}")
|
|
print(
|
|
"RGB Preview : CANONICAL | "
|
|
f"ROT={rgb_orientation['rotate_deg']} "
|
|
f"FH={int(rgb_orientation['flip_horizontal'])} "
|
|
f"FV={int(rgb_orientation['flip_vertical'])} | "
|
|
f"{rgb_orientation['native_size']} -> "
|
|
f"{rgb_orientation['oriented_size']}"
|
|
)
|
|
print(f"FrameType : {frame_type_requested} (fixo científico)")
|
|
print(f"CaptureMode : {effective_capture_mode} (module_params)")
|
|
print(f"RAW policy : {raw_policy} (fixo de dataset)")
|
|
print("============================================")
|
|
|
|
beauty_preview = False
|
|
radiometric_ae = True
|
|
auto_save = False
|
|
last_auto_t = 0.0
|
|
preview_upscale = args.preview_upscale
|
|
|
|
t_view_fps = time.time()
|
|
view_frames = 0
|
|
fps_view = 0.0
|
|
|
|
t_stream_fps = time.time()
|
|
last_stream_frame_id = None
|
|
stream_frames_accum = 0
|
|
fps_stream = 0.0
|
|
|
|
last_msg = ""
|
|
last_msg_t = 0.0
|
|
|
|
window_name = "Dataset Capture (C/SPACE=save | A=auto-save | M=preview | R=rad | Q=quit)"
|
|
cv2.namedWindow(window_name, cv2.WINDOW_NORMAL)
|
|
|
|
last_frame_id = -1
|
|
last_packed_raw_by_camera = None
|
|
last_preview_bgr = None
|
|
last_meta_stream = None
|
|
last_raw_integrity = None
|
|
last_saved_frame_id = None
|
|
last_valid_frame_t = 0.0
|
|
|
|
try:
|
|
with MultiSpectralClient(
|
|
mx_id=str(profile["device_mx_id"]),
|
|
width=raw_w,
|
|
height=raw_h,
|
|
bayer=bayer_pattern,
|
|
fps=args.fps,
|
|
frame_type=frame_type_requested,
|
|
output_dtype=output_dtype,
|
|
capture_mode=effective_capture_mode,
|
|
raw_policy=raw_policy,
|
|
module_calibration_json=str(module_params_path),
|
|
) as cam:
|
|
rad = getattr(cam, "radiometric_controller", None)
|
|
radiometric_ae = bool(rad is not None and rad.enabled)
|
|
while True:
|
|
t0 = time.time()
|
|
frame, meta, _decoded = cam.get_next_decoded(timeout=1.0)
|
|
if meta is not None and frame is not None and meta.get("frame_id") != last_frame_id:
|
|
last_frame_id = meta["frame_id"]
|
|
|
|
try:
|
|
frame_type = str(meta.get("frame_type") or "").upper()
|
|
if frame_type != "RAW_BRUTO":
|
|
raise RuntimeError(
|
|
f"Servidor retornou {frame_type!r}; este capturador aceita somente RAW_BRUTO."
|
|
)
|
|
raw_integrity = validate_raw_triplet(
|
|
frame,
|
|
meta,
|
|
hardware_signature,
|
|
)
|
|
packed_by_camera = frame
|
|
preview_bgr, _, _ = (
|
|
cam.build_preview_from_raw_payload(frame=frame, meta=meta)
|
|
)
|
|
|
|
if preview_bgr is None:
|
|
raise RuntimeError("OakFcc3Client não conseguiu montar o preview RAW.")
|
|
|
|
if beauty_preview:
|
|
preview_bgr, _, _ = build_preview_to_save(
|
|
cam=cam,
|
|
packed_raw_by_camera=packed_by_camera,
|
|
meta_stream=meta,
|
|
raw_integrity=raw_integrity,
|
|
expected_size=(raw_w, raw_h),
|
|
bayer_pattern=bayer_pattern,
|
|
rgb_orientation=rgb_orientation,
|
|
)
|
|
|
|
last_packed_raw_by_camera = {
|
|
cam_id: np.ascontiguousarray(arr).copy()
|
|
for cam_id, arr in packed_by_camera.items()
|
|
}
|
|
last_raw_integrity = raw_integrity
|
|
|
|
if preview_upscale and preview_upscale > 1:
|
|
preview_show = cv2.resize(
|
|
preview_bgr,
|
|
(preview_bgr.shape[1] * preview_upscale, preview_bgr.shape[0] * preview_upscale),
|
|
interpolation=cv2.INTER_NEAREST,
|
|
)
|
|
else:
|
|
preview_show = preview_bgr.copy()
|
|
|
|
curr_frame_id = meta.get("frame_id")
|
|
|
|
if curr_frame_id is not None:
|
|
if last_stream_frame_id != curr_frame_id:
|
|
stream_frames_accum += 1
|
|
|
|
last_stream_frame_id = curr_frame_id
|
|
|
|
dt_stream = time.time() - t_stream_fps
|
|
if dt_stream >= 1.0:
|
|
fps_stream = stream_frames_accum / dt_stream
|
|
stream_frames_accum = 0
|
|
t_stream_fps = time.time()
|
|
|
|
view_frames += 1
|
|
dt_view = time.time() - t_view_fps
|
|
if dt_view >= 1.0:
|
|
fps_view = view_frames / dt_view
|
|
view_frames = 0
|
|
t_view_fps = time.time()
|
|
|
|
active_sources = meta.get("payload_sources")
|
|
|
|
rad = getattr(cam, "radiometric_controller", None)
|
|
if rad and rad.enabled:
|
|
st = rad.state
|
|
line_rad = (
|
|
f"RAD | "
|
|
f"RGB(exp={st['rgb']['exp']}, g={st['rgb']['gain']:.2f}) | "
|
|
f"RE(exp={st['re']['exp']}, g={st['re']['gain']:.2f}) | "
|
|
f"NIR(exp={st['nir']['exp']}, g={st['nir']['gain']:.2f})"
|
|
)
|
|
else:
|
|
line_rad = "RAD | OFF"
|
|
|
|
camera_info = meta.get("camera_info", {}) or {}
|
|
frame_controls = meta.get("frame_controls", {}) or {}
|
|
|
|
role_controls = {}
|
|
for cam_id, info in camera_info.items():
|
|
role = info.get("role", cam_id)
|
|
role_controls[role] = frame_controls.get(cam_id, {})
|
|
|
|
line_ae = (
|
|
f"AE_REAL | "
|
|
f"RGB exp={role_controls.get('rgb', {}).get('exposure_time_us')} "
|
|
f"iso={role_controls.get('rgb', {}).get('sensitivity_iso')} | "
|
|
f"RE exp={role_controls.get('re', {}).get('exposure_time_us')} "
|
|
f"iso={role_controls.get('re', {}).get('sensitivity_iso')} | "
|
|
f"NIR exp={role_controls.get('nir', {}).get('exposure_time_us')} "
|
|
f"iso={role_controls.get('nir', {}).get('sensitivity_iso')}"
|
|
)
|
|
lines = [
|
|
f"CANA: {args.cana} | HORA: {args.horario} | Pasta: {os.path.basename(session_dir)}",
|
|
f"Type={meta.get('frame_type')} | CaptureMode={effective_capture_mode} | RAW policy={raw_policy}",
|
|
f"Sources={active_sources} | FPS_STREAM={fps_stream:.1f} | FPS_VIEW={fps_view:.1f}",
|
|
f"frame_id={meta.get('frame_id')} | layout={meta.get('output_layout')} | dtype={meta.get('dtype') or meta.get('output_dtype')}",
|
|
f"codec={meta.get('codec_name', meta.get('codec_family', '-'))} | comp={meta.get('dt_comp', 0):.4f}s | send={meta.get('dt_send_payload_prev', 0):.4f}s",
|
|
f"PROFILE={profile['profile_name']} | MP={module_params_path.name}",
|
|
line_ae,
|
|
line_rad,
|
|
"Keys: C/SPACE=save | A=auto-save | M=preview | R=rad | Q/Esc=quit"
|
|
]
|
|
overlay_hud(preview_show, lines, base_h=raw_h)
|
|
|
|
if last_msg and (time.time() - last_msg_t) < 2.0:
|
|
cv2.putText(preview_show, last_msg, (12, preview_show.shape[0] - 18),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2, cv2.LINE_AA)
|
|
|
|
cv2.imshow(window_name, preview_show)
|
|
|
|
last_preview_bgr = preview_bgr.copy()
|
|
last_meta_stream = dict(meta)
|
|
last_meta_stream["frame_type"] = frame_type
|
|
last_valid_frame_t = time.time()
|
|
|
|
except Exception as e:
|
|
last_packed_raw_by_camera = None
|
|
last_raw_integrity = None
|
|
err = np.zeros((500, 1200, 3), dtype=np.uint8)
|
|
cv2.putText(err, f"Erro ao processar frame: {e}", (20, 60),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2, cv2.LINE_AA)
|
|
cv2.imshow(window_name, err)
|
|
print(f"[ERRO FRAME] {e}")
|
|
|
|
now = time.time()
|
|
can_save = (
|
|
last_meta_stream is not None and
|
|
last_preview_bgr is not None and
|
|
last_meta_stream.get("frame_type") == "RAW_BRUTO" and
|
|
last_packed_raw_by_camera is not None and
|
|
last_raw_integrity is not None and
|
|
(now - last_valid_frame_t) <= 2.0
|
|
)
|
|
|
|
if (
|
|
auto_save
|
|
and can_save
|
|
and (now - last_auto_t) >= args.interval
|
|
and last_meta_stream.get("frame_id") != last_saved_frame_id
|
|
):
|
|
frame_type_save = last_meta_stream.get("frame_type")
|
|
frame_id_save = last_meta_stream.get("frame_id")
|
|
|
|
preview_to_save, preview_method, preview_report = (
|
|
build_preview_to_save(
|
|
cam=cam,
|
|
packed_raw_by_camera=last_packed_raw_by_camera,
|
|
meta_stream=last_meta_stream,
|
|
raw_integrity=last_raw_integrity,
|
|
expected_size=(raw_w, raw_h),
|
|
bayer_pattern=bayer_pattern,
|
|
rgb_orientation=rgb_orientation,
|
|
)
|
|
)
|
|
|
|
meta_save = build_capture_metadata(
|
|
args=args,
|
|
contract=contract,
|
|
stream_meta=last_meta_stream,
|
|
cam=cam,
|
|
effective_capture_mode=effective_capture_mode,
|
|
raw_policy=raw_policy,
|
|
raw_integrity=last_raw_integrity,
|
|
preview_source_id=last_raw_integrity["rgb"]["camera_id"],
|
|
annotation_preview=preview_report,
|
|
note="autosave",
|
|
)
|
|
|
|
meta_save["saved_preview_method"] = preview_method
|
|
|
|
save_sample(
|
|
session_dir,
|
|
frame_type=frame_type_save,
|
|
preview_bgr=preview_to_save,
|
|
meta=meta_save,
|
|
packed_raw_by_camera=last_packed_raw_by_camera,
|
|
)
|
|
|
|
last_msg = "SALVO (auto)"
|
|
last_msg_t = now
|
|
last_auto_t = now
|
|
last_saved_frame_id = frame_id_save
|
|
|
|
k = cv2.waitKey(1) & 0xFF
|
|
if k in (ord("q"), ord("Q"), 27):
|
|
break
|
|
|
|
elif k in (ord("a"), ord("A")):
|
|
auto_save = not auto_save
|
|
last_msg = f"AutoSave -> {'ON' if auto_save else 'OFF'}"
|
|
last_msg_t = time.time()
|
|
|
|
elif k in (ord("m"), ord("M")):
|
|
beauty_preview = False if beauty_preview else True
|
|
last_msg = f"Preview Beauty -> {beauty_preview}"
|
|
last_msg_t = time.time()
|
|
|
|
elif k in (ord("r"), ord("R")):
|
|
rad = getattr(cam, "radiometric_controller", None)
|
|
if rad is not None:
|
|
radiometric_ae = False if radiometric_ae else True
|
|
rad.enabled = radiometric_ae
|
|
last_msg = f"RAD -> {radiometric_ae}"
|
|
last_msg_t = time.time()
|
|
|
|
|
|
elif k in (ord("c"), ord("C"), 32):
|
|
if can_save:
|
|
frame_type_save = last_meta_stream.get("frame_type")
|
|
preview_to_save, preview_method, preview_report = (
|
|
build_preview_to_save(
|
|
cam=cam,
|
|
packed_raw_by_camera=last_packed_raw_by_camera,
|
|
meta_stream=last_meta_stream,
|
|
raw_integrity=last_raw_integrity,
|
|
expected_size=(raw_w, raw_h),
|
|
bayer_pattern=bayer_pattern,
|
|
rgb_orientation=rgb_orientation,
|
|
)
|
|
)
|
|
meta_save = build_capture_metadata(
|
|
args=args,
|
|
contract=contract,
|
|
stream_meta=last_meta_stream,
|
|
cam=cam,
|
|
effective_capture_mode=effective_capture_mode,
|
|
raw_policy=raw_policy,
|
|
raw_integrity=last_raw_integrity,
|
|
preview_source_id=last_raw_integrity["rgb"]["camera_id"],
|
|
annotation_preview=preview_report,
|
|
note="manual",
|
|
)
|
|
|
|
meta_save["saved_preview_method"] = preview_method
|
|
|
|
save_sample(
|
|
session_dir,
|
|
frame_type=frame_type_save,
|
|
preview_bgr=preview_to_save,
|
|
meta=meta_save,
|
|
packed_raw_by_camera=last_packed_raw_by_camera,
|
|
)
|
|
|
|
last_msg = "SALVO (manual)"
|
|
last_msg_t = time.time()
|
|
last_saved_frame_id = last_meta_stream.get("frame_id")
|
|
|
|
dt_loop = time.time() - t0
|
|
if dt_loop < 0.001:
|
|
time.sleep(0.001)
|
|
|
|
finally:
|
|
cv2.destroyAllWindows()
|
|
print("Fim da captura.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|