Antenna-coverage breakdown from network_observations. Returns per-interface: unique — distinct BSSIDs this interface ever saw only_here — BSSIDs ONLY this interface saw (no other adapter) best_rssi_avg / best_rssi_median — antenna sensitivity proxies
(self, interfaces=None)
| 721 | count = (existing[0] or 0) + 1 |
| 722 | conn.execute(""" |
| 723 | UPDATE cell_towers SET |
| 724 | signal_dbm = MAX(signal_dbm, ?), last_seen = ?, scan_count = ?, |
| 725 | latitude = COALESCE(?, latitude), longitude = COALESCE(?, longitude) |
| 726 | WHERE cell_id = ? |
| 727 | """, (signal_dbm, now, count, lat, lon, cell_id)) |
| 728 | else: |
| 729 | conn.execute(""" |
| 730 | INSERT INTO cell_towers |
| 731 | (cell_id, mcc, mnc, lac, tech, provider, signal_dbm, band_freq, |
| 732 | latitude, longitude, first_seen, last_seen, scan_count) |
| 733 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1) |
| 734 | """, (cell_id, mcc, mnc, lac, tech, provider, signal_dbm, band_freq, |
| 735 | lat, lon, now, now)) |
| 736 | except Exception as e: |
| 737 | logger.debug(f"Cell upsert error: {e}") |
| 738 | |
| 739 | def upsert_zigbee(self, addr, panid, short_addr, device_type, |
| 740 | rssi, lqi, channel, lat, lon, alt, proto='zigbee'): |
| 741 | """Insert or update a discovered 802.15.4 device (Zigbee / Thread). |
| 742 | |
| 743 | `addr` is the stable identity (EUI-64 or "<panid>:<short>"). Best |
| 744 | position is tracked at the strongest RSSI sighting, mirroring how |
| 745 | Bluetooth devices are recorded so map exports stay consistent. |
| 746 | |
| 747 | `proto` is the classified network layer ('zigbee' / 'thread' / |
| 748 | '802.15.4'). It's only upgraded to a *specific* protocol on re-sightings |
| 749 | so a later opaque (encrypted) frame can't downgrade a device that a |
| 750 | beacon already identified. |
| 751 | """ |
| 752 | if not addr: |
| 753 | return |
| 754 | proto = (proto or 'zigbee').lower() |
| 755 | now = datetime.now(timezone.utc).isoformat() |
| 756 | with self._lock: |
| 757 | try: |
| 758 | with sqlite3.connect(self.db_path) as conn: |
| 759 | existing = conn.execute( |
| 760 | "SELECT scan_count, best_rssi FROM zigbee_devices WHERE addr = ?", |
| 761 | (addr,) |
| 762 | ).fetchone() |
| 763 | if existing: |
| 764 | count = (existing[0] or 0) + 1 |
| 765 | old_best = existing[1] if existing[1] is not None else -100 |
| 766 | new_best = rssi if rssi > old_best else old_best |
| 767 | best_lat_val = lat if (rssi > old_best and lat) else None |
| 768 | best_lon_val = lon if (rssi > old_best and lon) else None |
| 769 | conn.execute(""" |
| 770 | UPDATE zigbee_devices SET |
| 771 | panid = COALESCE(NULLIF(?, ''), panid), |
| 772 | short_addr = COALESCE(NULLIF(?, ''), short_addr), |
| 773 | device_type = COALESCE(NULLIF(?, ''), device_type), |
| 774 | proto = CASE WHEN ? IN ('zigbee','thread') THEN ? ELSE proto END, |
| 775 | rssi = ?, lqi = ?, |
| 776 | channel = CASE WHEN ? > 0 THEN ? ELSE channel END, |
| 777 | best_rssi = ?, |
| 778 | best_lat = COALESCE(?, best_lat), |
| 779 | best_lon = COALESCE(?, best_lon), |
| 780 | latitude = COALESCE(?, latitude), |