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

Make a little robust with lessons learned later

Browse Source
This commit is contained in:
Danny Staple 2022-11-13 18:46:59 +00:00
parent e444040b86
commit 1d31bee775
2 changed files with 31 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
ch-13/1-modelling-space/computer/display_from_robot.py
View File

@ -1,6 +1,5 @@
import asyncio import asyncio
import json import json
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
from robot_ble_connection import BleConnection from robot_ble_connection import BleConnection
@ -10,7 +9,7 @@ class RobotDisplay:
def __init__(self): def __init__(self):
self.ble_connection = BleConnection(self.handle_data) self.ble_connection = BleConnection(self.handle_data)
self.line = "" self.line = ""
self.arena = {}
self.display_closed = False self.display_closed = False
def handle_close(self, _): def handle_close(self, _):
@ -18,41 +17,43 @@ class RobotDisplay:
def handle_data(self, data): def handle_data(self, data):
self.line += data.decode("utf-8") self.line += data.decode("utf-8")
if not self.line.endswith("\n"): while "\n" in self.line:
return line, self.line = self.line.split("\n", 1)
print(f"Received data: {self.line}") print(f"Received data: ```{line}```")
message = json.loads(self.line) try:
self.line = "" message = json.loads(line)
if "arena" in message: except ValueError:
self.update_arena(message) print("Error parsing JSON")
return
if "arena" in message:
self.arena = message
def update_arena(self, arena): def draw(self):
plt.gca().clear() plt.gca().clear()
for line in arena["arena"]: if self.arena:
plt.gca().plot( for line in self.arena["arena"]:
[line[0][0], line[1][0]], [line[0][1], line[1][1]], color="black" plt.gca().plot(
) [line[0][0], line[1][0]], [line[0][1], line[1][1]], color="black"
for line in arena["target_zone"]: )
plt.gca().plot( for line in self.arena["target_zone"]:
[line[0][0], line[1][0]], [line[0][1], line[1][1]], color="red" plt.gca().plot(
) [line[0][0], line[1][0]], [line[0][1], line[1][1]], color="red"
)
async def main(self): async def main(self):
plt.ion() plt.ion()
await self.ble_connection.connect()
try: try:
await self.ble_connection.connect()
request = json.dumps({"command": "arena"}).encode() request = json.dumps({"command": "arena"}).encode()
print(f"Sending request for arena: {request}") print(f"Sending request for arena: {request}")
await self.ble_connection.send_uart_data(request) await self.ble_connection.send_uart_data(request)
plt.gcf().canvas.mpl_connect("close_event", self.handle_close) plt.gcf().canvas.mpl_connect("close_event", self.handle_close)
while not self.display_closed: while not self.display_closed:
plt.pause(0.05) self.draw()
plt.draw() plt.draw()
plt.pause(0.05)
await asyncio.sleep(0.01) await asyncio.sleep(0.01)
plt.show()
finally: finally:
await self.ble_connection.close() await self.ble_connection.close()

11
ch-13/1-modelling-space/robot/code.py
View File

@ -9,10 +9,12 @@ async def command_handler():
while True: while True:
if robot.uart.in_waiting: if robot.uart.in_waiting:
print("Receiving data...") print("Receiving data...")
data = robot.uart.readline().decode() try:
print(f"Received data: {data}") data = robot.uart.readline().decode()
request = json.loads(data) request = json.loads(data)
print(f"Received command: {request}") except (UnicodeError, ValueError):
print("Invalid data")
continue
# {"command": "arena"} # {"command": "arena"}
if request["command"] == "arena": if request["command"] == "arena":
response = { response = {
@ -20,5 +22,6 @@ async def command_handler():
"target_zone": arena.target_zone, "target_zone": arena.target_zone,
} }
robot.uart.write((json.dumps(response)+"\n").encode()) robot.uart.write((json.dumps(response)+"\n").encode())
await asyncio.sleep(0.1)
asyncio.run(command_handler()) asyncio.run(command_handler())