itsDNNS
diff --git a/‎app/collectors/demo.py‎
Lines changed: 102 additions & 0 deletions b/‎app/collectors/demo.py‎
Lines changed: 102 additions & 0 deletions
diff --git a/‎app/modules/backup/backup.py‎
Lines changed: 35 additions & 24 deletions b/‎app/modules/backup/backup.py‎
Lines changed: 35 additions & 24 deletions
diff --git a/‎app/modules/bqm/manifest.json‎
Lines changed: 1 addition & 1 deletion b/‎app/modules/bqm/manifest.json‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎app/modules/connection_monitor/__init__.py‎ b/‎app/modules/connection_monitor/__init__.py‎
diff --git a/‎app/modules/connection_monitor/collector.py‎
Lines changed: 169 additions & 0 deletions b/‎app/modules/connection_monitor/collector.py‎
Lines changed: 169 additions & 0 deletions
@@ -206,6 +206,7 @@ def _seed_demo_data(self):
         self._seed_incident_containers(now)
         self._seed_bnetz_measurements(now)
         self._seed_weather_data(now)
+        self._seed_connection_monitor_data(now)
 
     def _seed_history(self, now):
         """Generate 9 months of historical snapshots (every 15 min)."""
@@ -888,6 +889,107 @@ def _seed_weather_data(self, now):
             )
         log.info("Demo: seeded %d weather records (%d days)", len(records), days)
 
+    def _seed_connection_monitor_data(self, now):
+        """Seed 7 days of Connection Monitor data showing a typical cable troubleshooting scenario.
+
+        Story: User notices evening lag and dropped video calls. Enables Connection Monitor.
+        Gateway is always fine (proves home network OK), but external targets show:
+        - Evening congestion (19-23h): latency spikes, occasional packet loss
+        - Two short outages (~1-3 min) on different days
+        - One longer outage (~8 min) that prompted the investigation
+        """
+        try:
+            from app.modules.connection_monitor.storage import ConnectionMonitorStorage
+        except ImportError:
+            log.debug("Demo: connection_monitor module not available, skipping")
+            return
+
+        data_dir = os.path.dirname(self._storage.db_path)
+        cm_db = os.path.join(data_dir, "connection_monitor.db")
+        cm = ConnectionMonitorStorage(cm_db)
+
+        # Purge existing demo targets/samples
+        with cm._connect() as conn:
+            conn.execute("DELETE FROM connection_samples")
+            conn.execute("DELETE FROM connection_targets")
+
+        # Create targets
+        gw_id = cm.create_target("Gateway", "192.168.178.1")
+        cf_id = cm.create_target("Cloudflare DNS", "1.1.1.1")
+        gg_id = cm.create_target("Google DNS", "8.8.8.8")
+
+        rng = random.Random(2026)
+        days = 7
+        interval_s = 10  # 10s between samples
+        samples_per_day = 86400 // interval_s  # 8640
+
+        # Outage windows (day_offset, hour_start, duration_minutes)
+        outages = [
+            (2, 20.5, 2),    # Day 3: short 2-min outage during evening
+            (4, 21.0, 3),    # Day 5: 3-min outage
+            (5, 19.75, 8),   # Day 6: the big one - 8 min outage that triggered investigation
+        ]
+
+        rows = []
+        for d in range(days):
+            day_start = now - timedelta(days=days - d)
+            for s in range(samples_per_day):
+                ts = day_start.timestamp() + s * interval_s
+                hour = (s * interval_s / 3600) % 24
+
+                # Check if we're in an outage window (external targets only)
+                in_outage = False
+                for o_day, o_hour, o_dur in outages:
+                    if d == o_day and o_hour <= hour < o_hour + o_dur / 60:
+                        in_outage = True
+                        break
+
+                # Evening congestion window
+                evening = 19 <= hour < 23
+                late_evening = 20 <= hour < 22  # worst window
+
+                # --- Gateway: always fast, 1-3ms ---
+                gw_lat = round(rng.uniform(0.8, 3.0), 2)
+                rows.append((gw_id, ts, gw_lat, False, "tcp"))
+
+                # --- External targets ---
+                for tid in (cf_id, gg_id):
+                    base = 11.0 if tid == cf_id else 14.0
+
+                    if in_outage:
+                        # Full timeout
+                        rows.append((tid, ts, None, True, "tcp"))
+                    elif late_evening:
+                        # Heavy congestion: spikes + occasional loss
+                        if rng.random() < 0.04:
+                            rows.append((tid, ts, None, True, "tcp"))
+                        else:
+                            spike = rng.uniform(30, 250) if rng.random() < 0.3 else rng.uniform(0, 20)
+                            lat = round(base + spike, 2)
+                            rows.append((tid, ts, lat, False, "tcp"))
+                    elif evening:
+                        # Moderate congestion: elevated latency, rare loss
+                        if rng.random() < 0.008:
+                            rows.append((tid, ts, None, True, "tcp"))
+                        else:
+                            spike = rng.uniform(5, 60) if rng.random() < 0.15 else rng.uniform(0, 8)
+                            lat = round(base + spike, 2)
+                            rows.append((tid, ts, lat, False, "tcp"))
+                    else:
+                        # Normal: stable low latency
+                        lat = round(base + rng.uniform(-2, 3), 2)
+                        rows.append((tid, ts, lat, False, "tcp"))
+
+        # Bulk insert
+        with cm._connect() as conn:
+            conn.executemany(
+                "INSERT INTO connection_samples (target_id, timestamp, latency_ms, timeout, probe_method) "
+                "VALUES (?, ?, ?, ?, ?)",
+                rows,
+            )
+
+        log.info("Demo: seeded %d connection monitor samples (%d days, 3 targets)", len(rows), days)
+
     @staticmethod
     def _generate_bqm_png(width=800, height=200, seed=0):
         """Generate a simple BQM-style quality graph as PNG bytes."""
 
@@ -18,7 +18,10 @@
 log = logging.getLogger("docsis.backup")
 
 # Files under data_dir to include in backups
-DATA_FILES = ["docsis_history.db", "config.json", ".config_key", ".session_key"]
+DATA_FILES = [
+    "docsis_history.db", "connection_monitor.db",
+    "config.json", ".config_key", ".session_key",
+]
 
 BACKUP_META_FILE = "backup_meta.json"
 FORMAT_VERSION = 1
@@ -54,32 +57,33 @@ def _get_table_counts(db_path):
     return counts
 
 
-def _vacuum_db(data_dir, dest_path):
-    """Create a consistent copy of the database using VACUUM INTO.
+def _vacuum_db(data_dir, db_name, dest_path):
+    """Create a consistent copy of a database using VACUUM INTO.
 
-    Also removes demo data (is_demo=1) from the copy.
+    Also removes demo data (is_demo=1) from the copy when applicable.
     """
-    src = os.path.join(data_dir, "docsis_history.db")
+    src = os.path.join(data_dir, db_name)
     if not os.path.exists(src):
         return False
 
     conn = sqlite3.connect(src)
     conn.execute(f"VACUUM INTO '{dest_path}'")
     conn.close()
 
-    # Remove demo data from copy
-    copy_conn = sqlite3.connect(dest_path)
-    demo_tables = [
-        "snapshots", "events", "journal_entries", "incidents",
-        "speedtest_results", "bqm_graphs", "bnetz_measurements",
-    ]
-    for table in demo_tables:
-        try:
-            copy_conn.execute(f"DELETE FROM [{table}] WHERE is_demo = 1")  # noqa: S608
-        except sqlite3.OperationalError:
-            pass  # table may not exist or lack is_demo column
-    copy_conn.commit()
-    copy_conn.close()
+    # Remove demo data from copy (only relevant for main DB)
+    if db_name == "docsis_history.db":
+        copy_conn = sqlite3.connect(dest_path)
+        demo_tables = [
+            "snapshots", "events", "journal_entries", "incidents",
+            "speedtest_results", "bqm_graphs", "bnetz_measurements",
+        ]
+        for table in demo_tables:
+            try:
+                copy_conn.execute(f"DELETE FROM [{table}] WHERE is_demo = 1")  # noqa: S608
+            except sqlite3.OperationalError:
+                pass  # table may not exist or lack is_demo column
+        copy_conn.commit()
+        copy_conn.close()
     return True
 
 
@@ -89,28 +93,35 @@ def create_backup(data_dir):
     Returns:
         BytesIO containing the .tar.gz archive.
     """
+    db_files = {"docsis_history.db", "connection_monitor.db"}
+
     buf = BytesIO()
     with tempfile.TemporaryDirectory() as tmp:
-        db_copy = os.path.join(tmp, "docsis_history.db")
-        has_db = _vacuum_db(data_dir, db_copy)
+        # Vacuum all databases for consistent copies
+        vacuumed = {}
+        for db_name in db_files:
+            db_copy = os.path.join(tmp, db_name)
+            vacuumed[db_name] = _vacuum_db(data_dir, db_name, db_copy)
 
+        main_copy = os.path.join(tmp, "docsis_history.db")
         meta = {
             "magic": MAGIC,
             "format_version": FORMAT_VERSION,
             "timestamp": datetime.now(timezone.utc).isoformat(),
             "app_version": _get_app_version(),
-            "tables": _get_table_counts(db_copy) if has_db else {},
+            "tables": _get_table_counts(main_copy) if vacuumed.get("docsis_history.db") else {},
         }
         meta_path = os.path.join(tmp, BACKUP_META_FILE)
         with open(meta_path, "w") as f:
             json.dump(meta, f, indent=2)
 
         with tarfile.open(fileobj=buf, mode="w:gz") as tar:
             tar.add(meta_path, arcname=BACKUP_META_FILE)
-            if has_db:
-                tar.add(db_copy, arcname="docsis_history.db")
+            for db_name, has_db in vacuumed.items():
+                if has_db:
+                    tar.add(os.path.join(tmp, db_name), arcname=db_name)
             for fname in DATA_FILES:
-                if fname == "docsis_history.db":
+                if fname in db_files:
                     continue  # already added via vacuum copy
                 fpath = os.path.join(data_dir, fname)
                 if os.path.exists(fpath):
 
@@ -19,7 +19,7 @@
   },
   "menu": {
     "label_key": "docsight.bqm.bqm_title",
-    "icon": "activity",
+    "icon": "chart-spline",
     "order": 22
   }
 }
@@ -0,0 +1,169 @@
+"""Collector for Connection Monitor - orchestrates probing, storage, and events."""
+
+import logging
+import os
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from app.collectors.base import Collector, CollectorResult
+from app.modules.connection_monitor.event_rules import ConnectionEventRules
+from app.modules.connection_monitor.probe import ProbeEngine
+from app.modules.connection_monitor.storage import ConnectionMonitorStorage
+
+logger = logging.getLogger(__name__)
+
+# Run retention cleanup every 15 minutes, not every collect cycle
+_CLEANUP_INTERVAL_S = 900
+
+
+class ConnectionMonitorCollector(Collector):
+    """Always-on latency collector with per-target timing."""
+
+    name = "connection_monitor"
+
+    def __init__(self, config_mgr, storage, web, **kwargs):
+        super().__init__(poll_interval_seconds=1)
+        self._config_mgr = config_mgr
+        self._core_storage = storage
+        self._web = web
+
+        method = config_mgr.get("connection_monitor_probe_method", "auto")
+        self._probe = ProbeEngine(method=method)
+        self._last_probe: dict[int, float] = {}
+        self._last_cleanup = 0.0
+        self._event_rules = ConnectionEventRules(
+            outage_threshold=int(config_mgr.get("connection_monitor_outage_threshold", 5)),
+            loss_warning_pct=float(config_mgr.get("connection_monitor_loss_warning_pct", 2.0)),
+        )
+
+        data_dir = os.environ.get("DATA_DIR", "/data")
+        db_path = os.path.join(data_dir, "connection_monitor.db")
+        self._cm_storage = ConnectionMonitorStorage(db_path)
+
+        self._seeded = False
+
+    def is_enabled(self) -> bool:
+        return bool(self._config_mgr.get("connection_monitor_enabled", False))
+
+    def should_poll(self) -> bool:
+        """Always return True - per-target timing is managed internally."""
+        return True
+
+    def collect(self) -> CollectorResult:
+        try:
+            self._ensure_default_targets()
+            targets = [
+                t for t in self._cm_storage.get_targets() if t["enabled"]
+            ]
+            if not targets:
+                return CollectorResult.ok(self.name, None)
+
+            # Determine which targets are due
+            now = time.time()
+            due = []
+            for t in targets:
+                interval_s = t["poll_interval_ms"] / 1000.0
+                last = self._last_probe.get(t["id"], 0)
+                if now - last >= interval_s:
+                    due.append(t)
+
+            if not due:
+                return CollectorResult.ok(self.name, None)
+
+            # Probe all due targets in parallel
+            samples = self._probe_targets(due, now)
+
+            # Save samples
+            if samples:
+                self._cm_storage.save_samples(samples)
+
+            # Check events
+            self._check_events(samples)
+
+            # Periodic retention cleanup
+            if now - self._last_cleanup >= _CLEANUP_INTERVAL_S:
+                retention = int(
+                    self._config_mgr.get("connection_monitor_retention_days", 0)
+                )
+                self._cm_storage.cleanup(retention)
+                self._last_cleanup = now
+
+            return CollectorResult.ok(self.name, {"probed": len(due)})
+        except Exception as exc:
+            logger.exception("Connection Monitor collect error")
+            return CollectorResult.failure(self.name, str(exc))
+
+    def _probe_targets(self, targets: list[dict], now: float) -> list[dict]:
+        """Probe targets in parallel and return sample dicts."""
+        samples = []
+        tcp_port = int(self._config_mgr.get("connection_monitor_tcp_port", 443))
+
+        with ThreadPoolExecutor(
+            max_workers=max(len(targets), 1),
+            thread_name_prefix="cm-probe",
+        ) as pool:
+            futures = {
+                pool.submit(self._probe.probe, t["host"], t.get("tcp_port", tcp_port)): t
+                for t in targets
+            }
+            for future in as_completed(futures, timeout=5):
+                target = futures[future]
+                try:
+                    result = future.result()
+                except Exception:
+                    result = type("R", (), {"latency_ms": None, "timeout": True, "method": "error"})()
+
+                self._last_probe[target["id"]] = now
+                samples.append({
+                    "target_id": target["id"],
+                    "timestamp": now,
+                    "latency_ms": result.latency_ms,
+                    "timeout": result.timeout,
+                    "probe_method": result.method,
+                })
+        return samples
+
+    def _check_events(self, samples: list[dict]):
+        """Run event rules and save any emitted events."""
+        all_events = []
+        for s in samples:
+            events = self._event_rules.check_probe_result(
+                target_id=s["target_id"], timeout=s["timeout"]
+            )
+            all_events.extend(events)
+
+        # Check windowed packet loss stats per probed target
+        window_seconds = 60
+        checked_targets = set()
+        for s in samples:
+            tid = s["target_id"]
+            if tid in checked_targets:
+                continue
+            checked_targets.add(tid)
+            summary = self._cm_storage.get_summary(tid, window_seconds=window_seconds)
+            loss_pct = summary.get("packet_loss_pct") or 0.0
+            events = self._event_rules.check_window_stats(
+                target_id=tid, packet_loss_pct=loss_pct, window_seconds=window_seconds,
+            )
+            all_events.extend(events)
+
+        if all_events and hasattr(self._core_storage, "save_events"):
+            self._core_storage.save_events(all_events)
+
+    def _ensure_default_targets(self):
+        """Seed default targets on first enable."""
+        if self._seeded:
+            return
+        self._seeded = True
+        if not self._cm_storage.get_targets():
+            self._cm_storage.create_target("Cloudflare DNS", "1.1.1.1")
+            self._cm_storage.create_target("Google DNS", "8.8.8.8")
+            logger.info("Connection Monitor: seeded default targets")
+
+    def get_storage(self) -> ConnectionMonitorStorage:
+        """Expose storage for routes."""
+        return self._cm_storage
+
+    def get_probe(self) -> ProbeEngine:
+        """Expose probe engine for capability endpoint."""
+        return self._probe
Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,7 @@`
`19`	`19`	`},`
`20`	`20`	`"menu": {`
`21`	`21`	`"label_key": "docsight.bqm.bqm_title",`
`22`		`- "icon": "activity",`
	`22`	`+ "icon": "chart-spline",`
`23`	`23`	`"order": 22`
`24`	`24`	`}`
`25`	`25`	`}`