"""Monitorización continua: ping, ARP, correlación de caídas.""" from __future__ import annotations import threading import time from collections import defaultdict, deque from dataclasses import dataclass, field from typing import Callable, Deque, Dict, List, Optional, Set from .arp_watch import ArpChange, ArpWatcher from .events import EventLog from .models import Device, DeviceStatus, epoch, now_short, now_str, status_from_stats from .oui import lookup_vendor from .pinger import ping_host, update_latency_stats @dataclass class TimelineEvent: ts: float ts_str: str ip: str kind: str # offline | online | conflict | loss details: str = "" class MonitorEngine: def __init__( self, event_log: EventLog, arp: ArpWatcher, on_update: Optional[Callable[[Device], None]] = None, on_timeline: Optional[Callable[[TimelineEvent], None]] = None, on_correlation: Optional[Callable[[str], None]] = None, ): self.event_log = event_log self.arp = arp self.on_update = on_update self.on_timeline = on_timeline self.on_correlation = on_correlation self.devices: Dict[str, Device] = {} self.interval = 2.0 self._stop = threading.Event() self._thread: Optional[threading.Thread] = None self._last_rtt: Dict[str, Optional[float]] = {} self.timeline: Deque[TimelineEvent] = deque(maxlen=2000) self.latency_series: Dict[str, Deque[tuple]] = defaultdict(lambda: deque(maxlen=600)) self.loss_series: Dict[str, Deque[tuple]] = defaultdict(lambda: deque(maxlen=600)) self.online_count_series: Deque[tuple] = deque(maxlen=600) self.arp_event_series: Deque[tuple] = deque(maxlen=600) self.conflicts: Set[str] = set() self._offline_burst: List[TimelineEvent] = [] def _on_arp(ch: ArpChange) -> None: if ch.kind == "conflict": self.conflicts.add(ch.ip) msg = f"{ch.old_mac} → {ch.new_mac}" self.event_log.critical(ch.ip, "IP CONFLICT", msg) te = TimelineEvent(epoch(), now_str(), ch.ip, "conflict", msg) self.timeline.append(te) self.arp_event_series.append((epoch(), 1)) if ch.ip in self.devices: d = self.devices[ch.ip] d.status = DeviceStatus.CONFLICT.value d.notes = f"⚠ CONFLICTO DE IP DETECTADO: {msg}" d.mac_history.append({"ts": ch.ts, "mac": ch.new_mac}) if self.on_update: self.on_update(d) if self.on_timeline: self.on_timeline(te) if self.on_correlation: self.on_correlation( f"POSIBLE CONFLICTO DE IP en {ch.ip}: MAC cambió de {ch.old_mac} a {ch.new_mac}" ) self.arp.on_change = _on_arp def set_targets(self, devices: List[Device]) -> None: for d in devices: self.devices[d.ip] = d def start(self, interval_sec: float = 2.0) -> None: self.interval = max(1.0, float(interval_sec)) self._stop.clear() if self._thread and self._thread.is_alive(): return self._thread = threading.Thread(target=self._loop, daemon=True) self._thread.start() self.event_log.info("-", "MONITOR START", f"intervalo={self.interval}s dispositivos={len(self.devices)}") def stop(self) -> None: self._stop.set() if self._thread: self._thread.join(timeout=3) self.event_log.info("-", "MONITOR STOP", "") @property def running(self) -> bool: return self._thread is not None and self._thread.is_alive() and not self._stop.is_set() def _loop(self) -> None: while not self._stop.is_set(): t0 = time.time() try: self.arp.refresh() online_n = 0 for ip, dev in list(self.devices.items()): if self._stop.is_set(): break self._tick(dev) if dev.status not in (DeviceStatus.OFFLINE.value,): online_n += 1 self.online_count_series.append((epoch(), online_n)) self._check_correlation() except Exception as e: self.event_log.error("-", "MONITOR ERROR", str(e)) elapsed = time.time() - t0 wait = max(0.1, self.interval - elapsed) self._stop.wait(wait) def _tick(self, dev: Device) -> None: was_online = dev.status != DeviceStatus.OFFLINE.value pr = ping_host(dev.ip, timeout_ms=800, count=1) # actualizar ARP tras ping table = ArpWatcher.read_arp_table() if dev.ip in table: mac = table[dev.ip].mac if mac: if not dev.mac: dev.mac = mac dev.vendor = lookup_vendor(mac) or dev.vendor self.arp.record_observation(dev.ip, mac) conflict = dev.ip in self.conflicts or any(c.ip == dev.ip for c in self.arp.conflicts()) if conflict: self.conflicts.add(dev.ip) dev.sent += pr.sent if pr.ok: dev.recv += pr.recv last = self._last_rtt.get(dev.ip) avg, rmin, rmax, jitter = update_latency_stats( dev.rtt_avg, dev.rtt_min, dev.rtt_max, dev.jitter, last, pr.rtt_ms, max(1, dev.recv) ) dev.rtt_avg, dev.rtt_min, dev.rtt_max, dev.jitter = avg, rmin, rmax, jitter self._last_rtt[dev.ip] = pr.rtt_ms dev.latency_ms = pr.rtt_ms self.latency_series[dev.ip].append((epoch(), pr.rtt_ms or 0)) if dev.sent: lost = max(0, dev.sent - dev.recv) dev.loss_pct = 100.0 * lost / dev.sent self.loss_series[dev.ip].append((epoch(), dev.loss_pct)) online = pr.ok new_status = status_from_stats(online, dev.loss_pct, dev.latency_ms, conflict) if online: if not was_online or (dev.offline_since is not None and not was_online): downtime = 0.0 if dev.offline_since is not None: downtime = epoch() - dev.offline_since self.event_log.info( dev.ip, "ONLINE", f"Ping {dev.latency_ms:.0f} ms — duración caída: {downtime:.0f} s" if downtime else f"Ping {dev.latency_ms:.0f} ms", ) te = TimelineEvent( epoch(), now_str(), dev.ip, "online", f"duración caída: {downtime:.0f} s" if downtime else "", ) self.timeline.append(te) if self.on_timeline: self.on_timeline(te) elif conflict: pass elif new_status == DeviceStatus.DEGRADED.value: self.event_log.warning(dev.ip, "PACKET LOSS", f"{dev.loss_pct:.0f}% lat={dev.latency_ms}") else: # no spamear INFO cada segundo pass dev.offline_since = None dev.last_seen = now_str() else: if was_online and (dev.status != DeviceStatus.OFFLINE.value or dev.offline_since is None): # transición a offline if dev.offline_since is None: dev.offline_since = epoch() self.event_log.error(dev.ip, "OFFLINE", f"{now_short()}") te = TimelineEvent(epoch(), now_str(), dev.ip, "offline", "") self.timeline.append(te) self._offline_burst.append(te) if self.on_timeline: self.on_timeline(te) if conflict: new_status = DeviceStatus.CONFLICT.value dev.status = new_status if self.on_update: self.on_update(dev) def _check_correlation(self) -> None: """Si varias cámaras caen en ventana corta → problema común.""" now = epoch() # limpiar burst viejo (>120s) self._offline_burst = [e for e in self._offline_burst if now - e.ts < 120] recent = [e for e in self._offline_burst if now - e.ts <= 60] ips = {e.ip for e in recent} if len(ips) >= 3: msg = ( f"Se detecta patrón de múltiples dispositivos afectados " f"({len(ips)} OFFLINE en ~60 s). Probable problema común de Wi-Fi/AP/router." ) self.event_log.critical("-", "CORRELACION", msg) if self.on_correlation: self.on_correlation(msg) # evitar spam: vaciar burst tras alerta self._offline_burst.clear() def diagnose_one(self, ip: str, ping_count: int = 20) -> str: """Diagnóstico detallado de una cámara.""" from .ports import scan_ports, format_port_report from .onvif_rtsp import camera_service_probe from .models import load_config cfg = load_config() ports_list = list(cfg.get("default_ports", [80, 443, 554, 37777])) lines = [f"DIAGNÓSTICO {ip}", "", f"IP: {ip}"] mac = self.arp.known_mac(ip) or (self.devices.get(ip).mac if ip in self.devices else "") if not mac: table = ArpWatcher.read_arp_table() mac = table.get(ip).mac if ip in table else "" vendor = lookup_vendor(mac) if mac else "" lines.append(f"MAC: {mac or '—'}") lines.append(f"Fabricante: {vendor or '—'}") lines.append("") # ping burst sent = recv = 0 rtts = [] for _ in range(ping_count): pr = ping_host(ip, timeout_ms=800, count=1) sent += pr.sent recv += pr.recv if pr.rtt_ms is not None: rtts.append(pr.rtt_ms) time.sleep(0.05) loss = 100.0 * (sent - recv) / sent if sent else 100.0 avg = sum(rtts) / len(rtts) if rtts else None mx = max(rtts) if rtts else None mn = min(rtts) if rtts else None lines.append(f"Ping: {sent} enviados / {recv} recibidos") lines.append(f"Pérdida: {loss:.0f}%") lines.append(f"Latencia media: {avg:.0f} ms" if avg is not None else "Latencia media: —") lines.append(f"Mínima: {mn:.0f} ms" if mn is not None else "Mínima: —") lines.append(f"Máxima: {mx:.0f} ms" if mx is not None else "Máxima: —") lines.append("") hist = self.arp.history_for(ip) if len(hist) >= 2 and hist[-1][1] != hist[0][1]: lines.append("ARP: INESTABLE / CAMBIOS DETECTADOS") for ts, m in hist: lines.append(f" {ts} {m}") conflict = True else: lines.append("ARP: ESTABLE") conflict = ip in self.conflicts lines.append(f"Conflicto IP: {'SÍ — DETECTADO' if conflict else 'NO DETECTADO'}") lines.append("") ports = scan_ports(ip, ports_list, timeout=1.5) lines.append(format_port_report(ip, ports)) lines.append("") try: svc = camera_service_probe(ip) rtsp = svc["rtsp"] onvif = svc["onvif"] lines.append(f"RTSP puerto 554: {'OPEN / servicio detectado' if rtsp.get('ok') else 'NO RESPONDE'}") if rtsp.get("needs_auth"): lines.append(" (RTSP requiere autenticación)") if onvif.get("ok"): lines.append( f"ONVIF: OK — {onvif.get('manufacturer','')} {onvif.get('model','')}".strip() ) else: lines.append(f"ONVIF: no disponible ({onvif.get('error','')})") except Exception as e: lines.append(f"ONVIF/RTSP: error {e}") lines.append("") online = recv > 0 status = status_from_stats(online, loss, avg, conflict) lines.append(f"Estado: {status}") lines.append("") # recomendación if conflict: rec = "Se detectó posible conflicto de IP (MAC cambiante). Reserve IPs fijas y revise DHCP." elif not online: rec = "La cámara no responde a ping. Compruebe Wi-Fi/AP, alimentación y si el NVR y la cámara están en la misma subred." elif loss >= 10: rec = "Se detecta pérdida de paquetes intermitente. Probable saturación o señal Wi-Fi débil/AP inestable." elif avg and avg >= 100: rec = "Latencia elevada. Revise canal Wi-Fi, interferencias y distancia al AP." elif not ports.get(554) and not ports.get(80): rec = "Ping OK pero puertos de cámara cerrados. ¿Firewall o IP de otro dispositivo?" else: rec = "Sin anomalías graves en esta prueba puntual. Si el NVR dice 'red inalcanzable', monitorice 15–30 min." lines.append(f"Recomendación:\n\"{rec}\"") lines.append("") # Video encoder (lag por config) try: from .onvif_rtsp import probe_video_encoders vid = probe_video_encoders(ip) lines.append("--- CONFIG VIDEO (posible causa de lag) ---") if vid.get("ok"): for i, p in enumerate(vid.get("profiles") or [], 1): lines.append( f" [{i}] {p.get('encoding','')} {p.get('width')}x{p.get('height')} " f"@ {p.get('fps')}fps bitrate={p.get('bitrate_kbps')} kbps I-frame/gov={p.get('gov_length')}" ) for fl in vid.get("lag_flags") or []: lines.append(f" FLAG: {fl}") for h in vid.get("manual_hints") or []: lines.append(f" QUE HACER: {h}") else: lines.append(f" No leido: {vid.get('error') or 'sin perfiles'}") lines.append( " Manual: http://" + ip + " → Video → Main CBR 1024–1536 kbps, 15 fps, I-frame=FPS; " "NVR live = sub-stream." ) except Exception as e: lines.append(f"Video probe error: {e}") lines.append("") from .manual_actions import MANUAL_GUIDES if conflict: guide_key = "conflicto_ip" elif not online: guide_key = "una_camara_offline" elif loss >= 10 or (avg and avg >= 100): guide_key = "degradado_wifi" elif online and loss < 10 and (ports.get(80) or ports.get(554)): guide_key = "fluidez_lag_wifi" else: guide_key = "monitorizar" lines.append("--- PASOS MANUALES ---") for step in MANUAL_GUIDES.get(guide_key, []): lines.append(step) return "\n".join(lines)