Black formatting

This commit is contained in:
jpk 2023-05-20 14:13:37 +02:00
parent 6020b7c38b
commit a20c35aefd
4 changed files with 25 additions and 23 deletions

View File

@ -1,6 +1,6 @@
from textual.message import Message
from bleak.backends.device import BLEDevice from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData from bleak.backends.scanner import AdvertisementData
from textual.message import Message
class DeviceDiscovered(Message): class DeviceDiscovered(Message):

View File

@ -1,21 +1,17 @@
import asyncio import asyncio
from binascii import hexlify from binascii import hexlify
from rich.live import Live
from rich.table import Table
import bleak.exc import bleak.exc
from bleak import BleakClient, BleakGATTCharacteristic from bleak import BleakClient, BleakGATTCharacteristic
from bleak.backends.bluezdbus.characteristic import BleakGATTCharacteristicBlueZDBus from bleak.backends.bluezdbus.characteristic import BleakGATTCharacteristicBlueZDBus
from pydispatch import Dispatcher from pydispatch import Dispatcher
from rich.live import Live
from rich.table import Table
from ...events.event import DeviceDiscovered from ...events.event import DeviceDiscovered
from ...tools import context from ...tools import context
from ...tools.context import BlattedEnvironment from ...tools.context import BlattedEnvironment
environment = context.get_environment() environment = context.get_environment()
console = context.get_console() console = context.get_console()
@ -39,7 +35,9 @@ class Monitor(Dispatcher):
self.bind(update_service=self.on_update_service) self.bind(update_service=self.on_update_service)
super().__init__() 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) data_received = (sender, data)
if environment == BlattedEnvironment.CLI: if environment == BlattedEnvironment.CLI:
self.emit("update_service", data=data_received) self.emit("update_service", data=data_received)
@ -71,7 +69,9 @@ class Monitor(Dispatcher):
for char in service.characteristics: for char in service.characteristics:
if "notify" in char.properties: if "notify" in char.properties:
try: try:
await ble_client.start_notify(char.uuid, self.notify_handler) await ble_client.start_notify(
char.uuid, self.notify_handler
)
if BlattedEnvironment.CLI: if BlattedEnvironment.CLI:
self.emit("acquire_notify", char) self.emit("acquire_notify", char)
self.acquired_notify.append(char.uuid) self.acquired_notify.append(char.uuid)
@ -88,6 +88,7 @@ class Monitor(Dispatcher):
# TODO notify error # TODO notify error
continue continue
def run(ble_address: str, service_uuids: list[str] = []): def run(ble_address: str, service_uuids: list[str] = []):
if environment == BlattedEnvironment.CLI: if environment == BlattedEnvironment.CLI:
live.start() live.start()

View File

@ -1,20 +1,18 @@
import asyncio import asyncio
from typing import Dict, Any from typing import Any, Dict
from rich.live import Live
from rich.table import Table
import bleak.exc import bleak.exc
from bleak import BleakScanner from bleak import BleakScanner
from bleak.backends.device import BLEDevice from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData from bleak.backends.scanner import AdvertisementData
from pydispatch import Dispatcher from pydispatch import Dispatcher
from rich.live import Live
from rich.table import Table
from ...events.event import DeviceDiscovered from ...events.event import DeviceDiscovered
from ...tools import context from ...tools import context
from ...tools.context import BlattedEnvironment from ...tools.context import BlattedEnvironment
environment = context.get_environment() environment = context.get_environment()
console = context.get_console() console = context.get_console()
@ -23,7 +21,7 @@ live = Live(table, console=console)
table.add_column("Address", max_width=18) table.add_column("Address", max_width=18)
table.add_column("Name", max_width=32) table.add_column("Name", max_width=32)
table.add_column("RSSI", max_width=5) table.add_column("RSSI", max_width=5)
table.add_column("Services", width=100-55) table.add_column("Services", width=100 - 55)
class Scanner(Dispatcher): class Scanner(Dispatcher):
@ -31,7 +29,7 @@ class Scanner(Dispatcher):
def __init__(self) -> None: def __init__(self) -> None:
self.environment = context.get_environment() self.environment = context.get_environment()
self.devices: Dict[str, Any] = {} self.devices: dict[str, Any] = {}
self.bind(device_discovered=self.on_device_discovered) self.bind(device_discovered=self.on_device_discovered)
super().__init__() super().__init__()
@ -39,7 +37,7 @@ class Scanner(Dispatcher):
self, device: BLEDevice, advertising_data: AdvertisementData self, device: BLEDevice, advertising_data: AdvertisementData
) -> None: ) -> None:
if len(advertising_data.service_uuids) > 0: if len(advertising_data.service_uuids) > 0:
#discovered = DiscoveredDevice(device, advertising_data) # discovered = DiscoveredDevice(device, advertising_data)
discovered = DeviceDiscovered(device, advertising_data) discovered = DeviceDiscovered(device, advertising_data)
if environment == BlattedEnvironment.CLI: if environment == BlattedEnvironment.CLI:
self.emit("device_discovered", data=discovered) self.emit("device_discovered", data=discovered)
@ -49,9 +47,12 @@ class Scanner(Dispatcher):
def on_device_discovered(self, data: DeviceDiscovered) -> None: def on_device_discovered(self, data: DeviceDiscovered) -> None:
if data.device.address not in self.devices: if data.device.address not in self.devices:
row = table.add_row(f"{data.device.address}", f"{data.device.name}", table.add_row(
f"{data.adverisement_data.rssi}", f"{data.device.address}",
"\n".join(data.adverisement_data.service_uuids)) 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} self.devices[data.device.address] = {"data": data, "seen": 1}
else: else:
known_data = self.devices[data.device.address] known_data = self.devices[data.device.address]

View File

@ -1,11 +1,11 @@
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
from textual.containers import Container from textual.containers import Container
from textual.widgets import Header, Footer from textual.widgets import Footer, Header
class BlattedApp(App): class BlattedApp(App):
CSS_PATH = "blatted.css" CSS_PATH = "blatted.css"
SCREENS = { } SCREENS = {}
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield Header() yield Header()