diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/tcp_streamer.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/tcp_streamer.py index 932726768..10ab74e4b 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/tcp_streamer.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/tcp_streamer.py @@ -1,155 +1,836 @@ import socket import struct +import threading import time + import cv2 -import numpy as np from shared.enums import T_Code from shared.utils import resize_frame from shared.contexto_global_redis import ContextoGlobalRedis + class CameraTcpStreamer: def __init__(self, porta: int, mx_id: str, mostrar_log=None): self.mx_id = mx_id self.mostrar_log = mostrar_log or (lambda msg: None) - self._sock = None + # --------------------------------------------------------- + # SOCKET TCP + # --------------------------------------------------------- + self.sock_port = porta + + self._sock = None self._sock_conectado = False + self._sock_lock = threading.RLock() + self._sock_ultima_tentativa_conexao = 0.0 self._sock_intervalo_reconexao = 1.0 - self._last_frame_sent_ts = 0.0 - self._last_mode_change_ts = 0.0 + self._sock_timeout_conexao_s = 1.0 + self._sock_timeout_envio_s = 0.75 + + # --------------------------------------------------------- + # CONTROLE ADAPTATIVO + # --------------------------------------------------------- + + agora_mono = time.monotonic() + self._op_mode = 1 self._op_fps = 2.0 - def fechar(self): - self._sock_conectado = False - if self._sock is not None: - try: - self._sock.close() - except: - pass + # Recupera qualidade lentamente. + self._last_mode_up_ts = agora_mono + self._mode_up_interval_s = 6.0 + + # Reduz qualidade rapidamente. + self._last_mode_down_ts = 0.0 + self._mode_down_interval_s = 1.0 + + # --------------------------------------------------------- + # MÉTRICAS DO STREAMING + # --------------------------------------------------------- + + self._last_frame_sent_mono = 0.0 + self._stream_kbps_ema = 0.0 + self._streaming_ativo = False + + # --------------------------------------------------------- + # CONTROLE DE LOG + # --------------------------------------------------------- + + self._log_intervalo_s = 5.0 + self._last_log_ts = {} + + # ============================================================= + # LOG + # ============================================================= + + def _log(self, msg: str): + try: + self.mostrar_log(msg) + except Exception: + pass + + def _log_throttled(self, chave: str, msg: str): + """ + Evita inundar os logs quando uma falha persiste. + """ + agora = time.monotonic() + ultimo = self._last_log_ts.get(chave, 0.0) + + if (agora - ultimo) >= self._log_intervalo_s: + self._last_log_ts[chave] = agora + self._log(msg) + + # ============================================================= + # CONTEXTO REDIS + # ============================================================= + + def _publicar_estado(self, **campos): + """ + Publicação de telemetria não pode derrubar o streaming caso + ocorra alguma falha temporária no contexto Redis. + """ + try: + ContextoGlobalRedis.atualizar_ctx_dict( + ContextoGlobalRedis.CamKey(self.mx_id), + **campos, + ) + except Exception as e: + self._log_throttled( + "redis_publish", + ( + f"[{self.mx_id}] Falha ao publicar estado " + f"do streaming: {e}" + ), + ) + + # ============================================================= + # SOCKET + # ============================================================= + + def _fechar_socket(self): + """ + Fecha somente o socket, sem publicar estado. + Pode ser usado durante reconexões internas. + """ + with self._sock_lock: + sock = self._sock + self._sock = None + self._sock_conectado = False + + if sock is not None: + try: + sock.shutdown(socket.SHUT_RDWR) + except Exception: + pass + + try: + sock.close() + except Exception: + pass + + def fechar(self): + """ + Encerramento público do streamer. + """ + self._fechar_socket() + + self._stream_kbps_ema = 0.0 + self._last_frame_sent_mono = 0.0 + self._streaming_ativo = False + + self._publicar_estado( + stream_active=False, + stream_connected=False, + stream_kbps=0.0, + stream_fps=0.0, + ) def _sock_tentar_conectar(self): - _cfg = ContextoGlobalRedis.get_equipamento() or {} - _ip = _cfg.get("base_ip") + try: + cfg = ContextoGlobalRedis.get_equipamento() or {} + ip_base = cfg.get("base_ip") - if not _ip or not self.sock_port: + except Exception as e: + self._log_throttled( + "base_config_error", + f"[{self.mx_id}] Falha ao obter IP da base: {e}", + ) return - agora = time.time() - if (agora - self._sock_ultima_tentativa_conexao) < self._sock_intervalo_reconexao: + if not ip_base or not self.sock_port: + return + + agora = time.monotonic() + + if ( + agora - self._sock_ultima_tentativa_conexao + ) < self._sock_intervalo_reconexao: return self._sock_ultima_tentativa_conexao = agora - self.fechar() + # Remove qualquer socket anterior antes de reconectar. + self._fechar_socket() + + novo_socket = None try: - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) - s.settimeout(1.0) - s.connect((_ip, self.sock_port)) - s.settimeout(None) + novo_socket = socket.socket( + socket.AF_INET, + socket.SOCK_STREAM, + ) + + novo_socket.setsockopt( + socket.IPPROTO_TCP, + socket.TCP_NODELAY, + 1, + ) + + novo_socket.setsockopt( + socket.SOL_SOCKET, + socket.SO_KEEPALIVE, + 1, + ) + + # Timeout usado somente durante a conexão. + novo_socket.settimeout( + self._sock_timeout_conexao_s + ) + + novo_socket.connect( + (ip_base, self.sock_port) + ) + + # Timeout mantido para impedir sendall bloqueado + # indefinidamente em caso de enlace congestionado. + novo_socket.settimeout( + self._sock_timeout_envio_s + ) + + with self._sock_lock: + self._sock = novo_socket + self._sock_conectado = True + + self._publicar_estado( + stream_connected=True, + stream_last_error=None, + ) + + self._log( + f"[{self.mx_id}] Socket de vídeo conectado." + ) + + except Exception as e: + if novo_socket is not None: + try: + novo_socket.close() + except Exception: + pass + + with self._sock_lock: + self._sock = None + self._sock_conectado = False + + self._publicar_estado( + stream_connected=False, + stream_kbps=0.0, + stream_fps=0.0, + stream_last_error=str(e), + ) + + self._log_throttled( + "connect_error", + ( + f"[{self.mx_id}] Falha ao conectar " + f"socket de vídeo: {e}" + ), + ) + + # ============================================================= + # PRESSÃO DO ENLACE + # ============================================================= + + def _get_ipb_bw_pressure(self): + """ + Retorna a pressão atual do enlace entre 0 e 100. + + None significa que ainda não existem dados confiáveis + suficientes para alterar o modo atual. + """ + try: + ipb = ( + ContextoGlobalRedis.get_modulo(T_Code.Ipb) + or {} + ) + + saude = ipb.get("saude") or {} + detalhes = saude.get("detalhes") or {} - self._sock = s - self._sock_conectado = True - self.mostrar_log(f"[{self.mx_id}] Socket de vídeo conectado.") except Exception: - self._sock_conectado = False - self._sock = None + return None - def _get_ipb_bw_util_pct(self) -> float: - ipb = ContextoGlobalRedis.get_modulo(T_Code.Ipb) or {} - saude = (ipb.get("saude") or {}) - detalhes = (saude.get("detalhes") or {}) - return float(detalhes.get("bw_pressure_pct") or detalhes.get("bw_util_pct") or 0.0) + # IPB ainda não publicou saúde. + if not saude: + return None + + conectado = bool( + saude.get("conectado", False) + ) + + valor = detalhes.get("bw_pressure_pct") + + if valor is None: + valor = detalhes.get("bw_util_pct") + + # O IPB publicou estado, mas está desconectado. + # Força comportamento conservador. + if valor is None: + return 90.0 if not conectado else None + + try: + pressure = float(valor) + + except (TypeError, ValueError): + return 90.0 if not conectado else None + + pressure = max( + 0.0, + min(100.0, pressure), + ) + + if not conectado: + pressure = max(pressure, 90.0) + + return pressure + + # ============================================================= + # MODOS DE STREAMING + # ============================================================= def _mode_params(self, mode: int): - table = { - 3: dict(fps=5.0, q=55, w=640, h=360), - 2: dict(fps=3.0, q=45, w=640, h=360), - 1: dict(fps=2.0, q=40, w=480, h=270), - 0: dict(fps=1.0, q=35, w=320, h=180), + """ + Os limites são máximos. A resolução real pode ser menor, + pois resize_frame preserva a proporção original. + """ + tabela = { + 3: { + "fps": 5.0, + "q": 55, + "w": 640, + "h": 360, + }, + 2: { + "fps": 3.0, + "q": 45, + "w": 640, + "h": 360, + }, + 1: { + "fps": 2.0, + "q": 40, + "w": 480, + "h": 270, + }, + 0: { + "fps": 1.0, + "q": 35, + "w": 320, + "h": 180, + }, } - return table.get(int(mode), table[1]) + + return tabela.get( + int(mode), + tabela[1], + ) def _maybe_update_mode(self): - util = self._get_ipb_bw_util_pct() - now = time.time() + pressure = self._get_ipb_bw_pressure() + agora = time.monotonic() - if (now - self._last_mode_change_ts) < 1.5: + if pressure is None: + return None + + # --------------------------------------------------------- + # EMERGÊNCIA + # Pressão superior a 85 pode reduzir até dois níveis. + # --------------------------------------------------------- + + if pressure > 85.0: + if ( + self._op_mode > 0 + and ( + agora - self._last_mode_down_ts + ) >= self._mode_down_interval_s + ): + self._op_mode = max( + 0, + self._op_mode - 2, + ) + + self._last_mode_down_ts = agora + + return pressure + + # --------------------------------------------------------- + # PRESSÃO ALTA + # Reduz um nível rapidamente. + # --------------------------------------------------------- + + if pressure > 75.0: + if ( + self._op_mode > 0 + and ( + agora - self._last_mode_down_ts + ) >= self._mode_down_interval_s + ): + self._op_mode -= 1 + self._last_mode_down_ts = agora + + return pressure + + # --------------------------------------------------------- + # ZONA NEUTRA + # Entre 50 e 75 não aumenta nem reduz. + # --------------------------------------------------------- + + if pressure >= 50.0: + return pressure + + # --------------------------------------------------------- + # ENLACE FOLGADO + # Recupera somente um nível por vez e lentamente. + # --------------------------------------------------------- + + if ( + self._op_mode < 3 + and ( + agora - self._last_mode_up_ts + ) >= self._mode_up_interval_s + ): + self._op_mode += 1 + self._last_mode_up_ts = agora + + return pressure + + # ============================================================= + # ESTADO INATIVO + # ============================================================= + + def _set_streaming_inativo(self): + """ + Executa a limpeza apenas na transição ligado -> desligado. + """ + if not self._streaming_ativo: return - if util < 55 and self._op_mode < 3: - self._op_mode += 1 - self._last_mode_change_ts = now - elif util > 75 and self._op_mode > 0: - self._op_mode -= 1 - self._last_mode_change_ts = now + self._streaming_ativo = False - if util > 85 and self._op_mode > 0: - self._op_mode = max(0, self._op_mode - 1) - self._last_mode_change_ts = now + self._fechar_socket() + + self._stream_kbps_ema = 0.0 + self._last_frame_sent_mono = 0.0 + + self._publicar_estado( + stream_active=False, + stream_connected=False, + stream_kbps=0.0, + stream_fps=0.0, + ) + + # ============================================================= + # ENVIO + # ============================================================= def enviar_frame_tcp(self, frame_bgr): - _camera = ContextoGlobalRedis.get_camera(self.mx_id) or {} - _stream_on = _camera.get("streaming", False) - if not _stream_on: - return - - self._maybe_update_mode() - params = self._mode_params(self._op_mode) - self._op_fps = params["fps"] - - now = time.time() - min_dt = 1.0 / max(0.1, self._op_fps) - if (now - self._last_frame_sent_ts) < min_dt: - return - - prev_sent_ts = self._last_frame_sent_ts - self._last_frame_sent_ts = now - - if not self._sock_conectado or self._sock is None: - self._sock_tentar_conectar() - if not self._sock_conectado or self._sock is None: - return - - if frame_bgr is None: - frame_bgr = np.zeros((params["h"], params["w"], 3), dtype=np.uint8) - else: - frame_bgr = resize_frame(frame_bgr, max_width=params["w"], max_height=params["h"]) - - encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), int(params["q"])] - ok, buf = cv2.imencode(".jpg", frame_bgr, encode_param) - if not ok: - return - data = buf.tobytes() - size = len(data) - header = struct.pack("!I", size) - - dt_real = max(0.001, now - prev_sent_ts) - fps_real = 1.0 / dt_real - kbps_inst = (size * 8) / dt_real / 1000.0 - - kbps_ant = float(_camera.get("stream_kbps", 0.0) or 0.0) - kbps_suave = kbps_inst if kbps_ant <= 0 else (kbps_ant * 0.8 + kbps_inst * 0.2) + # --------------------------------------------------------- + # CONFIGURAÇÃO DA CÂMERA + # --------------------------------------------------------- try: - self._sock.sendall(header + data) - except (BrokenPipeError, ConnectionResetError, OSError): - self._sock_conectado = False - kbps_suave = 0.0 - self.fechar() - finally: - ContextoGlobalRedis.atualizar_ctx_dict( - ContextoGlobalRedis.CamKey(self.mx_id), - stream_last_frame=time.time(), - stream_kbps=round(kbps_suave, 1), - fps=round(fps_real, 1), - stream_w=params["w"], - stream_h=params["h"], + camera = ( + ContextoGlobalRedis.get_camera(self.mx_id) + or {} ) + + except Exception as e: + self._log_throttled( + "camera_context_error", + ( + f"[{self.mx_id}] Falha ao obter " + f"contexto da câmera: {e}" + ), + ) + return + + stream_on = bool( + camera.get("streaming", False) + ) + + if not stream_on: + self._set_streaming_inativo() + return + + if not self._streaming_ativo: + self._streaming_ativo = True + + self._publicar_estado( + stream_active=True + ) + + # --------------------------------------------------------- + # MODO ADAPTATIVO + # --------------------------------------------------------- + + pressure = self._maybe_update_mode() + + params = self._mode_params( + self._op_mode + ) + + self._op_fps = float( + params["fps"] + ) + + # --------------------------------------------------------- + # LIMITADOR DE FPS + # --------------------------------------------------------- + + agora_mono = time.monotonic() + min_dt = 1.0 / max( + 0.1, + self._op_fps, + ) + + if ( + self._last_frame_sent_mono > 0.0 + and ( + agora_mono + - self._last_frame_sent_mono + ) < min_dt + ): + return + + # --------------------------------------------------------- + # CONEXÃO + # --------------------------------------------------------- + + with self._sock_lock: + conectado = ( + self._sock_conectado + and self._sock is not None + ) + + if not conectado: + self._sock_tentar_conectar() + + with self._sock_lock: + conectado = ( + self._sock_conectado + and self._sock is not None + ) + + if not conectado: + return + + # --------------------------------------------------------- + # VALIDAÇÃO DO FRAME + # --------------------------------------------------------- + + frame_valido = ( + frame_bgr is not None + and hasattr(frame_bgr, "shape") + and hasattr(frame_bgr, "size") + and frame_bgr.size > 0 + and len(frame_bgr.shape) >= 2 + ) + + if not frame_valido: + self._publicar_estado( + stream_frame_valido=False, + stream_last_error=( + "Frame ausente ou inválido." + ), + ) + return + + # --------------------------------------------------------- + # REDIMENSIONAMENTO + # --------------------------------------------------------- + + try: + frame_redimensionado = resize_frame( + frame_bgr, + max_width=int(params["w"]), + max_height=int(params["h"]), + ) + + except Exception as e: + self._publicar_estado( + stream_frame_valido=False, + stream_last_error=( + f"Falha ao redimensionar frame: {e}" + ), + ) + + self._log_throttled( + "resize_error", + ( + f"[{self.mx_id}] Falha ao " + f"redimensionar frame: {e}" + ), + ) + return + + frame_redimensionado_valido = ( + frame_redimensionado is not None + and hasattr( + frame_redimensionado, + "shape", + ) + and hasattr( + frame_redimensionado, + "size", + ) + and frame_redimensionado.size > 0 + and len( + frame_redimensionado.shape + ) >= 2 + ) + + if not frame_redimensionado_valido: + self._publicar_estado( + stream_frame_valido=False, + stream_last_error=( + "Frame inválido após redimensionamento." + ), + ) + return + + stream_h_real, stream_w_real = ( + frame_redimensionado.shape[:2] + ) + + # --------------------------------------------------------- + # JPEG + # --------------------------------------------------------- + + encode_inicio = time.monotonic() + encode_erro = None + + try: + encode_param = [ + int(cv2.IMWRITE_JPEG_QUALITY), + int(params["q"]), + ] + + ok, buffer_jpeg = cv2.imencode( + ".jpg", + frame_redimensionado, + encode_param, + ) + + except Exception as e: + ok = False + buffer_jpeg = None + encode_erro = e + + encode_ms = ( + time.monotonic() - encode_inicio + ) * 1000.0 + + if not ok or buffer_jpeg is None: + erro_txt = ( + str(encode_erro) + if encode_erro is not None + else "cv2.imencode retornou falha." + ) + + self._publicar_estado( + stream_frame_valido=False, + stream_last_error=( + f"Falha ao codificar JPEG: {erro_txt}" + ), + stream_encode_ms=round( + encode_ms, + 1, + ), + ) + + self._log_throttled( + "encode_error", + ( + f"[{self.mx_id}] Falha ao " + f"codificar JPEG: {erro_txt}" + ), + ) + return + + data = buffer_jpeg.tobytes() + size = len(data) + + # Cabeçalho de quatro bytes em big-endian. + header = struct.pack( + "!I", + size, + ) + + # --------------------------------------------------------- + # ENVIO TCP + # --------------------------------------------------------- + + send_inicio = time.monotonic() + + try: + with self._sock_lock: + if ( + not self._sock_conectado + or self._sock is None + ): + return + + self._sock.sendall( + header + data + ) + + except ( + BrokenPipeError, + ConnectionResetError, + socket.timeout, + OSError, + ) as e: + send_ms = ( + time.monotonic() + - send_inicio + ) * 1000.0 + + self._fechar_socket() + self._stream_kbps_ema = 0.0 + + # O frame era válido. O que falhou foi o transporte. + self._publicar_estado( + stream_connected=False, + stream_kbps=0.0, + stream_fps=0.0, + stream_frame_valido=True, + stream_send_ms=round( + send_ms, + 1, + ), + stream_last_error=str(e), + ) + + self._log_throttled( + "send_error", + ( + f"[{self.mx_id}] Falha ao " + f"enviar frame TCP: {e}" + ), + ) + return + + # --------------------------------------------------------- + # ENVIO CONCLUÍDO + # --------------------------------------------------------- + + enviado_mono = time.monotonic() + + send_ms = ( + enviado_mono - send_inicio + ) * 1000.0 + + if self._last_frame_sent_mono > 0.0: + dt_real = max( + 0.001, + enviado_mono + - self._last_frame_sent_mono, + ) + else: + # Estimativa inicial para o primeiro frame. + dt_real = 1.0 / max( + 0.1, + self._op_fps, + ) + + self._last_frame_sent_mono = enviado_mono + + fps_real = 1.0 / dt_real + + kbps_inst = ( + size * 8.0 + ) / dt_real / 1000.0 + + # EMA para evitar métrica tremendo a cada JPEG. + if self._stream_kbps_ema <= 0.0: + self._stream_kbps_ema = kbps_inst + else: + self._stream_kbps_ema = ( + self._stream_kbps_ema * 0.8 + + kbps_inst * 0.2 + ) + + # --------------------------------------------------------- + # PUBLICAÇÃO DAS MÉTRICAS + # --------------------------------------------------------- + + self._publicar_estado( + stream_active=True, + stream_connected=True, + stream_frame_valido=True, + + # Timestamp de parede somente após envio confirmado. + stream_last_frame=time.time(), + + stream_kbps=round( + self._stream_kbps_ema, + 1, + ), + stream_fps=round( + fps_real, + 1, + ), + stream_target_fps=round( + self._op_fps, + 1, + ), + + # Dimensões realmente transmitidas. + stream_w=int( + stream_w_real + ), + stream_h=int( + stream_h_real + ), + + stream_mode=int( + self._op_mode + ), + stream_quality=int( + params["q"] + ), + + stream_pressure_pct=( + round(pressure, 1) + if pressure is not None + else None + ), + + stream_jpeg_bytes=int( + size + ), + stream_encode_ms=round( + encode_ms, + 1, + ), + stream_send_ms=round( + send_ms, + 1, + ), + + stream_last_error=None, + ) \ No newline at end of file diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/ponte_ip.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/ponte_ip.py index f51cfcd75..354e47d62 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/ponte_ip.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/ponte_ip.py @@ -86,9 +86,13 @@ class ModuloIPBribge(ModuloDiagnosticoBase): self._hb_last_rx_monotonic = 0.0 self._last_ping_sample_applied_ts = 0.0 - self.bw_max_mbps = 10.0 # capacidade aproximada do link (ajuste por teste) - self.bw_safe_mbps = 05.0 # alvo saudável (até aqui não penaliza) - self.bw_hard_mbps = 2.5 # acima disso é zona vermelha + self._bw_nominal_max_mbps = 10.0 + self._bw_nominal_safe_mbps = 6.0 + self._bw_nominal_hard_mbps = 7.8 + + self.bw_max_mbps = self._bw_nominal_max_mbps + self.bw_safe_mbps = self._bw_nominal_safe_mbps + self.bw_hard_mbps = self._bw_nominal_hard_mbps self.link_state = "OK" # OK | DEGRADED | CRITICAL self._cycles_degraded = 0 @@ -675,48 +679,118 @@ class ModuloIPBribge(ModuloDiagnosticoBase): t = (math.log10(x) - math.log10(x0)) / (math.log10(x1) - math.log10(x0)) return y0 + t * (y1 - y0) - def _ajustar_limites_banda_por_distancia(self, distancia_m: float): + def _ajustar_limites_banda_por_distancia(self, distancia_m: float | None): """ - Ajusta bw_max/safe/hard com base na distância da base. - Curva conservadora para Wi-Fi HaLow 915 MHz em campo. + Ajusta os limites de banda usando a distância da base. + + Quando não há posição confiável, aplica o perfil nominal do enlace, + sem penalização baseada em distância. """ - # fallback quando não há posição válida - if distancia_m is None or distancia_m < 0: - self.bw_max_mbps = 10.0 - self.bw_safe_mbps = 05.0 - self.bw_hard_mbps = 02.5 + if ( + distancia_m is None + or not math.isfinite(float(distancia_m)) + or distancia_m < 0 + ): + self._aplicar_limites_nominais_banda() return - # limita a curva d = max(10.0, min(float(distancia_m), 1200.0)) - # 10 m -> 10 Mbps - # 1200 m -> 0.4 Mbps - bw_max = self._interp_log(d, 10.0, 1200.0, 10.0, 0.4) + bw_max = self._interp_log( + d, + 10.0, + 1200.0, + 10.0, + 0.4 + ) - # opcional: um pequeno piso e teto bw_max = max(0.35, min(bw_max, 10.0)) - bw_safe = bw_max * 0.60 - bw_hard = bw_max * 0.78 + bw_safe = max(0.20, bw_max * 0.60) + bw_hard = max(bw_safe + 0.05, bw_max * 0.78) - # garantias mínimas - bw_safe = max(0.20, bw_safe) - bw_hard = max(bw_safe + 0.05, bw_hard) - - self.bw_max_mbps = round(bw_max, 3) + self.bw_max_mbps = round(bw_max, 3) self.bw_safe_mbps = round(bw_safe, 3) self.bw_hard_mbps = round(bw_hard, 3) + def _normalizar_status_modulo(self, status): + if status is None: + return None + + # Caso tenha vindo como enum + return getattr(status, "value", status) + + def _obter_status_gnss(self): + """ + Retorna o status atual do módulo GNSS. + + Ajuste T_Code.Gps somente se o membro do enum tiver outro nome + no seu projeto, por exemplo T_Code.Gnss. + """ + modulo_gnss = ContextoGlobalRedis.get_modulo(T_Code.Gps) or {} + saude_gnss = modulo_gnss.get("saude") or {} + + status = saude_gnss.get( + "status", + modulo_gnss.get("status") + ) + + return self._normalizar_status_modulo(status) + + def _normalizar_distancia(self, valor): + try: + distancia = float(valor) + except (TypeError, ValueError): + return None + + if not math.isfinite(distancia) or distancia < 0: + return None + + return distancia + + def _posicao_confiavel(self, distancia_base): + status_gnss = self._obter_status_gnss() + + status_validos = { + StatusModulo.OPERANTE.value, + StatusModulo.ALERTA.value, + } + + confiavel = ( + status_gnss in status_validos + and distancia_base is not None + ) + + return confiavel, status_gnss + + def _aplicar_limites_nominais_banda(self): + self.bw_max_mbps = self._bw_nominal_max_mbps + self.bw_safe_mbps = self._bw_nominal_safe_mbps + self.bw_hard_mbps = self._bw_nominal_hard_mbps + def atualizar_saude_interno(self): #print("atualizando saude IPB...") try: m = ContextoGlobalRedis.get_modulo(self.t_code) or {} - distancia_base = m.get("distancia_base", -1) - self._ajustar_limites_banda_por_distancia(distancia_m=distancia_base) + + distancia_base_bruta = m.get("distancia_base") + distancia_base = self._normalizar_distancia(distancia_base_bruta) + + posicao_confiavel, status_gnss = self._posicao_confiavel( + distancia_base + ) + + regra_distancia_ativa = ( + posicao_confiavel + and distancia_base is not None + ) + + self._ajustar_limites_banda_por_distancia( + distancia_m=distancia_base if regra_distancia_ativa else None + ) self._check_mqtt_heartbeat() self._start_mqtt_heartbeat_async() @@ -1124,6 +1198,7 @@ class ModuloIPBribge(ModuloDiagnosticoBase): self._cycles_recovered = min(self._cycles_recovered, 30) prev_state = self.link_state + if self._cycles_critical >= 10: self.link_state = "CRITICAL" elif self._cycles_degraded >= 14: @@ -1131,33 +1206,71 @@ class ModuloIPBribge(ModuloDiagnosticoBase): elif self._cycles_recovered >= 24: self.link_state = "OK" - # entrou em DEGRADED agora - if prev_state != "DEGRADED" and self.link_state == "DEGRADED": - self._distancia_inicio_degradado = distancia_base if distancia_base >= 0 else 0.0 - self._max_distancia_desde_degradado = self._distancia_inicio_degradado - # enquanto estiver degradado, atualiza a maior distância observada - if self.link_state == "DEGRADED" and distancia_base is not None and distancia_base >= 0: - self._max_distancia_desde_degradado = max( - self._max_distancia_desde_degradado, - distancia_base - ) + # --------------------------------------------------------- + # MARCADORES DE DISTÂNCIA + # Só possuem validade quando o GNSS fornece posição confiável + # --------------------------------------------------------- - # entrou em CRITICAL agora - if prev_state != "CRITICAL" and self.link_state == "CRITICAL": - self._distancia_inicio_critico = distancia_base if distancia_base >= 0 else 0.0 - - # voltou para OK, limpa marcos - if prev_state != "OK" and self.link_state == "OK": + if not regra_distancia_ativa: + # Sem posição confiável, não podemos associar a degradação + # do enlace a uma distância física. self._distancia_inicio_degradado = None self._distancia_inicio_critico = None self._max_distancia_desde_degradado = 0.0 - if self.link_state == "OK" and distancia_base >= 0: - self._max_distancia_ok = max(self._max_distancia_ok, distancia_base) + else: + # O enlace está degradado e temos uma distância válida. + if self.link_state == "DEGRADED": + # Inicializa também se o GNSS voltou enquanto o enlace + # já estava no estado DEGRADED. + if self._distancia_inicio_degradado is None: + self._distancia_inicio_degradado = distancia_base + self._max_distancia_desde_degradado = distancia_base + + self._max_distancia_desde_degradado = max( + self._max_distancia_desde_degradado, + distancia_base + ) + + # Registra onde o enlace entrou ou foi observado como crítico. + if self.link_state == "CRITICAL": + if self._distancia_inicio_critico is None: + self._distancia_inicio_critico = distancia_base + + # Ao recuperar o enlace, limpa os marcos da degradação anterior. + if self.link_state == "OK": + if prev_state != "OK": + self._distancia_inicio_degradado = None + self._distancia_inicio_critico = None + self._max_distancia_desde_degradado = 0.0 + + self._max_distancia_ok = max( + self._max_distancia_ok, + distancia_base + ) + # Condição 1: Link degradado persistente if self.link_state == "DEGRADED": + if regra_distancia_ativa: + complemento_descricao = "Não aumentar distância da base." + acoes_degradacao = [ + "Bloquear avanço para longe da base", + "Reduzir vídeo/logs/telemetria não crítica", + "Permitir apenas manter posição ou retornar" + ] + else: + complemento_descricao = ( + "Posição GNSS indisponível. A degradação está sendo avaliada " + "somente pelas métricas reais do enlace." + ) + acoes_degradacao = [ + "Reduzir vídeo/logs/telemetria não crítica", + "Priorizar tráfego de controle e heartbeat", + "Verificar a estabilidade do enlace antes de continuar a operação" + ] + condicoes.append({ "label": "Enlace degradado persistente", "valor": round(risk_score, 1), @@ -1165,23 +1278,20 @@ class ModuloIPBribge(ModuloDiagnosticoBase): "descricao": ( f"Rede degradando de forma persistente. " f"Loss slow {loss_pct_s:.1f}%, timeout slow {tmout_pct_s:.1f}%, " - f"heartbeat {atraso_s:.1f}s. Não aumentar distância da base." + f"heartbeat {atraso_s:.1f}s. " + f"{complemento_descricao}" ), - "acoes": [ - "Bloquear avanço para longe da base", - "Reduzir vídeo/logs/telemetria não crítica", - "Permitir apenas manter posição ou retornar" - ] + "acoes": acoes_degradacao }) # Condição 2: Cerca invisível de rede violada dist_violando_cerca = ( - self.link_state == "DEGRADED" and - self._cycles_degraded >= 15 and - risk_score >= 70 and - self._distancia_inicio_degradado is not None and - distancia_base is not None and - distancia_base >= 0 and - distancia_base > self._distancia_inicio_degradado + 5.0 + regra_distancia_ativa + and self.link_state == "DEGRADED" + and self._cycles_degraded >= 15 + and risk_score >= 70 + and self._distancia_inicio_degradado is not None + and distancia_base is not None + and distancia_base > self._distancia_inicio_degradado + 5.0 ) if dist_violando_cerca: condicoes.append({ @@ -1255,7 +1365,6 @@ class ModuloIPBribge(ModuloDiagnosticoBase): "saude_individual": saude_individual, "condicoes_operacionais": condicoes, "detalhes": { - "d_base": distancia_base, "ls": ls, "tmout": tmout, "los": los, @@ -1290,6 +1399,19 @@ class ModuloIPBribge(ModuloDiagnosticoBase): "distancia_inicio_degradado": self._distancia_inicio_degradado, "distancia_inicio_critico": self._distancia_inicio_critico, "max_distancia_ok": self._max_distancia_ok, + + "status_gnss": status_gnss, + "posicao_confiavel": posicao_confiavel, + "regra_distancia_ativa": regra_distancia_ativa, + + "d_base": distancia_base if regra_distancia_ativa else -1.0, + "d_base_bruta": distancia_base_bruta, + + "perfil_limite_banda": ( + "adaptativo_por_distancia" + if regra_distancia_ativa + else "nominal_sem_posicao" + ), } }