ecki/Robotics-at-Home-with-Raspberry-Pi-Pico
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Do this without the asyncio with.

Browse Source
This commit is contained in:
Danny Staple 2022-11-03 16:34:28 +00:00
parent d03c48bf8d
commit 0a8a86960d
2 changed files with 34 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
ch-13/1-modelling-space/computer/display_arena_from_robot.py
View File

@ -9,7 +9,7 @@ from robot_ble_connection import BleConnection
class RobotDisplay: class RobotDisplay:
def __init__(self): def __init__(self):
self.fig, self.ax = plt.subplots() self.fig, self.ax = plt.subplots()
self.ble_connection = BleConnection(self.handle_data, self.connected) self.ble_connection = BleConnection(self.handle_data)
self.arena = None self.arena = None
self.display_closed = False self.display_closed = False
@ -24,7 +24,7 @@ class RobotDisplay:
line_part = data.decode("utf-8") line_part = data.decode("utf-8")
self.line += line_part self.line += line_part
if not self.line.endswith("\n"): if not self.line.endswith("\n"):
return return
print(f"Received data: {self.line}") print(f"Received data: {self.line}")
data = json.loads(self.line) data = json.loads(self.line)
self.line = "" self.line = ""
@ -35,29 +35,32 @@ class RobotDisplay:
self.arena = arena self.arena = arena
self.ax.clear() self.ax.clear()
for line in arena["arena"]: for line in arena["arena"]:
self.ax.plot([line[0][0], line[1][0]], self.ax.plot(
[line[0][1], line[1][1]], [line[0][0], line[1][0]], [line[0][1], line[1][1]], color="black"
color='black') )
for line in arena["target_zone"]: for line in arena["target_zone"]:
self.ax.plot([line[0][0], line[1][0]], self.ax.plot(
[line[0][1], line[1][1]], [line[0][0], line[1][0]], [line[0][1], line[1][1]], color="red"
color="red") )
def connected(self): async def main(self):
request = json.dumps({"command": "arena"}).encode()
print(f"Sending request for arena: {request}")
self.ble_connection.send_uart_data(request)
async def main(self):
plt.ion() plt.ion()
asyncio.create_task(self.ble_connection.connect()) try:
while not self.display_closed: await self.ble_connection.connect()
plt.pause(0.05) request = json.dumps({"type": "arena"}).encode()
plt.draw() print(f"Sending request for arena: {request}")
await asyncio.sleep(0.01) self.ble_connection.send_uart_data(request)
plt.show() while not self.display_closed:
plt.pause(0.05)
plt.draw()
await asyncio.sleep(0.01)
plt.show()
finally:
await self.ble_connection.close()
robot_display = RobotDisplay() robot_display = RobotDisplay()
asyncio.run(robot_display.main()) asyncio.run(robot_display.main())

22
ch-13/1-modelling-space/computer/robot_ble_connection.py
View File

@ -11,11 +11,9 @@ class BleConnection:
adafruit_tx_uuid = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E" adafruit_tx_uuid = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
ble_name = "Adafruit Bluefruit LE" ble_name = "Adafruit Bluefruit LE"
def __init__(self, receive_handler: typing.Callable[[bytes], None], def __init__(self, receive_handler: typing.Callable[[bytes], None]):
connected_handler: typing.Callable[[], None]):
self.ble_client : bleak.BleakClient = None self.ble_client : bleak.BleakClient = None
self.receive_handler = receive_handler self.receive_handler = receive_handler
self.connected_handler = connected_handler
def _uart_handler(self, _, data: bytes): def _uart_handler(self, _, data: bytes):
self.receive_handler(data) self.receive_handler(data)
@ -27,16 +25,16 @@ class BleConnection:
print([device.name for device in devices]) print([device.name for device in devices])
ble_device_info = [device for device in devices if device.name==self.ble_name][0] ble_device_info = [device for device in devices if device.name==self.ble_name][0]
print("Connecting to {}...".format(ble_device_info.name)) print("Connecting to {}...".format(ble_device_info.name))
async with bleak.BleakClient(ble_device_info.address) as ble_client: self.ble_client = bleak.BleakClient(ble_device_info.address)
self.ble_client = ble_client await self.ble_client.connect()
self.connected_handler() print("Connected to {}".format(ble_device_info.name))
print("Connected to {}".format(ble_device_info.name)) self.notify_task = asyncio.create_task(
asyncio.create_task( self.ble_client.start_notify(self.adafruit_rx_uuid, self._uart_handler)
self.ble_client.start_notify(self.adafruit_rx_uuid, self._uart_handler) )
)
while True:
await asyncio.sleep(1)
async def close(self):
await self.ble_client.disconnect()
def send_uart_data(self, data: bytes): def send_uart_data(self, data: bytes):
if self.ble_client: if self.ble_client:
asyncio.create_task( asyncio.create_task(