ajustes saude ipb e stream de video
This commit is contained in:
Diego Freitas
parent
769cf2a4d5
commit
0dfa272660
|
|
@ -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,
|
||||
)
|
||||
|
|
@ -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"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue