diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/camera_multispectral.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/camera_multispectral.py index 036842c5d..f17a1373f 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/camera_multispectral.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/camera_multispectral.py @@ -81,6 +81,10 @@ class CameraMultispectral: self.ultimo_meta = None self.ultimo_decoded = None + self._vida_lock = threading.RLock() + self._parando = False + self.client = None + self._ultimo_resultado_tensor = { "erro": "sem tensor em cache", "duracao": 0.0, @@ -99,6 +103,12 @@ class CameraMultispectral: "frame_valido": False, } + self._ultimo_pipeline_ia = { + "ok": False, + "timestamp": 0.0, + "motivos": ["pipeline IA ainda não validado"], + } + self.parametros = {} if module_calibration_json is None: @@ -203,9 +213,18 @@ class CameraMultispectral: ) except Exception as e: + client_parcial = getattr(self, "client", None) + + try: + if client_parcial is not None: + client_parcial.stop() + except Exception as stop_error: + self.mostrar_log(f"[CameraMultispectral] Erro ao limpar início parcial: {stop_error}") + self.client = None self.iniciado = False self.rodando = False + self.mostrar_log(f"[CameraMultispectral] Erro ao iniciar módulo: {e}") ContextoGlobalRedis.atualizar_ctx_dict( @@ -283,54 +302,79 @@ class CameraMultispectral: # ============================================================ def parar(self): - self.rodando = False - self.iniciado = False + with self._vida_lock: + if self._parando: + return + + self._parando = True + + client = self.client + imu = self.imu + stream = self.stream + + # Desacopla as referências primeiro. + self.client = None + self.imu = None + self.q_imu = None + self.stream = None + + self.tem_imu = False + self.rodando = False + self.iniciado = False try: - if self.imu is not None: - self.imu.parar() - except Exception: - pass - - self.imu = None - self.q_imu = None - self.tem_imu = False - - try: - if self.client is not None: - self.client.stop() - except Exception as e: try: - self.mostrar_log(f"[CameraMultispectral] Erro ao parar client: {e}") + if imu is not None: + imu.parar() except Exception: pass - self.client = None + try: + if client is not None: + client.stop() + except Exception as e: + self.mostrar_log( + f"[CameraMultispectral] Erro ao parar client: {e}" + ) - try: - if self.stream is not None and hasattr(self.stream, "parar"): - self.stream.parar() - except Exception: - pass + try: + if stream is not None and hasattr(stream, "parar"): + stream.parar() + except Exception: + pass - self.stream = None + with self._lock: + self.ultimo_tensor_multispec = None + self.ultimo_frame_rgb = None + self.ultimo_raw_multi = None + self.ultimo_meta = None + self.ultimo_decoded = None - with self._lock: - self.ultimo_tensor_multispec = None - self.ultimo_frame_rgb = None - self.ultimo_raw_multi = None - self.ultimo_meta = None - self.ultimo_decoded = None + self.timestamp_ultimo_tensor = None + self.timestamp_ultimo_frame_rgb = None + self.timestamp_ultimo_raw_multi = None - self.timestamp_ultimo_tensor = None - self.timestamp_ultimo_frame_rgb = None - self.timestamp_ultimo_raw_multi = None + self._ultimo_resultado_tensor = { + "erro": "pipeline parado", + "duracao": 0.0, + "frame_valido": False, + } - self._ultimo_resultado_tensor = { - "erro": "pipeline parado", - "duracao": 0.0, - "frame_valido": False, - } + self._ultimo_resultado_rgb = { + "erro": "pipeline parado", + "duracao": 0.0, + "frame_valido": False, + } + + self._ultimo_resultado_raw = { + "erro": "pipeline parado", + "duracao": 0.0, + "frame_valido": False, + } + + finally: + with self._vida_lock: + self._parando = False def _is_erro_fatal_depthai(self, erro): if erro is None: @@ -525,6 +569,9 @@ class CameraMultispectral: ts = time.time() dur = time.perf_counter() - t_total0 + meta = dict(meta or {}) + capture_perf = dict(meta.get("capture_perf", {}) or {}) + resultado = { "erro": None, "duracao": dur, @@ -532,8 +579,15 @@ class CameraMultispectral: "shape": list(tensor.shape), "dtype": str(tensor.dtype), "channels": ["R", "G", "B", "RE", "NIR"], - "sync_ok": bool(meta.get("sync_ok", True)) if isinstance(meta, dict) else True, - "sync_dt_ms": float(meta.get("sync_dt_ms", 0.0)) if isinstance(meta, dict) else 0.0, + + "frame_id": int(meta.get("frame_id", 0) or 0), + "async_packet_seq": int( + capture_perf.get("async_packet_seq", 0) or 0 + ), + + "sync_ok": bool(meta.get("sync_ok", True)), + "sync_dt_ms": float(meta.get("sync_dt_ms", 0.0) or 0.0), + "perf": { "origem_tensor": origem_tensor, "get_decoded_ms": t_get_decoded_ms, @@ -689,7 +743,7 @@ class CameraMultispectral: # Saúde # ============================================================ - def atualizar_saude(self): + def atualizar_saude(self, pipeline_ia=None): #self.mostrar_log(f"[{self.mx_id}] Atualizando saude {self.dispositivo.name} multispectral...") if self.imu is not None: @@ -708,6 +762,17 @@ class CameraMultispectral: "velocidade": "", "sync_dt_ms": 0.0 } + + if pipeline_ia is not None: + self._ultimo_pipeline_ia = dict(pipeline_ia) + pipeline_ia = dict(self._ultimo_pipeline_ia or {}) + pipeline_ia_informado = True + pipeline_ia_ok = ( + not pipeline_ia_informado + or bool(pipeline_ia.get("ok", False)) + ) + performance["pipeline_ia"] = pipeline_ia + imu = self.imu.get_dados() if self.imu else { "valido": False, "calibrado": False, @@ -820,8 +885,8 @@ class CameraMultispectral: saude += 50 if not frame_recente: - motivos.append("Tensor antigo") - saude -= 20 + motivos.append(f"Tensor sem atualização há {agora - ts_tensor:.2f}s") + saude = 0 dur = float(resultado.get("duracao", 0.0) or 0.0) if dur > 1.5: @@ -836,17 +901,34 @@ class CameraMultispectral: saude -= 10 motivos.append(f"Sincronismo alto: {sync_dt:.1f} ms") + + if pipeline_ia_informado and not pipeline_ia_ok: + motivos_pipeline = pipeline_ia.get("motivos", []) or [] + for motivo in motivos_pipeline: + texto = f"Pipeline IA: {motivo}" + if texto not in motivos: + motivos.append(texto) + saude = 0 + saude = min(max(saude, 0), 100) status_mod = StatusModulo.OPERANTE if not conectado: status_mod = StatusModulo.DESCONECTADO + elif not frame_recente: + status_mod = StatusModulo.FALHA + elif not pipeline_ia_ok: + status_mod = StatusModulo.FALHA elif saude <= 0: status_mod = StatusModulo.FALHA elif saude < 80: status_mod = StatusModulo.ALERTA - self.rodando = conectado and frame_recente + self.rodando = ( + conectado + and frame_recente + and pipeline_ia_ok + ) saude_geral = { "timestamp": agora, diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_manager.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_manager.py index e983aeb85..6a9142a44 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_manager.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_manager.py @@ -749,115 +749,132 @@ class OakFcc3Manager: if self.running: return - self.dev_info = self._resolve_device_info() - self.mx_id = self._device_id_from_info(self.dev_info) or self.mx_id + thread_anterior = self._capture_thread - self.device = dai.Device(self.dev_info) - self.pipeline = dai.Pipeline() + if (thread_anterior is not None and thread_anterior.is_alive()): + raise RuntimeError("Não é seguro iniciar novo pipeline: thread OakFcc3AsyncCapture anterior ainda está viva.") - features = self.device.getConnectedCameraFeatures() + try: + self.dev_info = self._resolve_device_info() + self.mx_id = self._device_id_from_info(self.dev_info) or self.mx_id - self.queues.clear() - self.buffers.clear() - self.camera_info.clear() - self.control_queues.clear() - self._last_raw_dims.clear() - self.aligned_geometry = None + self.device = dai.Device(self.dev_info) + self.pipeline = dai.Pipeline() - if self._is_multispec_mode(): - self._start_multispec_pipeline(features) - else: - for f in features: - socket = f.socket - socket_name = socket.name - if self.only_camera is not None and socket_name != self.only_camera: - continue + features = self.device.getConnectedCameraFeatures() - role = self.roles.get(socket_name, "unknown") + self.queues.clear() + self.buffers.clear() + self.camera_info.clear() + self.control_queues.clear() + self._last_raw_dims.clear() + self.aligned_geometry = None - print(f"[OAK] Criando câmera {socket_name} sensor={f.sensorName} role={role}") + if self._is_multispec_mode(): + self._start_multispec_pipeline(features) + else: + for f in features: + socket = f.socket + socket_name = socket.name + if self.only_camera is not None and socket_name != self.only_camera: + continue - cam, output = self._create_camera_node_classic( - socket=socket, - sensor_name=f.sensorName, - role=role, - ) + role = self.roles.get(socket_name, "unknown") - self.apply_initial_camera_controls_to_node(cam, socket_name) + print(f"[OAK] Criando câmera {socket_name} sensor={f.sensorName} role={role}") - xin_ctrl = self.pipeline.create(dai.node.XLinkIn) - xin_ctrl.setStreamName(f"{socket_name}_ctrl") - xin_ctrl.out.link(cam.inputControl) + cam, output = self._create_camera_node_classic( + socket=socket, + sensor_name=f.sensorName, + role=role, + ) - xout = self.pipeline.create(dai.node.XLinkOut) - xout.setStreamName(socket_name) - output.link(xout.input) + self.apply_initial_camera_controls_to_node(cam, socket_name) - cam_id = socket_name + xin_ctrl = self.pipeline.create(dai.node.XLinkIn) + xin_ctrl.setStreamName(f"{socket_name}_ctrl") + xin_ctrl.out.link(cam.inputControl) - self.queues[cam_id] = None - self.buffers[cam_id] = deque(maxlen=self.buffer_size) - self.control_queues[cam_id] = None + xout = self.pipeline.create(dai.node.XLinkOut) + xout.setStreamName(socket_name) + output.link(xout.input) - self.camera_info[cam_id] = { - "id": cam_id, - "socket": socket_name, - "sensor": f.sensorName, - "role": role, - } + cam_id = socket_name - self._validate_capture_mode() + self.queues[cam_id] = None + self.buffers[cam_id] = deque(maxlen=self.buffer_size) + self.control_queues[cam_id] = None - self._create_imu_node(self.pipeline) + self.camera_info[cam_id] = { + "id": cam_id, + "socket": socket_name, + "sensor": f.sensorName, + "role": role, + } - self.device.startPipeline(self.pipeline) + self._validate_capture_mode() - self.q_imu = None - self.tem_imu = False + self._create_imu_node(self.pipeline) - if self.has_imu_pipeline: - try: - self.q_imu = self.device.getOutputQueue( - name="imu", - maxSize=8, + self.device.startPipeline(self.pipeline) + + self.q_imu = None + self.tem_imu = False + + if self.has_imu_pipeline: + try: + self.q_imu = self.device.getOutputQueue( + name="imu", + maxSize=8, + blocking=False, + ) + self.tem_imu = True + print("[OAK] Fila IMU criada | maxSize=8") + except Exception as e: + self.q_imu = None + self.tem_imu = False + print(f"[WARN] Fila IMU indisponível: {e}") + + for cam_id in self.camera_info.keys(): + self.queues[cam_id] = self.device.getOutputQueue( + name=cam_id, + maxSize=self.buffer_size, blocking=False, ) - self.tem_imu = True - print("[OAK] Fila IMU criada | maxSize=8") - except Exception as e: - self.q_imu = None - self.tem_imu = False - print(f"[WARN] Fila IMU indisponível: {e}") - for cam_id in self.camera_info.keys(): - self.queues[cam_id] = self.device.getOutputQueue( - name=cam_id, - maxSize=self.buffer_size, - blocking=False, - ) + self.control_queues[cam_id] = self.device.getInputQueue( + name=f"{cam_id}_ctrl", + maxSize=4, + blocking=False, + ) - self.control_queues[cam_id] = self.device.getInputQueue( - name=f"{cam_id}_ctrl", - maxSize=4, - blocking=False, - ) + self.running = True - self.running = True + if not hasattr(self, "async_capture_enabled"): + self.async_capture_enabled = True + if not hasattr(self, "async_capture_mode"): + self.async_capture_mode = "latest" + if not hasattr(self, "async_capture_max_queue"): + self.async_capture_max_queue = 2 - if not hasattr(self, "async_capture_enabled"): - self.async_capture_enabled = True - if not hasattr(self, "async_capture_mode"): - self.async_capture_mode = "latest" - if not hasattr(self, "async_capture_max_queue"): - self.async_capture_max_queue = 2 - self._reset_async_capture_state() - self._start_async_capture_thread() - time.sleep(0.05) + self._reset_async_capture_state() + self._start_async_capture_thread() + time.sleep(0.05) + + except Exception: + # Garante limpeza mesmo se a falha acontecer + # no meio da construção do pipeline. + try: + self.stop() + except Exception: + pass + + raise def stop(self): - if not self.running: - return - + # Impede novas capturas imediatamente. + self.running = False + self._stop_async_capture_thread() try: @@ -874,16 +891,25 @@ class OakFcc3Manager: self.pipeline = None self.device = None + self.queues.clear() self.buffers.clear() self.camera_info.clear() self.control_queues.clear() self._last_raw_dims.clear() + self.aligned_geometry = None self.q_imu = None self.tem_imu = False self.has_imu_pipeline = False - self.running = False + + try: + with self._capture_cond: + self._latest_packet = None + self._packet_queue.clear() + self._capture_cond.notify_all() + except Exception: + pass def _is_fatal_depthai_error(self, erro): txt = str(erro) @@ -1734,8 +1760,28 @@ class OakFcc3Manager: self._capture_cond.notify_all() th = getattr(self, "_capture_thread", None) - if th is not None and th.is_alive(): + if ( + th is not None + and th.is_alive() + and th is not threading.current_thread() + ): th.join(timeout=1.0) + + if th.is_alive(): + try: + if self.device is not None: + self.device.close() + except Exception: + pass + + th.join(timeout=2.0) + + if th.is_alive(): + self._capture_thread_stats["last_error"] = ("thread de captura não encerrou após fechamento do device") + return False + + self._capture_thread = None + return True except Exception: pass @@ -1743,6 +1789,9 @@ class OakFcc3Manager: def _async_capture_loop(self): if not hasattr(self, "_capture_thread_stats"): + thread_anterior = self._capture_thread + if (thread_anterior is not None and thread_anterior.is_alive()): + raise RuntimeError("Não é seguro iniciar novo pipeline: thread OakFcc3AsyncCapture anterior ainda está viva.") self._reset_async_capture_state() self._capture_thread_stats["started"] = True diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/camera_manager.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/camera_manager.py index 1764ffee6..35cc34c33 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/camera_manager.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/camera_manager.py @@ -113,6 +113,14 @@ class CameraManager: self._tensor_ts = 0.0 self._tensor_consumido_ts = 0.0 + self.pipeline_ia_ok = False + self._ultimo_tensor_ok_ts = 0.0 + self._ultima_inferencia_ok_ts = 0.0 + self._ultima_deteccao_ok_ts = 0.0 + self._ultima_saude_pipeline_ia = {} + self._falhas_captura_consecutivas = 0 + self._primeira_falha_captura_ts = 0.0 + self._pred_cache = { "ts": 0.0, "tensor_ts": 0.0, @@ -189,14 +197,22 @@ class CameraManager: } def inicializar(self, mx_id): - if self.iniciando: - return - if mx_id is None: - return + return False with self._vida_lock: + if self.iniciando: + return False + + if ( + self.camera is not None + and self.operante + and self.pipeline_ia_ok + ): + return True + self.iniciando = True + self.pipeline_ia_ok = False try: self.mx_id = mx_id @@ -210,11 +226,9 @@ class CameraManager: if self.camera is None: self.mostrar_log(f"❌ Camera com ID {mx_id} não iniciada.") - return + return False - self.mostrar_log( - f"📷 Camera selecionada: {self.camera.modelo} - {self.camera.mx_id}" - ) + self.mostrar_log(f"📷 Camera selecionada: {self.camera.modelo} - {self.camera.mx_id}") self._reset_runtime_state() @@ -228,25 +242,41 @@ class CameraManager: self.operante = False self._em_warmup = True - warmup_ok = self._executar_warmup_camera_manager( - n_tensors=5, - n_infer=4, - n_detector=2, - timeout_s=8.0, - ) + warmup_ok = self._executar_warmup_camera_manager(n_ciclos=3, timeout_s=12.0) self._em_warmup = False if not warmup_ok: - self.mostrar_log( - "[weed][WARMUP] finalizou com alerta, liberando operação mesmo assim" - ) + self.mostrar_log("[weed][WARMUP] recuperação rejeitada: pipeline não concluiu validação ponta a ponta") + self.fechar_camera_manager("warmup incompleto após reconexão") + return False self.operante = True self._iniciar_loops_se_necessario(self.seg_config) - self.atualizar_saude_camera() + # A avaliação final precisa acontecer depois que a inicialização acabou. + self.iniciando = False + + pipeline_ia = self._avaliar_saude_pipeline_ia() + + if not pipeline_ia.get("ok", False): + self.mostrar_log( + "[weed] inicialização rejeitada após warmup: " + + "; ".join(pipeline_ia.get("motivos", [])) + ) + + self.fechar_camera_manager(motivo="pipeline IA inválido após warmup", camera_esperada=self.camera) + return False + + self.atualizar_saude_camera(pipeline_ia=pipeline_ia) + + return True + + except Exception as e: + self.mostrar_log(f"[weed] erro ao inicializar câmera: {e}") + self.fechar_camera_manager(motivo=f"falha de inicialização: {e}", camera_esperada=self.camera) + return False finally: self.iniciando = False @@ -325,37 +355,109 @@ class CameraManager: self._iniciar_loop_analise_continua(freq=freq_analise) self._loop_analise_iniciado = True - def fechar_camera_manager(self, motivo=""): - with self._vida_lock: - self._fechando_camera = True + def _registrar_falha_captura( + self, + camera_atual, + erro, + ): + if camera_atual is not self.camera: + return False - try: - if self.camera is not None: - self.camera.parar() - except Exception as e: - self.mostrar_log(f"[weed] Erro ao fechar câmera: {e}") + agora = time.monotonic() + + self._falhas_captura_consecutivas += 1 + + if self._primeira_falha_captura_ts <= 0: + self._primeira_falha_captura_ts = agora + + duracao_falha = ( + agora - self._primeira_falha_captura_ts + ) + + try: + fatal = camera_atual._is_erro_fatal_depthai(erro) + except Exception: + fatal = False + + deve_reiniciar = ( + fatal + or self._falhas_captura_consecutivas >= 5 + or duracao_falha >= 3.0 + ) + + if not deve_reiniciar: + self.mostrar_log( + f"[weed][CAPTURE] falha transitória " + f"{self._falhas_captura_consecutivas}/5: {erro}" + ) + return False + + self.mostrar_log( + f"[weed][CAPTURE] reiniciando câmera | " + f"falhas={self._falhas_captura_consecutivas} " + f"duração={duracao_falha:.2f}s " + f"erro={erro}" + ) + + self.fechar_camera_manager( + motivo=f"watchdog de captura: {erro}", + camera_esperada=camera_atual, + ) + + return True + + def fechar_camera_manager(self, motivo="", camera_esperada=None): + with self._vida_lock: + # Um erro atrasado da câmera antiga jamais pode fechar a nova. + if (camera_esperada is not None and self.camera is not camera_esperada): + self.mostrar_log("[weed] fechamento ignorado: a câmera ativa já foi substituída") + return False + + camera_alvo = self.camera self.camera = None self.operante = False - self._fechando_camera = False + self.pipeline_ia_ok = False + self._fechando_camera = True + + self._reset_runtime_state() + + try: + self._forcar_bicos_off(motivo=f"câmera fechada: {motivo}", intervalo_min_s=0.0) + except Exception: + pass + + try: + if camera_alvo is not None: + camera_alvo.parar() + + except Exception as e: + self.mostrar_log(f"[weed] Erro ao fechar câmera: {e}") + + finally: + self._fechando_camera = False self.mostrar_log(f"[weed] Camera Manager fechado: {motivo}") + return True + # ============================================================ # Warmup # ============================================================ def _executar_warmup_camera_manager( self, - n_tensors=5, - n_infer=4, - n_detector=2, - timeout_s=8.0, + n_ciclos=3, + timeout_s=12.0, ): - if self.camera is None: + camera_alvo = self.camera + + if camera_alvo is None: return False - self.mostrar_log("[weed][WARMUP] iniciando warmup do CameraManager...") + self.mostrar_log( + f"[weed][WARMUP] iniciando {n_ciclos} ciclos ponta a ponta..." + ) operante_anterior = self.operante warmup_anterior = self._em_warmup @@ -363,121 +465,178 @@ class CameraManager: self.operante = False self._em_warmup = True - tensor5 = None - res = None - predictions = None + deadline = time.monotonic() + float(timeout_s) + ultimo_frame_id = 0 + ciclos_ok = 0 + ultimo_ciclo = None try: - tensor_ok = 0 - t0 = time.time() + while ciclos_ok < n_ciclos and time.monotonic() < deadline: + if self.camera is not camera_alvo: + self.mostrar_log( + "[weed][WARMUP] câmera foi substituída durante o warmup" + ) + return False - while time.time() - t0 < timeout_s and tensor_ok < n_tensors: try: + # 1. Frame/tensor realmente novo t_cap0 = time.perf_counter() - tensor5, res = self.camera.requisitar_tensor_multispec(force=True) + + tensor5, res = camera_alvo.requisitar_tensor_multispec( + force=True + ) + t_cap1 = time.perf_counter() - if tensor5 is not None: - tensor_ok += 1 - - with self._lock_tensor: - self._tensor_pronto = tensor5 - self._tensor_res = res - self._tensor_ts = time.time() + erro = res.get("erro") if isinstance(res, dict) else None + if tensor5 is None or erro: self.mostrar_log( - f"[weed][WARMUP] tensor {tensor_ok}/{n_tensors} " - f"total={(t_cap1 - t_cap0) * 1000.0:.1f}ms " - f"shape={getattr(tensor5, 'shape', None)}" + f"[weed][WARMUP] captura inválida: {erro}" ) + time.sleep(0.05) + continue - except Exception as e: - self.mostrar_log(f"[weed][WARMUP] tensor falhou: {e}") + frame_id = int( + res.get("frame_id") + or res.get("async_packet_seq") + or 0 + ) - time.sleep(0.03) + if frame_id <= 0: + self.mostrar_log( + "[weed][WARMUP] frame sem identidade válida" + ) + time.sleep(0.03) + continue - if tensor5 is None: - self.mostrar_log("[weed][WARMUP] abortado: sem tensor válido") - return False + if frame_id <= ultimo_frame_id: + self.mostrar_log( + f"[weed][WARMUP] frame não avançou: " + f"anterior={ultimo_frame_id} atual={frame_id}" + ) + time.sleep(0.03) + continue - infer_ok = 0 - - for i in range(max(0, n_infer)): - try: + # 2. Inferência sobre este frame t_inf0 = time.perf_counter() + predictions = self.model_svc.infer_tensor_fast( tensor5, keep_probs=False, ) + t_inf1 = time.perf_counter() - infer_full = getattr(self.model_svc, "_ultimo_predictions_full", {}) or {} + if predictions is None: + self.mostrar_log( + f"[weed][WARMUP] inferência inválida no frame {frame_id}" + ) + continue - infer_forward_ms = infer_full.get( - "forward_ms", - infer_full.get("infer_ms", None), + infer_full = ( + getattr( + self.model_svc, + "_ultimo_predictions_full", + {}, + ) + or {} + ) + + infer_ms = (t_inf1 - t_inf0) * 1000.0 + infer_gpu_ms = infer_full.get( + "forward_ms", + infer_full.get("infer_ms", 0.0), ) - infer_prepare_ms = infer_full.get("prepare_ms", 0.0) - infer_post_ms = infer_full.get("post_ms", 0.0) self._atualizar_fps_inferencia( - infer_ms=(t_inf1 - t_inf0) * 1000.0, - infer_gpu_ms=infer_forward_ms, + infer_ms=infer_ms, + infer_gpu_ms=infer_gpu_ms, ) - if predictions is not None: - infer_ok += 1 - self._ultimo_raw_base = tensor5 - self._ultimo_raw_input = tensor5 - self._ultimo_predictions = predictions - self._ultimo_predictions_full = infer_full - - self.mostrar_log( - f"[weed][WARMUP] infer {i + 1}/{n_infer} " - f"total={(t_inf1 - t_inf0) * 1000.0:.1f}ms " - f"prep={infer_prepare_ms:.1f}ms " - f"fwd={infer_forward_ms if infer_forward_ms is not None else -1:.1f}ms " - f"post={infer_post_ms:.1f}ms" - ) - - except Exception as e: - self.mostrar_log(f"[weed][WARMUP] infer falhou: {e}") - - time.sleep(0.03) - - if predictions is None: - self.mostrar_log("[weed][WARMUP] sem predictions válidas") - return False - - det_ok = 0 - - for i in range(max(0, n_detector)): - try: + # 3. Detecção sobre a prediction deste mesmo frame t_det0 = time.perf_counter() analise_completa = self.detectar_ervas(predictions) t_det1 = time.perf_counter() - if isinstance(analise_completa, dict): - det_ok += 1 - analise = analise_completa.get("dados_visuais", {}) - self._ultima_analise = analise_completa.copy() - self._set_pub_cache("analise", analise) + if not isinstance(analise_completa, dict): + self.mostrar_log( + f"[weed][WARMUP] detecção inválida no frame {frame_id}" + ) + continue + + agora = time.time() + ciclos_ok += 1 + ultimo_frame_id = frame_id + + ultimo_ciclo = { + "tensor": tensor5, + "res": res, + "tensor_ts": agora, + "predictions": predictions, + "pred_ts": agora, + "infer_full": infer_full, + "analise": analise_completa, + } self.mostrar_log( - f"[weed][WARMUP] detector {i + 1}/{n_detector} " - f"total={(t_det1 - t_det0) * 1000.0:.1f}ms" + f"[weed][WARMUP] ciclo {ciclos_ok}/{n_ciclos} OK | " + f"frame_id={frame_id} " + f"captura={(t_cap1 - t_cap0) * 1000.0:.1f}ms " + f"infer={infer_ms:.1f}ms " + f"det={(t_det1 - t_det0) * 1000.0:.1f}ms" ) except Exception as e: - self.mostrar_log(f"[weed][WARMUP] detector falhou: {e}") + self.mostrar_log( + f"[weed][WARMUP] ciclo falhou: " + f"{type(e).__name__}: {e}" + ) + time.sleep(0.05) - time.sleep(0.03) + if ciclos_ok != n_ciclos or ultimo_ciclo is None: + self.mostrar_log( + f"[weed][WARMUP] rejeitado | " + f"ciclos_ok={ciclos_ok}/{n_ciclos}" + ) + return False + + # Publica o estado nos caches internos, mas marca como já consumido. + # Assim o warmup não gera uma análise operacional artificial. + with self._lock_tensor: + self._tensor_pronto = ultimo_ciclo["tensor"] + self._tensor_res = ultimo_ciclo["res"] + self._tensor_ts = ultimo_ciclo["tensor_ts"] + self._tensor_consumido_ts = ultimo_ciclo["tensor_ts"] + + with self._pred_lock: + self._pred_cache = { + "ts": ultimo_ciclo["pred_ts"], + "tensor_ts": ultimo_ciclo["tensor_ts"], + "predictions": ultimo_ciclo["predictions"], + "res": ultimo_ciclo["res"], + "infer_ms": 0.0, + "infer_gpu_ms": 0.0, + "infer_prepare_ms": 0.0, + "infer_post_ms": 0.0, + "fps_model": float(self._fps_infer_ema or 0.0), + } + self._pred_consumido_ts = ultimo_ciclo["pred_ts"] + + self._ultimo_raw_base = ultimo_ciclo["tensor"] + self._ultimo_raw_input = ultimo_ciclo["tensor"] + self._ultimo_predictions = ultimo_ciclo["predictions"] + self._ultimo_predictions_full = ultimo_ciclo["infer_full"] + self._ultima_analise = ultimo_ciclo["analise"].copy() + + agora = time.time() + self._ultimo_tensor_ok_ts = agora + self._ultima_inferencia_ok_ts = agora + self._ultima_deteccao_ok_ts = agora self.mostrar_log( - f"[weed][WARMUP] finalizado | " - f"tensor_ok={tensor_ok}/{n_tensors} " - f"infer_ok={infer_ok}/{n_infer} " - f"det_ok={det_ok}/{n_detector}" + f"[weed][WARMUP] aprovado | " + f"{n_ciclos} frames distintos processados ponta a ponta" ) return True @@ -491,6 +650,7 @@ class CameraManager: self._em_warmup = warmup_anterior self.perf = VisualPerfMonitor(janela=180) + if self.camera is not None: self.camera.perf = self.perf @@ -498,25 +658,19 @@ class CameraManager: # Saúde / config dinâmica # ============================================================ - def atualizar_saude_camera(self): + def atualizar_saude_camera(self, pipeline_ia=None): self._ultima_saude_ts = time.time() try: if self.camera is not None: - self.camera.atualizar_saude() + if pipeline_ia is None: + pipeline_ia = self._avaliar_saude_pipeline_ia() + + self.camera.atualizar_saude(pipeline_ia=pipeline_ia) + elif self.mx_id is not None: from camera_worker.manager import definir_saude_camera - - definir_saude_camera( - self.mx_id, - StatusModulo.DESCONECTADO, - 0, - ["desconectado"], - False, - {}, - disp=T_Code.Cam, - conectado=False, - ) + definir_saude_camera(self.mx_id, StatusModulo.DESCONECTADO, 0, ["desconectado"], False, {}, disp=T_Code.Cam, conectado=False) except Exception as e: self.mostrar_log(f"[saude] erro: {e}") @@ -554,6 +708,75 @@ class CameraManager: except Exception as e: self.mostrar_log(f"[weed] erro ao atualizar config dinâmica: {e}") + def _avaliar_saude_pipeline_ia(self): + agora = time.time() + + cfg = self.seg_config or {} + + timeout_tensor = float( + cfg.get("watchdog_tensor_timeout_s", 2.0) + ) + timeout_inferencia = float( + cfg.get("watchdog_inferencia_timeout_s", 2.0) + ) + timeout_deteccao = float( + cfg.get("watchdog_deteccao_timeout_s", 2.0) + ) + + def idade(ts): + if not ts: + return float("inf") + return max(0.0, agora - float(ts)) + + idade_tensor = idade(self._ultimo_tensor_ok_ts) + idade_inferencia = idade(self._ultima_inferencia_ok_ts) + idade_deteccao = idade(self._ultima_deteccao_ok_ts) + + motivos = [] + + if self.camera is None: + motivos.append("câmera indisponível") + elif self.iniciando: + motivos.append("câmera inicializando") + elif self._em_warmup: + motivos.append("pipeline em warmup") + elif not self.operante: + motivos.append("CameraManager não operante") + else: + if idade_tensor > timeout_tensor: + motivos.append( + f"tensor antigo: {idade_tensor:.2f}s" + ) + + if idade_inferencia > timeout_inferencia: + motivos.append( + f"inferência antiga: {idade_inferencia:.2f}s" + ) + + if idade_deteccao > timeout_deteccao: + motivos.append( + f"detecção antiga: {idade_deteccao:.2f}s" + ) + + ok = len(motivos) == 0 + + resultado = { + "ok": ok, + "timestamp": agora, + "motivos": motivos, + "idade_tensor_s": idade_tensor, + "idade_inferencia_s": idade_inferencia, + "idade_deteccao_s": idade_deteccao, + "timeout_tensor_s": timeout_tensor, + "timeout_inferencia_s": timeout_inferencia, + "timeout_deteccao_s": timeout_deteccao, + } + + self.pipeline_ia_ok = ok + self._ultima_saude_pipeline_ia = resultado + + return resultado + # ============================================================ # Caches # ============================================================ @@ -889,55 +1112,92 @@ class CameraManager: def _iniciar_loop_captura_tensor(self, freq=25.0): def loop(): + periodo = 1.0 / max(float(freq), 0.1) + while True: t0_wall = time.time() - t0 = time.perf_counter() + t0_perf = time.perf_counter() + camera_atual = None try: - if self.camera is None: - time.sleep(1.0) + if (self.iniciando or self._em_warmup or not self.operante or self.camera is None): + time.sleep(0.05) continue - tensor5, res = self.camera.requisitar_tensor_multispec(force=True) + camera_atual = self.camera - if isinstance(res, dict) and res.get("erro"): - erro = res.get("erro") - if self.camera._is_erro_fatal_depthai(erro): - self.fechar_camera_manager(f"falha fatal DepthAI: {erro}") - return + tensor5, res = (camera_atual.requisitar_tensor_multispec(force=True)) - if tensor5 is not None: - agora = time.time() - perf = res.get("perf", {}) if isinstance(res, dict) else {} + erro = (res.get("erro") if isinstance(res, dict) else None) - with self._lock_tensor: - self._tensor_pronto = tensor5 - self._tensor_res = res - self._tensor_ts = agora + if erro: + reiniciou = self._registrar_falha_captura(camera_atual, erro) - t1 = time.perf_counter() + time.sleep(0.50 if reiniciou else 0.05) + continue - self.perf.tick( - "tensor", - latencia_ms=(t1 - t0) * 1000.0, - tensor_core_ms=float(perf.get("total_ms", 0.0) or 0.0), - get_decoded_ms=float(perf.get("get_decoded_ms", 0.0) or 0.0), - build_ms=float(perf.get("build_ms", 0.0) or 0.0), - validate_ms=float(perf.get("validate_ms", 0.0) or 0.0), - post_ms=float(perf.get("post_ms", 0.0) or 0.0), - preview_ms=float(perf.get("preview_ms", 0.0) or 0.0), - frame_ts=agora, - idade_frame_ms=0.0, - ) + if tensor5 is None: + self.perf.inc("tensor_none") + + reiniciou = self._registrar_falha_captura(camera_atual, "tensor None sem erro explícito") + + time.sleep(0.50 if reiniciou else 0.05) + continue + + # Uma captura válida limpa a sequência de falhas. + self._falhas_captura_consecutivas = 0 + self._primeira_falha_captura_ts = 0.0 + + agora = time.time() + perf = ( + res.get("perf", {}) + if isinstance(res, dict) + else {} + ) + + with self._lock_tensor: + self._tensor_pronto = tensor5 + self._tensor_res = res + self._tensor_ts = agora + + self._ultimo_tensor_ok_ts = agora + + t1_perf = time.perf_counter() + + self.perf.tick( + "tensor", + latencia_ms=(t1_perf - t0_perf) * 1000.0, + tensor_core_ms=float(perf.get("total_ms", 0.0) or 0.0), + get_decoded_ms=float(perf.get("get_decoded_ms", 0.0) or 0.0), + build_ms=float(perf.get("build_ms", 0.0) or 0.0), + validate_ms=float(perf.get("validate_ms", 0.0) or 0.0), + post_ms=float(perf.get("post_ms", 0.0) or 0.0), + preview_ms=float(perf.get("preview_ms", 0.0) or 0.0), + frame_ts=agora, + idade_frame_ms=0.0, + ) except Exception as e: - self.mostrar_log(f"[CAPTURE] erro: {e}") + self.mostrar_log(f"[weed][CAPTURE] Erro inesperado: {type(e).__name__}: {e}") + + if camera_atual is None: + camera_atual = self.camera + + reiniciou = False + + if camera_atual is not None: + reiniciou = self._registrar_falha_captura(camera_atual, e) + + time.sleep(0.50 if reiniciou else 0.05) finally: - lat = time.time() - t0_wall - time.sleep(max(0.0, (1.0 / freq) - lat)) + gasto = time.time() - t0_wall + restante = periodo - gasto - threading.Thread(target=loop, daemon=True).start() + if restante > 0: + time.sleep(restante) + + threading.Thread(target=loop, name="WeedTensorCapture", daemon=True).start() def _iniciar_loop_inferencia(self, freq=25.0): def loop(): @@ -1010,6 +1270,7 @@ class CameraManager: "infer_post_ms": float(infer_post_ms or 0.0), "fps_model": float(self._fps_infer_ema or 0.0), } + self._ultima_inferencia_ok_ts = pred_ts self._ultimo_raw_base = tensor5 self._ultimo_raw_input = tensor5 @@ -1116,7 +1377,7 @@ class CameraManager: "infer_ms": pred_cache.get("infer_ms", 0.0), "infer_gpu_ms": pred_cache.get("infer_gpu_ms", 0.0), "analise": analise_convertida, - }) + }, ts_origem=pred_ts) t_pub_analise1 = time.perf_counter() # ==================================================== @@ -1165,6 +1426,9 @@ class CameraManager: self._ultima_analise = analise_completa.copy() + # Só considera o ciclo válido quando análise, gate e comando já foram preparados com sucesso. + self._ultima_deteccao_ok_ts = time.time() + t_loop1 = time.perf_counter() total_ms = (t_loop1 - t_loop0) * 1000.0 self._ultimo_loop_analise_fps = 1000.0 / max(total_ms, 1e-6) @@ -1360,52 +1624,81 @@ class CameraManager: def _iniciar_loop_analise_continua(self, freq=15.0): def loop(): + ultimo_health_ts = 0.0 + ultimo_perf_ts = 0.0 + while True: t0 = time.time() try: - if self.camera is None: - time.sleep(0.5) + camera_atual = self.camera + + if camera_atual is None: + self.pipeline_ia_ok = False + self._garantir_bicos_off_se_nao_permitido(intervalo_min_s=0.20) + time.sleep(0.20) continue self._verificar_desconexao_camera() agora = time.time() - if not hasattr(self, "_ultimo_perf_publish"): - self._ultimo_perf_publish = 0.0 - if agora - self._ultimo_perf_publish >= 1.0: - self._ultimo_perf_publish = agora + # Saúde completa a cada 500 ms. + if agora - ultimo_health_ts >= 0.50: + ultimo_health_ts = agora + + pipeline_ia = self._avaliar_saude_pipeline_ia() + + if not pipeline_ia.get("ok", False): + self._forcar_bicos_off( + motivo=( + "pipeline IA inválido: " + + "; ".join( + pipeline_ia.get("motivos", []) + ) + ), + intervalo_min_s=0.20, + ) + + self.atualizar_saude_camera(pipeline_ia=pipeline_ia) + + if agora - ultimo_perf_ts >= 1.0: + ultimo_perf_ts = agora self._publicar_e_logar_performance() except Exception as e: self.mostrar_log(f"Erro no loop supervisor weed: {e}") + try: + self._forcar_bicos_off(motivo="erro no supervisor do Weed Worker", intervalo_min_s=0.0) + except Exception: + pass + finally: lat = time.time() - t0 - time.sleep(max(0.0, (1.0 / freq) - lat)) + time.sleep(max(0.0, (1.0 / max(freq, 0.1)) - lat)) - threading.Thread(target=loop, daemon=True).start() + threading.Thread(target=loop, name="WeedSupervisor", daemon=True).start() def _verificar_desconexao_camera(self): - status = StatusModulo( - (self.camera.ultima_saude or {}).get( - "status", - StatusModulo.DESCONECTADO.value, - ) - ) + camera_atual = self.camera - ts_status = (self.camera.ultima_saude or {}).get("timestamp", 0) + if camera_atual is None: + return - if status == StatusModulo.DESCONECTADO and (time.time() - ts_status) > 10.0: - try: - if getattr(self.camera, "imu", None): - self.camera.imu.parar() - except Exception: - pass + try: + status = StatusModulo((camera_atual.ultima_saude or {}).get("status", StatusModulo.DESCONECTADO.value)) + except Exception: + status = StatusModulo.DESCONECTADO - self.camera = None - self.operante = False + ts_status = float((camera_atual.ultima_saude or {}).get("timestamp", 0.0) or 0.0) + + if ( + status == StatusModulo.DESCONECTADO + and ts_status > 0 + and (time.time() - ts_status) > 10.0 + ): + self.fechar_camera_manager(motivo="câmera permaneceu desconectada por mais de 10s", camera_esperada=camera_atual) # ============================================================ # Detecção / atuação @@ -1426,15 +1719,19 @@ class CameraManager: # Publicação / cache Redis # ============================================================ - def _set_pub_cache(self, chave, valor): + def _set_pub_cache(self, chave, valor, ts_origem=None): try: agora = time.time() + ts_origem = agora if ts_origem is None else float(ts_origem) with self._pub_lock: self._pub_cache[chave] = valor - self._pub_cache[f"ts_{chave}"] = agora + self._pub_cache[f"ts_{chave}"] = ts_origem self._pub_cache[f"dirty_{chave}"] = True - self._pub_cache["ts_analise"] = agora + + # Só uma análise real pode avançar ts_analise. + if chave == "analise": + self._pub_cache["ts_analise"] = ts_origem except Exception as e: self.mostrar_log(f"[weed] erro ao atualizar pub_cache[{chave}]: {e}") @@ -1444,7 +1741,8 @@ class CameraManager: cache = dict(self._pub_cache) payload_weed = { - "ts_analise": time.time(), + "ts_publicacao": time.time(), + "ts_analise": float(cache.get("ts_analise", 0.0) or 0.0), } payload_controle = None @@ -1634,10 +1932,11 @@ class CameraManager: ) camera_ok = ( - self.camera is not None and - self.operante and - not self.iniciando and - not self._em_warmup + self.camera is not None + and self.operante + and self.pipeline_ia_ok + and not self.iniciando + and not self._em_warmup ) retornando_base = ( @@ -2177,5 +2476,4 @@ class CameraManager: def _finalizar_salvamento_posprocessamento(self): with self._posproc_save_lock: - self._posproc_salvando = False - \ No newline at end of file + self._posproc_salvando = False \ No newline at end of file