Bluez DBus Emulator v0.1.0

Initial Release

Author: kdewald

.github/workflows/deploy.yml

@@ -0,0 +1,59 @@
+name: Run Deploy
+
+on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: New Version
+        default: ""
+        required: true
+      publish:
+        description: Make Public(true or false)
+        default: "false"
+        required: true
+
+jobs:
+  RunDeploy:
+    runs-on: windows-2019
+
+    steps:
+    # check users permission
+    - name: "Check Permissions"
+      uses: "lannonbr/[email protected]"
+      with:
+        permission: "admin"
+      env:
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+    # Install Dependencies
+    - name: Clone Repository
+      uses: actions/checkout@v2
+    - name: Install Python Deploy Tools
+      run: |
+        python -m pip install wheel
+        python -m pip install twine
+      shell: cmd
+    # Prepare package
+    - name: Prepare Python Package
+      run: |
+        cd $env:GITHUB_WORKSPACE
+        (gc .\setup.py).replace('0.0.1', $env:VERSION) | Out-File -encoding ASCII setup.py
+        type setup.py
+        python setup.py sdist bdist_wheel
+      env:
+        PYPI_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+        VERSION: ${{ github.event.inputs.version }}
+    # Publish package
+    - name: Publish Packages
+      if: github.event.inputs.publish == 'true'
+      run: |
+        cd $env:GITHUB_WORKSPACE
+        twine upload --skip-existing dist/*.whl --user $env:PYPI_USER --password $env:PYPI_PASSWORD
+      env:
+        PYPI_USER: $ {{ secrets.PYPI_USER }}
+        PYPI_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+    # To be able to test/share whl wo publishing
+    - name: Upload Artifacts
+      uses: actions/upload-artifact@v1
+      with:
+        name: dist
+        path: dist

CHANGELOG.md

@@ -0,0 +1,8 @@
+# Changelog
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [0.1.0] - 2021-07-24
+- First implementation with a basic example emulating a device exposing the Nordic UART Service.

README.md

@@ -1 +1,28 @@
-# python_bluez_dbus_emulator
+# bluez_dbus_emulator
+
+A simple set of libraries to allow emulating the behavior of a BlueZ
+Bluetooth device over DBus.
+
+## Prerequisites
+
+Before you begin, ensure you have met the following requirements:
+- [dbus_next](https://github.com/altdesktop/python-dbus-next)
+
+## Installation
+
+```
+pip3 install bluez_dbus_emulator
+```
+
+## Usage
+
+For usage instructions, just follow the examples provided in the `examples` folder.
+
+## Contributors
+
+Thanks to the following people who have contributed to this project:
+* [@Andrey1994](https://github.com/Andrey1994)
+
+## License
+
+This project is licensed under the terms of the [MIT Licence](LICENCE.md).

bluez_dbus_emulator/__init__.py

@@ -0,0 +1,5 @@
+from bluez_dbus_emulator.emulator import Emulator
+from bluez_dbus_emulator.adapter1 import Adapter1
+from bluez_dbus_emulator.device1 import Device1
+from bluez_dbus_emulator.gattservice1 import GattService1
+from bluez_dbus_emulator.gattcharacteristic1 import GattCharacteristic1

bluez_dbus_emulator/adapter1.py

@@ -0,0 +1,51 @@
+from dbus_next.service import ServiceInterface, method, dbus_property, PropertyAccess
+
+import asyncio
+import random
+
+
+class Adapter1(ServiceInterface):
+    def __init__(self, bus, path):
+        self.bus = bus
+        self.path = path
+        super().__init__("org.bluez.Adapter1")
+        self._discovering = False
+        self._address = "00:00:00:00:00:00"
+
+        self._devices = []
+
+    def export(self):
+        self.bus.export(f"/org/bluez/{self.path}", self)
+
+    def add_device(self, device):
+        self._devices.append(device)
+
+    @method()
+    def SetDiscoveryFilter(self, properties: "a{sv}"):
+        return
+
+    @method()
+    async def StartDiscovery(self):
+        print("StartDiscovery")
+        await self._update_discoverying(True)
+        for device in self._devices:
+            await device.task_scanning_start()
+        return
+
+    @method()
+    async def StopDiscovery(self):
+        print("StopDiscovery")
+        await self._update_discoverying(False)
+        for device in self._devices:
+            device.task_scanning_stop()
+        return
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Discovering(self) -> "b":
+        return self._discovering
+
+    async def _update_discoverying(self, new_value: bool):
+        await asyncio.sleep(random.uniform(0.5, 1.5))
+        self._discovering = new_value
+        self.emit_properties_changed({"Discovering": self._discovering})
+        print(f"Property changed: {self._discovering}")

bluez_dbus_emulator/device1.py

@@ -0,0 +1,115 @@
+from dbus_next.service import ServiceInterface, method, dbus_property, PropertyAccess
+
+import asyncio
+import random
+
+
+class Device1(ServiceInterface):
+    def __init__(self, bus, parent_path, mac_address="00:00:00:00:00:00"):
+        self.bus = bus
+        self.path = f"{parent_path}/dev_{'_'.join(mac_address.split(':'))}"
+        super().__init__("org.bluez.Device1")
+        self._exported = False
+        self._connected = False
+        self._services_resolved = False
+        self._rssi = -128
+        self._address = mac_address
+        self._name = f"dev_{'_'.join(mac_address.split(':'))}"
+        self._services = []
+
+        self.__task_scanning_active = False
+        self.__task_connected_active = False
+
+    async def export(self):
+        if not self._exported:
+            await asyncio.sleep(random.uniform(0.5, 1.5))
+            self.bus.export(f"/org/bluez/{self.path}", self)
+            self._exported = True
+
+    def add_service(self, service):
+        self._services.append(service)
+
+    def run_connected_task(self):
+        pass
+
+    async def task_scanning_start(self):
+        await self.export()
+        self.__task_scanning_active = True
+        asyncio.ensure_future(self._task_scanning_run())
+
+    def task_scanning_stop(self):
+        self.__task_scanning_active = False
+
+    def task_connected_start(self):
+        self.__task_connected_active = True
+        asyncio.ensure_future(self._task_connected_run())
+
+    def task_connected_stop(self):
+        self.__task_connected_active = False
+
+    async def _task_scanning_run(self):
+        await asyncio.sleep(random.uniform(0.02, 0.2))
+        # Execute scanning tasks
+        await self._update_rssi(random.uniform(-90, -60))
+        if self.__task_scanning_active:
+            asyncio.ensure_future(self._task_scanning_run())
+
+    async def _task_connected_run(self):
+        await asyncio.sleep(0.015)
+        # Execute connected tasks
+        self.run_connected_task()
+        if self.__task_connected_active:
+            asyncio.ensure_future(self._task_connected_run())
+
+    @method()
+    async def Connect(self):
+        print("Connect")
+        await self._update_connected(True)
+        for service in self._services:
+            service.export()
+        await self._update_services_resolved(True)
+        self.task_connected_start()
+        return
+
+    @method()
+    async def Disconnect(self):
+        print("Disconnect")
+        await self._update_services_resolved(False)
+        self.task_connected_stop()
+        await self._update_connected(False)
+        return
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Connected(self) -> "b":
+        return self._connected
+
+    @dbus_property(access=PropertyAccess.READ)
+    def ServicesResolved(self) -> "b":
+        return self._services_resolved
+
+    @dbus_property(access=PropertyAccess.READ)
+    def RSSI(self) -> "n":
+        return self._rssi
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Name(self) -> "s":
+        return self._name
+
+    async def _update_connected(self, new_value: bool):
+        await asyncio.sleep(random.uniform(0.5, 1.5))
+        self._connected = new_value
+        property_changed = {"Connected": self._connected}
+        self.emit_properties_changed(property_changed)
+        print(f"Property changed: {property_changed}")
+
+    async def _update_services_resolved(self, new_value: bool):
+        await asyncio.sleep(random.uniform(0.0, 0.5))
+        self._services_resolved = new_value
+        property_changed = {"ServicesResolved": self._services_resolved}
+        self.emit_properties_changed(property_changed)
+        print(f"Property changed: {property_changed}")
+
+    async def _update_rssi(self, new_value: int):
+        self._rssi = int(new_value)
+        property_changed = {"RSSI": self._rssi}
+        self.emit_properties_changed(property_changed)

bluez_dbus_emulator/emulator.py

@@ -0,0 +1,22 @@
+from dbus_next.service import ServiceInterface, method
+
+
+class Emulator(ServiceInterface):
+    """
+    Entry point for the bluez dbus emulator. Provides programmatic
+    control of the async application.
+    """
+
+    def __init__(self, bus):
+        self.bus = bus
+        super().__init__("emulator.bluez_dbus")
+
+    def export(self):
+        self.bus.export("/", self)
+
+    @method()
+    def Exit(self):
+        """
+        Finishes the emulation session by disconnecting from dbus.
+        """
+        self.bus.disconnect()

bluez_dbus_emulator/gattcharacteristic1.py

@@ -0,0 +1,65 @@
+from dbus_next.service import ServiceInterface, method, dbus_property, PropertyAccess
+
+import asyncio
+import random
+
+
+class GattCharacteristic1(ServiceInterface):
+    def __init__(self, bus, parent_path, id_num, uuid):
+        self.bus = bus
+        self.path = f"{parent_path}/char{id_num:04x}"
+        super().__init__("org.bluez.GattCharacteristic1")
+        self._uuid = uuid
+        self._value = bytes()
+        self._notifying = False
+        self._exported = False
+
+    def export(self):
+        if not self._exported:
+            self.bus.export(f"/org/bluez/{self.path}", self)
+            self._exported = True
+
+    def update_value(self, new_value: bytes):
+        self._update_value(new_value)
+
+    @method()
+    async def StartNotify(self):
+        await self._update_notifying(True)
+        return
+
+    @method()
+    async def StopNotify(self):
+        await self._update_notifying(False)
+        return
+
+    @method()
+    def ReadValue(self, options: "a{sv}") -> "ay":
+        return self._value
+
+    @method()
+    def WriteValue(self, value: "ay", options: "a{sv}"):
+        self._update_value(value)
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Notifying(self) -> "b":
+        return self._notifying
+
+    @dbus_property(access=PropertyAccess.READ)
+    def UUID(self) -> "s":
+        return self._uuid
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Value(self) -> "ay":
+        return self._value
+
+    def _update_value(self, new_value: bytes):
+        self._value = new_value
+        if self._notifying:
+            property_changed = {"Value": self._value}
+            self.emit_properties_changed(property_changed)
+
+    async def _update_notifying(self, new_value: bool):
+        await asyncio.sleep(random.uniform(0.0, 0.2))
+        self._notifying = new_value
+        property_changed = {"Notifying": self._notifying}
+        self.emit_properties_changed(property_changed)

bluez_dbus_emulator/gattservice1.py

@@ -0,0 +1,25 @@
+from dbus_next.service import ServiceInterface, dbus_property, PropertyAccess
+
+
+class GattService1(ServiceInterface):
+    def __init__(self, bus, parent_path, id_num, uuid):
+        self.bus = bus
+        self.path = f"{parent_path}/service{id_num:04x}"
+        super().__init__("org.bluez.GattService1")
+        self._uuid = uuid
+        self._exported = False
+        self._characteristics = []
+
+    def export(self):
+        if not self._exported:
+            self.bus.export(f"/org/bluez/{self.path}", self)
+            for char in self._characteristics:
+                char.export()
+            self._exported = True
+
+    def add_characteristic(self, characteristic):
+        self._characteristics.append(characteristic)
+
+    @dbus_property(access=PropertyAccess.READ)
+    def UUID(self) -> "s":
+        return self._uuid