Chapter 13 modelling space initial.

ch-13/1-modelling-space/computer/display_arena_from_robot.py

@@ -0,0 +1,68 @@
+import asyncio
+import json
+
+import matplotlib.pyplot as plt
+
+from robot_ble_connection import BleConnection
+
+
+class RobotDisplay:
+    def __init__(self, ble_connection: BleConnection):
+        self.fig, self.ax = plt.subplots()
+        self.ble_connection = ble_connection
+        self.arena = None
+        self.display_closed = False
+        # handle closed event
+        self.fig.canvas.mpl_connect("close_event", self.handle_close)
+        ble_connection.receive_handler = self.handle_data
+        ble_connection.connected_handler = self.connected
+        self.line = ""
+
+    def handle_close(self, _):
+        self.display_closed = True
+
+    def handle_data(self, data):
+        line_part = data.decode("utf-8")
+        self.line += line_part
+        if not self.line.endswith("\n"):
+          return
+        print(f"Received data: {self.line}")
+        data = json.loads(self.line)
+        self.line = ""
+        if "arena" in data:
+            self.update(data)
+
+    def update(self, arena):
+        self.arena = arena
+        self.ax.clear()
+        for line in arena["arena"]:
+            self.ax.plot([line[0][0], line[1][0]],
+                         [line[0][1], line[1][1]],
+                         color='black')
+        for line in arena["target_zone"]:
+            self.ax.plot([line[0][0], line[1][0]],
+                         [line[0][1], line[1][1]],
+                         color="red")
+
+    def connected(self):
+        request = json.dumps({"command": "arena"}).encode()
+        print(f"Sending request for arena: {request}")
+        self.ble_connection.send_uart_data(request)
+
+
+async def main():
+    
+    plt.ion()
+
+    ble_connection = BleConnection()
+    robot_display = RobotDisplay(ble_connection)
+    asyncio.create_task(ble_connection.connect())
+    while not robot_display.display_closed:
+        plt.pause(0.05)
+        plt.draw()
+        await asyncio.sleep(0.01)
+    
+    plt.show()
+
+
+asyncio.run(main())

ch-13/1-modelling-space/computer/find_devices.py

@@ -0,0 +1,8 @@
+import asyncio
+from robot_ble_connection import BleConnection
+
+async def run():
+    ble_connection = BleConnection()
+    await ble_connection.connect()
+
+asyncio.run(run())

ch-13/1-modelling-space/computer/pyproject.toml

@@ -0,0 +1,15 @@
+[tool.poetry]
+name = "modelling-space"
+version = "0.1.0"
+description = ""
+authors = ["Danny Staple <danny@orionrobots.co.uk>"]
+readme = "README.md"
+packages = [{include = "modelling_space"}]
+
+[tool.poetry.dependencies]
+python = "^3.9"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

ch-13/1-modelling-space/computer/requirements.txt

@@ -0,0 +1,3 @@
+matplotlib==3.6.1
+numpy==1.23.4
+bleak==0.19.0

ch-13/1-modelling-space/computer/robot_ble_connection.py

@@ -0,0 +1,49 @@
+import asyncio
+import typing
+
+import bleak
+
+
+class BleConnection:
+    # See https://learn.adafruit.com/introducing-adafruit-ble-bluetooth-low-energy-friend/uart-service
+    ble_uuid = "6E400001-B5A3-F393-E0A9-E50E24DCCA9E"
+    adafruit_rx_uuid = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"
+    adafruit_tx_uuid = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
+    ble_name = "Adafruit Bluefruit LE"
+
+    def __init__(self, receive_handler: typing.Callable[[bytes], None] = None, 
+            connected_handler: typing.Callable[[], None] = None):
+        self.ble_client : bleak.BleakClient = None
+        # receive_handler(bytes) -> None
+        self.receive_handler = receive_handler
+        # connected_handler() -> None
+        self.connected_handler = connected_handler
+
+    def _uart_handler(self, _, data: bytes):
+        # hmm - not receiving the whole message, only parts of it...
+        if self.receive_handler:
+          self.receive_handler(data)
+
+    async def connect(self):
+        print("Scanning for devices...")
+        devices = await bleak.BleakScanner.discover(service_uuids=[self.ble_uuid])
+        print("Found {} devices".format(len(devices)))
+        print([device.name for device in devices])
+        ble_device_info = [device for device in devices if device.name==self.ble_name][0]
+        print("Connecting to {}...".format(ble_device_info.name))
+        async with bleak.BleakClient(ble_device_info.address) as ble_client:
+            self.ble_client = ble_client
+            if self.connected_handler: 
+                self.connected_handler()
+            print("Connected to {}".format(ble_device_info.name))
+            asyncio.create_task(
+                self.ble_client.start_notify(self.adafruit_rx_uuid, self._uart_handler)
+            )
+            while True:
+                await asyncio.sleep(1)
+
+    def send_uart_data(self, data: bytes):
+        if self.ble_client:
+            asyncio.create_task(
+                self.ble_client.write_gatt_char(self.adafruit_tx_uuid, data)
+            )