|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
5 | 5 | import asyncio |
6 | | -import json |
7 | 6 | import logging |
8 | 7 | import time |
9 | 8 | from abc import ABC, abstractmethod |
|
18 | 17 |
|
19 | 18 | if TYPE_CHECKING: |
20 | 19 | from collections.abc import AsyncGenerator |
21 | | - from pathlib import Path |
22 | 20 |
|
| 21 | + from _typeshed import SupportsWrite |
23 | 22 | from bleak.backends.device import BLEDevice |
24 | 23 | from bleak.backends.scanner import AdvertisementData |
25 | 24 |
|
@@ -100,7 +99,7 @@ def is_from(self, other_device: HasPublicKey | RollingKeyPairSource) -> bool: |
100 | 99 | raise NotImplementedError |
101 | 100 |
|
102 | 101 | @abstractmethod |
103 | | - def print_device(self, out_file: Path | None = None) -> None: |
| 102 | + def print(self, file: SupportsWrite[str] | None = None) -> None: |
104 | 103 | """Print human-readable information about the device to stdout or file.""" |
105 | 104 | raise NotImplementedError |
106 | 105 |
|
@@ -254,33 +253,17 @@ def from_payload( |
254 | 253 | ) |
255 | 254 |
|
256 | 255 | @override |
257 | | - def print_device(self, out_file: Path | None = None) -> None: |
| 256 | + def print(self, file: SupportsWrite[str] | None = None) -> None: |
258 | 257 | """Print human-readable information about the device to stdout or file.""" |
259 | 258 | # ruff: noqa: T201 |
260 | | - if out_file: |
261 | | - data = { |
262 | | - "mac_address": self.mac_address, |
263 | | - "status": self.status, |
264 | | - "device_type": self.device_type, |
265 | | - "battery_level": self.battery_level, |
266 | | - "rssi": self.rssi, |
267 | | - "detected_at": self.detected_at.isoformat(), |
268 | | - "additional_data": self.additional_data, |
269 | | - } |
270 | | - |
271 | | - with out_file.open("a", encoding="utf-8") as f: |
272 | | - # Indent json for readability |
273 | | - json.dump(data, f, indent=4) |
274 | | - f.write("\n") |
275 | | - else: |
276 | | - print(f"Nearby {self.device_type} - {self.mac_address}") |
277 | | - print(f" Status byte: 0x{self.status:x}") |
278 | | - print(f" Battery lvl: {self.battery_level}") |
279 | | - print(f" RSSI: {self.rssi}") |
280 | | - print(" Extra data:") |
281 | | - for k, v in sorted(self.additional_data.items()): |
282 | | - print(f" {k:20}: {v}") |
283 | | - print("\n") |
| 259 | + print(f"Nearby {self.device_type} - {self.mac_address}", file=file) |
| 260 | + print(f" Status byte: 0x{self.status:x}", file=file) |
| 261 | + print(f" Battery lvl: {self.battery_level}", file=file) |
| 262 | + print(f" RSSI: {self.rssi}", file=file) |
| 263 | + print(" Extra data:", file=file) |
| 264 | + for k, v in sorted(self.additional_data.items()): |
| 265 | + print(f" {k:20}: {v}", file=file) |
| 266 | + print("\n", file=file) |
284 | 267 |
|
285 | 268 |
|
286 | 269 | class SeparatedOfflineFindingDevice(OfflineFindingDevice, HasPublicKey): |
@@ -383,38 +366,20 @@ def from_payload( |
383 | 366 | ) |
384 | 367 |
|
385 | 368 | @override |
386 | | - def print_device(self, out_file: Path | None = None) -> None: |
| 369 | + def print(self, file: SupportsWrite[str] | None = None) -> None: |
387 | 370 | """Print human-readable information about the device to stdout or file.""" |
388 | 371 | # ruff: noqa: T201 |
389 | | - if out_file: |
390 | | - data = { |
391 | | - "mac_address": self.mac_address, |
392 | | - "public_key": self.adv_key_b64, |
393 | | - "lookup_key": self.hashed_adv_key_b64, |
394 | | - "status": self.status, |
395 | | - "device_type": self.device_type, |
396 | | - "battery_level": self.battery_level, |
397 | | - "hint": self.hint, |
398 | | - "rssi": self.rssi, |
399 | | - "detected_at": self.detected_at.isoformat(), |
400 | | - "additional_data": self.additional_data, |
401 | | - } |
402 | | - |
403 | | - with out_file.open("a", encoding="utf-8") as f: |
404 | | - json.dump(data, f, indent=4) |
405 | | - f.write("\n") |
406 | | - else: |
407 | | - print(f"Separated {self.device_type} - {self.mac_address}") |
408 | | - print(f" Public key: {self.adv_key_b64}") |
409 | | - print(f" Lookup key: {self.hashed_adv_key_b64}") |
410 | | - print(f" Status byte: 0x{self.status:x}") |
411 | | - print(f" Battery lvl: {self.battery_level}") |
412 | | - print(f" Hint byte: 0x{self.hint:x}") |
413 | | - print(f" RSSI: {self.rssi}") |
414 | | - print(" Extra data:") |
415 | | - for k, v in sorted(self.additional_data.items()): |
416 | | - print(f" {k:20}: {v}") |
417 | | - print("\n") |
| 372 | + print(f"Separated {self.device_type} - {self.mac_address}", file=file) |
| 373 | + print(f" Public key: {self.adv_key_b64}", file=file) |
| 374 | + print(f" Lookup key: {self.hashed_adv_key_b64}", file=file) |
| 375 | + print(f" Status byte: 0x{self.status:x}", file=file) |
| 376 | + print(f" Battery lvl: {self.battery_level}", file=file) |
| 377 | + print(f" Hint byte: 0x{self.hint:x}", file=file) |
| 378 | + print(f" RSSI: {self.rssi}", file=file) |
| 379 | + print(" Extra data:", file=file) |
| 380 | + for k, v in sorted(self.additional_data.items()): |
| 381 | + print(f" {k:20}: {v}", file=file) |
| 382 | + print("\n", file=file) |
418 | 383 |
|
419 | 384 | @override |
420 | 385 | def __repr__(self) -> str: |
|
0 commit comments