diff --git a/blatted/events/event.py b/blatted/events/event.py index e46a20b..aa28f5c 100644 --- a/blatted/events/event.py +++ b/blatted/events/event.py @@ -1,6 +1,6 @@ -from textual.message import Message from bleak.backends.device import BLEDevice from bleak.backends.scanner import AdvertisementData +from textual.message import Message class DeviceDiscovered(Message): diff --git a/blatted/tools/ble/monitor.py b/blatted/tools/ble/monitor.py index 7ef23f4..c82d3d5 100644 --- a/blatted/tools/ble/monitor.py +++ b/blatted/tools/ble/monitor.py @@ -1,21 +1,17 @@ import asyncio - from binascii import hexlify -from rich.live import Live -from rich.table import Table - import bleak.exc from bleak import BleakClient, BleakGATTCharacteristic from bleak.backends.bluezdbus.characteristic import BleakGATTCharacteristicBlueZDBus - from pydispatch import Dispatcher +from rich.live import Live +from rich.table import Table from ...events.event import DeviceDiscovered from ...tools import context from ...tools.context import BlattedEnvironment - environment = context.get_environment() console = context.get_console() @@ -39,7 +35,9 @@ class Monitor(Dispatcher): self.bind(update_service=self.on_update_service) super().__init__() - async def notify_handler(self, sender: BleakGATTCharacteristic, data: bytearray) -> None: + async def notify_handler( + self, sender: BleakGATTCharacteristic, data: bytearray + ) -> None: data_received = (sender, data) if environment == BlattedEnvironment.CLI: self.emit("update_service", data=data_received) @@ -71,7 +69,9 @@ class Monitor(Dispatcher): for char in service.characteristics: if "notify" in char.properties: try: - await ble_client.start_notify(char.uuid, self.notify_handler) + await ble_client.start_notify( + char.uuid, self.notify_handler + ) if BlattedEnvironment.CLI: self.emit("acquire_notify", char) self.acquired_notify.append(char.uuid) @@ -88,6 +88,7 @@ class Monitor(Dispatcher): # TODO notify error continue + def run(ble_address: str, service_uuids: list[str] = []): if environment == BlattedEnvironment.CLI: live.start() diff --git a/blatted/tools/ble/scanner.py b/blatted/tools/ble/scanner.py index 534429b..7a78a1f 100644 --- a/blatted/tools/ble/scanner.py +++ b/blatted/tools/ble/scanner.py @@ -1,20 +1,18 @@ import asyncio -from typing import Dict, Any - -from rich.live import Live -from rich.table import Table +from typing import Any, Dict import bleak.exc from bleak import BleakScanner from bleak.backends.device import BLEDevice from bleak.backends.scanner import AdvertisementData from pydispatch import Dispatcher +from rich.live import Live +from rich.table import Table from ...events.event import DeviceDiscovered from ...tools import context from ...tools.context import BlattedEnvironment - environment = context.get_environment() console = context.get_console() @@ -23,7 +21,7 @@ live = Live(table, console=console) table.add_column("Address", max_width=18) table.add_column("Name", max_width=32) table.add_column("RSSI", max_width=5) -table.add_column("Services", width=100-55) +table.add_column("Services", width=100 - 55) class Scanner(Dispatcher): @@ -31,7 +29,7 @@ class Scanner(Dispatcher): def __init__(self) -> None: self.environment = context.get_environment() - self.devices: Dict[str, Any] = {} + self.devices: dict[str, Any] = {} self.bind(device_discovered=self.on_device_discovered) super().__init__() @@ -39,7 +37,7 @@ class Scanner(Dispatcher): self, device: BLEDevice, advertising_data: AdvertisementData ) -> None: if len(advertising_data.service_uuids) > 0: - #discovered = DiscoveredDevice(device, advertising_data) + # discovered = DiscoveredDevice(device, advertising_data) discovered = DeviceDiscovered(device, advertising_data) if environment == BlattedEnvironment.CLI: self.emit("device_discovered", data=discovered) @@ -49,9 +47,12 @@ class Scanner(Dispatcher): def on_device_discovered(self, data: DeviceDiscovered) -> None: if data.device.address not in self.devices: - row = table.add_row(f"{data.device.address}", f"{data.device.name}", - f"{data.adverisement_data.rssi}", - "\n".join(data.adverisement_data.service_uuids)) + table.add_row( + f"{data.device.address}", + f"{data.device.name}", + f"{data.adverisement_data.rssi}", + "\n".join(data.adverisement_data.service_uuids), + ) self.devices[data.device.address] = {"data": data, "seen": 1} else: known_data = self.devices[data.device.address] diff --git a/blatted/tui/app.py b/blatted/tui/app.py index 3788645..94cb42f 100644 --- a/blatted/tui/app.py +++ b/blatted/tui/app.py @@ -1,11 +1,11 @@ -from textual.app import App, ComposeResult +from textual.app import App, ComposeResult from textual.containers import Container -from textual.widgets import Header, Footer +from textual.widgets import Footer, Header class BlattedApp(App): CSS_PATH = "blatted.css" - SCREENS = { } + SCREENS = {} def compose(self) -> ComposeResult: yield Header()