Added support for custom Bluetooth device plugins.

This commit is contained in:
Fabio Manganiello 2023-03-23 17:10:37 +01:00
parent af125347d6
commit d6805a8b18
Signed by: blacklight
GPG key ID: D90FBA7F76362774
7 changed files with 103 additions and 3 deletions

View file

@ -19,6 +19,7 @@ from platypush.message.event.bluetooth import (
from .._cache import EntityCache from .._cache import EntityCache
from .._model import ServiceClass from .._model import ServiceClass
from .._plugins import BaseBluetoothPlugin
from ._cache import DeviceCache from ._cache import DeviceCache
from ._mappers import device_to_entity from ._mappers import device_to_entity
@ -89,7 +90,7 @@ event_matchers: Dict[
or (old.reachable is False and new.reachable is True), or (old.reachable is False and new.reachable is True),
BluetoothDeviceSignalUpdateEvent: lambda old, new: ( BluetoothDeviceSignalUpdateEvent: lambda old, new: (
(new.rssi is not None or new.tx_power is not None) (new.rssi is not None or new.tx_power is not None)
and (old and old.rssi) and bool(old and old.rssi)
and ( and (
_has_changed(old, new, 'rssi', tolerance=5) _has_changed(old, new, 'rssi', tolerance=5)
or _has_changed(old, new, 'tx_power', tolerance=5) or _has_changed(old, new, 'tx_power', tolerance=5)
@ -115,6 +116,7 @@ class EventHandler:
device_queue: Queue, device_queue: Queue,
device_cache: DeviceCache, device_cache: DeviceCache,
entity_cache: EntityCache, entity_cache: EntityCache,
plugins: Collection[BaseBluetoothPlugin],
exclude_known_noisy_beacons: bool, exclude_known_noisy_beacons: bool,
): ):
""" """
@ -126,6 +128,7 @@ class EventHandler:
self._device_queue = device_queue self._device_queue = device_queue
self._device_cache = device_cache self._device_cache = device_cache
self._entity_cache = entity_cache self._entity_cache = entity_cache
self._plugins = plugins
self._exclude_known_noisy_beacons = exclude_known_noisy_beacons self._exclude_known_noisy_beacons = exclude_known_noisy_beacons
def __call__(self, device: BLEDevice, data: AdvertisementData): def __call__(self, device: BLEDevice, data: AdvertisementData):
@ -155,6 +158,10 @@ class EventHandler:
) )
return return
# Extend the new entity with children entities added by the plugins
for plugin in self._plugins:
plugin.extend_device(new_entity)
events: List[BluetoothDeviceEvent] = [] events: List[BluetoothDeviceEvent] = []
existing_entity = self._entity_cache.get(device.address) existing_entity = self._entity_cache.get(device.address)
events += [ events += [
@ -195,7 +202,7 @@ class EventHandler:
return False return False
mapped_uuids = [ mapped_uuids = [
int(str(srv.uuid).split('-')[0], 16) & 0xFFFF int(str(srv.uuid).split('-', maxsplit=1)[0], 16) & 0xFFFF
if isinstance(srv.uuid, UUID) if isinstance(srv.uuid, UUID)
else srv.uuid else srv.uuid
for srv in device.services for srv in device.services

View file

@ -210,7 +210,7 @@ def _parse_services(
srv_cls = ServiceClass.get(uuid) srv_cls = ServiceClass.get(uuid)
services.append( services.append(
BluetoothService( BluetoothService(
id=f'{device.address}:{uuid}', id=f'{device.address}::{uuid}',
uuid=uuid, uuid=uuid,
name=f'[{uuid}]' if srv_cls == ServiceClass.UNKNOWN else str(srv_cls), name=f'[{uuid}]' if srv_cls == ServiceClass.UNKNOWN else str(srv_cls),
protocol=Protocol.L2CAP, protocol=Protocol.L2CAP,

View file

@ -43,6 +43,8 @@ class BaseBluetoothManager(ABC, threading.Thread):
:param device_cache: Cache used to keep track of discovered devices. :param device_cache: Cache used to keep track of discovered devices.
:param exclude_known_noisy_beacons: Exclude known noisy beacons. :param exclude_known_noisy_beacons: Exclude known noisy beacons.
""" """
from ._plugins import scan_plugins
kwargs['name'] = f'Bluetooth:{self.__class__.__name__}' kwargs['name'] = f'Bluetooth:{self.__class__.__name__}'
super().__init__(**kwargs) super().__init__(**kwargs)
@ -59,6 +61,8 @@ class BaseBluetoothManager(ABC, threading.Thread):
self._cache = device_cache or EntityCache() self._cache = device_cache or EntityCache()
""" Cache of discovered devices. """ """ Cache of discovered devices. """
self._plugins = scan_plugins(self)
""" Plugins compatible with this manager. """
def notify( def notify(
self, event_type: Type[BluetoothDeviceEvent], device: BluetoothDevice, **kwargs self, event_type: Type[BluetoothDeviceEvent], device: BluetoothDevice, **kwargs
@ -71,6 +75,10 @@ class BaseBluetoothManager(ABC, threading.Thread):
get_bus().post(event_type.from_device(device=device, **kwargs)) get_bus().post(event_type.from_device(device=device, **kwargs))
self._device_queue.put_nowait(device) self._device_queue.put_nowait(device)
@property
def plugins(self):
return self._plugins
def should_stop(self) -> bool: def should_stop(self) -> bool:
return self._stop_event.is_set() return self._stop_event.is_set()

View file

@ -0,0 +1,4 @@
from ._base import BaseBluetoothPlugin
from ._scanner import scan_plugins
__all__ = ['BaseBluetoothPlugin', 'scan_plugins']

View file

@ -0,0 +1,39 @@
from abc import ABC, abstractmethod
from typing import Iterable
from platypush.entities import Entity
from platypush.entities.bluetooth import BluetoothDevice
from platypush.plugins.bluetooth._manager import BaseBluetoothManager
class BaseBluetoothPlugin(ABC):
"""
Base class for Bluetooth plugins, like Switchbot or HID integrations.
"""
def __init__(self, manager: BaseBluetoothManager):
self._manager = manager
@abstractmethod
def supports_device(self, device: BluetoothDevice) -> bool:
"""
Returns True if the given device matches this plugin.
"""
@abstractmethod
def _extract_entities(self, device: BluetoothDevice) -> Iterable[Entity]:
"""
If :meth:`_matches_device` returns True, this method should return an
iterable of entities that will be used to extend the existing device -
like sensors, switches etc.
"""
def extend_device(self, device: BluetoothDevice):
"""
Extends the given device with entities extracted through
:meth:`_extract_entities`, if :meth:`_matches_device` returns True.
"""
if not self.supports_device(device):
return
device.children.extend(self._extract_entities(device))

View file

@ -0,0 +1,42 @@
import importlib
import inspect
import logging
import os
import pkgutil
from typing import List
from platypush.plugins.bluetooth._plugins import BaseBluetoothPlugin
logger = logging.getLogger(__name__)
def scan_plugins(manager) -> List[BaseBluetoothPlugin]:
"""
Initializes all the plugins associated to the given BluetoothManager by
scanning all the modules under the ``_plugins`` folder in the manager's
package.
"""
plugins = {}
base_dir = os.path.dirname(inspect.getfile(manager.__class__))
module = inspect.getmodule(manager.__class__)
assert module is not None
package = module.__package__
assert package is not None
for _, mod_name, _ in pkgutil.walk_packages([base_dir], prefix=package + '.'):
try:
module = importlib.import_module(mod_name)
except Exception as e:
logger.warning('Could not import module %s', mod_name)
logger.exception(e)
continue
for _, obj in inspect.getmembers(module):
if (
inspect.isclass(obj)
and not inspect.isabstract(obj)
and issubclass(obj, BaseBluetoothPlugin)
):
plugins[obj] = obj(manager)
return list(plugins.values())