Zendion/agrobot_base
Zendion
/
agrobot_base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ajustes no bayer padrao e extract bayer channels

This commit is contained in:
Diego Freitas 2026-05-12 10:23:09 -03:00
parent 4a04ab6b18
commit d886e282f8
11 changed files with 310 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Python/OAK/datasets/oak-fcc-3/_0_capture.py
View File

@ -144,7 +144,7 @@ def main():
parser.add_argument("--height", type=int, default=RAW_SIZE[1], help="Altura óptica da câmera.")
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 visual do preview.")
parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"], help="Padrão Bayer das câmeras.")
parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"], help="Padrão Bayer das câmeras.")
parser.add_argument("--output_dtype", default="float32", choices=["uint8", "uint16", "float32"], help="Dtype do payload processado no Pi.")
parser.add_argument("--frame_type", default="RAW_BRUTO", choices=["RAW_BRUTO", "RGB", "MULTISPEC"], help="Tipo de payload pedido ao Pi.")
parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"], help="Modo de captura desejado no módulo.")

2
Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_client.py
View File

@ -14,7 +14,7 @@ class OakFcc3Client:
self,
width=640,
height=400,
bayer="RGGB",
bayer="BGGR",
fps=30,
frame_type="RAW_BRUTO",
output_dtype="uint8",

4
Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_manager.py
View File

@ -33,8 +33,8 @@ class OakFcc3Manager:
self.roles = roles or {
"CAM_A": "rgb",
"CAM_B": "nir",
"CAM_C": "re",
"CAM_B": "re",
"CAM_C": "nir",
}
self.sync_mode = sync_mode

319
Python/OAK/datasets/oak-fcc-3/core/raw_processor_core.py
View File

@ -7,10 +7,13 @@ from typing import Optional
class RawProcessorCore:
def __init__(self, sensor_width: int, sensor_height: int, bayer_pattern: str = "GBRG", calibration_json_path=None):
def __init__(self, sensor_width: int, sensor_height: int, bayer_pattern: str = "BGGR", calibration_json_path=None):
self.sensor_width = sensor_width
self.sensor_height = sensor_height
self.bayer_pattern = bayer_pattern.upper()
self.rgb_processing_config = {
"mode": "linear_demosaic", # "linear_demosaic" ou "bayer_planes"
}
self.fusion_config = {
"alignment_mode": "manual_affine",
@ -174,39 +177,42 @@ class RawProcessorCore:
return raw16
def extract_bayer_channels(self, raw16: np.ndarray) -> dict:
p = self.bayer_pattern
p = self.bayer_pattern.upper()
if p == "GBRG":
g1 = raw16[0::2, 0::2]
b = raw16[0::2, 1::2]
r = raw16[1::2, 0::2]
g2 = raw16[1::2, 1::2]
elif p == "GRBG":
g1 = raw16[0::2, 0::2]
r = raw16[0::2, 1::2]
b = raw16[1::2, 0::2]
g2 = raw16[1::2, 1::2]
elif p == "RGGB":
b = raw16[0::2, 0::2]
if p == "RGGB":
r = raw16[0::2, 0::2]
g1 = raw16[0::2, 1::2]
g2 = raw16[1::2, 0::2]
r = raw16[1::2, 1::2]
b = raw16[1::2, 1::2]
elif p == "BGGR":
b = raw16[0::2, 0::2]
g1 = raw16[0::2, 1::2]
g2 = raw16[1::2, 0::2]
r = raw16[1::2, 1::2]
elif p == "GRBG":
g1 = raw16[0::2, 0::2]
r = raw16[0::2, 1::2]
b = raw16[1::2, 0::2]
g2 = raw16[1::2, 1::2]
elif p == "GBRG":
g1 = raw16[0::2, 0::2]
b = raw16[0::2, 1::2]
r = raw16[1::2, 0::2]
g2 = raw16[1::2, 1::2]
else:
raise ValueError(f"Padrão Bayer não suportado: {p}")
return {"R": r, "G1": g1, "G2": g2, "B": b}
def build_training_rgb(
def bayer_planes_to_rgb_linear(
self,
raw16: np.ndarray,
output_dtype: str = "float32",
bit_depth: int = 10,
) -> np.ndarray:
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
ch = self.extract_bayer_channels(raw16)
max_val = float((1 << bit_depth) - 1)
@ -215,12 +221,75 @@ class RawProcessorCore:
g = ((ch["G1"].astype(np.float32) + ch["G2"].astype(np.float32)) * 0.5) / max_val
b = ch["B"].astype(np.float32) / max_val
return (
np.clip(r, 0.0, 1.0),
np.clip(g, 0.0, 1.0),
np.clip(b, 0.0, 1.0),
)
def demosaic_raw16_to_rgb_linear(
self,
raw16: np.ndarray,
bit_depth: int = 10,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
code_map = {
"RGGB": cv2.COLOR_BayerRG2BGR_EA,
"BGGR": cv2.COLOR_BayerBG2BGR_EA,
"GRBG": cv2.COLOR_BayerGR2BGR_EA,
"GBRG": cv2.COLOR_BayerGB2BGR_EA,
}
p = self.bayer_pattern.upper()
if p not in code_map:
raise ValueError(f"Padrão Bayer não suportado para demosaic: {p}")
raw16 = np.asarray(raw16)
if raw16.dtype != np.uint16:
raw16 = raw16.astype(np.uint16)
out = cv2.cvtColor(raw16, code_map[p])
max_val = float((1 << bit_depth) - 1)
out = np.clip(out.astype(np.float32) / max_val, 0.0, 1.0)
r = out[:, :, 0]
g = out[:, :, 1]
b = out[:, :, 2]
return r, g, b
def build_training_rgb(
self,
raw16: np.ndarray,
output_dtype: str = "float32",
bit_depth: int = 10,
) -> np.ndarray:
rgb_mode = str(
(getattr(self, "rgb_processing_config", {}) or {}).get("mode", "linear_demosaic")
).lower()
if rgb_mode in ("linear_demosaic", "demosaic", "full_res"):
r, g, b = self.demosaic_raw16_to_rgb_linear(
raw16,
bit_depth=bit_depth,
)
elif rgb_mode in ("bayer_planes", "bayer", "half_res"):
r, g, b = self.bayer_planes_to_rgb_linear(
raw16,
bit_depth=bit_depth,
)
else:
raise ValueError(f"rgb_processing.mode inválido: {rgb_mode}")
rgb_cal = getattr(self, "rgb_calibration", {}) or {}
if rgb_cal.get("enabled", False):
gains = rgb_cal.get("gains", {}) or {}
r *= float(gains.get("R", 1.0))
g *= float(gains.get("G", 1.0))
b *= float(gains.get("B", 1.0))
r = r * float(gains.get("R", 1.0))
g = g * float(gains.get("G", 1.0))
b = b * float(gains.get("B", 1.0))
chw = np.stack([r, g, b], axis=0).astype(np.float32)
chw = np.clip(chw, 0.0, 1.0)
@ -767,6 +836,47 @@ class RawProcessorCore:
h, w = out.shape[1], out.shape[2]
result["tensor_stats_before"] = self._tensor_channel_stats(out, channel_names)
# Guarda anti-roxo para normalização por patches.
# Usa uma máscara comum RGB calculada antes de qualquer escala.
patch_sat_guard_enabled = bool(cfg.get("rgb_saturation_guard_enabled", True))
patch_sat_guard_mode = str(cfg.get("rgb_saturation_guard_mode", "fade_strength")).lower()
patch_sat_soft_start = float(cfg.get("rgb_saturation_soft_start", 0.88))
patch_sat_hard = float(cfg.get("rgb_saturation_hard", 0.97))
patch_sat_threshold = float(cfg.get("rgb_saturation_threshold", 0.97))
rgb_original = out[:3].copy() if out.shape[0] >= 3 else None
rgb_sat_mask = None
rgb_strength_mask = None
if patch_sat_guard_enabled and rgb_original is not None:
rgb_max = np.max(rgb_original, axis=0)
rgb_sat_mask = rgb_max >= patch_sat_threshold
denom = max(patch_sat_hard - patch_sat_soft_start, 1e-6)
t = (rgb_max - patch_sat_soft_start) / denom
t = np.clip(t, 0.0, 1.0)
# 1.0 = aplica normalização normal
# 0.0 = preserva original
rgb_strength_mask = 1.0 - t
result["rgb_saturation_guard"] = {
"enabled": True,
"mode": patch_sat_guard_mode,
"soft_start": patch_sat_soft_start,
"hard": patch_sat_hard,
"threshold": patch_sat_threshold,
"sat_pct": float(np.mean(rgb_sat_mask) * 100.0),
"mean_strength": float(np.mean(rgb_strength_mask)),
}
else:
result["rgb_saturation_guard"] = {
"enabled": False
}
# Para o tensor final fusionado, a geometria de referência é o espaço do RGB.
# Como RE/NIR são alinhados por homografia para casar no RGB, as ROIs usadas
# na normalização final devem ser as ROIs da role rgb, com suporte a múltiplas
@ -876,6 +986,26 @@ class RawProcessorCore:
output_after_clip_stats = self._array01_stats(out_ch)
out[ci] = out_ch
# Anti-roxo: não deixa patch_normalization recolorir pixels RGB saturados.
if (
patch_sat_guard_enabled
and ci < 3
and rgb_original is not None
and rgb_strength_mask is not None
):
original_ch = rgb_original[ci]
if patch_sat_guard_mode == "skip":
if rgb_sat_mask is not None:
out_ch = np.where(rgb_sat_mask, original_ch, out_ch)
elif patch_sat_guard_mode == "fade_strength":
# Mistura entre canal original e canal normalizado.
# Em região normal: strength=1 -> usa out_ch.
# Em saturação: strength=0 -> preserva original.
s = rgb_strength_mask.astype(np.float32)
out_ch = original_ch * (1.0 - s) + out_ch * s
result["scales"][ch_name] = {
"scale": scale,
"scale_raw_gray": scale_raw,
@ -1478,6 +1608,12 @@ class RawProcessorCore:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
self.bayer_pattern = data.get("bayer_pattern", self.bayer_pattern)
rgb_proc = data.get("rgb_processing")
if isinstance(rgb_proc, dict):
self.rgb_processing_config = self._merge_config(self.rgb_processing_config, rgb_proc)
fusion = data.get("fusion_config")
if isinstance(fusion, dict):
self.fusion_config = self._merge_config(self.fusion_config, fusion)
@ -1693,6 +1829,11 @@ class RawProcessorCore:
clip_output = bool(cfg.get("clip_output", True))
corrected = {}
# Guarda anti-roxo / anti-artefato em saturação
sat_guard_enabled = bool(cfg.get("saturation_guard_enabled", True))
sat_mode = str(cfg.get("saturation_guard_mode", "fade_strength")).lower()
sat_threshold = float(cfg.get("saturation_guard_threshold", 0.97))
for cam_id, item in decoded.items():
role = str(item.get("role") or item.get("meta", {}).get("role") or "").lower()
img = item.get("image")
@ -1711,11 +1852,20 @@ class RawProcessorCore:
out = img.astype(np.float32).copy()
# Máscara comum RGB:
# se qualquer canal estiver perto de saturar, tratamos os 3 canais juntos.
# Isso evita R/G/B receberem correções diferentes e criarem magenta/roxo.
saturation_mask = None
if sat_guard_enabled:
rgb_max = np.max(out[:, :, :3], axis=2)
saturation_mask = rgb_max >= sat_threshold
for idx, ch in enumerate(("R", "G", "B")):
out[:, :, idx] = self._apply_flat_gain_single_channel(
out[:, :, idx],
ch,
clip_output=clip_output,
saturation_mask=saturation_mask,
)
new_item["image"] = out
@ -1723,10 +1873,18 @@ class RawProcessorCore:
elif role in ("re", "nir"):
ch = "RE" if role == "re" else "NIR"
# Para RE/NIR a guarda pode ser por canal mesmo.
# Não existe cor roxa aqui, mas ainda evita mexer em pixels clipados.
saturation_mask = None
img_f = img.astype(np.float32)
if sat_guard_enabled:
saturation_mask = img_f >= sat_threshold
new_item["image"] = self._apply_flat_gain_single_channel(
img.astype(np.float32),
img_f,
ch,
clip_output=clip_output,
saturation_mask=saturation_mask,
)
else:
@ -1735,6 +1893,12 @@ class RawProcessorCore:
new_meta["flatfield_applied"] = True
new_meta["flatfield_map_type"] = cfg.get("map_type", "gain")
new_meta["flatfield_saturation_guard"] = {
"enabled": sat_guard_enabled,
"mode": sat_mode,
"threshold": sat_threshold,
}
new_item["meta"] = new_meta
corrected[cam_id] = new_item
@ -1768,7 +1932,7 @@ class RawProcessorCore:
out = np.maximum(base - dark, 0.0)
return out.astype(np.float32, copy=False)
def _apply_flat_gain_single_channel(
def _apply_flat_gain_single_channel_bkp(
self,
img: np.ndarray,
channel_name: str,
@ -1801,6 +1965,115 @@ class RawProcessorCore:
return out.astype(np.float32, copy=False)
def _apply_flat_gain_single_channel(
self,
img: np.ndarray,
channel_name: str,
clip_output: bool = True,
saturation_mask: np.ndarray | None = None,
) -> np.ndarray:
ch = str(channel_name).upper()
entry = self.flatfield_maps.get(ch)
if not entry:
return img.astype(np.float32, copy=False)
gain = entry.get("gain")
if gain is None:
return img.astype(np.float32, copy=False)
cfg = self.flatfield_config or {}
base = img.astype(np.float32)
gain = gain.astype(np.float32)
if gain.shape[:2] != base.shape[:2]:
gain = cv2.resize(
gain,
(base.shape[1], base.shape[0]),
interpolation=cv2.INTER_LINEAR,
)
# Suavização extra em runtime.
# Útil para corrigir apenas o borrão grande, não microtextura/ruído.
runtime_smooth_ksize = int(cfg.get("runtime_smooth_ksize", 0) or 0)
if runtime_smooth_ksize >= 3:
if runtime_smooth_ksize % 2 == 0:
runtime_smooth_ksize += 1
gain = cv2.GaussianBlur(
gain,
(runtime_smooth_ksize, runtime_smooth_ksize),
0,
)
# Intensidade global e por canal:
# strength=0.0 -> não aplica flatfield
# strength=1.0 -> aplica mapa integral
strength = float(cfg.get("strength", 1.0))
strength_by_channel = cfg.get("strength_by_channel", {}) or {}
if ch in strength_by_channel:
try:
strength = float(strength_by_channel[ch])
except Exception:
pass
gain_min_runtime = float(cfg.get("gain_min_runtime", 0.0))
gain_max_runtime = float(cfg.get("gain_max_runtime", 999.0))
# Guarda de saturação.
sat_guard_enabled = bool(cfg.get("saturation_guard_enabled", True))
sat_mode = str(cfg.get("saturation_guard_mode", "fade_strength")).lower()
sat_soft_start = float(cfg.get("saturation_guard_soft_start", 0.90))
sat_hard = float(cfg.get("saturation_guard_hard", 0.98))
# Se não veio uma máscara RGB comum, usa máscara do próprio canal.
if saturation_mask is None and sat_guard_enabled:
sat_threshold = float(cfg.get("saturation_guard_threshold", 0.97))
saturation_mask = base >= sat_threshold
# ============================================================
# Calcula ganho efetivo
# ============================================================
if sat_guard_enabled and sat_mode == "fade_strength":
# Reduz gradualmente a força do flatfield conforme aproxima saturação.
# A força cai de 1.0 para 0.0 entre soft_start e hard.
denom = max(sat_hard - sat_soft_start, 1e-6)
t = (base - sat_soft_start) / denom
t = np.clip(t, 0.0, 1.0)
# strength_mask:
# 1.0 longe da saturação
# 0.0 perto/acima de sat_hard
strength_mask = 1.0 - t
# Se uma máscara comum RGB foi passada, zera força nesses pixels.
# Isso mantém a neutralidade entre R/G/B em pixels suspeitos.
if saturation_mask is not None:
strength_mask = np.where(saturation_mask, 0.0, strength_mask)
gain_eff = 1.0 + (strength * strength_mask) * (gain - 1.0)
else:
# Modo normal com força fixa.
gain_eff = 1.0 + strength * (gain - 1.0)
gain_eff = np.clip(gain_eff, gain_min_runtime, gain_max_runtime)
out = base * gain_eff
# Modo skip: onde saturou, não aplica flatfield.
# O pixel continua queimado, mas não vira magenta/roxo artificial.
if sat_guard_enabled and sat_mode == "skip" and saturation_mask is not None:
out = np.where(saturation_mask, base, out)
if clip_output:
out = np.clip(out, 0.0, 1.0)
return out.astype(np.float32, copy=False)
def normalize_decoded_by_capture_controls(self, decoded: dict, meta: dict | None = None) -> dict:
cfg = self.radiometric_normalization_config or {}

2
Python/OAK/datasets/oak-fcc-3/core/raw_processor_preview.py
View File

@ -48,7 +48,7 @@ class RawProcessorPreview:
"GRBG": cv2.COLOR_BayerGR2BGR,
"RGGB": cv2.COLOR_BayerRG2BGR,
"BGGR": cv2.COLOR_BayerBG2BGR,
}
}
if self.bayer_pattern not in mapping:
raise ValueError(f"Padrão Bayer não suportado: {self.bayer_pattern}")

8
Python/OAK/datasets/oak-fcc-3/utils/check_saved_files.py
View File

@ -234,7 +234,7 @@ def build_multispec_from_raw_native_multi(group: dict, meta: dict):
sensor_width = int(meta.get("sensor_width", 1280))
sensor_height = int(meta.get("sensor_height", 800))
bayer = meta.get("bayer_pattern", "RGGB")
bayer = meta.get("bayer_pattern", "BGGR")
# Tenta usar o mesmo module_params que foi usado na captura.
calib_path = meta.get("camera_params_json") or "calibration/module_params.json"
@ -316,7 +316,7 @@ def build_visual_from_saved_payload(payload_path: Path, meta: dict, cam_id: str
role = cam_meta.get("role", cam_id)
interface = cam_meta.get("interface", "")
bit_depth = int(cam_meta.get("bit_depth", 10))
bayer = cam_meta.get("bayer_pattern", meta.get("bayer_pattern", "GBRG"))
bayer = cam_meta.get("bayer_pattern", meta.get("bayer_pattern", "BGGR"))
# USB RGB nativo
if interface.upper() == "USB" or (arr.ndim == 3 and arr.shape[2] == 3 and arr.dtype == np.uint8):
@ -411,7 +411,7 @@ def build_visual_from_saved_payload(payload_path: Path, meta: dict, cam_id: str
stream_meta = meta.get("stream_meta", {})
source_camera = stream_meta.get("source_camera", {}) or {}
bayer = source_camera.get("bayer_pattern", meta.get("bayer_pattern", "GBRG"))
bayer = source_camera.get("bayer_pattern", meta.get("bayer_pattern", "BGGR"))
bit_depth = int(source_camera.get("bit_depth", 10))
sensor_height = int(meta.get("sensor_height"))
@ -442,7 +442,7 @@ def build_visual_from_saved_payload(payload_path: Path, meta: dict, cam_id: str
stream_meta = meta.get("stream_meta", {}) or {}
source_camera = stream_meta.get("source_camera", {}) or {}
bayer = source_camera.get("bayer_pattern", meta.get("bayer_pattern", "GBRG"))
bayer = source_camera.get("bayer_pattern", meta.get("bayer_pattern", "BGGR"))
bit_depth = int(source_camera.get("bit_depth", 10))
sensor_height = int(meta.get("sensor_height"))

2
Python/OAK/datasets/oak-fcc-3/utils/flatfield_calibration_tool.py
View File

@ -664,7 +664,7 @@ def main():
parser.add_argument("--fps", type=int, default=20)
parser.add_argument("--width", type=int, default=1280)
parser.add_argument("--height", type=int, default=800)
parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"])
parser.add_argument("--raw_policy", default="require_triple", choices=["allow_single", "require_triple"])
parser.add_argument("--module_calibration_json", default="calibration/module_params.json")

2
Python/OAK/datasets/oak-fcc-3/utils/flatfield_calibration_tool_quality.py
View File

@ -1110,7 +1110,7 @@ def main():
parser.add_argument("--fps", type=int, default=20)
parser.add_argument("--width", type=int, default=1280)
parser.add_argument("--height", type=int, default=800)
parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"])
parser.add_argument("--raw_policy", default="require_triple", choices=["allow_single", "require_triple"])
parser.add_argument("--module_calibration_json", default="calibration/module_params.json")

2
Python/OAK/datasets/oak-fcc-3/utils/focus_calibration_tool.py
View File

@ -505,7 +505,7 @@ def main():
parser.add_argument("--fps", type=int, default=20)
parser.add_argument("--width", type=int, default=1280)
parser.add_argument("--height", type=int, default=800)
parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"])
parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"])
parser.add_argument("--preview_scale", type=float, default=1.0)

2
Python/OAK/datasets/oak-fcc-3/utils/manual_fusion_calibrator.py
View File

@ -233,7 +233,7 @@ def main():
parser.add_argument("--fps", type=int, default=20)
parser.add_argument("--width", type=int, default=1280)
parser.add_argument("--height", type=int, default=800)
parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"])
parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"])
parser.add_argument("--baseline_mm", type=float, default=75.0)

2
Python/OAK/datasets/oak-fcc-3/utils/sensor_calibration_tool.py
View File

@ -873,7 +873,7 @@ def main():
parser.add_argument("--fps", type=int, default=20)
parser.add_argument("--width", type=int, default=1280)
parser.add_argument("--height", type=int, default=800)
parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"])
parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"])
parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"])
parser.add_argument("--preview_scale", type=float, default=1.0)