Method upsert_network

wardriving.py:362–483

Insert or update a discovered network (thread-safe).

(self, bssid, ssid, security, channel, frequency,
                       rssi, lat, lon, alt, speed, hdop, interface='')

                    mcc TEXT DEFAULT '',
                    mnc TEXT DEFAULT '',
                    lac TEXT DEFAULT '',
                    tech TEXT DEFAULT '',
                    provider TEXT DEFAULT '',
                    signal_dbm INTEGER DEFAULT -120,
                    band_freq TEXT DEFAULT '',
                    latitude REAL,
                    longitude REAL,
                    first_seen TEXT NOT NULL,
                    last_seen TEXT NOT NULL,
                    scan_count INTEGER DEFAULT 1,
                    gps_backfilled INTEGER DEFAULT 0
                )
            """)
            conn.execute("""
                CREATE UNIQUE INDEX IF NOT EXISTS idx_cell ON cell_towers(cell_id)
            """)
            # Zigbee / IEEE 802.15.4 devices seen by a companion with a
            # 2.4 GHz 802.15.4 radio (e.g. HuginnESP on an ESP32-C5). Keyed by
            # the device identity: its 64-bit extended address (EUI-64) when the
            # frame carries one, otherwise "<panid>:<short>" as a fallback.
            conn.execute("""
                CREATE TABLE IF NOT EXISTS zigbee_devices (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    addr TEXT NOT NULL,
                    panid TEXT DEFAULT '',
                    short_addr TEXT DEFAULT '',
                    device_type TEXT DEFAULT '',
                    proto TEXT DEFAULT 'zigbee',
                    rssi INTEGER DEFAULT -100,
                    best_rssi INTEGER DEFAULT -100,
                    lqi INTEGER DEFAULT 0,
                    channel INTEGER DEFAULT 0,
                    latitude REAL,
                    longitude REAL,
                    altitude REAL,
                    best_lat REAL,
                    best_lon REAL,
                    first_seen TEXT NOT NULL,
                    last_seen TEXT NOT NULL,
                    scan_count INTEGER DEFAULT 1,
                    gps_backfilled INTEGER DEFAULT 0
                )
            """)
            conn.execute("""
                CREATE UNIQUE INDEX IF NOT EXISTS idx_zigbee_addr ON zigbee_devices(addr)
            """)
            # Migrate older networks / cell_towers tables that predate the
            # GPS-backfill flag. Table names here are hardcoded literals, so
            # the f-string interpolation carries no injection risk.
            for _tbl in ('networks', 'cell_towers'):
                try:
                    cols = {r[1] for r in conn.execute(f"PRAGMA table_info({_tbl})").fetchall()}
                    if 'gps_backfilled' not in cols:
                        conn.execute(f"ALTER TABLE {_tbl} ADD COLUMN gps_backfilled INTEGER DEFAULT 0")
                except Exception:
                    pass
            # Add the 802.15.4 protocol column (zigbee/thread/802.15.4) to
            # zigbee_devices tables created before Thread classification existed.