import os import json import time import argparse from datetime import datetime import cv2 import numpy as np from cam_3.multispectral_client import MultiSpectralClient # ============================================================ # Helpers # ============================================================ def now_str() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def ensure_dir(path: str): os.makedirs(path, exist_ok=True) def overlay_hud( img_bgr, lines, x=12, y=22, area_h=None, max_font_scale=None, min_font_scale=None, max_line_step=None, min_line_step=None, bottom_margin=12, ): h, w = img_bgr.shape[:2] if area_h is None: area_h = h - y - bottom_margin scale = max(1.0, min(1.45, area_h / 480.0)) if max_font_scale is None: max_font_scale = 0.62 * scale if min_font_scale is None: min_font_scale = 0.34 * scale if max_line_step is None: max_line_step = int(22 * scale) if min_line_step is None: min_line_step = int(13 * scale) available_h = max(1, area_h - bottom_margin) n = max(1, len(lines)) font_scale = max_font_scale line_step = max_line_step needed_h = n * line_step if needed_h > available_h: shrink = available_h / float(needed_h) font_scale = max(min_font_scale, max_font_scale * shrink) line_step = max(min_line_step, int(max_line_step * shrink)) yy = y for s in lines: if yy > y + area_h - bottom_margin: break cv2.putText(img_bgr, s, (x, yy), cv2.FONT_HERSHEY_SIMPLEX, font_scale, (0, 0, 0), 3, cv2.LINE_AA) cv2.putText(img_bgr, s, (x, yy), cv2.FONT_HERSHEY_SIMPLEX, font_scale, (255, 255, 255), 1, cv2.LINE_AA) yy += line_step def to_bgr_u8_from_rgb01(rgb01: np.ndarray) -> np.ndarray: rgb_u8 = np.clip(rgb01 * 255.0, 0, 255).astype(np.uint8) return cv2.cvtColor(rgb_u8, cv2.COLOR_RGB2BGR) def gray_to_bgr_u8(gray01: np.ndarray) -> np.ndarray: g = np.clip(gray01 * 255.0, 0, 255).astype(np.uint8) return cv2.cvtColor(g, cv2.COLOR_GRAY2BGR) def resize_if_needed(img: np.ndarray, target_hw: tuple[int, int]) -> np.ndarray: target_h, target_w = target_hw if img.shape[:2] == (target_h, target_w): return img return cv2.resize(img, (target_w, target_h), interpolation=cv2.INTER_LINEAR) def validate_module_ready(status: dict, frame_type: str, raw_policy: str, capture_mode: str): if not status.get("ok", True): raise RuntimeError(f"Status inválido retornado pelo módulo: {status}") active_ids = list(status.get("active_camera_ids", [])) active_count = int(status.get("camera_count_active", 0)) if frame_type == "RAW_BRUTO": if raw_policy == "require_triple": missing = [cid for cid in ("cam0", "cam1", "cam2") if cid not in active_ids] if missing: raise RuntimeError( f"RAW_BRUTO com política require_triple exige três câmeras ativas. " f"Faltando: {missing}. Ativas atuais: {active_ids}" ) else: if active_count < 1: raise RuntimeError("RAW_BRUTO requer ao menos uma câmera ativa, mas nenhuma foi detectada.") return raise RuntimeError(f"frame_type desconhecido para validação: {frame_type}") def build_empty_panel(shape_hw: tuple[int, int], title: str) -> np.ndarray: h, w = shape_hw img = np.zeros((h, w, 3), dtype=np.uint8) overlay_hud(img, [title, "sem frame disponivel"]) return img def color_for_index(idx: int) -> tuple[int, int, int]: palette = [ (0, 255, 255), (0, 255, 0), (255, 255, 0), (255, 0, 255), (255, 128, 0), (128, 255, 0), (0, 128, 255), (200, 200, 255), ] return palette[idx % len(palette)] def compute_stats_from_roi(img01: np.ndarray, rect: tuple[int, int, int, int]) -> dict: x0, y0, x1, y1 = rect x0, x1 = sorted((int(x0), int(x1))) y0, y1 = sorted((int(y0), int(y1))) roi = img01[y0:y1, x0:x1] if roi.size == 0: return { "valid": False, "mean": 0.0, "std": 0.0, "min": 0.0, "max": 0.0, "p05": 0.0, "p95": 0.0, "pct_saturated": 0.0, "pct_dark": 0.0, "pixels": 0, } arr = roi.astype(np.float32).reshape(-1) return { "valid": True, "mean": float(arr.mean()), "std": float(arr.std()), "min": float(arr.min()), "max": float(arr.max()), "p05": float(np.percentile(arr, 5)), "p95": float(np.percentile(arr, 95)), "pct_saturated": float((arr >= 0.98).mean() * 100.0), "pct_dark": float((arr <= 0.02).mean() * 100.0), "pixels": int(arr.size), } def compute_scene_health(img01: np.ndarray) -> dict: arr = img01.astype(np.float32).reshape(-1) mean = float(arr.mean()) std = float(arr.std()) pct_sat = float((arr >= 0.98).mean() * 100.0) pct_dark = float((arr <= 0.02).mean() * 100.0) p05 = float(np.percentile(arr, 5)) p95 = float(np.percentile(arr, 95)) comments = [] if pct_sat > 5.0: comments.append("saturando") if pct_dark > 40.0: comments.append("muito escuro") if std < 0.05: comments.append("baixo contraste") if not comments: comments.append("ok") return { "mean": mean, "std": std, "p05": p05, "p95": p95, "pct_saturated": pct_sat, "pct_dark": pct_dark, "comment": ", ".join(comments), } def draw_rois(panel_bgr: np.ndarray, rois: list[dict]): for idx, roi in enumerate(rois): color = roi.get("color", color_for_index(idx)) label = roi.get("name", f"roi_{idx+1}") if roi.get("type") == "polygon": pts = np.array(roi.get("points", []), dtype=np.int32) if len(pts) >= 2: cv2.polylines(panel_bgr, [pts], isClosed=True, color=color, thickness=2) if len(pts) >= 1: x, y = pts[0] cv2.putText( panel_bgr, label, (int(x) + 4, max(18, int(y) - 6)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2, cv2.LINE_AA, ) continue rect = roi.get("rect") if rect is None: continue x0, y0, x1, y1 = rect cv2.rectangle(panel_bgr, (x0, y0), (x1, y1), color, 2) cv2.putText( panel_bgr, label, (x0 + 4, max(18, y0 - 6)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2, cv2.LINE_AA, ) def compute_stats_from_polygon_roi(img01: np.ndarray, points: list) -> dict: h, w = img01.shape[:2] if len(points) < 3: return compute_stats_from_roi(img01, (0, 0, 0, 0)) pts = np.array(points, dtype=np.int32) mask = np.zeros((h, w), dtype=np.uint8) cv2.fillPoly(mask, [pts], 255) arr = img01[mask > 0].astype(np.float32).reshape(-1) if arr.size == 0: return { "valid": False, "mean": 0.0, "std": 0.0, "min": 0.0, "max": 0.0, "p05": 0.0, "p95": 0.0, "pct_saturated": 0.0, "pct_dark": 0.0, "pixels": 0, } return { "valid": True, "mean": float(arr.mean()), "std": float(arr.std()), "min": float(arr.min()), "max": float(arr.max()), "p05": float(np.percentile(arr, 5)), "p95": float(np.percentile(arr, 95)), "pct_saturated": float((arr >= 0.98).mean() * 100.0), "pct_dark": float((arr <= 0.02).mean() * 100.0), "pixels": int(arr.size), } def compute_stats_for_roi(img01: np.ndarray, roi: dict) -> dict: if roi.get("type") == "polygon": return compute_stats_from_polygon_roi(img01, roi.get("points", [])) return compute_stats_from_roi(img01, roi.get("rect", (0, 0, 0, 0))) def draw_current_polygon(panel_bgr: np.ndarray, points: list): if not points: return pts = np.array(points, dtype=np.int32) for p in pts: cv2.circle(panel_bgr, tuple(p), 4, (0, 255, 255), -1) if len(pts) >= 2: cv2.polylines(panel_bgr, [pts], isClosed=False, color=(0, 255, 255), thickness=1) # ============================================================ # MOCK # ============================================================ def load_mock_image_rgb(path: str, fallback_shape=(480, 640)): if not path: h, w = fallback_shape img = np.zeros((h, w, 3), dtype=np.float32) return img bgr = cv2.imread(path, cv2.IMREAD_COLOR) if bgr is None: raise RuntimeError(f"Falha ao carregar mock RGB: {path}") rgb = bgr[:, :, ::-1].astype(np.float32) / 255.0 return rgb def load_mock_image_gray(path: str, fallback_shape=(480, 640)): if not path: h, w = fallback_shape return np.zeros((h, w), dtype=np.float32) gray = cv2.imread(path, cv2.IMREAD_GRAYSCALE) if gray is None: raise RuntimeError(f"Falha ao carregar mock mono: {path}") return gray.astype(np.float32) / 255.0 def build_mock_decoded(args): shape = (args.height, args.width) cam2 = load_mock_image_rgb(args.mock_cam2, fallback_shape=shape) cam0 = load_mock_image_gray(args.mock_cam0, fallback_shape=shape) cam1 = load_mock_image_gray(args.mock_cam1, fallback_shape=shape) return { "cam2": {"name": "RGB", "image": cam2, "meta": {"mock": True}}, "cam0": {"name": "RE", "image": cam0, "meta": {"mock": True}}, "cam1": {"name": "NIR", "image": cam1, "meta": {"mock": True}}, } # ============================================================ # Análise de dados offline # ============================================================ def ts_name() -> str: return datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] def save_offline_sample( base_dir: str, preview_bgr: np.ndarray, meta: dict, packed_raw_by_camera: dict, ): 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") payload_files = {} payload_shapes = {} payload_dtypes = {} for cam_id, arr in packed_raw_by_camera.items(): path = os.path.join(base_dir, f"{name}_{cam_id}.bin") arr.tofile(path) payload_files[cam_id] = os.path.basename(path) payload_shapes[cam_id] = list(arr.shape) payload_dtypes[cam_id] = str(arr.dtype) meta_save = dict(meta) meta_save["saved_payload_type"] = "raw_native_multi" meta_save["saved_payload_paths"] = payload_files meta_save["saved_payload_shapes"] = payload_shapes meta_save["saved_payload_dtypes"] = payload_dtypes meta_save["saved_preview_path"] = os.path.basename(png_path) cv2.imwrite(png_path, preview_bgr) with open(json_path, "w", encoding="utf-8") as f: json.dump(meta_save, f, ensure_ascii=False, indent=2) return png_path, json_path def load_offline_sample_decoded(json_path: str, cam: MultiSpectralClient): if not os.path.isfile(json_path): raise FileNotFoundError(f"Sample offline não encontrado: {json_path}") with open(json_path, "r", encoding="utf-8") as f: meta = json.load(f) base_dir = os.path.dirname(json_path) payload_paths = meta.get("saved_payload_paths") or {} payload_shapes = meta.get("saved_payload_shapes") or {} payload_dtypes = meta.get("saved_payload_dtypes") or {} if not payload_paths: raise RuntimeError("Sample offline inválido: saved_payload_paths ausente") frame = {} for cam_id, rel_path in payload_paths.items(): bin_path = os.path.join(base_dir, rel_path) if not os.path.isfile(bin_path): raise FileNotFoundError(f"Payload não encontrado para {cam_id}: {bin_path}") dtype_str = payload_dtypes.get(cam_id, "uint8") shape = payload_shapes.get(cam_id) if shape is None: raise RuntimeError(f"Shape ausente para {cam_id}") arr = np.fromfile(bin_path, dtype=np.dtype(dtype_str)).reshape(tuple(shape)) frame[cam_id] = arr stream_meta = meta.get("stream_meta") or meta # Garante campos mínimos usados pelo decoder. stream_meta.setdefault("camera_frames", meta.get("camera_frames", {})) stream_meta.setdefault("frame_type", "RAW_BRUTO") decoded = cam.core.decode_stream_cameras(frame, stream_meta) preview_path = meta.get("saved_preview_path") preview_bgr = None if preview_path: preview_full = os.path.join(base_dir, preview_path) if os.path.isfile(preview_full): preview_bgr = cv2.imread(preview_full, cv2.IMREAD_COLOR) return decoded, stream_meta, frame, preview_bgr # ============================================================ # Persistência dos parâmetros/snapshots # ============================================================ def default_payload(args, effective_capture_mode: str): return { "schema": "manual_sensor_calibration_v1", "saved_at": now_str(), "pi_host": args.pi_host, "pc_host": args.pc_host, "stream_port": args.stream_port, "frame_type": "RAW_BRUTO", "capture_mode_requested": args.capture_mode, "capture_mode_effective": effective_capture_mode, "raw_policy": args.raw_policy, "sensor_width": args.width, "sensor_height": args.height, "bayer_pattern": args.bayer, "notes": args.notes or "", "camera_settings": { "cam0": {}, "cam1": {}, "cam2": {}, }, "snapshots": [], "calibration_guidance_log": [], } def load_payload(path: str, args, effective_capture_mode: str): if not path or not os.path.isfile(path): return default_payload(args, effective_capture_mode) with open(path, "r", encoding="utf-8") as f: data = json.load(f) data.setdefault("schema", "manual_sensor_calibration_v1") data.setdefault("camera_settings", {"cam0": {}, "cam1": {}, "cam2": {}}) data.setdefault("snapshots", []) data.setdefault("calibration_guidance_log", []) return data def save_payload(path: str, data: dict): ensure_dir(os.path.dirname(path) or ".") data = dict(data) data["saved_at"] = now_str() with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def build_camera_params_payload(args, effective_capture_mode, camera_controls, rois=None, snapshots=None, guidance_log=None): return { "schema": "multispec_camera_params_v1", "saved_at": now_str(), "pi_host": args.pi_host, "pc_host": args.pc_host, "stream_port": args.stream_port, "frame_type": "RAW_BRUTO", "capture_mode_requested": args.capture_mode, "capture_mode_effective": effective_capture_mode, "raw_policy": args.raw_policy, "sensor_width": args.width, "sensor_height": args.height, "bayer_pattern": args.bayer, "camera_settings": json.loads(json.dumps(camera_controls)), "rois": rois or {}, "snapshots": snapshots or [], "notes": args.notes or "", "calibration_guidance_log": guidance_log or [], } # ============================================================ # Guia automática de ajustes dos controles # ============================================================ def normalize_class_name(name: str) -> str: s = (name or "").strip().lower() if s.startswith("cana"): return "cana" if s.startswith("erva"): return "erva" if s.startswith("solo") or s.startswith("chao") or s.startswith("chão"): return "solo" if s.startswith("palha"): return "palha" return s def collect_roi_metrics_by_class(img01: np.ndarray, rois_for_cam: list[dict]) -> dict: grouped = {} for roi in rois_for_cam: cls = normalize_class_name(roi.get("name", "")) if not cls: continue stats = compute_stats_for_roi(img01, roi) if not stats.get("valid"): continue grouped.setdefault(cls, []).append(stats) summary = {} for cls, items in grouped.items(): summary[cls] = { "count": len(items), "mean": float(np.mean([x["mean"] for x in items])), "std": float(np.mean([x["std"] for x in items])), "p05": float(np.mean([x["p05"] for x in items])), "p95": float(np.mean([x["p95"] for x in items])), "pct_saturated": float(np.mean([x["pct_saturated"] for x in items])), "pct_dark": float(np.mean([x["pct_dark"] for x in items])), "pixels": int(sum(x["pixels"] for x in items)), } return summary def mean_of_classes(summary: dict, classes: list[str], key: str = "mean"): vals = [summary[c][key] for c in classes if c in summary] if not vals: return None return float(np.mean(vals)) def analyze_spectral_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: summary = collect_roi_metrics_by_class(img01, rois_for_cam) veg_mean = mean_of_classes(summary, ["cana", "erva"], "mean") veg_p95 = mean_of_classes(summary, ["cana", "erva"], "p95") veg_sat = mean_of_classes(summary, ["cana", "erva"], "pct_saturated") solo_mean = mean_of_classes(summary, ["solo", "palha"], "mean") before = json.loads(json.dumps(ctrl)) new_ctrl = json.loads(json.dumps(ctrl)) action = "keep" status = "ok" reason = "Parametros parecem aceitaveis." if veg_mean is None: return { "status": "need_rois", "action": "none", "reason": "Crie pelo menos uma ROI de cana ou erva para analisar canal espectral.", "class_metrics": summary, "before_settings": before, "after_settings": new_ctrl, } separation = None if solo_mean is not None: separation = float(veg_mean - solo_mean) exp = new_ctrl.get("exposure_time_us") gain = new_ctrl.get("analogue_gain") if exp is None: exp = 15000 if gain is None: gain = 1.0 new_ctrl["ae_enable"] = False new_ctrl["awb_enable"] = False # 1) Proteção contra estouro MIN_EXP_US = 100 MIN_GAIN = 1.0 if veg_sat is not None and veg_sat > 1.0: if exp > MIN_EXP_US: new_ctrl["exposure_time_us"] = int(max(exp - exp_step, MIN_EXP_US)) action = "decrease_exposure" status = "adjust" reason = f"Vegetacao saturando ({veg_sat:.2f}%). Reduzir exposicao." elif gain > MIN_GAIN: new_ctrl["analogue_gain"] = float(max(gain / (1.0 + gain_step), MIN_GAIN)) action = "decrease_gain" status = "adjust" reason = ( f"Vegetacao saturando ({veg_sat:.2f}%), mas exposicao ja esta no minimo. " "Reduzir ganho." ) else: action = "keep" status = "limit" reason = ( f"Vegetacao saturando ({veg_sat:.2f}%), mas exposicao e ganho ja estao no minimo. " "Nao ha ajuste possivel por software." ) # 2) Vegetação pouco iluminada elif veg_p95 is not None and veg_p95 < 0.75: new_ctrl["exposure_time_us"] = int(min(exp + exp_step, 200000)) action = "increase_exposure" status = "adjust" reason = f"p95 da vegetacao baixo ({veg_p95:.3f}). Aumentar exposicao." # 3) Vegetação muito perto do teto elif veg_p95 is not None and veg_p95 > 0.96: new_ctrl["exposure_time_us"] = int(max(exp - exp_step, 100)) action = "decrease_exposure" status = "adjust" reason = f"p95 da vegetacao alto ({veg_p95:.3f}). Reduzir exposicao." # 4) Separação ruim elif separation is not None and separation < 0.25: if veg_p95 is not None and veg_p95 < 0.90: new_ctrl["exposure_time_us"] = int(min(exp + exp_step, 200000)) action = "increase_exposure" status = "adjust" reason = f"Separacao baixa ({separation:.3f}) e ha margem no p95. Aumentar exposicao." else: new_ctrl["analogue_gain"] = float(min(gain * (1.0 + gain_step), 32.0)) action = "increase_gain" status = "adjust" reason = f"Separacao baixa ({separation:.3f}) sem muita margem de exposicao. Aumentar ganho levemente." return { "status": status, "action": action, "reason": reason, "channel": selected_cam, "class_metrics": summary, "veg_mean": veg_mean, "solo_mean": solo_mean, "separation": separation, "veg_p95": veg_p95, "veg_sat": veg_sat, "before_settings": before, "after_settings": new_ctrl, } def analyze_rgb_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: scene = compute_scene_health(img01) summary = collect_roi_metrics_by_class(img01, rois_for_cam) before = json.loads(json.dumps(ctrl)) new_ctrl = json.loads(json.dumps(ctrl)) action = "keep" status = "ok" reason = "RGB parece aceitável." exp = new_ctrl.get("exposure_time_us") gain = new_ctrl.get("analogue_gain") if exp is None: exp = 15000 if gain is None: gain = 1.0 # Para RGB calibrado fixo: desligar AE/AWB quando for aplicar preset final. new_ctrl["ae_enable"] = False new_ctrl["awb_enable"] = False if scene["pct_saturated"] > 2.0 or scene["p95"] > 0.97: new_ctrl["exposure_time_us"] = int(max(exp - exp_step, 100)) action = "decrease_exposure" status = "adjust" reason = f"RGB muito próximo de saturar. sat={scene['pct_saturated']:.2f}%, p95={scene['p95']:.3f}." elif scene["pct_dark"] > 20.0 and scene["p95"] < 0.85: new_ctrl["exposure_time_us"] = int(min(exp + exp_step, 200000)) action = "increase_exposure" status = "adjust" reason = f"RGB escuro. dark={scene['pct_dark']:.2f}%, p95={scene['p95']:.3f}." elif scene["std"] < 0.08: new_ctrl["analogue_gain"] = float(min(gain * (1.0 + gain_step), 32.0)) action = "increase_gain" status = "adjust" reason = f"RGB com baixo contraste global. std={scene['std']:.3f}." return { "status": status, "action": action, "reason": reason, "channel": selected_cam, "scene_health": scene, "class_metrics": summary, "before_settings": before, "after_settings": new_ctrl, } def run_guidance_analysis(selected_cam: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: if img01 is None: return { "status": "error", "action": "none", "reason": "Sem imagem ativa para análise.", "before_settings": json.loads(json.dumps(ctrl)), "after_settings": json.loads(json.dumps(ctrl)), } if selected_cam == "cam2": return analyze_rgb_guidance(selected_cam, img01, rois_for_cam, ctrl, exp_step, gain_step) return analyze_spectral_guidance(selected_cam, img01, rois_for_cam, ctrl, exp_step, gain_step) # ============================================================ # Main # ============================================================ def main(): parser = argparse.ArgumentParser( description="Ferramenta de calibração dos sensores RGB/RE/NIR com controle manual e ROIs em tempo real.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--pi_host", default="192.168.105.6") parser.add_argument("--pc_host", default="192.168.105.5") parser.add_argument("--stream_port", type=int, default=6001) parser.add_argument("--server_port", type=int, default=5000) parser.add_argument("--fps", type=int, default=20) parser.add_argument("--width", type=int, default=640) parser.add_argument("--height", type=int, default=480) parser.add_argument("--bayer", default="GBRG", 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) parser.add_argument("--exp_step", type=int, default=1000, help="Passo de exposição em us") parser.add_argument("--gain_step", type=float, default=0.10, help="Passo multiplicativo do ganho") parser.add_argument("--out_json", default="calibration/sensor_calibration.json") parser.add_argument("--load_json", default="") parser.add_argument("--notes", default="") parser.add_argument("--mock", action="store_true") parser.add_argument("--mock_cam0", default="", help="Imagem mock para cam0 / RE") parser.add_argument("--mock_cam1", default="", help="Imagem mock para cam1 / NIR") parser.add_argument("--mock_cam2", default="", help="Imagem mock para cam2 / RGB") parser.add_argument("--offline_sample_json", default="", help="JSON de sample salvo para análise offline") parser.add_argument("--offline_save_dir", default="calibration/offline_samples", help="Pasta para salvar frames brutos offline") args = parser.parse_args() cam = MultiSpectralClient( pi_host=args.pi_host, pc_host=args.pc_host, server_port=args.server_port, stream_port=args.stream_port, width=args.width, height=args.height, bayer=args.bayer, fps=args.fps, frame_type="RAW_BRUTO", output_dtype="uint8", capture_mode=args.capture_mode, raw_policy=args.raw_policy, module_calibration_json=None, ) offline_mode = bool(args.offline_sample_json) live_mode = not args.mock and not offline_mode effective_capture_mode = args.capture_mode data_payload = load_payload(args.load_json, args, effective_capture_mode) selected_cam = "cam2" last_msg = "" last_msg_t = 0.0 last_frame_id = -1 fps_view = 0.0 fps_stream = 0.0 t_view_fps = time.time() t_stream_fps = time.time() view_frames = 0 stream_frames_accum = 0 last_stream_frame_id = None decoded_last = {} last_meta_stream = None last_raw_frame = None last_preview_bgr = None roi_name_input_active = False roi_name_buffer = "" roi_name_points_pending = [] guidance_log = data_payload.get("calibration_guidance_log", []) last_guidance = guidance_log[-1]["result"] if guidance_log else None window_name = "Sensor Calibration Tool" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) panel_rects = { "cam2": None, "cam0": None, "cam1": None, "data": None, } # Controle de câmera camera_controls = { "cam0": { "ae_enable": False, "awb_enable": False, "exposure_time_us": 15000, "analogue_gain": 1.0, "colour_gains": None, }, "cam1": { "ae_enable": False, "awb_enable": False, "exposure_time_us": 15000, "analogue_gain": 1.0, "colour_gains": None, }, "cam2": { "ae_enable": True, "awb_enable": True, "exposure_time_us": 15000, "analogue_gain": 1.0, "colour_gains": [1.0, 1.0], }, } rois = { "cam2": [], "cam0": [], "cam1": [], } current_polygon_points = [] def get_active_rect_for_mouse(): return panel_rects.get(selected_cam) def on_mouse(event, x, y, flags, param): nonlocal current_polygon_points, last_msg, last_msg_t rect = get_active_rect_for_mouse() if rect is None: return x0, y0, x1, y1 = rect inside = (x0 <= x < x1 and y0 <= y < y1) if not inside: return lx = int(x - x0) ly = int(y - y0) if event == cv2.EVENT_LBUTTONDOWN: current_polygon_points.append((lx, ly)) last_msg = f"{selected_cam}: ponto #{len(current_polygon_points)} adicionado" last_msg_t = time.time() cv2.setMouseCallback(window_name, on_mouse) if args.mock: decoded_last = build_mock_decoded(args) if offline_mode: decoded_last, last_meta_stream, last_raw_frame, last_preview_bgr = load_offline_sample_decoded(args.offline_sample_json, cam) def apply_controls_to_selected_cam(): nonlocal cam, last_msg, last_msg_t ctrl = camera_controls[selected_cam] try: resp = cam.svc.set_ae_enable(selected_cam, bool(ctrl["ae_enable"])) ctrl["ae_enable"] = bool(resp.get("ae_enable", ctrl["ae_enable"])) if selected_cam == "cam2": resp = cam.svc.set_awb_enable(selected_cam, bool(ctrl["awb_enable"])) ctrl["awb_enable"] = bool(resp.get("awb_enable", ctrl["awb_enable"])) if not ctrl["ae_enable"]: if ctrl["exposure_time_us"] is not None: resp = cam.svc.set_exposure_time(selected_cam, int(ctrl["exposure_time_us"])) exp_val = resp.get("exposure_time_us", ctrl["exposure_time_us"]) ctrl["exposure_time_us"] = int(exp_val) if exp_val is not None else None if ctrl["analogue_gain"] is not None: resp = cam.svc.set_analogue_gain(selected_cam, float(ctrl["analogue_gain"])) gain_val = resp.get("analogue_gain", ctrl["analogue_gain"]) ctrl["analogue_gain"] = float(gain_val) if gain_val is not None else None last_msg = f"Controles aplicados em {selected_cam}" last_msg_t = time.time() except Exception as e: last_msg = f"Falha ao aplicar controles: {e}" last_msg_t = time.time() def snapshot_current_state(): active_img = None if selected_cam in decoded_last: active_img = decoded_last[selected_cam]["image"] if active_img is None: return None roi_entries = [] for roi in rois[selected_cam]: stats = compute_stats_for_roi(active_img, roi) entry = { "name": roi["name"], "type": roi.get("type", "rect"), "metrics": stats, } if roi.get("type") == "polygon": entry["points"] = [[int(x), int(y)] for x, y in roi.get("points", [])] else: entry["rect"] = list(map(int, roi["rect"])) roi_entries.append(entry) snap = { "timestamp": now_str(), "camera": selected_cam, "camera_settings": json.loads(json.dumps(camera_controls[selected_cam])), "scene_health": compute_scene_health(active_img), "rois": roi_entries, } return snap def sync_camera_controls_from_pi(): nonlocal cam, camera_controls if not live_mode: return for cam_id in camera_controls.keys(): try: initial_ctrl = cam.svc.get_camera_controls(cam_id) camera_controls[cam_id]["ae_enable"] = bool( initial_ctrl.get("ae_enable", camera_controls[cam_id]["ae_enable"]) ) camera_controls[cam_id]["awb_enable"] = bool( initial_ctrl.get("awb_enable", camera_controls[cam_id]["awb_enable"]) ) exp_val = initial_ctrl.get("exposure_time_us", camera_controls[cam_id]["exposure_time_us"]) camera_controls[cam_id]["exposure_time_us"] = int(exp_val) if exp_val is not None else None gain_val = initial_ctrl.get("analogue_gain", camera_controls[cam_id]["analogue_gain"]) camera_controls[cam_id]["analogue_gain"] = float(gain_val) if gain_val is not None else None camera_controls[cam_id]["colour_gains"] = initial_ctrl.get( "colour_gains", camera_controls[cam_id]["colour_gains"] ) except Exception as e: print(f"[WARN] Falha ao ler controles iniciais de {cam_id}: {e}") try: if live_mode: cam.start(print_debug=True) sync_camera_controls_from_pi() else: last_msg = "MODO OFFLINE ativo" if offline_mode else "MODO MOCK ativo" last_msg_t = time.time() while True: t0 = time.time() if live_mode: frame, meta, decoded = cam.get_next_decoded(timeout=2.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"] if not isinstance(frame, dict): raise RuntimeError("Este calibrador espera RAW_BRUTO multi-payload como dict de câmeras.") decoded_last = decoded last_meta_stream = dict(meta) last_raw_frame = {cam_id: arr.copy() for cam_id, arr in frame.items()} curr_frame_id = meta.get("frame_id") if curr_frame_id is not None and 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() else: fps_stream = 0.0 fps_view = 0.0 if decoded_last: rgb01 = decoded_last.get("cam2", {}).get("image") re01 = decoded_last.get("cam0", {}).get("image") nir01 = decoded_last.get("cam1", {}).get("image") if rgb01 is None: rgb_panel = build_empty_panel((args.height, args.width), "RGB") base_h, base_w = args.height, args.width else: rgb_panel = to_bgr_u8_from_rgb01(rgb01) base_h, base_w = rgb01.shape[:2] re_panel = gray_to_bgr_u8(resize_if_needed(re01, (base_h, base_w))) if re01 is not None else build_empty_panel((base_h, base_w), "RE") nir_panel = gray_to_bgr_u8(resize_if_needed(nir01, (base_h, base_w))) if nir01 is not None else build_empty_panel((base_h, base_w), "NIR") draw_rois(rgb_panel, rois["cam2"]) draw_rois(re_panel, rois["cam0"]) draw_rois(nir_panel, rois["cam1"]) active_panel = {"cam2": rgb_panel, "cam0": re_panel, "cam1": nir_panel}.get(selected_cam) if active_panel is not None: draw_current_polygon(active_panel, current_polygon_points) overlay_hud(rgb_panel, ["RGB (cam2)", f"ativo={selected_cam == 'cam2'}"]) overlay_hud(re_panel, ["RE (cam0)", f"ativo={selected_cam == 'cam0'}"]) overlay_hud(nir_panel, ["NIR (cam1)", f"ativo={selected_cam == 'cam1'}"]) ph = max(rgb_panel.shape[0], re_panel.shape[0], nir_panel.shape[0], base_h) pw = max(rgb_panel.shape[1], re_panel.shape[1], nir_panel.shape[1], base_w) def fit_panel(img): if img.shape[:2] != (ph, pw): return cv2.resize(img, (pw, ph), interpolation=cv2.INTER_NEAREST) return img rgb_panel = fit_panel(rgb_panel) re_panel = fit_panel(re_panel) nir_panel = fit_panel(nir_panel) if rgb01 is not None: last_preview_bgr = to_bgr_u8_from_rgb01(rgb01) else: last_preview_bgr = rgb_panel.copy() data_panel = np.zeros((ph, pw, 3), dtype=np.uint8) panel_rects["cam2"] = (0, 0, pw, ph) panel_rects["cam0"] = (pw, 0, pw * 2, ph) panel_rects["cam1"] = (0, ph, pw, ph * 2) panel_rects["data"] = (pw, ph, pw * 2, ph * 2) top = np.hstack([rgb_panel, re_panel]) bottom = np.hstack([nir_panel, data_panel]) board = np.vstack([top, bottom]) active_img = decoded_last.get(selected_cam, {}).get("image") global_stats = compute_scene_health(active_img) if active_img is not None else None ctrl = camera_controls[selected_cam] lines = [ f"CAM ATIVA: {selected_cam}", f"AE={'ON' if ctrl['ae_enable'] else 'OFF'} | AWB={'ON' if ctrl['awb_enable'] else 'OFF'}", f"EXP={ctrl['exposure_time_us']} us", f"GAIN={ctrl['analogue_gain']:.2f}", f"fps_stream={fps_stream:.1f} | fps_view={fps_view:.1f}", ] if global_stats is not None: lines.extend([ f"mean={global_stats['mean']:.3f} | std={global_stats['std']:.3f}", f"p05={global_stats['p05']:.3f} | p95={global_stats['p95']:.3f}", f"sat={global_stats['pct_saturated']:.2f}% | dark={global_stats['pct_dark']:.2f}%", f"scene={global_stats['comment']}", ]) else: lines.append("sem stats da cena") if last_guidance is not None: lines.extend([ "-", f"GUIDE: {last_guidance.get('status')} | {last_guidance.get('action')}", f"{last_guidance.get('reason', '')[:46]}", ]) sep = last_guidance.get("separation") if sep is not None: lines.append(f"sep_veg_solo={sep:.3f}") if roi_name_input_active: lines.extend([ "-", "NOME DA ROI:", f"> {roi_name_buffer}_", "ENTER confirma | ESC cancela | BACKSPACE apaga", ]) lines.append("-") lines.append(f"ROIs: {len(rois[selected_cam])}") for idx, roi in enumerate(rois[selected_cam][:6]): if active_img is None: break stats = compute_stats_for_roi(active_img, roi) lines.append(f"{roi['name']}: mean={stats['mean']:.3f} std={stats['std']:.3f}") lines.append(f" p95={stats['p95']:.3f} sat={stats['pct_saturated']:.1f}% dark={stats['pct_dark']:.1f}%") lines.extend([ "-", "1=RGB | 2=RE | 3=NIR | E=AE | B=AWB", "I/K exp +/- | O/L gain +/- | G guia | A aplica", "mouse: clique pontos | ENTER fecha ROI | U desfaz ponto/ROI | X limpa poligono", "F salva frame bruto | SPACE salva PARAMS | S snapshot | Q sai", ]) x0, y0, x1, y1 = panel_rects["data"] overlay_hud( board, lines, x=x0 + 12, y=y0 + 22, area_h=(y1 - y0) - 22, ) if last_msg and (time.time() - last_msg_t) < 2.5: cv2.putText(board, last_msg, (12, board.shape[0] - 16), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 0), 2, cv2.LINE_AA) if args.preview_scale != 1.0: board = cv2.resize( board, (int(board.shape[1] * args.preview_scale), int(board.shape[0] * args.preview_scale)), interpolation=cv2.INTER_NEAREST, ) cv2.imshow(window_name, board) else: blank = np.zeros((720, 1280, 3), dtype=np.uint8) overlay_hud(blank, ["Aguardando frames do módulo..."], x=40, y=80) cv2.imshow(window_name, blank) k = cv2.waitKey(1) & 0xFF if roi_name_input_active: if k in (13, 10): # ENTER name = roi_name_buffer.strip() if not name: name = f"roi_{len(rois[selected_cam]) + 1}" roi = { "name": name, "type": "polygon", "points": list(roi_name_points_pending), "color": color_for_index(len(rois[selected_cam])), } rois[selected_cam].append(roi) current_polygon_points = [] roi_name_points_pending = [] roi_name_buffer = "" roi_name_input_active = False last_msg = f"ROI criada em {selected_cam}: {name}" last_msg_t = time.time() elif k in (27,): # ESC roi_name_input_active = False roi_name_buffer = "" roi_name_points_pending = [] last_msg = "Criacao de ROI cancelada" last_msg_t = time.time() elif k in (8, 127): # BACKSPACE roi_name_buffer = roi_name_buffer[:-1] elif 32 <= k <= 126: roi_name_buffer += chr(k) continue if k in (ord("q"), ord("Q"), 27): break elif k == ord("1"): selected_cam = "cam2" last_msg = "Selecionada: cam2 / RGB" last_msg_t = time.time() elif k == ord("2"): selected_cam = "cam0" last_msg = "Selecionada: cam0 / RE" last_msg_t = time.time() elif k == ord("3"): selected_cam = "cam1" last_msg = "Selecionada: cam1 / NIR" last_msg_t = time.time() elif k in (ord("e"), ord("E")): camera_controls[selected_cam]["ae_enable"] = not camera_controls[selected_cam]["ae_enable"] last_msg = f"AE {selected_cam} -> {'ON' if camera_controls[selected_cam]['ae_enable'] else 'OFF'}" last_msg_t = time.time() elif k in (ord("b"), ord("B")): if selected_cam == "cam2": camera_controls[selected_cam]["awb_enable"] = not camera_controls[selected_cam]["awb_enable"] last_msg = f"AWB {selected_cam} -> {'ON' if camera_controls[selected_cam]['awb_enable'] else 'OFF'}" else: last_msg = "AWB só se aplica ao RGB" last_msg_t = time.time() elif k in (ord("i"), ord("I")): if camera_controls[selected_cam]["exposure_time_us"] is None: camera_controls[selected_cam]["exposure_time_us"] = 15000 else: camera_controls[selected_cam]["exposure_time_us"] = int( min(camera_controls[selected_cam]["exposure_time_us"] + args.exp_step, 200000) ) last_msg = f"EXP {selected_cam} -> {camera_controls[selected_cam]['exposure_time_us']} us" last_msg_t = time.time() elif k in (ord("k"), ord("K")): if camera_controls[selected_cam]["exposure_time_us"] is None: camera_controls[selected_cam]["exposure_time_us"] = 15000 else: camera_controls[selected_cam]["exposure_time_us"] = int( max(camera_controls[selected_cam]["exposure_time_us"] - args.exp_step, 100) ) last_msg = f"EXP {selected_cam} -> {camera_controls[selected_cam]['exposure_time_us']} us" last_msg_t = time.time() elif k in (ord("o"), ord("O")): if camera_controls[selected_cam]["analogue_gain"] is None: camera_controls[selected_cam]["analogue_gain"] = 1.0 else: camera_controls[selected_cam]["analogue_gain"] = float( min(camera_controls[selected_cam]["analogue_gain"] * (1.0 + args.gain_step), 32.0) ) last_msg = f"GAIN {selected_cam} -> {camera_controls[selected_cam]['analogue_gain']:.2f}" last_msg_t = time.time() elif k in (ord("l"), ord("L")): if camera_controls[selected_cam]["analogue_gain"] is None: camera_controls[selected_cam]["analogue_gain"] = 1.0 else: camera_controls[selected_cam]["analogue_gain"] = float( max(camera_controls[selected_cam]["analogue_gain"] / (1.0 + args.gain_step), 1.0) ) last_msg = f"GAIN {selected_cam} -> {camera_controls[selected_cam]['analogue_gain']:.2f}" last_msg_t = time.time() elif k in (ord("g"), ord("G")): active_img = decoded_last.get(selected_cam, {}).get("image") ctrl = camera_controls[selected_cam] result = run_guidance_analysis( selected_cam=selected_cam, img01=active_img, rois_for_cam=rois[selected_cam], ctrl=ctrl, exp_step=args.exp_step, gain_step=args.gain_step, ) last_guidance = result guidance_entry = { "timestamp": now_str(), "camera": selected_cam, "result": result, } guidance_log.append(guidance_entry) data_payload.setdefault("calibration_guidance_log", []).append(guidance_entry) after = result.get("after_settings") if isinstance(after, dict): camera_controls[selected_cam].update(after) last_msg = f"GUIDE {selected_cam}: {result.get('action')} | {result.get('status')}" last_msg_t = time.time() elif k in (ord("a"), ord("A")): if live_mode: apply_controls_to_selected_cam() else: last_msg = "Controles só aplicam no modo ao vivo" last_msg_t = time.time() elif k in (ord("u"), ord("U")): if current_polygon_points: current_polygon_points.pop() last_msg = f"Ponto removido | restantes={len(current_polygon_points)}" last_msg_t = time.time() elif rois[selected_cam]: removed = rois[selected_cam].pop() last_msg = f"ROI removida: {removed['name']}" last_msg_t = time.time() elif k in (ord("x"), ord("X")): current_polygon_points = [] last_msg = "Polígono atual limpo" last_msg_t = time.time() elif k in (ord("c"), ord("C")): rois[selected_cam] = [] last_msg = f"ROIs limpas em {selected_cam}" last_msg_t = time.time() elif k in (ord("s"), ord("S")): snap = snapshot_current_state() if snap is not None: data_payload.setdefault("snapshots", []).append(snap) last_msg = f"Snapshot salvo: {selected_cam} | rois={len(snap['rois'])}" else: last_msg = "Sem frame ativo para snapshot" last_msg_t = time.time() elif k in (ord("f"), ord("F")): if not live_mode: last_msg = "Salvar frame bruto só faz sentido no modo ao vivo" last_msg_t = time.time() elif last_raw_frame is None or last_meta_stream is None: last_msg = "Sem frame bruto atual para salvar" last_msg_t = time.time() else: preview_to_save = last_preview_bgr if preview_to_save is None: preview_to_save = np.zeros((args.height, args.width, 3), dtype=np.uint8) meta_save = { "ts": datetime.now().isoformat(timespec="milliseconds"), "schema": "multispec_offline_sample_v1", "frame_type": "RAW_BRUTO", "sensor_width": args.width, "sensor_height": args.height, "bayer_pattern": args.bayer, "capture_mode_requested": args.capture_mode, "capture_mode_effective": effective_capture_mode, "raw_policy": args.raw_policy, "stream_meta": last_meta_stream, "camera_settings": json.loads(json.dumps(camera_controls)), "note": "offline_sample_from_sensor_calibration_tool", } png_path, json_path = save_offline_sample( base_dir=args.offline_save_dir, preview_bgr=preview_to_save, meta=meta_save, packed_raw_by_camera=last_raw_frame, ) last_msg = f"FRAME salvo offline: {os.path.basename(json_path)}" last_msg_t = time.time() elif k == 32: # SPACE payload_to_save = build_camera_params_payload( args=args, effective_capture_mode=effective_capture_mode, camera_controls=camera_controls, rois=rois, snapshots=data_payload.get("snapshots", []), guidance_log=guidance_log, ) save_payload(args.out_json, payload_to_save) data_payload = payload_to_save last_msg = f"PARAMS salvos em: {args.out_json}" last_msg_t = time.time() elif k == 13: # ENTER if len(current_polygon_points) < 3: last_msg = "ROI poligonal precisa de pelo menos 3 pontos" last_msg_t = time.time() else: roi_name_input_active = True roi_name_buffer = "" roi_name_points_pending = list(current_polygon_points) last_msg = "Digite o nome da ROI na tela" last_msg_t = time.time() dt_loop = time.time() - t0 if dt_loop < 0.001: time.sleep(0.001) finally: if live_mode: cam.stop() cv2.destroyAllWindows() print("Fim da calibração dos sensores.") if __name__ == "__main__": main()