Poll the USB-serial device set ~every 1.5 s; attach/detach GPS and companion handlers as devices appear, disappear, or re-enumerate.
(self)
| 2067 | companion = _CompanionState(port) |
| 2068 | self._companions[port] = companion |
| 2069 | t = threading.Thread( |
| 2070 | target=self._serial_listen_loop, |
| 2071 | args=(companion,), |
| 2072 | daemon=True, |
| 2073 | name=f"wardriving-serial-{port}" |
| 2074 | ) |
| 2075 | companion.thread = t |
| 2076 | t.start() |
| 2077 | return companion |
| 2078 | |
| 2079 | def start_serial(self, port: str): |
| 2080 | """Start serial listener on the given port (adds to existing companions).""" |
| 2081 | gps_port = self._gps.port if self._gps else None |
| 2082 | if gps_port and port == gps_port: |
| 2083 | return {'error': f'Port {port} is already in use by GPS'} |
| 2084 | companion = self._start_companion_thread(port) |
| 2085 | if companion is None: |
| 2086 | return {'error': f'Could not start companion on {port}'} |
| 2087 | return {'success': True, 'port': port} |
| 2088 | |
| 2089 | def stop_serial(self, port: str | None = None): |
| 2090 | """Stop one companion (by port) or all companions.""" |
| 2091 | if port: |
| 2092 | c = self._companions.pop(port, None) |
| 2093 | if c: |
| 2094 | c.connected = False |
| 2095 | return {'success': True, 'stopped': port} |
| 2096 | # Stop all |
| 2097 | for c in list(self._companions.values()): |
| 2098 | c.connected = False |
| 2099 | self._companions.clear() |
| 2100 | return {'success': True, 'stopped': 'all'} |
| 2101 | |
| 2102 | # ------------------------------------------------------------------ |
| 2103 | # Huginn runtime config push (matches HuginnESP src/runtime_config.cpp). |
| 2104 | # Knobs: |
| 2105 | # wifi_scan_duration_ms : 500..600000 (firmware default 15000) |
| 2106 | # ble_spam_threshold : 1..10000 (firmware default 20) |
| 2107 | # skimmer_names : CSV string (firmware has built-in defaults) |
nothing calls this directly
no test coverage detected