Merge branch 'op_control' of http://zendioninc.com.br:3000/Zendion/agrobot_base into op_control

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._sock_timeout_conexao_s = 1.0
-        self._last_mode_change_ts = 0.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):
+        # Recupera qualidade lentamente.
-        self._sock_conectado = False
+        self._last_mode_up_ts = agora_mono
-        if self._sock is not None:
+        self._mode_up_interval_s = 6.0
-            try:
+
-                self._sock.close()
+        # Reduz qualidade rapidamente.
-            except:
+        self._last_mode_down_ts = 0.0
-                pass
+        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 {}
+        try:
-        _ip = _cfg.get("base_ip")
+            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 not ip_base or not self.sock_port:
-        if (agora - self._sock_ultima_tentativa_conexao) < self._sock_intervalo_reconexao:
+            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)
+            novo_socket = socket.socket(
-            s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+                socket.AF_INET,
-            s.settimeout(1.0)
+                socket.SOCK_STREAM,
-            s.connect((_ip, self.sock_port))
+            )
-            s.settimeout(None)
+
+            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
+            return None
-            self._sock = None
 
-    def _get_ipb_bw_util_pct(self) -> float:
+        # IPB ainda não publicou saúde.
-        ipb = ContextoGlobalRedis.get_modulo(T_Code.Ipb) or {}
+        if not saude:
-        saude = (ipb.get("saude") or {})
+            return None
-        detalhes = (saude.get("detalhes") or {})
+
-        return float(detalhes.get("bw_pressure_pct") or detalhes.get("bw_util_pct") or 0.0)
+        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),
+        Os limites são máximos. A resolução real pode ser menor,
-            2: dict(fps=3.0, q=45, w=640, h=360),
+        pois resize_frame preserva a proporção original.
-            1: dict(fps=2.0, q=40, w=480, h=270),
+        """
-            0: dict(fps=1.0, q=35, w=320, h=180),
+        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()
+        pressure = self._get_ipb_bw_pressure()
-        now = time.time()
+        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._streaming_ativo = False
-            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
 
-        if util > 85 and self._op_mode > 0:
+        self._fechar_socket()
-            self._op_mode = max(0, self._op_mode - 1)
+
-            self._last_mode_change_ts = now
+        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)
+        # CONFIGURAÇÃO DA CÂMERA
-        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)
 
         try:
-            self._sock.sendall(header + data)
+            camera = (
-        except (BrokenPipeError, ConnectionResetError, OSError):
+                ContextoGlobalRedis.get_camera(self.mx_id)
-            self._sock_conectado = False
+                or {}
-            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"],
             )
+
+        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,
+        )

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_nominal_max_mbps = 10.0
-        self.bw_safe_mbps = 05.0   # alvo saudável (até aqui não penaliza)
+        self._bw_nominal_safe_mbps = 6.0
-        self.bw_hard_mbps = 2.5   # acima disso é zona vermelha
+        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.
+        Ajusta os limites de banda usando a distância da base.
-        Curva conservadora para Wi-Fi HaLow 915 MHz em campo.
+
+        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 (
-        if distancia_m is None or distancia_m < 0:
+            distancia_m is None
-            self.bw_max_mbps  = 10.0
+            or not math.isfinite(float(distancia_m))
-            self.bw_safe_mbps = 05.0
+            or distancia_m < 0
-            self.bw_hard_mbps = 02.5
+        ):
+            self._aplicar_limites_nominais_banda()
             return
 
-        # limita a curva
         d = max(10.0, min(float(distancia_m), 1200.0))
 
-        # 10 m -> 10 Mbps
+        bw_max = self._interp_log(
-        # 1200 m -> 0.4 Mbps
+            d,
-        bw_max = self._interp_log(d, 10.0, 1200.0, 10.0, 0.4)
+            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_safe = max(0.20, bw_max * 0.60)
-        bw_hard = bw_max * 0.78
+        bw_hard = max(bw_safe + 0.05, bw_max * 0.78)
 
-        # garantias mínimas
+        self.bw_max_mbps = round(bw_max, 3)
-        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_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:
+            # MARCADORES DE DISTÂNCIA
-                self._max_distancia_desde_degradado = max(
+            # Só possuem validade quando o GNSS fornece posição confiável
-                    self._max_distancia_desde_degradado,
+            # ---------------------------------------------------------
-                    distancia_base
-                )
 
-            # entrou em CRITICAL agora
+            if not regra_distancia_ativa:
-            if prev_state != "CRITICAL" and self.link_state == "CRITICAL":
+                # Sem posição confiável, não podemos associar a degradação
-                self._distancia_inicio_critico = distancia_base if distancia_base >= 0 else 0.0
+                # do enlace a uma distância física.
-
-            # voltou para OK, limpa marcos
-            if prev_state != "OK" and self.link_state == "OK":
                 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:
+            else:
-                self._max_distancia_ok = max(self._max_distancia_ok, distancia_base)
+                # 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": [
+                    "acoes": 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"
-                    ]
                 })
             # Condição 2: Cerca invisível de rede violada
             dist_violando_cerca = (
-                self.link_state == "DEGRADED" and
+                regra_distancia_ativa
-                self._cycles_degraded >= 15 and
+                and self.link_state == "DEGRADED"
-                risk_score >= 70 and
+                and self._cycles_degraded >= 15
-                self._distancia_inicio_degradado is not None and
+                and risk_score >= 70
-                distancia_base is not None and
+                and self._distancia_inicio_degradado is not None
-                distancia_base >= 0 and
+                and distancia_base is not None
-                distancia_base > self._distancia_inicio_degradado + 5.0
+                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"
+                    ),
                 }
             }