Main scanning loop — runs fast continuous scans.
(self)
| 2403 | # it accumulated during the prior session — e.g. "No GPS device |
| 2404 | # detected" from a session where the receiver wasn't plugged in |
| 2405 | # yet) is what callers see, even after the user plugs in the GPS. |
| 2406 | self._gps = None |
| 2407 | # Force re-probe on the next status call instead of serving cached |
| 2408 | # "not detected" from before the user plugged in the receiver. |
| 2409 | if hasattr(self, '_gps_probe_cache'): |
| 2410 | self._gps_probe_cache = None |
| 2411 | self._gps_probe_time = 0 |
| 2412 | |
| 2413 | stats = self.session.get_stats() if self.session else {} |
| 2414 | logger.info(f"Wardriving stopped. Networks: {stats.get('total_networks', 0)}") |
| 2415 | return {'success': True, 'stats': stats} |
| 2416 | |
| 2417 | def _start_companion_thread(self, port: str) -> '_CompanionState | None': |
| 2418 | """Register a companion and start its listener thread. Returns the state object.""" |
| 2419 | gps_port = self._gps.port if self._gps else None |
| 2420 | if gps_port and port == gps_port: |
| 2421 | logger.warning(f"Skipping companion on {port}: same port as GPS") |
| 2422 | return None |
| 2423 | existing = self._companions.get(port) |
| 2424 | if existing and existing.thread and existing.thread.is_alive(): |
| 2425 | return existing |
| 2426 | companion = _CompanionState(port) |
| 2427 | self._companions[port] = companion |
| 2428 | t = threading.Thread( |
| 2429 | target=self._serial_listen_loop, |
| 2430 | args=(companion,), |
| 2431 | daemon=True, |
| 2432 | name=f"wardriving-serial-{port}" |
| 2433 | ) |
| 2434 | companion.thread = t |
| 2435 | t.start() |
| 2436 | return companion |
| 2437 | |
| 2438 | def start_serial(self, port: str): |
| 2439 | """Start serial listener on the given port (adds to existing companions).""" |
| 2440 | gps_port = self._gps.port if self._gps else None |
| 2441 | if gps_port and port == gps_port: |
| 2442 | return {'error': f'Port {port} is already in use by GPS'} |
| 2443 | companion = self._start_companion_thread(port) |
| 2444 | if companion is None: |
| 2445 | return {'error': f'Could not start companion on {port}'} |
| 2446 | return {'success': True, 'port': port} |
| 2447 | |
| 2448 | def stop_serial(self, port: str | None = None): |
| 2449 | """Stop one companion (by port) or all companions.""" |
| 2450 | if port: |
| 2451 | c = self._companions.pop(port, None) |
| 2452 | if c: |
| 2453 | c.connected = False |
| 2454 | return {'success': True, 'stopped': port} |
| 2455 | # Stop all |
| 2456 | for c in list(self._companions.values()): |
| 2457 | c.connected = False |
| 2458 | self._companions.clear() |
| 2459 | return {'success': True, 'stopped': 'all'} |
| 2460 | |
| 2461 | # ------------------------------------------------------------------ |
| 2462 | # Huginn runtime config push (matches HuginnESP src/runtime_config.cpp). |
nothing calls this directly
no test coverage detected