Add UARTClient; refactor common code

Author: dhalbert

adafruit_ble/scanner.py

@@ -54,6 +54,21 @@ class Scanner:
     def __init__(self):
         self._scanner = bleio.Scanner()
 
+    def scan_unique(self, timeout, *, interval=0.1, window=0.1):
+        """Scan for advertisements from BLE devices. Suppress duplicates
+        in returned `ScanEntry` objects.
+
+        :param int timeout: how long to scan for (in seconds)
+        :param float interval: the interval (in seconds) between the start
+           of two consecutive scan windows.
+           Must be in the range 0.0025 - 40.959375 seconds.
+        :param float window: the duration (in seconds) to scan a single BLE channel.
+          `window` must be <= `interval`.
+        :returns a list of `adafruit_ble.ScanEntry` objects.
+
+        """
+        return ScanEntry.unique(self.scan(timeout, interval=interval, window=window))
+
     def scan(self, timeout, *, interval=0.1, window=0.1):
         """Scan for advertisements from BLE devices.
 
@@ -122,23 +137,25 @@ def name(self):
     @property
     def service_uuids(self):
         """List of all the service UUIDs in the advertisement."""
+        uuid_values = []
+
         concat_uuids = self.item(AdvertisingPacket.ALL_16_BIT_SERVICE_UUIDS)
         concat_uuids = concat_uuids if concat_uuids else self.item(
             AdvertisingPacket.SOME_16_BIT_SERVICE_UUIDS)
 
-        uuid_values = []
         if concat_uuids:
-            for i in range(0, len(uuid_values), 2):
+            for i in range(0, len(concat_uuids), 2):
                 uuid_values.append(struct.unpack("<H", concat_uuids[i:i+2]))
 
         concat_uuids = self.item(AdvertisingPacket.ALL_128_BIT_SERVICE_UUIDS)
         concat_uuids = concat_uuids if concat_uuids else self.item(
             AdvertisingPacket.SOME_128_BIT_SERVICE_UUIDS)
 
         if concat_uuids:
-            for i in range(0, len(uuid_values), 16):
+            for i in range(0, len(concat_uuids), 16):
                 uuid_values.append(concat_uuids[i:i+16])
 
+        print(uuid_values)
         return [bleio.UUID(value) for value in uuid_values]
 
     @property
@@ -150,10 +167,16 @@ def matches(self, other):
         """True if two scan entries appear to be from the same device. Their
         addresses and advertisement_bytes must match.
         """
-        return self.address == other.address and self.advertisement_bytes == other.advertisement_bytes
+        return (self.address == other.address and
+                self.advertisement_bytes == other.advertisement_bytes)
+
+    @staticmethod
+    def with_service_uuid(scan_entries, service_uuid):
+        """Return all scan entries advertising the given service_uuid."""
+        return [se for se in scan_entries if service_uuid in se.service_uuids]
 
-    @classmethod
-    def unique(self, scan_entries):
+    @staticmethod
+    def unique(scan_entries):
         """Discard duplicate scan entries that appear to be from the same device.
 
         :param sequence scan_entries: ScanEntry objects
@@ -162,5 +185,5 @@ def unique(self, scan_entries):
         unique = []
         for entry in scan_entries:
             if not any(entry.matches(unique_entry) for unique_entry in unique):
-                unique.append(entry);
+                unique.append(entry)
         return unique

adafruit_ble/uart.py

@@ -1,6 +1,6 @@
 # The MIT License (MIT)
 #
-# Copyright (c) 2018 Dan Halbert for Adafruit Industries
+# Copyright (c) 2019 Dan Halbert for Adafruit Industries
 #
 # Permission is hereby granted, free of charge, to any person obtaining a copy
 # of this software and associated documentation files (the "Software"), to deal
@@ -23,68 +23,42 @@
 `adafruit_ble.uart`
 ====================================================
 
-UART-style communication.
+BLE UART-style communication. Common definitions.
 
 * Author(s): Dan Halbert for Adafruit Industries
 
 """
-from bleio import UUID, Characteristic, Service, Peripheral, CharacteristicBuffer
-from .advertising import ServerAdvertisement
+from bleio import UUID, CharacteristicBuffer
 
-class UARTServer:
+
+
+
+class UART:
     """
-    Provide UART-like functionality via the Nordic NUS service.
+    Common superclass for Nordic UART Service (NUS) clients or servers.
+    Not for general use: use `UARTServer` and `UARTClient` for Peripheral and Central,
+    respectively.
 
+    :param read_characteristic Characteristic: Characteristic to read from
+    :param write_characteristic Characteristic: Characteristic to write to
     :param int timeout:  the timeout in seconds to wait
-      for the first character and between subsequent characters.
+      for the first character and between subsequent characters
     :param int buffer_size: buffer up to this many bytes.
       If more bytes are received, older bytes will be discarded.
-    :param str name: Name to advertise for server. If None, use default Peripheral name.
-
-    Example::
-
-        from adafruit_ble.uart import UARTServer
-        uart = UARTServer()
-        uart.start_advertising()
-
-        # Wait for a connection.
-        while not uart.connected:
-            pass
-
-        uart.write('abc')
     """
 
     NUS_SERVICE_UUID = UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
+    """Nordic UART Service UUID"""
     NUS_RX_CHAR_UUID = UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")
+    """Nordic UART Service RX Characteristic UUID"""
     NUS_TX_CHAR_UUID = UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E")
+    """Nordic UART Service TX Characteristic UUID"""
 
-    def __init__(self, *, timeout=1.0, buffer_size=64, name=None):
-        self._nus_tx_char = Characteristic(self.NUS_TX_CHAR_UUID, notify=True)
-        self._nus_rx_char = Characteristic(self.NUS_RX_CHAR_UUID,
-                                           write=True, write_no_response=True)
-
-        nus_uart_service = Service(self.NUS_SERVICE_UUID, (self._nus_tx_char, self._nus_rx_char))
-
-        self._periph = Peripheral((nus_uart_service,), name=name)
-        self._rx_buffer = CharacteristicBuffer(self._nus_rx_char,
-                                               timeout=timeout, buffer_size=buffer_size)
-        self._advertisement = ServerAdvertisement(self._periph)
-
-    def start_advertising(self):
-        """Start advertising the service. When a client connects, advertising will stop.
-        When the client disconnects, restart advertising by calling ``start_advertising()`` again.
-        """
-        self._periph.start_advertising(data=self._advertisement.advertising_data_bytes,
-                                       scan_response=self._advertisement.scan_response_bytes)
-
-    def stop_advertising(self):
-        """Stop advertising the service."""
-        self._periph.stop_advertising()
-
-    @property
-    def connected(self):
-        """True if someone connected to the server."""
-        return self._periph.connected
+    def __init__(self, *, read_characteristic, write_characteristic, timeout=5.0, buffer_size=64):
+        self._read_char = read_characteristic
+        self._write_char = write_characteristic
+        self._read_buffer = CharacteristicBuffer(self._read_char,
+                                                 timeout=timeout, buffer_size=buffer_size)
 
     def read(self, nbytes=None):
         """
@@ -95,7 +69,7 @@ def read(self, nbytes=None):
         :return: Data read
         :rtype: bytes or None
         """
-        return self._rx_buffer.read(nbytes)
+        return self._read_buffer.read(nbytes)
 
     def readinto(self, buf, nbytes=None):
         """
@@ -105,7 +79,7 @@ def readinto(self, buf, nbytes=None):
         :return: number of bytes read and stored into ``buf``
         :rtype: int or None (on a non-blocking error)
         """
-        return self._rx_buffer.readinto(buf, nbytes)
+        return self._read_buffer.readinto(buf, nbytes)
 
     def readline(self):
         """
@@ -114,21 +88,21 @@ def readline(self):
         :return: the line read
         :rtype: int or None
         """
-        return self._rx_buffer.readline()
+        return self._read_buffer.readline()
 
     @property
     def in_waiting(self):
         """The number of bytes in the input buffer, available to be read."""
-        return self._rx_buffer.in_waiting
+        return self._read_buffer.in_waiting
 
     def reset_input_buffer(self):
         """Discard any unread characters in the input buffer."""
-        self._rx_buffer.reset_input_buffer()
+        self._read_buffer.reset_input_buffer()
 
     def write(self, buf):
         """Write a buffer of bytes."""
         # We can only write 20 bytes at a time.
         offset = 0
         while offset < len(buf):
-            self._nus_tx_char.value = buf[offset:offset+20]
+            self._write_char.value = buf[offset:offset+20]
             offset += 20

adafruit_ble/uart_client.py

@@ -0,0 +1,84 @@
+ # The MIT License (MIT)
+#
+# Copyright (c) 2019 Dan Halbert for Adafruit Industries
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+"""
+`adafruit_ble.uart_client`
+====================================================
+
+UART-style communication by a Central as a GATT Client
+
+* Author(s): Dan Halbert for Adafruit Industries
+
+"""
+from bleio import Characteristic, Central
+from .uart import UART
+
+class UARTClient(UART):
+    """
+    Provide UART-like functionality via the Nordic NUS service.
+
+    :param int timeout:  the timeout in seconds to wait
+      for the first character and between subsequent characters.
+    :param int buffer_size: buffer up to this many bytes.
+      If more bytes are received, older bytes will be discarded.
+    :param str name: Name to advertise for server. If None, use default Peripheral name.
+
+    Example::
+
+        from adafruit_ble.uart_client import UARTClient
+        from adafruit_ble.scanner import Scanner, ScanEntry
+
+        scanner = Scanner()
+        uarts = ScanEntry.with_service_uuid(scanner.scan_unique(3), UART.NUS_SERVICE_UUID)
+        if not uarts:
+            raise ValueError("No UART for connection")
+
+        uart_client = UARTClient()
+        uart_client.connect(uarts[0].address, 5, service_uuids=(UART.NUS_SERVICE_UUID,))
+
+        uart_client.write('abc')
+    """
+
+    def __init__(self, *, timeout=5.0, buffer_size=64):
+        # Since we're remote we receive on tx and send on rx. The names
+        # are from the point of view of the server.
+        super().__init__(read_characteristic=Characteristic(UART.NUS_TX_CHAR_UUID),
+                         write_characteristic=Characteristic(UART.NUS_RX_CHAR_UUID),
+                         timeout=timeout, buffer_size=buffer_size)
+
+        self._central = Central()
+
+    @property
+    def connected(self):
+        """True if we are connected to a peripheral."""
+        return self._central.connected
+
+    def connect(self, address, timeout):
+        """Try to connect to the peripheral at the given address.
+
+        :param bleio.Address address: The address of the peripheral to connect to
+        :param float/int timeout: Try to connect for timeout seconds.
+        """
+        self._central.connect(address, timeout, service_uuids=(UART.NUS_SERVICE_UUID,))
+
+    def disconnect(self):
+        """Disconnect from the peripheral."""
+        self._central.disconnect()

adafruit_ble/uart_server.py

@@ -0,0 +1,83 @@
+# The MIT License (MIT)
+#
+# Copyright (c) 2018 Dan Halbert for Adafruit Industries
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+"""
+`adafruit_ble.uart_server`
+====================================================
+
+UART-style communication: Peripheral acting as a GATT Server.
+
+* Author(s): Dan Halbert for Adafruit Industries
+
+"""
+from bleio import Characteristic, Service, Peripheral
+from .advertising import ServerAdvertisement
+from .uart import UART
+
+class UARTServer(UART):
+    """
+    Provide UART-like functionality via the Nordic NUS service.
+
+    :param int timeout:  the timeout in seconds to wait
+      for the first character and between subsequent characters.
+    :param int buffer_size: buffer up to this many bytes.
+      If more bytes are received, older bytes will be discarded.
+    :param str name: Name to advertise for server. If None, use default Peripheral name.
+
+    Example::
+
+        from adafruit_ble.uart_server import UARTServer
+        uart = UARTServer()
+        uart.start_advertising()
+
+        # Wait for a connection.
+        while not uart.connected:
+            pass
+
+        uart.write('abc')
+    """
+
+    def __init__(self, *, timeout=1.0, buffer_size=64, name=None):
+        read_char = Characteristic(UART.NUS_RX_CHAR_UUID, write=True, write_no_response=True)
+        write_char = Characteristic(UART.NUS_TX_CHAR_UUID, notify=True)
+        super().__init__(read_characteristic=read_char, write_characteristic=write_char,
+                         timeout=timeout, buffer_size=buffer_size)
+
+        nus_uart_service = Service(UART.NUS_SERVICE_UUID, (read_char, write_char))
+
+        self._periph = Peripheral((nus_uart_service,), name=name)
+        self._advertisement = ServerAdvertisement(self._periph)
+
+    def start_advertising(self):
+        """Start advertising the service. When a client connects, advertising will stop.
+        When the client disconnects, restart advertising by calling ``start_advertising()`` again.
+        """
+        self._periph.start_advertising(self._advertisement.advertising_data_bytes,
+                                       scan_response=self._advertisement.scan_response_bytes)
+
+    def stop_advertising(self):
+        """Stop advertising the service."""
+        self._periph.stop_advertising()
+
+    @property
+    def connected(self):
+        """True if someone connected to the server."""
+        return self._periph.connected