Add Pico UART client and tests
This commit is contained in:
@@ -0,0 +1,40 @@
|
||||
import json
|
||||
|
||||
|
||||
class UartClient:
|
||||
def __init__(self, uart):
|
||||
self.uart = uart
|
||||
self.buffer = ""
|
||||
|
||||
def send_message(self, message):
|
||||
payload = json.dumps(message, separators=(",", ":")) + "\n"
|
||||
self.uart.write(payload.encode("utf-8"))
|
||||
|
||||
def read_available_messages(self):
|
||||
messages = []
|
||||
|
||||
while self.uart.any():
|
||||
chunk = self.uart.read()
|
||||
|
||||
if not chunk:
|
||||
break
|
||||
|
||||
self.buffer += chunk.decode("utf-8")
|
||||
|
||||
while "\n" in self.buffer:
|
||||
line, self.buffer = self.buffer.split("\n", 1)
|
||||
line = line.strip()
|
||||
|
||||
if not line:
|
||||
continue
|
||||
|
||||
try:
|
||||
messages.append(json.loads(line))
|
||||
except ValueError:
|
||||
messages.append({
|
||||
"type": "error",
|
||||
"message": "invalid_json",
|
||||
"raw": line,
|
||||
})
|
||||
|
||||
return messages
|
||||
|
||||
Reference in New Issue
Block a user