MCPcopy Create free account
hub / github.com/PierreGode/Ragnar / import_wigle_csv

Method import_wigle_csv

wardriving.py:1150–1312  ·  view source on GitHub ↗

Import a WiGLE-format CSV file into this session. Supports Piglet, HuginnESP, and standard WiGLE CSV exports. Returns dict with counts of imported records.

(self, csv_path_or_stream)

Source from the content-addressed store, hash-verified

1148 Rows whose position was estimated via backfill_gps_from_track
1149 (gps_backfilled = 1) are excluded — they are interpolated coordinates,
1150 not real observations, and must not be submitted to WiGLE.
1151
1152 Zigbee / 802.15.4 devices are excluded by default: WiGLE has no
1153 standard 802.15.4 record type, so those rows are only useful for the
1154 user's own tooling. Set `include_zigbee=True` (opt-in from the config
1155 tab) to append them with a `ZIGBEE` type token."""
1156 lines = []
1157 dn = device_name or 'Ragnar'
1158 lines.append(f'WigleWifi-1.4,appRelease=Ragnar,model=RaspberryPi,release=1.0,device={dn},display=EPD,board=RPi,brand=Ragnar')
1159 lines.append(','.join(WIGLE_HEADER))
1160
1161 with self._lock:
1162 try:
1163 with sqlite3.connect(self.db_path) as conn:
1164 conn.row_factory = sqlite3.Row
1165 # WiFi networks (skip backfilled positions)
1166 rows = conn.execute(
1167 "SELECT * FROM networks WHERE COALESCE(gps_backfilled, 0) = 0 "
1168 "ORDER BY first_seen"
1169 ).fetchall()
1170 for row in rows:
1171 r = dict(row)
1172 lines.append(','.join([
1173 r.get('bssid', ''),
1174 r.get('ssid', '').replace(',', '\\,'),
1175 _map_security(r.get('security', '')),
1176 r.get('first_seen', ''),
1177 str(r.get('channel', 0)),
1178 str(r.get('best_rssi', -100)),
1179 str(r.get('best_lat', '') or ''),
1180 str(r.get('best_lon', '') or ''),
1181 str(r.get('altitude', '') or ''),
1182 str(r.get('hdop', '') or ''),
1183 'WIFI'
1184 ]))
1185 # Bluetooth devices
1186 try:
1187 bt_rows = conn.execute(
1188 "SELECT * FROM bluetooth_devices WHERE COALESCE(gps_backfilled, 0) = 0 "
1189 "ORDER BY first_seen"
1190 ).fetchall()
1191 for row in bt_rows:
1192 r = dict(row)
1193 lines.append(','.join([
1194 r.get('mac', ''),
1195 r.get('name', '').replace(',', '\\,'),
1196 '[BT]',
1197 r.get('first_seen', ''),
1198 '0',
1199 str(r.get('rssi', -100)),
1200 str(r.get('latitude', '') or ''),
1201 str(r.get('longitude', '') or ''),
1202 str(r.get('altitude', '') or ''),
1203 '',
1204 'BT'
1205 ]))
1206 except Exception:
1207 pass

Callers 1

wardriving_import_csvFunction · 0.95

Calls 8

upsert_bluetoothMethod · 0.95
upsert_cell_towerMethod · 0.95
upsert_networkMethod · 0.95
appendMethod · 0.80
errorMethod · 0.80
infoMethod · 0.80
closeMethod · 0.45
getMethod · 0.45

Tested by

no test coverage detected