"""Interfaz gráfica de campo para NVR Net Diag.""" from __future__ import annotations import threading import tkinter as tk from tkinter import messagebox, simpledialog, ttk from typing import Dict, List, Optional from .arp_watch import ArpWatcher from .diagnose import run_full_diagnostic from .events import EventLog from .models import Device, NetInfo, load_config, now_str from .monitor import MonitorEngine, TimelineEvent from .netinfo import get_net_info, suspicious_dhcp_notes from .onvif_rtsp import camera_service_probe from .ports import format_port_report, scan_ports from .manual_actions import format_manual_section from .report import generate_html_report from .conflict_intel import analyze_conflicts from .ip816_visor import MANUAL_IP816_VISOR, run_ip816_visor_test from .perfect_lan import build_perfect_lan_plan from .fluency import run_fluency_analysis from .camera_caudal import ( PROFILES, apply_caudal, caudal_math_text, filter_cameras, pick_profile, ) from .site_inventory import ( enrich_device, enrich_devices, inventory_text, label_for, principal_hint_ip, ) from .switch_l2 import switch_l2_guide_text from .topology import run_topology_analysis from .nvr_focus_report import ( build_nvr_focus_report, generate_nvr_html_report, is_nvr_device, ) from .ws_discovery import ws_discovery_probe_both from .scanner import NetworkScanner APP_TITLE = "NVR Net Diag — Diagnóstico de red CCTV" class App(tk.Tk): def __init__(self): super().__init__() self.title(APP_TITLE) self.geometry("1280x820") self.minsize(1024, 700) self.cfg = load_config() self.net = NetInfo() self.devices: Dict[str, Device] = {} self.event_log = EventLog() self.arp = ArpWatcher() self.scanner: Optional[NetworkScanner] = None self.monitor: Optional[MonitorEngine] = None self._stop_full = False self._busy = False self._last_topology_text = "" self._discovery_ips: set = set() self._last_fluency_text = "" self._last_conflict_text = "" self._last_perfect_text = "" self._last_ip816_text = "" self._last_caudal_text = "" self._last_switch_text = "" self._build_ui() self.refresh_netinfo() self.event_log.subscribe(self._on_event) self.protocol("WM_DELETE_WINDOW", self._on_close) # ---------- UI ---------- def _build_ui(self) -> None: style = ttk.Style(self) try: style.theme_use("vista") except Exception: pass top = ttk.Frame(self, padding=8) top.pack(fill=tk.X) ttk.Label(top, text="Adaptador:").grid(row=0, column=0, sticky="w") self.cmb_adapter = ttk.Combobox(top, width=42, state="readonly") self.cmb_adapter.grid(row=0, column=1, padx=4) self.cmb_adapter.bind("<>", lambda e: self.refresh_netinfo(self.cmb_adapter.get())) ttk.Label(top, text="IP local:").grid(row=0, column=2, sticky="w", padx=(12, 0)) self.var_ip = tk.StringVar() ttk.Entry(top, textvariable=self.var_ip, width=16, state="readonly").grid(row=0, column=3) ttk.Label(top, text="Máscara:").grid(row=0, column=4, sticky="w", padx=(12, 0)) self.var_mask = tk.StringVar() ttk.Entry(top, textvariable=self.var_mask, width=16, state="readonly").grid(row=0, column=5) ttk.Label(top, text="Gateway:").grid(row=1, column=0, sticky="w", pady=4) self.var_gw = tk.StringVar() ttk.Entry(top, textvariable=self.var_gw, width=16, state="readonly").grid(row=1, column=1, sticky="w") ttk.Label(top, text="Red (CIDR):").grid(row=1, column=2, sticky="w", padx=(12, 0)) self.var_cidr = tk.StringVar() ttk.Entry(top, textvariable=self.var_cidr, width=18).grid(row=1, column=3) ttk.Label(top, text="DHCP:").grid(row=1, column=4, sticky="w", padx=(12, 0)) self.var_dhcp = tk.StringVar() ttk.Entry(top, textvariable=self.var_dhcp, width=18, state="readonly").grid(row=1, column=5) ttk.Label(top, text="Intervalo mon. (s):").grid(row=2, column=0, sticky="w") self.var_interval = tk.StringVar(value=str(self.cfg.get("monitor_interval_sec", 2))) ttk.Combobox( top, textvariable=self.var_interval, values=[str(x) for x in self.cfg.get("monitor_intervals_allowed", [1, 2, 5, 10])], width=6, state="readonly", ).grid(row=2, column=1, sticky="w") ttk.Label(top, text="Diag. completo (min):").grid(row=2, column=2, sticky="w", padx=(12, 0)) self.var_diag_min = tk.StringVar(value=str(self.cfg.get("default_full_diag_min", 15))) ttk.Combobox( top, textvariable=self.var_diag_min, values=["5", "15", "30", "60", "0 (continuo)"], width=14, state="readonly", ).grid(row=2, column=3, sticky="w") self.var_admin = tk.StringVar(value="") ttk.Label(top, textvariable=self.var_admin, foreground="#a60").grid( row=2, column=4, columnspan=2, sticky="w", padx=8 ) btns = ttk.Frame(self, padding=8) btns.pack(fill=tk.X) ttk.Button(btns, text="ESCANEAR RED", command=self.start_scan).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="INICIAR MONITORIZACIÓN", command=self.start_monitor).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="DETENER", command=self.stop_all).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="EXPORTAR INFORME", command=self.export_report).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="DIAGNÓSTICO COMPLETO", command=self.start_full_diag).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Diag. cámara", command=self.diagnose_selected).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Test puertos", command=self.test_ports_selected).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="ONVIF/RTSP", command=self.test_onvif_selected).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Actualizar red", command=self.refresh_netinfo).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Guía manual", command=self.show_manual_guide).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Switches / 7 cams", command=self.show_switch_l2).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Topología / Vista NVR", command=self.start_topology).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Informe desde NVR", command=self.start_nvr_focus_report).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Fluidez / Lag", command=self.start_fluency).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Conflictos IP / DHCP", command=self.start_conflict_intel).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="Plan red perfecta", command=self.start_perfect_lan).pack(side=tk.LEFT, padx=3) ttk.Button(btns, text="IP816 visores (7)", command=self.start_ip816_visors).pack(side=tk.LEFT, padx=3) btns2 = ttk.Frame(self, padding=(8, 0, 8, 4)) btns2.pack(fill=tk.X) ttk.Button(btns2, text="Caudal 12-16 (aplicar)", command=self.start_caudal).pack(side=tk.LEFT, padx=3) ttk.Label( btns2, text="Acota Main/Extra en las Dahua para que el IP816 (100M) vea 12–16. Por defecto SIMULA; no guarda la clave.", ).pack(side=tk.LEFT, padx=8) self.var_status = tk.StringVar(value="Listo. Conecte el portátil a la misma red que NVR/cámaras.") ttk.Label(self, textvariable=self.var_status, padding=4).pack(fill=tk.X) self.var_alert = tk.StringVar(value="") self.lbl_alert = ttk.Label(self, textvariable=self.var_alert, foreground="#b71c1c", padding=4) self.lbl_alert.pack(fill=tk.X) paned = ttk.Panedwindow(self, orient=tk.VERTICAL) paned.pack(fill=tk.BOTH, expand=True, padx=8, pady=4) # Tabla dispositivos frame_table = ttk.LabelFrame(paned, text="Dispositivos (marque filas para monitorizar)", padding=4) paned.add(frame_table, weight=3) cols = ("sel", "ip", "mac", "vendor", "host", "tipo", "estado", "lat", "loss", "last") self.tree = ttk.Treeview(frame_table, columns=cols, show="headings", selectmode="extended") headings = { "sel": "✓", "ip": "IP", "mac": "MAC", "vendor": "Fabricante", "host": "Hostname", "tipo": "Tipo", "estado": "Estado", "lat": "Latencia", "loss": "Pérdida", "last": "Última respuesta", } widths = { "sel": 36, "ip": 110, "mac": 130, "vendor": 110, "host": 120, "tipo": 80, "estado": 110, "lat": 80, "loss": 70, "last": 140, } for c in cols: self.tree.heading(c, text=headings[c]) self.tree.column(c, width=widths[c], anchor="w") scroll_y = ttk.Scrollbar(frame_table, orient=tk.VERTICAL, command=self.tree.yview) self.tree.configure(yscrollcommand=scroll_y.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scroll_y.pack(side=tk.RIGHT, fill=tk.Y) self.tree.bind("", self._toggle_select) self.tree.bind("", self._on_tree_click) self.tree.tag_configure("ONLINE", foreground="#1b5e20") self.tree.tag_configure("DEGRADADO", foreground="#e65100") self.tree.tag_configure("OFFLINE", foreground="#b71c1c") self.tree.tag_configure("CONFLICTO IP", foreground="#b71c1c", background="#ffcdd2") # Inferior: notebook eventos / timeline / gráficas texto bottom = ttk.Notebook(paned) paned.add(bottom, weight=2) # Eventos fr_ev = ttk.Frame(bottom, padding=4) bottom.add(fr_ev, text="Registro de eventos") filt = ttk.Frame(fr_ev) filt.pack(fill=tk.X) self.var_f_info = tk.BooleanVar(value=True) self.var_f_warn = tk.BooleanVar(value=True) self.var_f_err = tk.BooleanVar(value=True) self.var_f_crit = tk.BooleanVar(value=True) for text, var in ( ("INFO", self.var_f_info), ("WARNING", self.var_f_warn), ("ERROR", self.var_f_err), ("CRITICAL", self.var_f_crit), ): ttk.Checkbutton(filt, text=text, variable=var, command=self._refresh_events_view).pack( side=tk.LEFT, padx=4 ) self.txt_events = tk.Text(fr_ev, height=12, wrap=tk.NONE, font=("Consolas", 9)) self.txt_events.pack(fill=tk.BOTH, expand=True) sx = ttk.Scrollbar(self.txt_events, orient=tk.HORIZONTAL, command=self.txt_events.xview) self.txt_events.configure(xscrollcommand=sx.set) # Timeline / correlación fr_tl = ttk.Frame(bottom, padding=4) bottom.add(fr_tl, text="Línea temporal / correlación") self.txt_timeline = tk.Text(fr_tl, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_timeline.pack(fill=tk.BOTH, expand=True) # Stats / gráficas simples (texto + canvas) fr_g = ttk.Frame(bottom, padding=4) bottom.add(fr_g, text="Estadísticas / gráficas") self.canvas = tk.Canvas(fr_g, bg="#ffffff", height=200) self.canvas.pack(fill=tk.BOTH, expand=True) self.txt_stats = tk.Text(fr_g, height=6, font=("Consolas", 9)) self.txt_stats.pack(fill=tk.BOTH, expand=True) ttk.Button(fr_g, text="Redibujar gráficas", command=self.redraw_charts).pack(anchor="e") # Preguntas clave fr_q = ttk.Frame(bottom, padding=4) bottom.add(fr_q, text="Preguntas clave (técnico)") self.txt_answers = tk.Text(fr_q, height=12, wrap=tk.WORD, font=("Segoe UI", 10)) self.txt_answers.pack(fill=tk.BOTH, expand=True) self._write_answers_placeholder() # Topología / vista dispositivos fr_topo = ttk.Frame(bottom, padding=4) bottom.add(fr_topo, text="Topología / vista NVR") self.txt_topology = tk.Text(fr_topo, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_topology.pack(fill=tk.BOTH, expand=True) self.txt_topology.insert( "1.0", "Pulse «Topología / Vista NVR» tras escanear.\n\n" "Compara inventario (ping/puertos) con ONVIF WS-Discovery (lo que el NVR ve al Buscar)\n" "y estima, por cada NVR, cuántas cámaras debería encontrar en la misma LAN.\n\n" "Limitación: no se ejecuta código dentro del NVR; el PC debe estar en el mismo switch/LAN.", ) self.txt_topology.configure(state=tk.DISABLED) self._last_topology_text = "" self.nb_bottom = bottom # Fluidez / lag fr_flu = ttk.Frame(bottom, padding=4) bottom.add(fr_flu, text="Fluidez / Lag") self.txt_fluency = tk.Text(fr_flu, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_fluency.pack(fill=tk.BOTH, expand=True) self.txt_fluency.insert( "1.0", "Pulse «Fluidez / Lag» tras escanear.\n\n" "Mide latencia/jitter/pérdida y lee bitrate/resolución/FPS por ONVIF.\n" "Sirve para saber por qué hay retardo: red WiFi, config de cámara o carga multi-NVR.\n" "Puede usarse con el técnico en WiFi si la LAN es plana (llega a todo).", ) self.txt_fluency.configure(state=tk.DISABLED) self._last_fluency_text = "" fr_cf = ttk.Frame(bottom, padding=4) bottom.add(fr_cf, text="Conflictos IP / DHCP") self.txt_conflicts = tk.Text(fr_cf, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_conflicts.pack(fill=tk.BOTH, expand=True) self.txt_conflicts.insert( "1.0", "Pulse «Conflictos IP / DHCP» tras escanear (mejor con monitorizacion si hay cortes a ratos).\n\n" "Identifica QUE dispositivo choca (camara vs movil de la familia vs Tenda)\n" "y QUIEN reparte las IPs (Askey 192.168.1.1 vs Tenda/Cudy con DHCP mal puesto).", ) self.txt_conflicts.configure(state=tk.DISABLED) self._last_conflict_text = "" fr_pf = ttk.Frame(bottom, padding=4) bottom.add(fr_pf, text="Plan red perfecta") self.txt_perfect = tk.Text(fr_pf, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_perfect.pack(fill=tk.BOTH, expand=True) self.txt_perfect.insert( "1.0", "Pulse «Plan red perfecta» tras escanear.\n\n" "Calcula si 8 NVR x ~22 camaras pueden ir fluidos con Askey+Cudy+Tenda+switch.\n" "Si todos piden MAIN WiFi, es imposible; live=SUB es la regla con este equipo.", ) self.txt_perfect.configure(state=tk.DISABLED) self._last_perfect_text = "" fr_816 = ttk.Frame(bottom, padding=4) bottom.add(fr_816, text="IP816 visores (7)") self.txt_ip816 = tk.Text(fr_816, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_ip816.pack(fill=tk.BOTH, expand=True) self.txt_ip816.insert( "1.0", "Marque SOLO el NVR PRINCIPAL (☑) y pulse «IP816 visores (7)».\n\n" "Testea los otros PNI House IP816 (TVs) y muestra el manual HDMI:\n" "16x5MP, Extra/Sub en vivo, grabacion OFF, ONVIF por IP, NIC 100 Mbps.\n\n" + MANUAL_IP816_VISOR[:900] + "\n… (manual completo al ejecutar el test)", ) self.txt_ip816.configure(state=tk.DISABLED) self._last_ip816_text = "" self.nb_bottom = bottom fr_caudal = ttk.Frame(bottom, padding=4) bottom.add(fr_caudal, text="Caudal 12-16") self.txt_caudal = tk.Text(fr_caudal, height=12, wrap=tk.WORD, font=("Consolas", 9)) self.txt_caudal.pack(fill=tk.BOTH, expand=True) self.txt_caudal.insert( "1.0", "Pulse «Caudal 12-16 (aplicar)» tras escanear.\n\n" "Calcula si 12-16 camaras caben en el IP816 (10/100) y, con usuario/clave temporal " "de las Dahua, puede APLICAR Main/Extra (CBR acotado).\n" "Por defecto es SIMULACION. Extra/Sub en HDMI de los 7 visores sigue siendo obligatorio.\n\n" + caudal_math_text(), ) self.txt_caudal.configure(state=tk.DISABLED) self._last_caudal_text = "" self.nb_bottom = bottom def _write_answers_placeholder(self) -> None: self.txt_answers.delete("1.0", tk.END) self.txt_answers.insert( tk.END, "Tras escanear y monitorizar, aquí se responderán automáticamente:\n\n" "1. ¿Hay dos dispositivos con la misma IP?\n" "2. ¿Qué MAC corresponde a cada IP?\n" "3. ¿Alguna cámara pierde paquetes?\n" "4. ¿Cuándo empezó a perder conectividad?\n" "5. ¿Fallan individualmente o por problema común?\n" "6. ¿Hay cambios ARP sospechosos?\n" "7. ¿Siguen accesibles por red cuando el NVR dice 'red inalcanzable'?\n" "8. ¿El problema parece cámara, Wi-Fi/AP, router o IP?\n", ) # ---------- Net info ---------- def refresh_netinfo(self, preferred: Optional[str] = None) -> None: self.net = get_net_info(preferred) names = [a["name"] for a in self.net.adapters] or [self.net.adapter] self.cmb_adapter["values"] = names if self.net.adapter: self.cmb_adapter.set(self.net.adapter) self.var_ip.set(self.net.ip) self.var_mask.set(self.net.mask) self.var_gw.set(self.net.gateway) if not self.var_cidr.get(): self.var_cidr.set(self.net.cidr) elif preferred: self.var_cidr.set(self.net.cidr) dhcp = self.net.dhcp_server or ("(no detectado)" if self.net.ip else "—") self.var_dhcp.set(dhcp) if self.net.is_admin: self.var_admin.set("Privilegios: Administrador") else: self.var_admin.set("Privilegios: usuario (ARP/ping OK; sin elevación)") # ---------- Table helpers ---------- def _on_tree_click(self, event) -> None: region = self.tree.identify("region", event.x, event.y) if region != "cell": return col = self.tree.identify_column(event.x) if col == "#1": item = self.tree.identify_row(event.y) if item: self._toggle_item(item) def _toggle_select(self, event) -> None: item = self.tree.identify_row(event.y) if item: self._toggle_item(item) def _toggle_item(self, item: str) -> None: vals = list(self.tree.item(item, "values")) ip = vals[1] if ip in self.devices: self.devices[ip].selected = not self.devices[ip].selected vals[0] = "☑" if self.devices[ip].selected else "☐" self.tree.item(item, values=vals) def _upsert_device_row(self, d: Device) -> None: enrich_device(d, self.cfg) self.devices[d.ip] = d tag = d.status if d.status in ("ONLINE", "DEGRADADO", "OFFLINE", "CONFLICTO IP") else "" host = d.hostname or label_for(d.ip, self.cfg) or "—" values = ( "☑" if d.selected else "☐", d.ip, d.mac or "—", d.vendor or "—", host, d.device_type, d.status, f"{d.latency_ms:.0f} ms" if d.latency_ms is not None else "—", f"{d.loss_pct:.0f}%" if d.sent else "—", d.last_seen or "—", ) if self.tree.exists(d.ip): self.tree.item(d.ip, values=values, tags=(tag,)) else: self.tree.insert("", tk.END, iid=d.ip, values=values, tags=(tag,)) def _scan_done(self, n: int) -> None: enrich_devices(list(self.devices.values()), self.cfg) for d in list(self.devices.values()): self._upsert_device_row(d) known = sum(1 for ip in ("192.168.1.62", "192.168.1.64", "192.168.1.76", "192.168.1.57") if ip in self.devices) extra = f" | NVR conocidos vistos: {known}/4" if known else "" self.var_status.set(f"Escaneo terminado: {n} dispositivos activos.{extra}") self.update_key_answers() self.redraw_charts() def show_switch_l2(self) -> None: text = switch_l2_guide_text(list(self.devices.values()), self.net) self._last_switch_text = text self._show_text_result("Switches / sintoma 7 cámaras", text) self.update_key_answers() def _selected_devices(self) -> List[Device]: sel = [d for d in self.devices.values() if d.selected] if sel: return sel out = [] for iid in self.tree.selection(): if iid in self.devices: out.append(self.devices[iid]) return out # ---------- Scan ---------- def start_scan(self) -> None: if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operación actual.") return cidr = self.var_cidr.get().strip() if not cidr: messagebox.showerror("Error", "Indique una red CIDR (ej. 192.168.1.0/24).") return self._busy = True self.var_status.set(f"Escaneando {cidr}…") self.tree.delete(*self.tree.get_children()) self.devices.clear() def work(): try: self.scanner = NetworkScanner(arp=self.arp) devices = self.scanner.scan( cidr, on_progress=lambda d, t, m: self.after(0, lambda: self.var_status.set(f"{m} ({d}/{t})")), on_device=lambda dev: self.after(0, lambda: self._upsert_device_row(dev)), ) for c in self.arp.conflicts(): self.event_log.critical(c.ip, "IP CONFLICT", f"{c.old_mac} → {c.new_mac}") self.after(0, lambda c=c: self._raise_conflict(c.ip, c.old_mac, c.new_mac)) notes = suspicious_dhcp_notes(self.net, [d.ip for d in devices if d.device_type == "Camara"]) for n in notes: self.event_log.warning("-", "DHCP", n) self.after(0, lambda: self._scan_done(len(devices))) except Exception as e: self.after(0, lambda: messagebox.showerror("Error escaneo", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def _raise_conflict(self, ip: str, old: str, new: str) -> None: from .conflict_intel import _role_for_mac gw = self.net.gateway if self.net else "" devices = list(self.devices.values()) a = _role_for_mac(old, devices, gw) b = _role_for_mac(new, devices, gw) self.var_alert.set( f"⚠ CONFLICTO IP {ip}: {a.role} ({old}) vs {b.role} ({new})" ) messagebox.showwarning( "Conflicto de IP — dispositivos identificados", f"IP en conflicto: {ip}\n\n" f"Dispositivo A:\n MAC {old}\n {a.vendor or '-'} → {a.role}\n\n" f"Dispositivo B:\n MAC {new}\n {b.vendor or '-'} → {b.role}\n\n" "Pulse «Conflictos IP / DHCP» para pasos manuales concretos.\n" f"Hora: {now_str()}", ) # ---------- Monitor ---------- def start_monitor(self) -> None: targets = self._selected_devices() if not targets: # por defecto cámaras + NVR + gateway targets = [ d for d in self.devices.values() if d.device_type in ("Camara", "NVR", "Router/AP") or d.ip == self.net.gateway ] if not targets: messagebox.showinfo( "Sin objetivos", "Seleccione dispositivos (doble clic / columna ✓) o escanee primero la red.", ) return try: interval = float(self.var_interval.get().split()[0]) except Exception: interval = 2.0 if self.monitor and self.monitor.running: messagebox.showinfo("Monitor", "Ya hay una monitorización en curso.") return self.monitor = MonitorEngine( event_log=self.event_log, arp=self.arp, on_update=lambda d: self.after(0, lambda: self._upsert_device_row(d)), on_timeline=lambda e: self.after(0, lambda: self._append_timeline(e)), on_correlation=lambda m: self.after(0, lambda: self._on_correlation(m)), ) # preservar selección for t in targets: if t.ip in self.devices: t.selected = self.devices[t.ip].selected self.monitor.set_targets(targets) self.monitor.start(interval) self.var_status.set(f"Monitorizando {len(targets)} dispositivos cada {interval}s…") self.event_log.info("-", "UI", f"Monitor start {len(targets)} hosts") def _append_timeline(self, e: TimelineEvent) -> None: line = f"{e.ts_str} {e.ip} {e.kind.upper()}" if e.details: line += f" — {e.details}" self.txt_timeline.insert(tk.END, line + "\n") self.txt_timeline.see(tk.END) if e.kind == "conflict": self.var_alert.set(f"⚠ CONFLICTO DE IP — {e.ip}: {e.details}") self.update_key_answers() self.redraw_charts() def _on_correlation(self, msg: str) -> None: self.var_alert.set(msg) self.txt_timeline.insert(tk.END, f"\n>>> {msg}\n\n") self.txt_timeline.see(tk.END) self.event_log.critical("-", "CORRELACION", msg) def stop_all(self) -> None: self._stop_full = True if self.scanner: self.scanner.stop() if self.monitor and self.monitor.running: self.monitor.stop() self.var_status.set("Monitorización detenida.") else: self.var_status.set("Detenido.") # ---------- Events ---------- def _on_event(self, ev) -> None: self.after(0, self._refresh_events_view) def _refresh_events_view(self) -> None: levels = [] if self.var_f_info.get(): levels.append("INFO") if self.var_f_warn.get(): levels.append("WARNING") if self.var_f_err.get(): levels.append("ERROR") if self.var_f_crit.get(): levels.append("CRITICAL") self.txt_events.delete("1.0", tk.END) for e in self.event_log.all(levels): self.txt_events.insert(tk.END, e.as_line() + "\n") self.txt_events.see(tk.END) # ---------- Diagnose one ---------- def diagnose_selected(self) -> None: targets = self._selected_devices() if not targets: messagebox.showinfo("Diag.", "Seleccione al menos una cámara/dispositivo.") return ip = targets[0].ip self.var_status.set(f"Diagnóstico detallado {ip}…") def work(): mon = self.monitor or MonitorEngine(self.event_log, self.arp) if ip in self.devices: mon.devices[ip] = self.devices[ip] text = mon.diagnose_one(ip) self.after(0, lambda: self._show_text_result(f"Diagnóstico {ip}", text)) threading.Thread(target=work, daemon=True).start() def test_ports_selected(self) -> None: targets = self._selected_devices() if not targets: messagebox.showinfo("Puertos", "Seleccione un dispositivo.") return ip = targets[0].ip custom = simpledialog.askstring( "Puertos", "Lista de puertos (separados por coma):", initialvalue=",".join(str(p) for p in self.cfg.get("default_ports", [80, 554, 37777])), ) if not custom: return try: ports = [int(x.strip()) for x in custom.split(",") if x.strip()] except Exception: messagebox.showerror("Error", "Lista de puertos inválida.") return def work(): res = scan_ports(ip, ports) if ip in self.devices: self.devices[ip].ports.update(res) self.after(0, lambda: self._upsert_device_row(self.devices[ip])) text = format_port_report(ip, res) self.after(0, lambda: self._show_text_result(f"Puertos {ip}", text)) threading.Thread(target=work, daemon=True).start() def test_onvif_selected(self) -> None: targets = self._selected_devices() if not targets: messagebox.showinfo("ONVIF", "Seleccione una cámara.") return ip = targets[0].ip user = simpledialog.askstring("ONVIF (opcional)", "Usuario (vacío = sin auth, no se guarda):") or "" password = "" if user: password = simpledialog.askstring("ONVIF", "Contraseña (temporal, no se guarda):", show="*") or "" def work(): svc = camera_service_probe(ip, user=user, password=password) lines = [ f"ONVIF/RTSP — {ip}", "", f"RTSP ok: {svc['rtsp'].get('ok')} port_open={svc['rtsp'].get('port_open')}", f"RTSP status: {svc['rtsp'].get('status_line')}", f"RTSP auth: {svc['rtsp'].get('needs_auth')}", f"RTSP error: {svc['rtsp'].get('error')}", "", f"ONVIF ok: {svc['onvif'].get('ok')}", f"Endpoint: {svc['onvif'].get('endpoint')}", f"Fabricante: {svc['onvif'].get('manufacturer')}", f"Modelo: {svc['onvif'].get('model')}", f"Firmware: {svc['onvif'].get('firmware')}", f"Error: {svc['onvif'].get('error')}", "", "VIDEO / ENCODER (lag por configuracion)", f" ok: {svc.get('video', {}).get('ok')}", f" error: {svc.get('video', {}).get('error')}", ] for i, p in enumerate(svc.get("video", {}).get("profiles") or [], 1): lines.append( f" [{i}] {p.get('encoding','')} {p.get('width')}x{p.get('height')} " f"@ {p.get('fps')}fps {p.get('bitrate_kbps')} kbps gov={p.get('gov_length')}" ) for fl in svc.get("video", {}).get("lag_flags") or []: lines.append(f" FLAG: {fl}") for h in svc.get("video", {}).get("manual_hints") or []: lines.append(f" QUE HACER: {h}") self.after(0, lambda: self._show_text_result(f"ONVIF/RTSP {ip}", "\n".join(lines))) threading.Thread(target=work, daemon=True).start() def _show_text_result(self, title: str, text: str) -> None: self.var_status.set("Listo.") win = tk.Toplevel(self) win.title(title) win.geometry("640x520") txt = tk.Text(win, wrap=tk.WORD, font=("Consolas", 10)) txt.pack(fill=tk.BOTH, expand=True) txt.insert("1.0", text) txt.configure(state=tk.DISABLED) def show_manual_guide(self) -> None: """Muestra pasos manuales según el estado actual de la sesión.""" conflicts = self.arp.conflicts() timeline = list(self.monitor.timeline) if self.monitor else [] devices = list(self.devices.values()) dhcp_notes = ( suspicious_dhcp_notes( self.net, [d.ip for d in devices if d.device_type == "Camara"], ) if self.net else [] ) text = format_manual_section(conflicts, timeline, dhcp_notes, devices) text = switch_l2_guide_text(devices, self.net) + "\n\n" + text self._show_text_result("Guía manual — pasos y comandos", text) def _selected_nvr(self) -> Optional[Device]: """NVR marcado o seleccionado; si no, hint .62 (NVR1 Salon).""" for d in self._selected_devices(): if is_nvr_device(d): return d for iid in self.tree.selection(): if iid in self.devices and is_nvr_device(self.devices[iid]): return self.devices[iid] hint = principal_hint_ip(self.cfg) if hint and hint in self.devices and is_nvr_device(self.devices[hint]): return self.devices[hint] for d in self.devices.values(): if is_nvr_device(d): return d return None def start_nvr_focus_report(self) -> None: """Informe + checklist manual para un NVR concreto.""" if not self.devices: messagebox.showinfo( "Informe NVR", "Primero ESCANEAR RED.\n" "Conecte el portatil por cable al mismo switch que ese NVR.", ) return nvr = self._selected_nvr() if not nvr: messagebox.showinfo( "Informe NVR", "No se detecto ningun NVR en el escaneo.\n" "Marque (☑) la fila del NVR o pulse en su IP y vuelva a intentar.\n" "Si el NVR no aparece: escanee de nuevo con el PC en el mismo switch.", ) return if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operacion actual.") return if not messagebox.askyesno( "Informe desde NVR", f"Generar informe y checklist manual para NVR {nvr.ip}?\n\n" "Incluye IPs a anadir manualmente, pasos en NVR PNI y comandos Windows.\n" "El PC debe estar en la misma LAN/switch que ese NVR.", ): return self._busy = True self.var_status.set(f"Generando informe desde NVR {nvr.ip} (discovery ONVIF)…") devices = list(self.devices.values()) net = self.net def work(): try: disc = ws_discovery_probe_both( local_ip=net.ip if net else "", listen_sec=5.0, ) discovery_ips = set(disc.get("ips") or []) self._discovery_ips = discovery_ips report = build_nvr_focus_report(nvr, devices, net, discovery_ips) html_path = generate_nvr_html_report(report, net) text = report.as_text() def done(): self._last_topology_text = text self.txt_topology.configure(state=tk.NORMAL) self.txt_topology.delete("1.0", tk.END) self.txt_topology.insert("1.0", text) self.txt_topology.configure(state=tk.DISABLED) try: self.nb_bottom.select(4) except Exception: pass add_n = sum(1 for c in report.checklist if c.reachable and not c.in_discovery) self.var_status.set(f"Informe NVR {nvr.ip} -> {html_path}") self.var_alert.set( report.summary if add_n else f"NVR {nvr.ip}: sin IPs pendientes de anadir manual" ) self.update_key_answers() self._show_text_result(f"Informe NVR {nvr.ip}", text) messagebox.showinfo( "Informe NVR generado", f"{report.summary}\n\n" f"HTML: {html_path}\n\n" f"IPs a anadir manual: {add_n}\n" "Abra el HTML o la pestaña Topologia para el checklist completo.", ) self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Informe NVR", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def start_fluency(self) -> None: """Diagnostico de lag/retardo: red + config video + carga multi-NVR.""" if not self.devices: messagebox.showinfo( "Fluidez / Lag", "Primero ESCANEAR RED.\n" "Puede estar en WiFi si llega a todas las camaras; " "para medir como el NVR, preferible cable al switch.", ) return if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operacion actual.") return if not messagebox.askyesno( "Fluidez / Lag", "Medira ping/jitter de camaras y leera bitrate/resolucion ONVIF " "(puede tardar 1–3 min con ~22 camaras).\n\n" "No modifica ninguna camara; solo detecta y da pasos manuales.\n¿Continuar?", ): return self._busy = True self.var_status.set("Midiendo fluidez (ping + ONVIF video)…") devices = list(self.devices.values()) net = self.net def work(): try: report = run_fluency_analysis( devices, net, probe_video=True, ping_count=10, max_video_probes=18, ) text = report.as_text() def done(): self._last_fluency_text = text # Tambien en topologia si no hay informe NVR reciente largo self.txt_fluency.configure(state=tk.NORMAL) self.txt_fluency.delete("1.0", tk.END) self.txt_fluency.insert("1.0", text) self.txt_fluency.configure(state=tk.DISABLED) try: # pestaña Fluidez es la ultima (indice 5) self.nb_bottom.select(5) except Exception: pass bad = [c for c in report.cams if c.score < 60] self.var_alert.set( report.summary if not bad else f"{len(bad)} camara(s) con fluidez critica — ver pestaña Fluidez / Lag" ) self.var_status.set("Fluidez lista — pestaña «Fluidez / Lag»") self.update_key_answers() self._show_text_result("Fluidez / Lag", text) self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Fluidez", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def start_conflict_intel(self) -> None: if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operacion actual.") return devices = list(self.devices.values()) if not devices: messagebox.showinfo( "Conflictos IP / DHCP", "Primero ESCANEAR RED.\n" "Si hay cortes a ratos, INICIAR MONITORIZACION 15 min y luego este boton.", ) return self._busy = True self.var_status.set("Sondeando DHCP y analizando conflictos ARP…") net = self.net arp = self.arp def work(): try: intel = analyze_conflicts(devices, net, arp, run_dhcp_probe=True) text = intel.as_text() def done(): self._last_conflict_text = text self.txt_conflicts.configure(state=tk.NORMAL) self.txt_conflicts.delete("1.0", tk.END) self.txt_conflicts.insert("1.0", text) self.txt_conflicts.configure(state=tk.DISABLED) try: self.nb_bottom.select(6) except Exception: pass dhcp = intel.dhcp.ipconfig_dhcp or "?" gw = intel.dhcp.gateway or "?" self.var_dhcp.set(f"{dhcp} (GW {gw})") if intel.cases or intel.dhcp.dual or intel.dhcp.tenda_suspect: self.var_alert.set(intel.summary[:180]) self.var_status.set("Conflictos/DHCP listos — pestaña «Conflictos IP / DHCP»") self.update_key_answers() self._show_text_result("Conflictos IP / DHCP", text) self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Conflictos", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def start_perfect_lan(self) -> None: if not self.devices: messagebox.showinfo( "Plan red perfecta", "Primero ESCANEAR RED.\n" "Calcula capacidad 8 NVR x camaras WiFi/cable y el checklist Askey/Tenda/Cudy.", ) return if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operacion actual.") return self._busy = True self.var_status.set("Calculando plan de capacidad y checklist…") devices = list(self.devices.values()) net = self.net dhcp = net.dhcp_server if net else "" gw = net.gateway if net else "" dual = bool(dhcp and gw and dhcp != gw) if self._last_conflict_text and "DOBLE DHCP" in self._last_conflict_text: dual = True def work(): try: report = build_perfect_lan_plan( devices, net, dhcp_dual=dual, dhcp_server=dhcp, gateway=gw, ) text = report.as_text() def done(): self._last_perfect_text = text self.txt_perfect.configure(state=tk.NORMAL) self.txt_perfect.delete("1.0", tk.END) self.txt_perfect.insert("1.0", text) self.txt_perfect.configure(state=tk.DISABLED) try: self.nb_bottom.select(7) except Exception: pass self.var_alert.set(report.verdict[:200]) self.var_status.set(f"Plan red perfecta — checklist {report.score}/100") self.update_key_answers() self._show_text_result("Plan red perfecta", text) self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Plan red perfecta", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def start_ip816_visors(self) -> None: """Manual HDMI + test de los 7 IP816 que NO son el principal.""" if not self.devices: messagebox.showinfo( "IP816 visores", "Primero ESCANEAR RED.\n" "Luego marque SOLO el NVR PRINCIPAL (☑) y pulse de nuevo.", ) return principal = self._selected_nvr() nvrs = [d for d in self.devices.values() if is_nvr_device(d)] if not nvrs: messagebox.showinfo( "IP816 visores", "No hay NVR en el escaneo. Compruebe cable y que el IP816 responde ping.", ) return if not principal: messagebox.showinfo( "IP816 visores", "Marque (☑) o seleccione en la tabla el NVR PRINCIPAL (el que ya ve las camaras bien).\n" "Los otros 7 IP816 se testean como visores de TV y se muestra su manual HDMI.", ) return visor_n = sum(1 for d in nvrs if d.ip != principal.ip) if visor_n == 0: messagebox.showinfo( "IP816 visores", f"Solo esta el NVR {principal.ip}. Encienda los 7 visores IP816 por cable y vuelva a escanear.", ) return if not messagebox.askyesno( "IP816 visores (7)", f"Principal: {principal.ip}\n" f"Visores a testear: {visor_n} (objetivo 7)\n\n" "Se comprueba ping/subred/puertos y se muestra el manual HDMI " "(16x5MP, Extra/Sub, grabacion OFF, ONVIF).\n¿Continuar?", ): return if self._busy: return self._busy = True self.var_status.set("Testeando visores IP816…") devices = list(self.devices.values()) net = self.net pip = principal.ip def work(): try: report = run_ip816_visor_test(devices, net, pip) text = report.as_text() def done(): self._last_ip816_text = text self.txt_ip816.configure(state=tk.NORMAL) self.txt_ip816.delete("1.0", tk.END) self.txt_ip816.insert("1.0", text) self.txt_ip816.configure(state=tk.DISABLED) try: self.nb_bottom.select(8) except Exception: pass self.var_alert.set(report.summary[:220]) self.var_status.set(f"IP816 visores: {len(report.visors)} testados") self.update_key_answers() self._show_text_result("IP816 visores (7)", text) self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("IP816 visores", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def _ask_caudal_options(self, n_cams: int, suggested: str) -> Optional[dict]: win = tk.Toplevel(self) win.title("Caudal 12-16 — configurar camaras") win.transient(self) win.geometry("640x560") win.grab_set() holder: List[Optional[dict]] = [None] ttk.Label( win, text=( f"{n_cams} camara(s) objetivo. El IP816 es 10/100: 16 x Extra 512 kbps cabe; " "16 x Main 5MP no. Usuario/clave de la CAMARA Dahua (no del NVR). No se guardan." ), wraplength=600, ).pack(anchor="w", padx=12, pady=(12, 6)) frm = ttk.Frame(win, padding=12) frm.pack(fill=tk.X) ttk.Label(frm, text="Perfil:").grid(row=0, column=0, sticky="w") choices = [("auto", "Auto (WiFi si latencia alta; si no visor 16ch)")] for k, meta in PROFILES.items(): choices.append((k, meta["label"])) labels = [c[1] for c in choices] key_by_label = {c[1]: c[0] for c in choices} sug_label = next((lab for k, lab in choices if k == suggested), choices[0][1]) var_prof = tk.StringVar(value=sug_label) cmb = ttk.Combobox(frm, textvariable=var_prof, values=labels, state="readonly", width=52) cmb.grid(row=0, column=1, sticky="we", padx=8, pady=4) ttk.Label(frm, text="Usuario:").grid(row=1, column=0, sticky="w") var_user = tk.StringVar(value="admin") ttk.Entry(frm, textvariable=var_user, width=32).grid(row=1, column=1, sticky="w", padx=8, pady=4) ttk.Label(frm, text="Clave:").grid(row=2, column=0, sticky="w") var_pass = tk.StringVar(value="") ttk.Entry(frm, textvariable=var_pass, width=32, show="*").grid(row=2, column=1, sticky="w", padx=8, pady=4) var_dry = tk.BooleanVar(value=True) ttk.Checkbutton( frm, text="Solo simular (recomendado primero: lee encode, NO escribe)", variable=var_dry, ).grid(row=3, column=0, columnspan=2, sticky="w", pady=4) var_res = tk.BooleanVar(value=True) ttk.Checkbutton( frm, text="Tambien bajar resolucion (si falla, la app reintenta solo bitrate/FPS)", variable=var_res, ).grid(row=4, column=0, columnspan=2, sticky="w", pady=2) txt = tk.Text(win, height=12, wrap=tk.WORD, font=("Consolas", 8)) txt.pack(fill=tk.BOTH, expand=True, padx=12, pady=6) txt.insert("1.0", caudal_math_text()) txt.configure(state=tk.DISABLED) def ok() -> None: holder[0] = { "profile": key_by_label.get(var_prof.get(), "auto"), "user": var_user.get().strip(), "password": var_pass.get(), "dry_run": bool(var_dry.get()), "also_resolution": bool(var_res.get()), } win.destroy() def cancel() -> None: holder[0] = None win.destroy() bf = ttk.Frame(win, padding=8) bf.pack(fill=tk.X) ttk.Button(bf, text="Continuar", command=ok).pack(side=tk.RIGHT, padx=4) ttk.Button(bf, text="Cancelar", command=cancel).pack(side=tk.RIGHT) win.bind("", lambda e: ok()) win.bind("", lambda e: cancel()) win.wait_window() return holder[0] def start_caudal(self) -> None: if not self.devices: messagebox.showinfo( "Caudal 12-16", "Primero ESCANEAR RED.\n" "Luego este boton calcula el caudal y puede aplicar Main/Extra en las Dahua.", ) return selected = self._selected_devices() cams = filter_cameras(selected) scope = "marcadas" if not cams: cams = filter_cameras(list(self.devices.values())) scope = "todas las del escaneo" if not cams: messagebox.showinfo("Caudal 12-16", "No hay camaras en el escaneo (tipo Camara o puerto 554).") return if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operacion actual.") return suggested = pick_profile(cams) opts = self._ask_caudal_options(len(cams), suggested) if not opts: return if not opts["dry_run"]: if not opts["user"] or not opts["password"]: messagebox.showerror( "Caudal 12-16", "Para APLICAR de verdad hace falta usuario y clave de las camaras Dahua.\n" "Deje «Solo simular» para probar sin escribir.", ) return if not messagebox.askyesno( "Aplicar encode en camaras", f"Esto CAMBIA bitrate/FPS (y quiza resolucion) en {len(cams)} camara(s) ({scope}).\n" "No se guarda la clave. Los 7 visores IP816 deben quedar en Extra/Sub despues.\n\n" "¿Aplicar ahora?", ): return self._busy = True mode = "Simulando caudal" if opts["dry_run"] else "Aplicando encode en camaras" self.var_status.set(f"{mode} ({len(cams)} {scope})…") user = opts["user"] password = opts["password"] profile = opts["profile"] dry = opts["dry_run"] also_res = opts["also_resolution"] def work(): try: text = apply_caudal( cams, profile, user, password, dry_run=dry, also_resolution=also_res, ) def done(): self._last_caudal_text = text self.txt_caudal.configure(state=tk.NORMAL) self.txt_caudal.delete("1.0", tk.END) self.txt_caudal.insert("1.0", text) self.txt_caudal.configure(state=tk.DISABLED) try: self.nb_bottom.select(9) except Exception: pass fail = text.count("[FAIL]") ok_n = text.count("[OK]") if dry: self.var_alert.set(f"Simulacion caudal: {ok_n} leidas, {fail} sin CGI/ONVIF. Desmarque simular para aplicar.") else: self.var_alert.set(f"Caudal aplicado: {ok_n} OK, {fail} fallos. HDMI Extra/Sub en los 7 visores.") self.var_status.set("Caudal 12-16 listo — pestaña «Caudal 12-16»") self.update_key_answers() self._show_text_result("Caudal 12-16", text) self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Caudal 12-16", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() def start_topology(self) -> None: """Inventario por rol + WS-Discovery + vista estimada desde cada NVR.""" if not self.devices: messagebox.showinfo( "Topología", "Primero pulse ESCANEAR RED para inventariar dispositivos.\n" "Conecte el portátil por cable al mismo switch que el NVR a diagnosticar.", ) return if self._busy: messagebox.showinfo("Ocupado", "Espere a que termine la operación actual.") return self._busy = True self.var_status.set("Analizando topología y WS-Discovery (simulando Buscar del NVR)…") devices = list(self.devices.values()) net = self.net def work(): try: report = run_topology_analysis( devices, net, enrich=True, discovery_sec=5.0, probe_services=False, ) text = report.as_text() def done(): self._last_topology_text = text self._discovery_ips = set(report.discovery_ips) self.txt_topology.configure(state=tk.NORMAL) self.txt_topology.delete("1.0", tk.END) self.txt_topology.insert("1.0", text) self.txt_topology.configure(state=tk.DISABLED) try: self.nb_bottom.select(4) except Exception: pass # refrescar tipos/puertos enriquecidos for d in devices: self._upsert_device_row(d) crit = [f for f in report.findings if f.severity == "CRITICAL"] if crit: self.var_alert.set(crit[0].title) self.var_status.set("Topología lista — pestaña «Topología / vista NVR»") self.update_key_answers() self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Topología", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() # ---------- Full diag ---------- def start_full_diag(self) -> None: if self._busy: return raw = self.var_diag_min.get() try: duration = int(raw.split()[0]) except Exception: duration = 15 cidr = self.var_cidr.get().strip() or self.net.cidr if not messagebox.askyesno( "Diagnóstico completo", f"Se escaneará {cidr} y se monitorizará " f"{'de forma continua' if duration <= 0 else f'{duration} minutos'}.\n¿Continuar?", ): return self._busy = True self._stop_full = False self.var_status.set("Diagnóstico completo en curso…") def work(): try: net, devices, path, conclusion = run_full_diagnostic( cidr=cidr, duration_min=duration, interval_sec=float(self.var_interval.get().split()[0]), on_progress=lambda m: self.after(0, lambda: self.var_status.set(m)), stop_flag=lambda: self._stop_full, ) self.net = net def done(): self.tree.delete(*self.tree.get_children()) self.devices.clear() for d in devices: self._upsert_device_row(d) self.var_status.set(f"Informe: {path}") self.var_alert.set(conclusion) self.update_key_answers() messagebox.showinfo("Informe generado", f"{conclusion}\n\n{path}") self.after(0, done) except Exception as e: self.after(0, lambda: messagebox.showerror("Error", str(e))) finally: self._busy = False threading.Thread(target=work, daemon=True).start() # ---------- Report ---------- def export_report(self) -> None: devices = list(self.devices.values()) notes = suspicious_dhcp_notes(self.net, [d.ip for d in devices if d.device_type == "Camara"]) path = generate_html_report( self.net, devices, self.event_log, monitor=self.monitor, arp=self.arp, dhcp_notes=notes, topology_text=(self._last_topology_text or "") + ("\n\n" + self._last_fluency_text if self._last_fluency_text else "") + ("\n\n" + self._last_conflict_text if self._last_conflict_text else "") + ("\n\n" + self._last_perfect_text if self._last_perfect_text else "") + ("\n\n" + self._last_ip816_text if getattr(self, "_last_ip816_text", "") else "") + ("\n\n" + self._last_caudal_text if getattr(self, "_last_caudal_text", "") else "") + ("\n\n" + getattr(self, "_last_switch_text", "") if getattr(self, "_last_switch_text", "") else ""), ) self.var_status.set(f"Informe guardado: {path}") messagebox.showinfo("Informe", f"Informe HTML generado:\n{path}") # ---------- Key answers ---------- def update_key_answers(self) -> None: conflicts = self.arp.conflicts() devices = list(self.devices.values()) cams = [d for d in devices if d.device_type == "Camara"] degraded = [d for d in devices if d.status == "DEGRADADO"] offline = [d for d in devices if d.status == "OFFLINE"] online_cams = [d for d in cams if d.status in ("ONLINE", "DEGRADADO")] lines = ["RESPUESTAS RÁPIDAS PARA EL TÉCNICO", "=" * 40, ""] # 1 if conflicts: from .conflict_intel import _role_for_mac gw = self.net.gateway if self.net else "" lines.append("1. ¿IP duplicada? → SÍ (CONFIRMADO: dos MAC en la misma IP)") for c in conflicts: pa = _role_for_mac(c.old_mac, devices, gw) pb = _role_for_mac(c.new_mac, devices, gw) lines.append(f" • IP {c.ip} ({c.ts})") lines.append(f" A: {c.old_mac} {pa.vendor or '-'} → {pa.role}") lines.append(f" B: {c.new_mac} {pb.vendor or '-'} → {pb.role}") lines.append(" Pulse «Conflictos IP / DHCP» para pasos manuales.") else: lines.append( "1. ¿IP duplicada? → No en esta sesión " "(no es 100% que no ocurra luego: monitorice 15 min o pulse Conflictos IP / DHCP)." ) dhcp = self.net.dhcp_server if self.net else "" gw = self.net.gateway if self.net else "" if dhcp and gw: if dhcp == gw: lines.append(f" DHCP de este PC: {dhcp} = gateway (probable Askey). Compruebe Tenda DHCP OFF.") else: lines.append(f" ⚠ DHCP ({dhcp}) ≠ gateway ({gw}) → posible Tenda/Cudy dando IPs.") lines.append("") # 2 lines.append("2. MAC por IP (ARP conocido):") table = self.arp.table if not table: table = {d.ip: d.mac for d in devices if d.mac} for ip in sorted(table.keys(), key=lambda x: tuple(int(p) for p in x.split("."))): lines.append(f" {ip} → {table[ip]}") if not table: lines.append(" (sin datos ARP todavía — ejecute escaneo)") lines.append("") # 3 if degraded: lines.append("3. ¿Pérdida de paquetes? → SÍ") for d in degraded: lines.append(f" • {d.ip}: pérdida {d.loss_pct:.0f}%, lat {d.latency_ms} ms") else: lines.append("3. ¿Pérdida de paquetes? → No relevante ahora (o sin monitor).") lines.append("") # 4-5 timeline timeline = list(self.monitor.timeline) if self.monitor else [] offs = [t for t in timeline if t.kind == "offline"] if offs: lines.append("4. ¿Cuándo empezó? → Primer OFFLINE:") lines.append(f" {offs[0].ts_str} {offs[0].ip}") recent_ips = {t.ip for t in offs if True} if len(recent_ips) >= 3: lines.append("5. ¿Individual o común? → PROBLEMA COMÚN (múltiples cámaras).") lines.append(" Revise AP Wi-Fi / router / saturación / canal.") else: lines.append("5. ¿Individual o común? → Parece fallo puntual / pocas cámaras.") else: lines.append("4. ¿Cuándo empezó? → Sin caídas registradas en monitor.") lines.append("5. ¿Individual o común? → Sin datos de correlación aún.") lines.append("") # 6 if conflicts or any(len(self.arp.history_for(d.ip)) > 1 for d in devices): lines.append("6. ¿ARP sospechoso? → SÍ (cambios de MAC o conflicto).") else: lines.append("6. ¿ARP sospechoso? → No en esta sesión.") lines.append("") # 7 if offline and online_cams: lines.append( "7. NVR 'red inalcanzable' vs ping: hay cámaras ONLINE desde el portátil. " "Si el NVR sigue sin verlas, revise ruta NVR↔cámaras, VLAN o Wi-Fi del NVR (debe ir por cable)." ) elif offline and not online_cams and cams: lines.append( "7. Cámaras OFFLINE también desde el portátil → el problema es de red/Wi-Fi/cámara, no solo del NVR." ) else: lines.append("7. Sin contraste NVR/ping aún (monitorice mientras el NVR falla).") lines.append("") # 8 lines.append("8. Hipótesis principal:") if conflicts: lines.append(" → Configuración IP / conflicto DHCP. Fije IPs y un solo DHCP.") elif len({t.ip for t in offs}) >= 3: lines.append(" → Wi-Fi/AP/router (fallo común). Tenda/AP dedicado, no repetidor para cámaras.") elif len(offline) == 1: lines.append(f" → Cámara concreta ({offline[0].ip}): alimentación, Wi-Fi o hardware.") elif degraded: lines.append(" → Enlace degradado (interferencia/saturación Wi-Fi).") else: lines.append(" → Red estable en esta ventana. Si el NVR falla, capture momento con monitor activo.") lines.append("") lines.append("9. ¿Un NVR ve solo la mitad de cámaras?") if self._last_topology_text: lines.append( " → Hay análisis de topología. Abra la pestaña «Topología / vista NVR» " "y compare unicast vs WS-Discovery por cada NVR." ) if "PROBLEMA CRITICO" in self._last_topology_text or "PROBLEMA CRÍTICO" in self._last_topology_text: lines.append(" → La topología marcó PROBLEMA CRÍTICO (multicast/subred/alcance).") elif "MULTICAST" in self._last_topology_text or "PARTIAL_DISCOVERY" in self._last_topology_text: lines.append(" → Brecha típica: ping OK pero Buscar/ONVIF multicast incompleto.") else: lines.append( " → Pulse «Topología / Vista NVR» tras escanear (PC por cable al mismo switch que ese NVR)." ) lines.append("") lines.append("10. ¿Lag / retardo / no fluido (sobre todo WiFi)?") if self._last_fluency_text: lines.append(" → Hay diagnostico Fluidez/Lag. Abra la pestana y baje bitrates en camaras con score bajo.") if "multi-NVR" in self._last_fluency_text or "MULTI-NVR" in self._last_fluency_text.upper() or "CARGA MULTI" in self._last_fluency_text: lines.append(" → Carga multi-NVR: use sub-stream en vivo en TODOS los NVR.") else: lines.append(" → Pulse «Fluidez / Lag» tras escanear (detecta bitrate/resolucion y jitter WiFi).") lines.append("") lines.append("11. ¿Se puede dejar fluido en TODOS los NVR con este equipo?") if self._last_perfect_text: lines.append(" → Ver pestaña «Plan red perfecta»: capacidad + checklist + 12 pasos.") else: lines.append( " → Pulse «Plan red perfecta». " "8 NVR pidiendo MAIN WiFi no cabe; live=SUB sí. Cudy+Tenda+Askey bastan si se usan bien." ) lines.append("") lines.append("12. ¿Los 7 NVR IP816 (TVs, no el principal) estan bien configurados?") if getattr(self, "_last_ip816_text", ""): lines.append(" → Ver pestaña «IP816 visores (7)»: test + manual HDMI Extra/Sub, 16x5MP, grabacion OFF.") else: lines.append( " → Marque el NVR PRINCIPAL y pulse «IP816 visores (7)». " "El IP816 es 10/100 y 16x5MP o 10x8MP; visores = Extra stream, no Main." ) lines.append("") lines.append("13. ¿Se pueden ver 12-16 camaras en un visor sin saturar?") if getattr(self, "_last_caudal_text", ""): lines.append( " → Ver pestaña «Caudal 12-16»: math + encode leido/aplicado. " "Luego Extra/Sub en HDMI de los 7 visores." ) if "[FAIL]" in self._last_caudal_text: lines.append(" → Alguna camara no acepto CGI/ONVIF: configurarla a mano en http://IP.") else: lines.append( " → Pulse «Caudal 12-16 (aplicar)»: Extra 384-512 kbps cabe 16 ch en 100M. " "Primero simule; luego aplique con admin de la camara." ) lines.append("") lines.append("14. ¿NVR solo ven ~7 camaras y ConfigTool tarda en ver el resto?") if getattr(self, "_last_switch_text", ""): lines.append(" → Ver «Switches / 7 cams»: Cudy FS1018 VLAN OFF + contadores D-Link.") else: lines.append( " → Pulse «Switches / 7 cams». NO toque Movistar. " "Cudy PoE: VLAN OFF. D-Link: VLAN 1 plana. Prueba ping -t a .126." ) lines.append("") lines.append("15. Inventario NVR de esta casa:") lines.append(" .62 NVR1 Salon | .64 NVR2 Salon | .76 Dormitorio | .57 Escalera") lines.append(" (aparecen con nombre tras ESCANEAR si estan online)") self.txt_answers.delete("1.0", tk.END) self.txt_answers.insert(tk.END, "\n".join(lines)) # ---------- Charts ---------- def redraw_charts(self) -> None: self.canvas.delete("all") w = max(self.canvas.winfo_width(), 400) h = max(self.canvas.winfo_height(), 180) self.canvas.create_text(8, 10, anchor="nw", text="Latencia (ms) — últimas muestras", font=("Segoe UI", 9, "bold")) mon = self.monitor if not mon: self.canvas.create_text(8, 40, anchor="nw", text="Inicie monitorización para ver gráficas.") self.txt_stats.delete("1.0", tk.END) return colors = ["#1565c0", "#2e7d32", "#c62828", "#6a1b9a", "#ef6c00", "#00838f", "#4527a0", "#558b2f"] legend_y = 28 plot_top, plot_bot = 50, h - 20 plot_left, plot_right = 40, w - 20 # escala all_vals = [] series = [] for i, (ip, dq) in enumerate(list(mon.latency_series.items())[:8]): pts = list(dq)[-80:] if pts: series.append((ip, pts, colors[i % len(colors)])) all_vals.extend(v for _, v in pts) vmax = max(all_vals) if all_vals else 1 vmax = max(vmax, 1) self.canvas.create_line(plot_left, plot_bot, plot_right, plot_bot, fill="#999") self.canvas.create_line(plot_left, plot_top, plot_left, plot_bot, fill="#999") for ip, pts, color in series: self.canvas.create_text(plot_right - 8, legend_y, anchor="ne", text=ip, fill=color, font=("Consolas", 8)) legend_y += 12 if len(pts) < 2: continue coords = [] for idx, (_, val) in enumerate(pts): x = plot_left + (plot_right - plot_left) * idx / max(1, len(pts) - 1) y = plot_bot - (plot_bot - plot_top) * (val / vmax) coords.extend([x, y]) self.canvas.create_line(*coords, fill=color, width=2) # online count if mon.online_count_series: last_n = list(mon.online_count_series)[-1][1] else: last_n = sum(1 for d in mon.devices.values() if d.status != "OFFLINE") self.txt_stats.delete("1.0", tk.END) lines = [f"Dispositivos monitorizados online (último ciclo): {last_n}/{len(mon.devices)}", ""] for ip, d in sorted(mon.devices.items()): lines.append( f"{ip}: {d.status} | sent={d.sent} recv={d.recv} loss={d.loss_pct:.0f}% " f"| rtt avg={d.rtt_avg} min={d.rtt_min} max={d.rtt_max} jitter={d.jitter}" ) self.txt_stats.insert("1.0", "\n".join(lines)) def _on_close(self) -> None: self._stop_full = True try: if self.monitor: self.monitor.stop() if self.scanner: self.scanner.stop() self.event_log.close() except Exception: pass self.destroy() def run_app() -> None: app = App() app.mainloop()