Represents a single wardriving session with its own SQLite DB.
| 160 | |
| 161 | Band comes from an explicit `Band:` field when the companion sends one |
| 162 | (the HuginnESP text format does), otherwise from the channel number: |
| 163 | 1-14 is 2.4 GHz, 32+ is 5 GHz. Channel 0/unknown still counts toward the |
| 164 | total but no band, so the two buckets never over-report. |
| 165 | """ |
| 166 | self.networks += 1 |
| 167 | label = str(band or '').strip().lower() |
| 168 | if label.startswith('2'): |
| 169 | self.networks_24 += 1 |
| 170 | return |
| 171 | if label.startswith('5'): |
| 172 | self.networks_5 += 1 |
| 173 | return |
| 174 | try: |
| 175 | ch = int(channel) |
| 176 | except (TypeError, ValueError): |
| 177 | return |
| 178 | if 1 <= ch <= 14: |
| 179 | self.networks_24 += 1 |
| 180 | elif ch >= 32: |
| 181 | self.networks_5 += 1 |
| 182 | |
| 183 | def active_coordinator_nodes(self) -> list: |
| 184 | now_ts = time.time() |
| 185 | expired = [m for m, n in self.coordinator_nodes.items() |
| 186 | if (now_ts - n.get('last_update', 0)) > 30.0] |
| 187 | for m in expired: |
| 188 | self.coordinator_nodes.pop(m, None) |
| 189 | self.mesh_node_count = len(self.coordinator_nodes) |
| 190 | return sorted(self.coordinator_nodes.values(), key=lambda n: n.get('idx', 0)) |
| 191 | |
| 192 | def as_dict(self) -> dict: |
| 193 | return { |
| 194 | 'port': self.port, |
| 195 | 'connected': self.connected, |
| 196 | 'name': self.name, |
| 197 | 'networks': self.networks, |
| 198 | 'networks_24': self.networks_24, |
| 199 | 'networks_5': self.networks_5, |
| 200 | 'mesh_node_count': self.mesh_node_count, |
| 201 | 'coordinator_board': self.coordinator_board, |
| 202 | 'coordinator_fw': self.coordinator_fw, |
| 203 | 'coordinator_nodes': self.active_coordinator_nodes(), |
| 204 | 'esp_mode': 'wardrive' if self.in_wardrive else self.current_esp_mode, |
| 205 | 'esp_ble_count': self.esp_ble_count, |
| 206 | 'esp_zigbee_count': self.esp_zigbee_count, |
| 207 | 'esp_alerts': self.esp_alerts[-5:], |
| 208 | } |
| 209 | |
| 210 | |
| 211 | class WardrivingSession: |
| 212 | """Represents a single wardriving session with its own SQLite DB.""" |
| 213 | |
| 214 | def __init__(self, data_dir, session_id=None): |
| 215 | self.session_id = session_id or datetime.now().strftime('%Y%m%d_%H%M%S') |
| 216 | self.data_dir = os.path.join(data_dir, 'wardriving') |
| 217 | os.makedirs(self.data_dir, exist_ok=True) |
| 218 | self.db_path = os.path.join(self.data_dir, f'session_{self.session_id}.db') |
| 219 | self.start_time = time.time() |
no outgoing calls
no test coverage detected