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

robustez no camera manager do weed e multiesctral

This commit is contained in:
Diego Freitas 2026-07-14 19:14:54 -03:00
parent a59bee710d
commit 4ee8a1e76a
5 changed files with 662 additions and 425 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

366
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/camera_multispectral.py
View File

@ -83,6 +83,7 @@ class CameraMultispectral:
self._vida_lock = threading.RLock() self._vida_lock = threading.RLock()
self._parando = False self._parando = False
self._falha_fatal_pendente = False
self.client = None self.client = None
self._ultimo_resultado_tensor = { self._ultimo_resultado_tensor = {
@ -452,15 +453,22 @@ class CameraMultispectral:
pass pass
def _falha_fatal_depthai(self, motivo): def _falha_fatal_depthai(self, motivo):
self.mostrar_log(f"[CameraMultispectral] Falha fatal DepthAI. Fechando pipeline: {motivo}") """
Marca a falha, mas não fecha o dai.Device nesta camada.
O CameraManager é o único dono do ciclo de vida e fará o fechamento.
Isso evita CameraMultispectral.parar() e CameraManager.fechar_camera_manager()
concorrendo sobre o mesmo objeto nativo.
"""
if self._falha_fatal_pendente:
return
self._falha_fatal_pendente = True
self.mostrar_log(
f"[CameraMultispectral] Falha fatal DepthAI detectada: {motivo}"
)
self._marcar_desconectada(motivo) self._marcar_desconectada(motivo)
try:
self.parar()
except Exception as e:
self.mostrar_log(f"[CameraMultispectral] Erro ao parar após falha fatal: {e}")
# ============================================================ # ============================================================
# Captura principal # Captura principal
# ============================================================ # ============================================================
@ -567,6 +575,7 @@ class CameraMultispectral:
t_preview_ms = 0.0 t_preview_ms = 0.0
ts = time.time() ts = time.time()
ts_raw_perf = time.perf_counter()
dur = time.perf_counter() - t_total0 dur = time.perf_counter() - t_total0
meta = dict(meta or {}) meta = dict(meta or {})
@ -606,6 +615,24 @@ class CameraMultispectral:
self.timestamp_ultimo_tensor = ts self.timestamp_ultimo_tensor = ts
self._ultimo_resultado_tensor = resultado self._ultimo_resultado_tensor = resultado
# O mesmo pacote RAW usado na inferência alimenta o cache científico.
# O salvamento não precisa consumir outra vez o cursor DepthAI.
if isinstance(frame, dict) and frame:
# Arrays do pacote são imutáveis após a captura.
# Mantemos referências e copiamos somente no salvamento.
self.ultimo_raw_multi = dict(frame)
self.timestamp_ultimo_raw_multi = ts_raw_perf
self._ultimo_resultado_raw = {
"erro": None,
"duracao": dur,
"frame_valido": True,
"cameras": list(frame.keys()),
"sync_ok": bool(meta.get("sync_ok", True)),
"sync_dt_ms": float(meta.get("sync_dt_ms", 0.0) or 0.0),
"frame_id": int(meta.get("frame_id", 0) or 0),
"origem": "cache_da_captura_operacional",
}
self.ultimo_frame_rgb = preview_bgr self.ultimo_frame_rgb = preview_bgr
self.timestamp_ultimo_frame_rgb = ts self.timestamp_ultimo_frame_rgb = ts
self._ultimo_resultado_rgb = { self._ultimo_resultado_rgb = {
@ -679,50 +706,28 @@ class CameraMultispectral:
} }
def requisitar_frame_raw_multi(self, force: bool = False, max_age_s: float = None): def requisitar_frame_raw_multi(self, force: bool = False, max_age_s: float = None):
""" """Retorna cópia do último RAW capturado pelo fluxo operacional."""
Debug opcional.
Retorna dict por câmera:
{"CAM_A": raw, "CAM_B": raw, "CAM_C": raw}, resultado
O weed_worker normal não precisa usar isso.
"""
try: try:
if max_age_s is None or float(max_age_s) <= 0:
max_age_s = max(1.0, self._cache_max_age_s)
agora = time.perf_counter() agora = time.perf_counter()
if max_age_s is None:
max_age_s = self._cache_max_age_s
with self._lock: with self._lock:
cache_ok = ( if self.ultimo_raw_multi is None or self.timestamp_ultimo_raw_multi is None:
self.ultimo_raw_multi is not None raise RuntimeError("cache RAW ainda não disponível")
and self.timestamp_ultimo_raw_multi is not None
and (agora - self.timestamp_ultimo_raw_multi) < max_age_s
)
if cache_ok and not force: idade = agora - self.timestamp_ultimo_raw_multi
return self.ultimo_raw_multi, dict(self._ultimo_resultado_raw) if idade > float(max_age_s):
raise RuntimeError(f"cache RAW antigo: {idade:.2f}s")
if self.client is None: raw_frame = {
raise RuntimeError("OakFcc3Client não inicializado") cam_id: arr.copy()
for cam_id, arr in self.ultimo_raw_multi.items()
t0 = time.perf_counter() }
raw_frame, raw_meta = self.client.get_next_raw_frame(timeout=self.timeout_s) resultado = dict(self._ultimo_resultado_raw)
ts = time.perf_counter() resultado["cache_age_s"] = float(idade)
dur = ts - t0 resultado["force_ignorado"] = bool(force)
resultado = {
"erro": None,
"duracao": dur,
"frame_valido": bool(raw_frame),
"cameras": list(raw_frame.keys()) if isinstance(raw_frame, dict) else [],
"sync_ok": bool(raw_meta.get("sync_ok", True)) if isinstance(raw_meta, dict) else True,
"sync_dt_ms": float(raw_meta.get("sync_dt_ms", 0.0)) if isinstance(raw_meta, dict) else 0.0,
}
with self._lock:
self.ultimo_raw_multi = raw_frame
self.ultimo_meta = raw_meta
self.timestamp_ultimo_raw_multi = ts
self._ultimo_resultado_raw = resultado
return raw_frame, resultado return raw_frame, resultado
@ -732,11 +737,9 @@ class CameraMultispectral:
"duracao": 0.0, "duracao": 0.0,
"frame_valido": False, "frame_valido": False,
} }
with self._lock: with self._lock:
self._ultimo_resultado_raw = resultado self._ultimo_resultado_raw = resultado
self.mostrar_log(f"[CameraMultispectral] RAW cache indisponível: {e}")
self.mostrar_log(f"[CameraMultispectral] Erro ao requisitar RAW multi: {e}")
return None, resultado return None, resultado
# ============================================================ # ============================================================
@ -744,33 +747,35 @@ class CameraMultispectral:
# ============================================================ # ============================================================
def atualizar_saude(self, pipeline_ia=None): 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: if self.imu is not None:
self.imu.atualizar_saude() self.imu.atualizar_saude()
conectado_ctx = (ContextoGlobalRedis.get_cameras() or {}).get(self.mx_id) is not None conectado_ctx = (ContextoGlobalRedis.get_cameras() or {}).get(self.mx_id) is not None
conectado = bool(self.iniciado and self.client is not None)
motivos = [] motivos = []
saude = 50 saude = 50
status = self._safe_get_status()
running = bool(status.get("running", False))
conectado = bool(
self.iniciado
and self.client is not None
and running
and not self._falha_fatal_pendente
)
performance = { performance = {
"temperatura": 0.0, "temperatura": 0.0,
"memoria_usada": 0.0, "memoria_usada": 0.0,
"executando": False, "executando": running,
"velocidade": "", "velocidade": "",
"sync_dt_ms": 0.0 "sync_dt_ms": 0.0,
"async_capture": status.get("async_capture") or {},
} }
if pipeline_ia is not None: if pipeline_ia is not None:
self._ultimo_pipeline_ia = dict(pipeline_ia) self._ultimo_pipeline_ia = dict(pipeline_ia)
pipeline_ia = dict(self._ultimo_pipeline_ia or {}) pipeline_ia = dict(self._ultimo_pipeline_ia or {})
pipeline_ia_informado = True pipeline_ia_ok = bool(pipeline_ia.get("ok", False))
pipeline_ia_ok = (
not pipeline_ia_informado
or bool(pipeline_ia.get("ok", False))
)
performance["pipeline_ia"] = pipeline_ia performance["pipeline_ia"] = pipeline_ia
imu = self.imu.get_dados() if self.imu else { imu = self.imu.get_dados() if self.imu else {
@ -794,59 +799,55 @@ class CameraMultispectral:
resultado = dict(self._ultimo_resultado_tensor) resultado = dict(self._ultimo_resultado_tensor)
try: try:
status = self._safe_get_status() metrics = (
running = bool(status.get("running", False)) self.client.get_device_metrics()
performance["executando"] = running if self.client is not None
else {"ok": False, "error": "client indisponível"}
)
performance["device_metrics"] = metrics
# Tenta chegar no device real por baixo do service/manager. temp = float(metrics.get("temperature_average_c", 0.0) or 0.0)
dev = None performance["temperatura"] = temp
try:
dev = self.client.svc.manager.device
except Exception:
dev = None
if dev is not None: if temp >= 80:
try: motivos.append(f"Temperatura crítica: {temp:.1f} °C")
temp = float(dev.getChipTemperature().average) saude -= 30
performance["temperatura"] = temp elif temp >= 70:
motivos.append(f"Temperatura elevada: {temp:.1f} °C")
saude -= 15
elif temp >= 60:
motivos.append(f"Temperatura acima do ideal: {temp:.1f} °C")
saude -= 5
if temp >= 80: ddr = float(metrics.get("ddr_used_bytes", 0) or 0) / 1024.0 / 1024.0
motivos.append(f"Temperatura crítica: {temp:.1f} °C") performance["memoria_usada"] = ddr
saude -= 30
elif temp >= 70:
motivos.append(f"Temperatura elevada: {temp:.1f} °C")
saude -= 15
elif temp >= 60:
motivos.append(f"Temperatura acima do ideal: {temp:.1f} °C")
saude -= 5
except Exception:
pass
try: if ddr >= 500:
ddr = float(dev.getDdrMemoryUsage().used) / 1024.0 / 1024.0 motivos.append(f"Memória DDR crítica: {ddr:.1f} MB")
performance["memoria_usada"] = ddr saude -= 30
elif ddr >= 400:
motivos.append(f"Memória DDR elevada: {ddr:.1f} MB")
saude -= 15
elif ddr >= 300:
motivos.append(f"Memória DDR acima do ideal: {ddr:.1f} MB")
saude -= 5
if ddr >= 500: speed = str(metrics.get("usb_speed", "") or "")
motivos.append(f"Memória DDR crítica: {ddr:.1f} MB") performance["velocidade"] = speed
saude -= 30 if speed and speed.lower() not in ("super", "superplus"):
elif ddr >= 400: motivos.append(f"USB lenta: {speed}")
motivos.append(f"Memória DDR elevada: {ddr:.1f} MB") saude -= 15
saude -= 15
elif ddr >= 300:
motivos.append(f"Memória DDR acima do ideal: {ddr:.1f} MB")
saude -= 5
except Exception:
pass
try: async_status = metrics.get("async_capture") or performance["async_capture"]
speed = dev.getUsbSpeed().name performance["async_capture"] = async_status
performance["velocidade"] = speed
if str(speed).lower() not in ["super", "superplus"]: if running and async_status:
motivos.append(f"USB lenta: {speed}") if not bool(async_status.get("thread_alive", False)):
saude -= 15 motivos.append("Thread assíncrona de captura parada")
except Exception: saude = 0
pass erro_async = async_status.get("last_error")
if erro_async:
motivos.append(f"Captura assíncrona: {erro_async}")
if not running: if not running:
motivos.append("Pipeline parada") motivos.append("Pipeline parada")
@ -858,19 +859,17 @@ class CameraMultispectral:
"duracao": 0.0, "duracao": 0.0,
"frame_valido": False, "frame_valido": False,
} }
if self._is_erro_fatal_depthai(e):
if "Communication exception" in str(e) or "X_LINK_ERROR" in str(e):
conectado = False conectado = False
agora = time.time() agora = time.time()
ts_tensor = self.timestamp_ultimo_tensor or 0.0 ts_tensor = self.timestamp_ultimo_tensor or 0.0
frame_recente = (agora - ts_tensor) <= 2.0 frame_recente = ts_tensor > 0 and (agora - ts_tensor) <= 2.0
if resultado.get("sync_dt_ms") is not None: try:
try: performance["sync_dt_ms"] = float(resultado.get("sync_dt_ms", 0.0) or 0.0)
performance["sync_dt_ms"] = float(resultado.get("sync_dt_ms", 0.0)) except Exception:
except Exception: pass
pass
if not conectado and not conectado_ctx: if not conectado and not conectado_ctx:
motivos.append("desconectado") motivos.append("desconectado")
@ -901,34 +900,25 @@ class CameraMultispectral:
saude -= 10 saude -= 10
motivos.append(f"Sincronismo alto: {sync_dt:.1f} ms") motivos.append(f"Sincronismo alto: {sync_dt:.1f} ms")
if not pipeline_ia_ok:
if pipeline_ia_informado and not pipeline_ia_ok: for motivo in pipeline_ia.get("motivos", []) or []:
motivos_pipeline = pipeline_ia.get("motivos", []) or []
for motivo in motivos_pipeline:
texto = f"Pipeline IA: {motivo}" texto = f"Pipeline IA: {motivo}"
if texto not in motivos: if texto not in motivos:
motivos.append(texto) motivos.append(texto)
saude = 0 saude = 0
motivos = list(dict.fromkeys(str(m) for m in motivos if m))
saude = min(max(saude, 0), 100) saude = min(max(saude, 0), 100)
status_mod = StatusModulo.OPERANTE status_mod = StatusModulo.OPERANTE
if not conectado: if not conectado:
status_mod = StatusModulo.DESCONECTADO status_mod = StatusModulo.DESCONECTADO
elif not frame_recente: elif not frame_recente or not pipeline_ia_ok or saude <= 0:
status_mod = StatusModulo.FALHA
elif not pipeline_ia_ok:
status_mod = StatusModulo.FALHA
elif saude <= 0:
status_mod = StatusModulo.FALHA status_mod = StatusModulo.FALHA
elif saude < 80: elif saude < 80:
status_mod = StatusModulo.ALERTA status_mod = StatusModulo.ALERTA
self.rodando = ( self.rodando = bool(conectado and frame_recente and pipeline_ia_ok)
conectado
and frame_recente
and pipeline_ia_ok
)
saude_geral = { saude_geral = {
"timestamp": agora, "timestamp": agora,
@ -938,7 +928,6 @@ class CameraMultispectral:
"motivos": motivos, "motivos": motivos,
"saude_individual": [], "saude_individual": [],
} }
self.ultima_saude = saude_geral self.ultima_saude = saude_geral
from camera_worker.manager import definir_saude_camera from camera_worker.manager import definir_saude_camera
@ -953,7 +942,7 @@ class CameraMultispectral:
conectado, conectado,
agora, agora,
self.dispositivo, self.dispositivo,
imu=imu imu=imu,
) )
def _definir_heartbeat(self): def _definir_heartbeat(self):
@ -1175,129 +1164,54 @@ class CameraMultispectral:
def requisitar_bundle_raw_multispec(self, force: bool = True, max_age_s: float = None): def requisitar_bundle_raw_multispec(self, force: bool = True, max_age_s: float = None):
""" """
Retorna um bundle científico RAW_BRUTO completo. Monta o bundle a partir do cache do fluxo operacional.
Retorno: Não chama client.get_next_raw_frame(), portanto não disputa o cursor
bundle, resultado assíncrono com a inferência. force é mantido por compatibilidade.
bundle = {
"raw_frame": {
"CAM_A": np.ndarray RAW10 packed,
"CAM_B": np.ndarray RAW10 packed,
"CAM_C": np.ndarray RAW10 packed,
},
"raw_meta": dict,
"preview_bgr": np.ndarray BGR uint8 ou None,
"preview_method": str,
}
Este método não salva nada em disco.
Ele apenas coleta e organiza o pacote bruto.
""" """
try: try:
agora = time.perf_counter() if max_age_s is None or float(max_age_s) <= 0:
max_age_s = max(1.0, self._cache_max_age_s)
if max_age_s is None: raw_frame, resultado = self.requisitar_frame_raw_multi(
max_age_s = self._cache_max_age_s force=False,
max_age_s=max_age_s,
with self._lock:
cache_ok = (
self.ultimo_raw_multi is not None
and self.ultimo_meta is not None
and self.timestamp_ultimo_raw_multi is not None
and (agora - self.timestamp_ultimo_raw_multi) < max_age_s
)
if cache_ok and not force:
raw_frame = {
cam_id: arr.copy()
for cam_id, arr in self.ultimo_raw_multi.items()
}
raw_meta = dict(self.ultimo_meta)
preview_bgr, preview_method = self._build_preview_raw_multispec(
raw_frame=raw_frame,
raw_meta=raw_meta,
)
bundle = {
"raw_frame": raw_frame,
"raw_meta": raw_meta,
"preview_bgr": preview_bgr,
"preview_method": preview_method,
}
return bundle, dict(self._ultimo_resultado_raw)
if self.client is None:
raise RuntimeError("OakFcc3Client não inicializado")
t0 = time.perf_counter()
raw_frame, raw_meta = self.client.get_next_raw_frame(
timeout=self.timeout_s
) )
dur = time.perf_counter() - t0 if raw_frame is None:
raise RuntimeError(resultado.get("erro") or "RAW cache indisponível")
if not isinstance(raw_frame, dict) or not raw_frame: with self._lock:
raise RuntimeError( raw_meta = dict(self.ultimo_meta or {})
f"RAW_BRUTO inválido. Esperado dict por câmera, veio {type(raw_frame)}"
)
raw_meta = dict(raw_meta or {})
frame_type = str(raw_meta.get("frame_type", "")).upper()
if frame_type and frame_type != "RAW_BRUTO":
raise RuntimeError(
f"Bundle científico esperado em RAW_BRUTO, mas veio frame_type={frame_type}"
)
required = {"CAM_A", "CAM_B", "CAM_C"} required = {"CAM_A", "CAM_B", "CAM_C"}
presentes = set(raw_frame.keys()) presentes = set(raw_frame.keys())
faltando = sorted(required - presentes) faltando = sorted(required - presentes)
if faltando: if faltando:
raise RuntimeError( raise RuntimeError(
f"RAW_BRUTO incompleto. Faltando câmeras: {faltando}. Presentes: {sorted(presentes)}" f"RAW_BRUTO incompleto. Faltando câmeras: {faltando}. "
f"Presentes: {sorted(presentes)}"
) )
raw_frame_copy = {
cam_id: arr.copy()
for cam_id, arr in raw_frame.items()
}
preview_bgr, preview_method = self._build_preview_raw_multispec( preview_bgr, preview_method = self._build_preview_raw_multispec(
raw_frame=raw_frame_copy, raw_frame=raw_frame,
raw_meta=raw_meta, raw_meta=raw_meta,
) )
resultado = { resultado = dict(resultado)
resultado.update({
"erro": None, "erro": None,
"duracao": dur,
"frame_valido": True, "frame_valido": True,
"cameras": list(raw_frame_copy.keys()),
"sync_ok": bool(raw_meta.get("sync_ok", True)),
"sync_dt_ms": float(raw_meta.get("sync_dt_ms", 0.0) or 0.0),
"frame_id": raw_meta.get("frame_id"),
"preview_method": preview_method, "preview_method": preview_method,
} "origem": "cache_da_captura_operacional",
})
with self._lock:
self.ultimo_raw_multi = raw_frame_copy
self.ultimo_meta = raw_meta
self.timestamp_ultimo_raw_multi = agora
self._ultimo_resultado_raw = resultado
bundle = { bundle = {
"raw_frame": raw_frame_copy, "raw_frame": raw_frame,
"raw_meta": raw_meta, "raw_meta": raw_meta,
"preview_bgr": preview_bgr, "preview_bgr": preview_bgr,
"preview_method": preview_method, "preview_method": preview_method,
} }
return bundle, resultado return bundle, resultado
except Exception as e: except Exception as e:
@ -1306,17 +1220,11 @@ class CameraMultispectral:
"duracao": 0.0, "duracao": 0.0,
"frame_valido": False, "frame_valido": False,
} }
with self._lock: with self._lock:
self._ultimo_resultado_raw = resultado self._ultimo_resultado_raw = resultado
self.mostrar_log( self.mostrar_log(
f"[CameraMultispectral] Erro ao requisitar bundle RAW multispec: {e}" f"[CameraMultispectral] Erro ao montar bundle RAW cacheado: {e}"
) )
if self._is_erro_fatal_depthai(e):
self._falha_fatal_depthai(e)
return None, resultado return None, resultado
def _build_preview_raw_multispec(self, raw_frame: dict, raw_meta: dict): def _build_preview_raw_multispec(self, raw_frame: dict, raw_meta: dict):

9
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_client.py
View File

@ -173,10 +173,11 @@ class OakFcc3Client:
return resp return resp
def stop(self): def stop(self):
try: # disconnect() já para o manager. Evita manager.stop() duplicado.
self.svc.stop() return self.svc.disconnect()
finally:
self.svc.disconnect() def get_device_metrics(self):
return self.svc.get_device_metrics()
def get_status(self): def get_status(self):
return self.svc.get_status() return self.svc.get_status()

159
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_manager.py
View File

@ -92,6 +92,10 @@ class OakFcc3Manager:
self.running = False self.running = False
self.frame_id = 0 self.frame_id = 0
# Serializa start/stop e leituras nativas de telemetria.
# Evita getChipTemperature/getUsbSpeed concorrendo com device.close().
self._device_lock = threading.RLock()
self.control_queues = {} self.control_queues = {}
self.camera_controls = { self.camera_controls = {
cam_id: self._default_controls_for_role(role) cam_id: self._default_controls_for_role(role)
@ -746,13 +750,20 @@ class OakFcc3Manager:
# ============================================================ # ============================================================
def start(self): def start(self):
with self._device_lock:
return self._start_impl()
def _start_impl(self):
if self.running: if self.running:
return return
thread_anterior = self._capture_thread thread_anterior = self._capture_thread
if (thread_anterior is not None and thread_anterior.is_alive()): 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.") raise RuntimeError(
"Não é seguro iniciar novo pipeline: "
"thread OakFcc3AsyncCapture anterior ainda está viva."
)
try: try:
self.dev_info = self._resolve_device_info() self.dev_info = self._resolve_device_info()
@ -781,7 +792,10 @@ class OakFcc3Manager:
role = self.roles.get(socket_name, "unknown") role = self.roles.get(socket_name, "unknown")
print(f"[OAK] Criando câmera {socket_name} sensor={f.sensorName} role={role}") print(
f"[OAK] Criando câmera {socket_name} "
f"sensor={f.sensorName} role={role}"
)
cam, output = self._create_camera_node_classic( cam, output = self._create_camera_node_classic(
socket=socket, socket=socket,
@ -813,9 +827,7 @@ class OakFcc3Manager:
} }
self._validate_capture_mode() self._validate_capture_mode()
self._create_imu_node(self.pipeline) self._create_imu_node(self.pipeline)
self.device.startPipeline(self.pipeline) self.device.startPipeline(self.pipeline)
self.q_imu = None self.q_imu = None
@ -861,21 +873,30 @@ class OakFcc3Manager:
self._start_async_capture_thread() self._start_async_capture_thread()
time.sleep(0.05) time.sleep(0.05)
if not self.running:
erro = self._capture_thread_stats.get("last_error")
raise RuntimeError(
"Pipeline DepthAI caiu imediatamente após iniciar. "
f"Último erro: {erro}"
)
except Exception: except Exception:
# Garante limpeza mesmo se a falha acontecer # _device_lock é RLock, então a limpeza pode reutilizar stop().
# no meio da construção do pipeline.
try: try:
self.stop() self.stop()
except Exception: except Exception:
pass pass
raise raise
def stop(self): def stop(self):
with self._device_lock:
return self._stop_impl()
def _stop_impl(self):
# Impede novas capturas imediatamente. # Impede novas capturas imediatamente.
self.running = False self.running = False
self._stop_async_capture_thread() thread_ok = self._stop_async_capture_thread()
try: try:
if self.pipeline is not None and hasattr(self.pipeline, "stop"): if self.pipeline is not None and hasattr(self.pipeline, "stop"):
@ -891,6 +912,7 @@ class OakFcc3Manager:
self.pipeline = None self.pipeline = None
self.device = None self.device = None
self.dev_info = None
self.queues.clear() self.queues.clear()
self.buffers.clear() self.buffers.clear()
@ -911,15 +933,23 @@ class OakFcc3Manager:
except Exception: except Exception:
pass pass
return bool(thread_ok is not False)
def _is_fatal_depthai_error(self, erro): def _is_fatal_depthai_error(self, erro):
txt = str(erro) txt = str(erro).lower()
sinais = [ sinais = [
"X_LINK_ERROR", "x_link_error",
"Communication exception", "communication exception",
"Couldn't read data from stream", "couldn't read data from stream",
"Device already closed", "couldn't open stream",
"device already closed",
"device has been closed", "device has been closed",
"x_link_device_not_found",
"failed to find device",
"no available devices",
"nenhum dispositivo depthai",
"device crashed",
] ]
return any(s in txt for s in sinais) return any(s in txt for s in sinais)
@ -928,29 +958,86 @@ class OakFcc3Manager:
# Status # Status
# ============================================================ # ============================================================
def get_device_metrics(self):
"""Lê métricas nativas sem concorrer com start/stop/device.close()."""
with self._device_lock:
dev = self.device
if dev is None or not self.running:
return {
"ok": False,
"running": bool(self.running),
"error": "device indisponível",
}
out = {
"ok": True,
"running": True,
"timestamp": time.time(),
}
try:
speed = dev.getUsbSpeed()
out["usb_speed"] = getattr(speed, "name", str(speed))
except Exception as e:
out["usb_error"] = str(e)
try:
temp = dev.getChipTemperature()
for attr in ("average", "css", "mss", "upa", "dss"):
if hasattr(temp, attr):
out[f"temperature_{attr}_c"] = float(getattr(temp, attr))
except Exception as e:
out["temperature_error"] = str(e)
for nome, metodo in (
("ddr", "getDdrMemoryUsage"),
("cmx", "getCmxMemoryUsage"),
):
try:
uso = getattr(dev, metodo)()
out[f"{nome}_used_bytes"] = int(getattr(uso, "used", 0))
out[f"{nome}_total_bytes"] = int(getattr(uso, "total", 0))
except Exception as e:
out[f"{nome}_error"] = str(e)
for nome, metodo in (
("leon_css", "getLeonCssCpuUsage"),
("leon_mss", "getLeonMssCpuUsage"),
):
try:
uso = getattr(dev, metodo)()
out[f"{nome}_average"] = float(getattr(uso, "average", uso))
except Exception as e:
out[f"{nome}_error"] = str(e)
out["async_capture"] = self.get_async_capture_status()
return out
def get_status(self): def get_status(self):
return { with self._device_lock:
"mx_id": self.mx_id, return {
"backend": "oak_fcc3", "mx_id": self.mx_id,
"running": self.running, "backend": "oak_fcc3",
"fps": self.fps, "running": bool(self.running),
"width": self.width, "fps": self.fps,
"height": self.height, "width": self.width,
"sensor_width": self.sensor_width, "height": self.height,
"sensor_height": self.sensor_height, "sensor_width": self.sensor_width,
"frame_type": self.frame_type, "sensor_height": self.sensor_height,
"output_dtype": self.output_dtype, "frame_type": self.frame_type,
"capture_mode": self.capture_mode, "output_dtype": self.output_dtype,
"raw_policy": self.raw_policy, "capture_mode": self.capture_mode,
"sync_tolerance_ms": self.sync_tolerance_ms, "raw_policy": self.raw_policy,
"buffer_size": self.buffer_size, "sync_tolerance_ms": self.sync_tolerance_ms,
"geometry_stage": "oak" if self._is_multispec_mode() else "pc", "buffer_size": self.buffer_size,
"aligned_geometry": self._serializable_aligned_geometry(), "geometry_stage": "oak" if self._is_multispec_mode() else "pc",
"cameras": list(self.camera_info.values()), "aligned_geometry": self._serializable_aligned_geometry(),
"async_capture": self.get_async_capture_status() if hasattr(self, "get_async_capture_status") else None, "cameras": list(self.camera_info.values()),
"tem_imu": bool(getattr(self, "tem_imu", False)), "async_capture": self.get_async_capture_status(),
"has_imu_pipeline": bool(getattr(self, "has_imu_pipeline", False)), "tem_imu": bool(getattr(self, "tem_imu", False)),
} "has_imu_pipeline": bool(getattr(self, "has_imu_pipeline", False)),
}
def _serializable_aligned_geometry(self): def _serializable_aligned_geometry(self):
if not isinstance(self.aligned_geometry, dict): if not isinstance(self.aligned_geometry, dict):

19
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_service.py
View File

@ -13,8 +13,12 @@ class OakFcc3Service:
return {"ok": True, "backend": "oak_fcc3", "connected": True} return {"ok": True, "backend": "oak_fcc3", "connected": True}
def disconnect(self): def disconnect(self):
self.stop() # Único caminho de fechamento do manager.
self.connected = False try:
self.manager.stop()
finally:
self.connected = False
return {"ok": True, "connected": False} return {"ok": True, "connected": False}
def ping(self): def ping(self):
@ -33,9 +37,13 @@ class OakFcc3Service:
for c in status.get("cameras", []) for c in status.get("cameras", [])
} }
running = bool(status.get("running", False))
realmente_ativo = bool(self.connected and running)
status.update({ status.update({
"ok": True, "ok": realmente_ativo,
"connected": self.connected, "connected": bool(self.connected),
"running": running,
"active_camera_ids": active_ids, "active_camera_ids": active_ids,
"active_roles": active_roles, "active_roles": active_roles,
"camera_count_active": len(active_ids), "camera_count_active": len(active_ids),
@ -43,6 +51,9 @@ class OakFcc3Service:
return status return status
def get_device_metrics(self):
return self.manager.get_device_metrics()
def get_config(self): def get_config(self):
return { return {
"mx_id": self.manager.mx_id, "mx_id": self.manager.mx_id,

534
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/camera_manager.py
View File

@ -75,6 +75,9 @@ class CameraManager:
self._loop_publicacao_iniciado = False self._loop_publicacao_iniciado = False
self._vida_lock = threading.RLock() self._vida_lock = threading.RLock()
self._camera_generation = 0
self._falhas_inicializacao = 0
self._proxima_inicializacao_monotonic = 0.0
self._fechando_camera = False self._fechando_camera = False
self._em_warmup = False self._em_warmup = False
@ -111,6 +114,7 @@ class CameraManager:
self._tensor_pronto = None self._tensor_pronto = None
self._tensor_res = None self._tensor_res = None
self._tensor_ts = 0.0 self._tensor_ts = 0.0
self._tensor_generation = int(getattr(self, "_camera_generation", 0))
self._tensor_consumido_ts = 0.0 self._tensor_consumido_ts = 0.0
self.pipeline_ia_ok = False self.pipeline_ia_ok = False
@ -131,6 +135,7 @@ class CameraManager:
"infer_prepare_ms": 0.0, "infer_prepare_ms": 0.0,
"infer_post_ms": 0.0, "infer_post_ms": 0.0,
"fps_model": 0.0, "fps_model": 0.0,
"generation": int(getattr(self, "_camera_generation", 0)),
} }
self._pred_consumido_ts = 0.0 self._pred_consumido_ts = 0.0
@ -196,39 +201,85 @@ class CameraManager:
"total_ms": 0.0, "total_ms": 0.0,
} }
def _avancar_geracao_camera(self):
self._camera_generation += 1
return self._camera_generation
def _sessao_camera_valida(self, camera, generation):
with self._vida_lock:
return bool(
camera is not None
and camera is self.camera
and int(generation) == int(self._camera_generation)
)
def _registrar_falha_inicializacao(self):
self._falhas_inicializacao += 1
atraso = min(10.0, 1.0 * (2 ** min(self._falhas_inicializacao - 1, 3)))
self._proxima_inicializacao_monotonic = time.monotonic() + atraso
return atraso
def inicializar(self, mx_id): def inicializar(self, mx_id):
if mx_id is None: if mx_id is None:
return False return False
with self._vida_lock: with self._vida_lock:
if self.iniciando: if self.iniciando or self._fechando_camera:
return False return False
if ( agora_mono = time.monotonic()
self.camera is not None if agora_mono < self._proxima_inicializacao_monotonic:
and self.operante return False
and self.pipeline_ia_ok
): camera_existente = self.camera
return True
if camera_existente is not None:
status_existente = camera_existente._safe_get_status()
ativa = bool(
camera_existente.iniciado
and status_existente.get("running", False)
)
if ativa and self.operante and self.pipeline_ia_ok:
return True
# Nunca tenta abrir outra instância sobre uma referência existente.
self.fechar_camera_manager(
motivo="reinicialização solicitada sobre câmera não saudável",
camera_esperada=camera_existente,
)
# Dá tempo para o runtime/USB liberar a sessão anterior.
time.sleep(1.0)
self.iniciando = True self.iniciando = True
self.pipeline_ia_ok = False self.pipeline_ia_ok = False
camera_nova = None
try: try:
self.mx_id = mx_id self.mx_id = str(mx_id)
from weed_worker.config import load_seg_config from weed_worker.config import load_seg_config
self.seg_config = load_seg_config() self.seg_config = load_seg_config()
self.posproc_intervalo_min_s = float(
self.seg_config.get("posproc_intervalo_min_s", 5.0)
)
self.posproc_intervalo_min_s = float(self.seg_config.get("posproc_intervalo_min_s", 5.0)) camera_nova = self._inicializar_camera(mx_id, self.seg_config)
if camera_nova is None:
self._inicializar_camera(mx_id, self.seg_config) atraso = self._registrar_falha_inicializacao()
self.mostrar_log(
if self.camera is None: f"❌ Camera com ID {mx_id} não iniciada. "
self.mostrar_log(f"❌ Camera com ID {mx_id} não iniciada.") f"Nova tentativa após backoff de {atraso:.1f}s."
)
return False return False
self.mostrar_log(f"📷 Camera selecionada: {self.camera.modelo} - {self.camera.mx_id}") self.camera = camera_nova
generation = self._avancar_geracao_camera()
self.mostrar_log(
f"📷 Camera selecionada: {camera_nova.modelo} - "
f"{camera_nova.mx_id} | geração={generation}"
)
self._reset_runtime_state() self._reset_runtime_state()
@ -242,22 +293,34 @@ class CameraManager:
self.operante = False self.operante = False
self._em_warmup = True self._em_warmup = True
warmup_ok = self._executar_warmup_camera_manager(n_ciclos=3, timeout_s=12.0) warmup_ok = self._executar_warmup_camera_manager(
n_ciclos=3,
timeout_s=12.0,
)
self._em_warmup = False self._em_warmup = False
if not warmup_ok: if not warmup_ok:
self.mostrar_log("[weed][WARMUP] recuperação rejeitada: pipeline não concluiu validação ponta a ponta") self.mostrar_log(
self.fechar_camera_manager("warmup incompleto após reconexão") "[weed][WARMUP] recuperação rejeitada: "
"pipeline não concluiu validação ponta a ponta"
)
self.fechar_camera_manager(
"warmup incompleto após reconexão",
camera_esperada=camera_nova,
)
atraso = self._registrar_falha_inicializacao()
self.mostrar_log(f"[weed] backoff de reconexão: {atraso:.1f}s")
return False
if not self._sessao_camera_valida(camera_nova, generation):
return False return False
self.operante = True self.operante = True
self._iniciar_loops_se_necessario(self.seg_config) self._iniciar_loops_se_necessario(self.seg_config)
# A avaliação final precisa acontecer depois que a inicialização acabou. # Avaliação final precisa ocorrer fora do estado iniciando.
self.iniciando = False self.iniciando = False
pipeline_ia = self._avaliar_saude_pipeline_ia() pipeline_ia = self._avaliar_saude_pipeline_ia()
if not pipeline_ia.get("ok", False): if not pipeline_ia.get("ok", False):
@ -265,23 +328,44 @@ class CameraManager:
"[weed] inicialização rejeitada após warmup: " "[weed] inicialização rejeitada após warmup: "
+ "; ".join(pipeline_ia.get("motivos", [])) + "; ".join(pipeline_ia.get("motivos", []))
) )
self.fechar_camera_manager(
self.fechar_camera_manager(motivo="pipeline IA inválido após warmup", camera_esperada=self.camera) motivo="pipeline IA inválido após warmup",
camera_esperada=camera_nova,
)
atraso = self._registrar_falha_inicializacao()
self.mostrar_log(f"[weed] backoff de reconexão: {atraso:.1f}s")
return False return False
self._falhas_inicializacao = 0
self._proxima_inicializacao_monotonic = 0.0
self.atualizar_saude_camera(pipeline_ia=pipeline_ia) self.atualizar_saude_camera(pipeline_ia=pipeline_ia)
return True return True
except Exception as e: except Exception as e:
self.mostrar_log(f"[weed] erro ao inicializar câmera: {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)
if self.camera is camera_nova and camera_nova is not None:
self.fechar_camera_manager(
motivo=f"falha de inicialização: {e}",
camera_esperada=camera_nova,
)
elif camera_nova is not None:
try:
camera_nova.parar()
except Exception:
pass
atraso = self._registrar_falha_inicializacao()
self.mostrar_log(f"[weed] backoff de reconexão: {atraso:.1f}s")
return False return False
finally: finally:
self.iniciando = False self.iniciando = False
self._em_warmup = False
def _inicializar_camera(self, mx_id, seg_config): def _inicializar_camera(self, mx_id, seg_config):
nova = None
try: try:
nova = CameraMultispectral( nova = CameraMultispectral(
self.mostrar_log, self.mostrar_log,
@ -289,18 +373,28 @@ class CameraManager:
module_calibration_json=seg_config.get("module_calibration_json"), module_calibration_json=seg_config.get("module_calibration_json"),
width=int(seg_config.get("camera_width", 1280)), width=int(seg_config.get("camera_width", 1280)),
height=int(seg_config.get("camera_height", 800)), height=int(seg_config.get("camera_height", 800)),
fps=int(seg_config.get("camera_fps", 40)), # Default conservador para campo. Se o JSON definir 40, ele será respeitado.
fps=int(seg_config.get("camera_fps", 20)),
target_size=seg_config.get("ia_resolution", [1024, 640]), target_size=seg_config.get("ia_resolution", [1024, 640]),
) )
if nova.iniciado: if not nova.iniciado:
self.camera = nova try:
else: nova.parar()
self.camera = None except Exception:
pass
return None
return nova
except Exception as e: except Exception as e:
self.camera = None if nova is not None:
try:
nova.parar()
except Exception:
pass
self.mostrar_log(f"⚠️ Camera com ID {mx_id} não conectada: {e}") self.mostrar_log(f"⚠️ Camera com ID {mx_id} não conectada: {e}")
return None
def _inicializar_modelo(self): def _inicializar_modelo(self):
if self.model_svc is not None: if self.model_svc is not None:
@ -408,42 +502,123 @@ class CameraManager:
def fechar_camera_manager(self, motivo="", camera_esperada=None): def fechar_camera_manager(self, motivo="", camera_esperada=None):
with self._vida_lock: 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:
if (camera_esperada is not None and self.camera is not camera_esperada): self.mostrar_log(
self.mostrar_log("[weed] fechamento ignorado: a câmera ativa já foi substituída") "[weed] fechamento ignorado: a câmera ativa já foi substituída"
)
return False
if self._fechando_camera:
return False return False
camera_alvo = self.camera camera_alvo = self.camera
geracao_antiga = self._camera_generation
# Invalida primeiro todos os trabalhos em voo.
self._avancar_geracao_camera()
self.camera = None self.camera = None
self.operante = False self.operante = False
self.pipeline_ia_ok = False self.pipeline_ia_ok = False
self._fechando_camera = True self._fechando_camera = True
self._reset_runtime_state() self._reset_runtime_state()
try: try:
self._forcar_bicos_off(motivo=f"câmera fechada: {motivo}", intervalo_min_s=0.0) self._forcar_bicos_off(
motivo=f"câmera fechada: {motivo}",
intervalo_min_s=0.0,
)
except Exception: except Exception:
pass pass
try: try:
if camera_alvo is not None: if camera_alvo is not None:
camera_alvo.parar() camera_alvo.parar()
except Exception as e: except Exception as e:
self.mostrar_log(f"[weed] Erro ao fechar câmera: {e}") self.mostrar_log(f"[weed] Erro ao fechar câmera: {e}")
finally: finally:
self._fechando_camera = False self._fechando_camera = False
self.mostrar_log(f"[weed] Camera Manager fechado: {motivo}") # Evita reabrir enquanto o dispositivo ainda está reenlistando no USB.
self._proxima_inicializacao_monotonic = max(
self._proxima_inicializacao_monotonic,
time.monotonic() + 1.0,
)
self.mostrar_log(
f"[weed] Camera Manager fechado: {motivo} | "
f"geração={geracao_antiga}->{self._camera_generation}"
)
return True return True
def reiniciar_deteccoes(self): def reiniciar_deteccoes(self):
self._reset_runtime_state() """Limpa somente estado da operação, sem tocar na OAK ou no ONNX."""
return True, "estado operacional limpo" with self._vida_lock:
detector = self.weed_detector
# Descarta resultados pendentes, mas mantém saúde física e timestamps vivos.
with self._lock_tensor:
self._tensor_consumido_ts = self._tensor_ts
with self._pred_lock:
self._pred_cache = {
"ts": 0.0,
"tensor_ts": 0.0,
"predictions": None,
"res": None,
"infer_ms": 0.0,
"infer_gpu_ms": 0.0,
"infer_prepare_ms": 0.0,
"infer_post_ms": 0.0,
"fps_model": 0.0,
"generation": self._camera_generation,
}
self._pred_consumido_ts = 0.0
self._ultima_analise = {}
self._ultimo_predictions = None
self._ultimo_predictions_full = None
self._ultimo_controle = None
self._ultimo_cmd_bicos = None
with self._pub_lock:
for chave in ("analise", "controle_bicos", "performance_weed", "cmd_controle"):
self._pub_cache[chave] = None
for chave in (
"dirty_analise",
"dirty_controle_bicos",
"dirty_performance_weed",
"dirty_cmd_controle",
):
self._pub_cache[chave] = False
self._pub_cache["ts_analise"] = 0.0
reset_nome = None
if detector is not None:
for nome in (
"reiniciar_estado_operacional",
"reiniciar_deteccoes",
"reset_estado_operacional",
"reset",
):
fn = getattr(detector, nome, None)
if callable(fn):
fn()
reset_nome = nome
break
if detector is None or reset_nome is None:
# Fallback seguro: recria apenas o detector, nunca câmera/modelo.
self.weed_detector = WeedDetector()
if self.seg_config and hasattr(self.weed_detector, "atualizar_config"):
self.weed_detector.atualizar_config(self.seg_config)
reset_nome = "recriado"
self._forcar_bicos_off(
motivo="reinício do estado operacional das detecções",
intervalo_min_s=0.0,
)
return True, f"estado operacional limpo | detector={reset_nome}"
# ============================================================ # ============================================================
# Warmup # Warmup
@ -455,6 +630,7 @@ class CameraManager:
timeout_s=12.0, timeout_s=12.0,
): ):
camera_alvo = self.camera camera_alvo = self.camera
generation = self._camera_generation
if camera_alvo is None: if camera_alvo is None:
return False return False
@ -476,7 +652,7 @@ class CameraManager:
try: try:
while ciclos_ok < n_ciclos and time.monotonic() < deadline: while ciclos_ok < n_ciclos and time.monotonic() < deadline:
if self.camera is not camera_alvo: if not self._sessao_camera_valida(camera_alvo, generation):
self.mostrar_log( self.mostrar_log(
"[weed][WARMUP] câmera foi substituída durante o warmup" "[weed][WARMUP] câmera foi substituída durante o warmup"
) )
@ -498,6 +674,11 @@ class CameraManager:
self.mostrar_log( self.mostrar_log(
f"[weed][WARMUP] captura inválida: {erro}" f"[weed][WARMUP] captura inválida: {erro}"
) )
try:
if camera_alvo._is_erro_fatal_depthai(erro):
return False
except Exception:
pass
time.sleep(0.05) time.sleep(0.05)
continue continue
@ -563,6 +744,10 @@ class CameraManager:
analise_completa = self.detectar_ervas(predictions) analise_completa = self.detectar_ervas(predictions)
t_det1 = time.perf_counter() t_det1 = time.perf_counter()
if not self._sessao_camera_valida(camera_alvo, generation):
self.perf.inc("warmup_descartado_geracao")
continue
if not isinstance(analise_completa, dict): if not isinstance(analise_completa, dict):
self.mostrar_log( self.mostrar_log(
f"[weed][WARMUP] detecção inválida no frame {frame_id}" f"[weed][WARMUP] detecção inválida no frame {frame_id}"
@ -581,6 +766,7 @@ class CameraManager:
"pred_ts": agora, "pred_ts": agora,
"infer_full": infer_full, "infer_full": infer_full,
"analise": analise_completa, "analise": analise_completa,
"generation": generation,
} }
self.mostrar_log( self.mostrar_log(
@ -611,6 +797,7 @@ class CameraManager:
self._tensor_pronto = ultimo_ciclo["tensor"] self._tensor_pronto = ultimo_ciclo["tensor"]
self._tensor_res = ultimo_ciclo["res"] self._tensor_res = ultimo_ciclo["res"]
self._tensor_ts = ultimo_ciclo["tensor_ts"] self._tensor_ts = ultimo_ciclo["tensor_ts"]
self._tensor_generation = ultimo_ciclo["generation"]
self._tensor_consumido_ts = ultimo_ciclo["tensor_ts"] self._tensor_consumido_ts = ultimo_ciclo["tensor_ts"]
with self._pred_lock: with self._pred_lock:
@ -624,6 +811,7 @@ class CameraManager:
"infer_prepare_ms": 0.0, "infer_prepare_ms": 0.0,
"infer_post_ms": 0.0, "infer_post_ms": 0.0,
"fps_model": float(self._fps_infer_ema or 0.0), "fps_model": float(self._fps_infer_ema or 0.0),
"generation": ultimo_ciclo["generation"],
} }
self._pred_consumido_ts = ultimo_ciclo["pred_ts"] self._pred_consumido_ts = ultimo_ciclo["pred_ts"]
@ -666,15 +854,28 @@ class CameraManager:
self._ultima_saude_ts = time.time() self._ultima_saude_ts = time.time()
try: try:
if self.camera is not None: # Mantém a referência estável durante leitura/publicação da saúde.
if pipeline_ia is None: # fechar_camera_manager aguarda esta seção curta terminar.
pipeline_ia = self._avaliar_saude_pipeline_ia() with self._vida_lock:
camera_atual = self.camera
self.camera.atualizar_saude(pipeline_ia=pipeline_ia) if camera_atual is not None:
if pipeline_ia is None:
pipeline_ia = self._avaliar_saude_pipeline_ia()
camera_atual.atualizar_saude(pipeline_ia=pipeline_ia)
elif self.mx_id is not None: elif self.mx_id is not None:
from camera_worker.manager import definir_saude_camera 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: except Exception as e:
self.mostrar_log(f"[saude] erro: {e}") self.mostrar_log(f"[saude] erro: {e}")
@ -790,23 +991,28 @@ class CameraManager:
tensor5 = self._tensor_pronto tensor5 = self._tensor_pronto
res = self._tensor_res res = self._tensor_res
ts_tensor = self._tensor_ts ts_tensor = self._tensor_ts
generation = self._tensor_generation
if tensor5 is None or ts_tensor <= 0: if tensor5 is None or ts_tensor <= 0:
return None, None, None return None, None, None, None
if int(generation) != int(self._camera_generation):
return None, None, res, generation
if ts_tensor == self._tensor_consumido_ts: if ts_tensor == self._tensor_consumido_ts:
return None, None, res return None, None, res, generation
self._tensor_consumido_ts = ts_tensor self._tensor_consumido_ts = ts_tensor
return tensor5, ts_tensor, res return tensor5, ts_tensor, res, generation
def _get_prediction_nova_para_deteccao(self): def _get_prediction_nova_para_deteccao(self):
with self._pred_lock: with self._pred_lock:
cache = dict(self._pred_cache) cache = dict(self._pred_cache)
pred_ts = float(cache.get("ts", 0.0) or 0.0) pred_ts = float(cache.get("ts", 0.0) or 0.0)
generation = int(cache.get("generation", -1))
if pred_ts <= 0: if pred_ts <= 0 or generation != int(self._camera_generation):
return None return None
if pred_ts == self._pred_consumido_ts: if pred_ts == self._pred_consumido_ts:
@ -1122,50 +1328,51 @@ class CameraManager:
t0_wall = time.time() t0_wall = time.time()
t0_perf = time.perf_counter() t0_perf = time.perf_counter()
camera_atual = None camera_atual = None
generation = -1
try: try:
if (self.iniciando or self._em_warmup or not self.operante or self.camera is None): if self.iniciando or self._em_warmup or not self.operante or self.camera is None:
time.sleep(0.05) time.sleep(0.05)
continue continue
camera_atual = self.camera with self._vida_lock:
camera_atual = self.camera
generation = self._camera_generation
tensor5, res = (camera_atual.requisitar_tensor_multispec(force=True)) tensor5, res = camera_atual.requisitar_tensor_multispec(force=True)
erro = res.get("erro") if isinstance(res, dict) else None
erro = (res.get("erro") if isinstance(res, dict) else None)
if erro: if erro:
reiniciou = self._registrar_falha_captura(camera_atual, erro) reiniciou = self._registrar_falha_captura(camera_atual, erro)
time.sleep(0.50 if reiniciou else 0.05) time.sleep(0.50 if reiniciou else 0.05)
continue continue
if tensor5 is None: if tensor5 is None:
self.perf.inc("tensor_none") self.perf.inc("tensor_none")
reiniciou = self._registrar_falha_captura(
reiniciou = self._registrar_falha_captura(camera_atual, "tensor None sem erro explícito") camera_atual,
"tensor None sem erro explícito",
)
time.sleep(0.50 if reiniciou else 0.05) time.sleep(0.50 if reiniciou else 0.05)
continue continue
# Uma captura válida limpa a sequência de falhas. if not self._sessao_camera_valida(camera_atual, generation):
self.perf.inc("tensor_descartado_geracao")
continue
self._falhas_captura_consecutivas = 0 self._falhas_captura_consecutivas = 0
self._primeira_falha_captura_ts = 0.0 self._primeira_falha_captura_ts = 0.0
agora = time.time() agora = time.time()
perf = ( perf = res.get("perf", {}) if isinstance(res, dict) else {}
res.get("perf", {})
if isinstance(res, dict)
else {}
)
with self._lock_tensor: with self._lock_tensor:
self._tensor_pronto = tensor5 self._tensor_pronto = tensor5
self._tensor_res = res self._tensor_res = res
self._tensor_ts = agora self._tensor_ts = agora
self._tensor_generation = generation
self._ultimo_tensor_ok_ts = agora self._ultimo_tensor_ok_ts = agora
t1_perf = time.perf_counter() t1_perf = time.perf_counter()
self.perf.tick( self.perf.tick(
@ -1179,25 +1386,23 @@ class CameraManager:
preview_ms=float(perf.get("preview_ms", 0.0) or 0.0), preview_ms=float(perf.get("preview_ms", 0.0) or 0.0),
frame_ts=agora, frame_ts=agora,
idade_frame_ms=0.0, idade_frame_ms=0.0,
generation=generation,
) )
except Exception as e: except Exception as e:
self.mostrar_log(f"[weed][CAPTURE] Erro inesperado: {type(e).__name__}: {e}") self.mostrar_log(
f"[weed][CAPTURE] Erro inesperado: {type(e).__name__}: {e}"
)
if camera_atual is None: if camera_atual is None:
camera_atual = self.camera camera_atual = self.camera
reiniciou = False reiniciou = False
if camera_atual is not None: if camera_atual is not None:
reiniciou = self._registrar_falha_captura(camera_atual, e) reiniciou = self._registrar_falha_captura(camera_atual, e)
time.sleep(0.50 if reiniciou else 0.05) time.sleep(0.50 if reiniciou else 0.05)
finally: finally:
gasto = time.time() - t0_wall gasto = time.time() - t0_wall
restante = periodo - gasto restante = periodo - gasto
if restante > 0: if restante > 0:
time.sleep(restante) time.sleep(restante)
@ -1215,21 +1420,26 @@ class CameraManager:
time.sleep(0.02) time.sleep(0.02)
continue continue
if not self.operante or self.camera is None: with self._vida_lock:
camera_atual = self.camera
if not self.operante or camera_atual is None:
time.sleep(0.2) time.sleep(0.2)
continue continue
t_get0 = time.perf_counter() t_get0 = time.perf_counter()
tensor5, tensor_ts, res = self._get_tensor_novo_para_inferencia() tensor5, tensor_ts, res, generation = self._get_tensor_novo_para_inferencia()
t_get1 = time.perf_counter() t_get1 = time.perf_counter()
if tensor5 is None or tensor_ts is None: if tensor5 is None or tensor_ts is None:
self.perf.inc("infer_sem_tensor_novo") self.perf.inc("infer_sem_tensor_novo")
# Espera curta, sem aplicar novamente um período inteiro.
time.sleep(0.001) time.sleep(0.001)
continue continue
if not self._sessao_camera_valida(camera_atual, generation):
self.perf.inc("infer_descartada_geracao_antes")
continue
t_inf0 = time.perf_counter() t_inf0 = time.perf_counter()
predictions = self.model_svc.infer_tensor_fast( predictions = self.model_svc.infer_tensor_fast(
tensor5, tensor5,
@ -1237,17 +1447,13 @@ class CameraManager:
) )
t_inf1 = time.perf_counter() t_inf1 = time.perf_counter()
infer_ms = (t_inf1 - t_inf0) * 1000.0 if not self._sessao_camera_valida(camera_atual, generation):
infer_full = getattr( self.perf.inc("infer_descartada_geracao_depois")
self.model_svc, continue
"_ultimo_predictions_full",
{},
) or {}
infer_forward_ms = infer_full.get( infer_ms = (t_inf1 - t_inf0) * 1000.0
"forward_ms", infer_full = getattr(self.model_svc, "_ultimo_predictions_full", {}) or {}
infer_full.get("infer_ms", 0.0), infer_forward_ms = infer_full.get("forward_ms", infer_full.get("infer_ms", 0.0))
)
infer_prepare_ms = infer_full.get("prepare_ms", 0.0) infer_prepare_ms = infer_full.get("prepare_ms", 0.0)
infer_post_ms = infer_full.get("post_ms", 0.0) infer_post_ms = infer_full.get("post_ms", 0.0)
@ -1273,6 +1479,7 @@ class CameraManager:
"infer_prepare_ms": float(infer_prepare_ms or 0.0), "infer_prepare_ms": float(infer_prepare_ms or 0.0),
"infer_post_ms": float(infer_post_ms or 0.0), "infer_post_ms": float(infer_post_ms or 0.0),
"fps_model": float(self._fps_infer_ema or 0.0), "fps_model": float(self._fps_infer_ema or 0.0),
"generation": generation,
} }
self._ultima_inferencia_ok_ts = pred_ts self._ultima_inferencia_ok_ts = pred_ts
@ -1282,7 +1489,6 @@ class CameraManager:
self._ultimo_predictions_full = infer_full self._ultimo_predictions_full = infer_full
t_loop1 = time.perf_counter() t_loop1 = time.perf_counter()
self.perf.tick( self.perf.tick(
"inferencia", "inferencia",
latencia_ms=(t_loop1 - t_loop0) * 1000.0, latencia_ms=(t_loop1 - t_loop0) * 1000.0,
@ -1294,26 +1500,20 @@ class CameraManager:
infer_gpu_ms=float(infer_forward_ms or 0.0), infer_gpu_ms=float(infer_forward_ms or 0.0),
tensor_ts=tensor_ts, tensor_ts=tensor_ts,
frame_ts=pred_ts, frame_ts=pred_ts,
idade_tensor_ms=( idade_tensor_ms=(time.time() - tensor_ts) * 1000.0 if tensor_ts else None,
(time.time() - tensor_ts) * 1000.0 generation=generation,
if tensor_ts else None
),
) )
# Só limita depois de uma inferência realmente executada.
gasto = time.perf_counter() - t_loop0 gasto = time.perf_counter() - t_loop0
restante = periodo - gasto restante = periodo - gasto
if restante > 0: if restante > 0:
time.sleep(restante) time.sleep(restante)
except Exception as e: except Exception as e:
self.mostrar_log( self.mostrar_log(f"❌ Erro no loop_inferencia weed: {e}")
f"❌ Erro no loop_inferencia weed: {e}"
)
time.sleep(0.02) time.sleep(0.02)
threading.Thread(target=loop, daemon=True).start() threading.Thread(target=loop, name="WeedInference", daemon=True).start()
def _iniciar_loop_deteccao_weed(self, freq=25.0): def _iniciar_loop_deteccao_weed(self, freq=25.0):
def loop(): def loop():
@ -1351,6 +1551,14 @@ class CameraManager:
predictions = pred_cache.get("predictions") predictions = pred_cache.get("predictions")
pred_ts = float(pred_cache.get("ts", 0.0) or 0.0) pred_ts = float(pred_cache.get("ts", 0.0) or 0.0)
tensor_ts = float(pred_cache.get("tensor_ts", 0.0) or 0.0) tensor_ts = float(pred_cache.get("tensor_ts", 0.0) or 0.0)
generation = int(pred_cache.get("generation", -1))
with self._vida_lock:
camera_atual = self.camera
if not self._sessao_camera_valida(camera_atual, generation):
self.perf.inc("det_descartada_geracao_antes")
continue
if predictions is None: if predictions is None:
self.perf.inc("det_predictions_none") self.perf.inc("det_predictions_none")
@ -1373,65 +1581,59 @@ class CameraManager:
analise_convertida = converter_valores_numpy(analise) analise_convertida = converter_valores_numpy(analise)
t_conv1 = time.perf_counter() t_conv1 = time.perf_counter()
t_pub_analise0 = time.perf_counter()
self._set_pub_cache("analise", {
"ts_analise": pred_ts,
"fps_model": pred_cache.get("fps_model", 0.0),
"fps_inferencia": self._fps_infer_ema,
"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()
# ==================================================== # ====================================================
# Gate de pulverização # Gate de pulverização
# ==================================================== # ====================================================
# O detector pode enxergar erva, mas só liberamos bicos se:
# - pulverizador automático ligado;
# - operação EmAndamento;
# - não finalizando/pausado/emergência/calibrando;
# - trajetória/status do carro = CaminhandoRua;
# - há comando/velocidade de movimento.
#
# Se qualquer item falhar, TODOS os bicos são OFF.
# Isso corrige o caso: operação ainda EmAndamento, mas Trajetoria.status=Parado.
t_ctrl0 = time.perf_counter() t_ctrl0 = time.perf_counter()
controle_detectado = analise.get("controle") or {} controle_detectado = analise.get("controle") or {}
atuacao_bicos, debug_pulverizacao = self._aplicar_gate_pulverizacao( atuacao_bicos, debug_pulverizacao = self._aplicar_gate_pulverizacao(
controle_detectado controle_detectado
) )
analise["controle"] = atuacao_bicos analise["controle"] = atuacao_bicos
analise["pulverizacao"] = debug_pulverizacao analise["pulverizacao"] = debug_pulverizacao
self._ultimo_controle = atuacao_bicos
self._ultimo_cmd_bicos = atuacao_bicos
t_ctrl1 = time.perf_counter() t_ctrl1 = time.perf_counter()
t_pub_ctrl0 = time.perf_counter() # Publicação atômica em relação a fechar/substituir câmera.
self._set_pub_cache("controle_bicos", atuacao_bicos) # Se a geração mudou, nenhum dado antigo chega aos caches/bicos.
t_pub_ctrl1 = time.perf_counter() with self._vida_lock:
if not self._sessao_camera_valida(camera_atual, generation):
self.perf.inc("det_descartada_geracao_depois")
continue
t_pub_cmd0 = time.perf_counter() t_pub_analise0 = time.perf_counter()
self._set_pub_cache("cmd_controle", { self._set_pub_cache("analise", {
"cmd": WeedWorkerCommandType.EnviarDadosControle.value, "ts_analise": pred_ts,
"params": atuacao_bicos, "fps_model": pred_cache.get("fps_model", 0.0),
"ts": time.time(), "fps_inferencia": self._fps_infer_ema,
"origem": "weed_worker", "infer_ms": pred_cache.get("infer_ms", 0.0),
"pulverizacao_permitida": bool(debug_pulverizacao.get("permitida", False)), "infer_gpu_ms": pred_cache.get("infer_gpu_ms", 0.0),
"motivo": debug_pulverizacao.get("motivo", ""), "analise": analise_convertida,
"forcado_off": not bool(debug_pulverizacao.get("permitida", False)), "generation": generation,
}) }, ts_origem=pred_ts)
t_pub_cmd1 = time.perf_counter() t_pub_analise1 = time.perf_counter()
self._ultima_analise = analise_completa.copy() self._ultimo_controle = atuacao_bicos
self._ultimo_cmd_bicos = atuacao_bicos
# Só considera o ciclo válido quando análise, gate e comando já foram preparados com sucesso. t_pub_ctrl0 = time.perf_counter()
self._ultima_deteccao_ok_ts = time.time() self._set_pub_cache("controle_bicos", atuacao_bicos)
t_pub_ctrl1 = time.perf_counter()
t_pub_cmd0 = time.perf_counter()
self._set_pub_cache("cmd_controle", {
"cmd": WeedWorkerCommandType.EnviarDadosControle.value,
"params": atuacao_bicos,
"ts": time.time(),
"origem": "weed_worker",
"pulverizacao_permitida": bool(debug_pulverizacao.get("permitida", False)),
"motivo": debug_pulverizacao.get("motivo", ""),
"forcado_off": not bool(debug_pulverizacao.get("permitida", False)),
"generation": generation,
})
t_pub_cmd1 = time.perf_counter()
self._ultima_analise = analise_completa.copy()
self._ultima_deteccao_ok_ts = time.time()
t_loop1 = time.perf_counter() t_loop1 = time.perf_counter()
total_ms = (t_loop1 - t_loop0) * 1000.0 total_ms = (t_loop1 - t_loop0) * 1000.0
@ -1458,6 +1660,7 @@ class CameraManager:
"pred_age_ms": (time.time() - pred_ts) * 1000.0 if pred_ts else None, "pred_age_ms": (time.time() - pred_ts) * 1000.0 if pred_ts else None,
"tensor_age_ms": (time.time() - tensor_ts) * 1000.0 if tensor_ts else None, "tensor_age_ms": (time.time() - tensor_ts) * 1000.0 if tensor_ts else None,
"pulverizacao": debug_pulverizacao, "pulverizacao": debug_pulverizacao,
"generation": generation,
} }
pub_cache_ms = ( pub_cache_ms = (
@ -2327,13 +2530,19 @@ class CameraManager:
) )
continue continue
def salvar_pp_async(nome_frame_async, pasta_async): with self._vida_lock:
camera_save = self.camera
generation_save = self._camera_generation
def salvar_pp_async(nome_frame_async, pasta_async, camera_esperada, generation_esperada):
try: try:
t_pp0 = time.perf_counter() t_pp0 = time.perf_counter()
caminhos = self._salvar_bruto_multiespectral( caminhos = self._salvar_bruto_multiespectral(
nome_base=nome_frame_async, nome_base=nome_frame_async,
pasta=pasta_async pasta=pasta_async,
camera_esperada=camera_esperada,
generation_esperada=generation_esperada,
) )
t_pp1 = time.perf_counter() t_pp1 = time.perf_counter()
@ -2355,7 +2564,7 @@ class CameraManager:
threading.Thread( threading.Thread(
target=salvar_pp_async, target=salvar_pp_async,
args=(nome_frame, pasta), args=(nome_frame, pasta, camera_save, generation_save),
daemon=True daemon=True
).start() ).start()
@ -2438,23 +2647,44 @@ class CameraManager:
except Exception: except Exception:
return frame return frame
def _salvar_bruto_multiespectral(self, nome_base: str, pasta: str): def _salvar_bruto_multiespectral(
self,
nome_base: str,
pasta: str,
camera_esperada=None,
generation_esperada=None,
):
try: try:
if not hasattr(self, "camera") or self.camera is None: with self._vida_lock:
camera_atual = self.camera
generation_atual = self._camera_generation
if camera_esperada is not None and camera_atual is not camera_esperada:
self.mostrar_log("[weed][SAVE_PP] cancelado: câmera foi substituída")
return []
if generation_esperada is not None and int(generation_atual) != int(generation_esperada):
self.mostrar_log("[weed][SAVE_PP] cancelado: geração da câmera mudou")
return []
if camera_atual is None:
self.mostrar_log("⚠️ CameraMultispectral indisponível para salvar RAW multispec.") self.mostrar_log("⚠️ CameraMultispectral indisponível para salvar RAW multispec.")
return [] return []
if not hasattr(self.camera, "salvar_bundle_raw_multispec"): if not hasattr(camera_atual, "salvar_bundle_raw_multispec"):
self.mostrar_log("⚠️ CameraMultispectral ainda não possui salvar_bundle_raw_multispec().") self.mostrar_log(
"⚠️ CameraMultispectral ainda não possui salvar_bundle_raw_multispec()."
)
return [] return []
return self.camera.salvar_bundle_raw_multispec( return camera_atual.salvar_bundle_raw_multispec(
pasta=pasta, pasta=pasta,
nome=nome_base, nome=nome_base,
nota="pos_processamento_operacao", nota="pos_processamento_operacao",
extra_meta={ extra_meta={
"origem_chamada": "weed_worker.salvar_frames", "origem_chamada": "weed_worker.salvar_frames",
"tipo_requisitado": "PosProcessamento", "tipo_requisitado": "PosProcessamento",
"camera_generation": generation_atual,
}, },
) )