christian
3905647b7f
Firmware, Bluetooth-Brücke und Agent-Skill für den KOSMOS-Proxi: einmal flashen, danach hört der Roboter über Bluetooth LE auf Textbefehle. Was die Befehle schickt, ist ihm egal — ein Skript, ein curl, oder ein KI-Agent, der aus "mach mal was Lustiges" eine Choreografie baut. firmware/ MakeCode-Projekt (TypeScript) und fertige .hex für V1 und V2 bridge/ BLE-Brücke und HTTP-API, einzige Abhängigkeit ist bleak agent/ Skill-Definition: Alltagssprache -> Befehle tools/ Quelltext aus .hex extrahieren, Musik-Tabellen prüfen docs/ Protokoll, Einrichtung, Hardware, Lizenzen Zwei Dinge, die sonst nirgends stehen und in docs/HARDWARE.md dokumentiert sind: die Pin-Belegung des Roboters ist nicht öffentlich, sie wurde aus den mitgelieferten .hex-Dateien rekonstruiert (tools/hex_source.py holt sie heraus). Und der Ton braucht zwingend eine ganzzahlige Frequenz sowie Pin P0 — sonst kracht der Lautsprecher, der Aufruf hängt und die Bluetooth- Verbindung stirbt. Am echten Roboter durchgetestet: Verbindung, Sensoren, Display, 41 Sekunden Musik am Stück, alle Bewegungsrichtungen, Musik und Fahren gleichzeitig, Tanzfiguren, Not-Stopp mitten in der Bewegung, Autostart nach Neustart. MIT-Lizenz. Die Hardware-Extension stammt von kaku111 (TobbieII) und die Notendaten aus pxt-microbit, beide ebenfalls MIT — siehe THIRD-PARTY-NOTICES.
413 lines
15 KiB
Python
Executable File
413 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
proxi_bridge.py — Bluetooth-LE-Brücke zum Proxi-Roboter.
|
|
|
|
Verbindet sich per BLE mit dem micro:bit im Roboter und schiebt Textbefehle
|
|
über dessen UART-Service. Läuft auf allem mit Linux + Bluetooth (Pi Zero W,
|
|
Laptop, NUC). Die einzige Abhängigkeit ist `bleak`, der Rest ist Stdlib.
|
|
|
|
pip install bleak
|
|
|
|
python3 proxi_bridge.py --scan Geräte suchen
|
|
python3 proxi_bridge.py --ping Lebenszeichen
|
|
python3 proxi_bridge.py "MOVE:FWD:1500" einzelner Befehl
|
|
python3 proxi_bridge.py -i interaktiv
|
|
python3 proxi_bridge.py --script demo.txt Skript abspielen
|
|
python3 proxi_bridge.py --serve 8080 HTTP-API für die KI
|
|
|
|
Befehlsreferenz: ../docs/PROTOCOL.md
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
try:
|
|
from bleak import BleakClient, BleakScanner
|
|
except ImportError:
|
|
sys.exit("bleak fehlt. pip install bleak")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# micro:bit BLE UART
|
|
#
|
|
# Aus codal-microbit-v2/source/bluetooth/MicroBitUARTService.cpp:
|
|
# mbbs_cIdxTX = 0x0002 -> propINDICATE micro:bit -> Host
|
|
# mbbs_cIdxRX = 0x0003 -> propWRITE | WRITE_WITHOUT Host -> micro:bit
|
|
#
|
|
# Die Namen sind aus Sicht des micro:bit vergeben. Wer sie mit den Nordic-UART-
|
|
# Konventionen verwechselt, schreibt auf das Indicate-Handle und wartet auf
|
|
# Benachrichtigungen vom Write-Handle — es passiert dann schlicht nichts.
|
|
#
|
|
# MICROBIT_UART_S_ATTRSIZE ist 20 Byte, längere Zeilen müssen gestückelt werden.
|
|
# --------------------------------------------------------------------------
|
|
|
|
UART_SERVICE = "6e400001-b5a3-f393-e0a9-e50e24dcca9e"
|
|
UART_FROM_MICROBIT = "6e400002-b5a3-f393-e0a9-e50e24dcca9e" # indicate
|
|
UART_TO_MICROBIT = "6e400003-b5a3-f393-e0a9-e50e24dcca9e" # write
|
|
ATTR_SIZE = 20
|
|
|
|
#: Zeilen mit diesen Präfixen sind unaufgefordert und keine Antwort auf einen Befehl.
|
|
UNSOLICITED = ("EVENT:", "DONE:", "READY:")
|
|
|
|
|
|
class ProxiBLE:
|
|
"""Asynchrone BLE-Verbindung zum Roboter."""
|
|
|
|
def __init__(self, address=None, timeout=15.0):
|
|
self.address = address
|
|
#: vom Aufrufer vorgegebene Adresse; None heißt "jedes Mal neu suchen"
|
|
self.fixed_address = address
|
|
self.timeout = timeout
|
|
self.client = None
|
|
self._rx = bytearray()
|
|
self._responses = None
|
|
self.events = []
|
|
self.on_event = None
|
|
|
|
# -- Verbindung ---------------------------------------------------------
|
|
|
|
@staticmethod
|
|
async def discover(timeout=8.0):
|
|
"""Alle micro:bits in Reichweite."""
|
|
found = []
|
|
for dev in await BleakScanner.discover(timeout=timeout):
|
|
name = dev.name or ""
|
|
if "micro:bit" in name.lower() or "microbit" in name.lower():
|
|
found.append((name, dev.address))
|
|
return found
|
|
|
|
async def connect(self):
|
|
if not self.address:
|
|
found = await self.discover()
|
|
if not found:
|
|
raise RuntimeError(
|
|
"kein micro:bit gefunden — ist Proxi an und nicht schon "
|
|
"mit einem anderen Gerät verbunden?"
|
|
)
|
|
name, self.address = found[0]
|
|
print(f"gefunden: {name} — {self.address}", file=sys.stderr)
|
|
|
|
self._responses = asyncio.Queue()
|
|
self.client = BleakClient(self.address, timeout=self.timeout)
|
|
await self.client.connect()
|
|
|
|
# start_notify deckt auch Indications ab — der micro:bit nutzt Indicate.
|
|
await self.client.start_notify(UART_FROM_MICROBIT, self._on_data)
|
|
return self.address
|
|
|
|
async def disconnect(self):
|
|
if self.client and self.client.is_connected:
|
|
try:
|
|
await self.send("STOP", wait=False)
|
|
await asyncio.sleep(0.2)
|
|
except Exception:
|
|
pass
|
|
await self.client.disconnect()
|
|
|
|
@property
|
|
def connected(self):
|
|
return bool(self.client and self.client.is_connected)
|
|
|
|
# -- Datenstrom ---------------------------------------------------------
|
|
|
|
def _on_data(self, _handle, data):
|
|
"""Bytes einsammeln und in Zeilen zerlegen."""
|
|
self._rx.extend(data)
|
|
while b"\n" in self._rx:
|
|
head, _, rest = bytes(self._rx).partition(b"\n")
|
|
self._rx = bytearray(rest)
|
|
line = head.decode("utf-8", "replace").strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith(UNSOLICITED):
|
|
self.events.append(line)
|
|
if self.on_event:
|
|
self.on_event(line)
|
|
else:
|
|
self._responses.put_nowait(line)
|
|
|
|
# -- Senden -------------------------------------------------------------
|
|
|
|
async def send(self, command, wait=True, timeout=10.0):
|
|
if not self.connected:
|
|
return "ERR:not_connected"
|
|
|
|
while not self._responses.empty():
|
|
self._responses.get_nowait()
|
|
|
|
payload = (command.strip() + "\n").encode("utf-8")
|
|
for i in range(0, len(payload), ATTR_SIZE):
|
|
await self.client.write_gatt_char(
|
|
UART_TO_MICROBIT, payload[i:i + ATTR_SIZE], response=False
|
|
)
|
|
|
|
if not wait:
|
|
return None
|
|
try:
|
|
return await asyncio.wait_for(self._responses.get(), timeout=timeout)
|
|
except asyncio.TimeoutError:
|
|
return "ERR:timeout"
|
|
|
|
async def send_many(self, commands, gap=0.15):
|
|
"""Befehle der Reihe nach senden.
|
|
|
|
`WAIT:<ms>` ist kein Roboter-Befehl, sondern eine Pause hier in der
|
|
Brücke. Bewegungsbefehle kommen sofort mit OK zurück und laufen dann
|
|
weiter — ohne WAIT dazwischen würde der nächste Befehl die laufende
|
|
Bewegung sofort überschreiben.
|
|
"""
|
|
results = []
|
|
for raw in commands:
|
|
cmd = raw.strip()
|
|
if not cmd or cmd.startswith("#"):
|
|
continue
|
|
|
|
head, sep, rest = cmd.partition(":")
|
|
if head.upper() == "WAIT":
|
|
try:
|
|
ms = int(rest) if sep else 0
|
|
except ValueError:
|
|
results.append((cmd, "ERR:WAIT:bad_ms"))
|
|
continue
|
|
await asyncio.sleep(max(0, min(ms, 60000)) / 1000)
|
|
results.append((cmd, f"OK:WAIT:{ms}"))
|
|
continue
|
|
|
|
results.append((cmd, await self.send(cmd)))
|
|
await asyncio.sleep(gap)
|
|
return results
|
|
|
|
|
|
class Bridge:
|
|
"""Synchrone Hülle: hält die Verbindung in einem eigenen Event-Loop-Thread."""
|
|
|
|
def __init__(self, address=None):
|
|
self.ble = ProxiBLE(address)
|
|
self._loop = asyncio.new_event_loop()
|
|
self._thread = threading.Thread(target=self._run_loop, daemon=True)
|
|
self._thread.start()
|
|
|
|
def _run_loop(self):
|
|
asyncio.set_event_loop(self._loop)
|
|
self._loop.run_forever()
|
|
|
|
def _call(self, coro, timeout=60):
|
|
return asyncio.run_coroutine_threadsafe(coro, self._loop).result(timeout)
|
|
|
|
def connect(self):
|
|
return self._call(self.ble.connect(), timeout=90)
|
|
|
|
def start_autoconnect(self, interval=10.0):
|
|
"""Im Hintergrund verbinden und verbunden halten.
|
|
|
|
Für den Dauerbetrieb als Dienst: der Roboter ist nicht immer an. Ohne
|
|
das beendet sich die Brücke beim Startversuch, systemd startet sie neu,
|
|
und der HTTP-Server kommt nie hoch — die API wäre also genau dann tot,
|
|
wenn jemand den Roboter einschaltet.
|
|
"""
|
|
asyncio.run_coroutine_threadsafe(self._keep_connected(interval), self._loop)
|
|
|
|
async def _keep_connected(self, interval):
|
|
while True:
|
|
if not self.ble.connected:
|
|
try:
|
|
await self.ble.connect()
|
|
print(f"verbunden mit {self.ble.address}", file=sys.stderr)
|
|
except Exception as exc:
|
|
# Roboter aus oder außer Reichweite — gleich nochmal probieren
|
|
print(f"warte auf Proxi ({exc})", file=sys.stderr)
|
|
self.ble.address = self.ble.fixed_address
|
|
await asyncio.sleep(interval)
|
|
|
|
def send(self, command, wait=True):
|
|
return self._call(self.ble.send(command, wait=wait))
|
|
|
|
def send_many(self, commands):
|
|
return self._call(self.ble.send_many(commands), timeout=600)
|
|
|
|
def close(self):
|
|
try:
|
|
self._call(self.ble.disconnect(), timeout=20)
|
|
finally:
|
|
self._loop.call_soon_threadsafe(self._loop.stop)
|
|
|
|
@property
|
|
def connected(self):
|
|
return self.ble.connected
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# HTTP-API
|
|
# --------------------------------------------------------------------------
|
|
|
|
def make_handler(bridge):
|
|
class Handler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, fmt, *args):
|
|
print(f"[http] {fmt % args}", file=sys.stderr)
|
|
|
|
def _send(self, code, payload):
|
|
body = json.dumps(payload).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_OPTIONS(self):
|
|
self._send(204, {})
|
|
|
|
def do_GET(self):
|
|
if self.path.rstrip("/") in ("", "/status"):
|
|
self._send(200, {
|
|
"connected": bridge.connected,
|
|
"address": bridge.ble.address,
|
|
"events": bridge.ble.events[-20:],
|
|
})
|
|
else:
|
|
self._send(404, {"error": "not found"})
|
|
|
|
def do_POST(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
try:
|
|
data = json.loads(self.rfile.read(length) or b"{}")
|
|
except json.JSONDecodeError:
|
|
return self._send(400, {"error": "kein gültiges JSON"})
|
|
|
|
path = self.path.rstrip("/")
|
|
if path == "/command":
|
|
cmd = data.get("command")
|
|
if not isinstance(cmd, str):
|
|
return self._send(400, {"error": "Feld 'command' fehlt"})
|
|
response = bridge.send(cmd)
|
|
self._send(200, {
|
|
"command": cmd,
|
|
"response": response,
|
|
"ok": not str(response).startswith("ERR:"),
|
|
})
|
|
elif path == "/commands":
|
|
cmds = data.get("commands")
|
|
if not isinstance(cmds, list):
|
|
return self._send(400, {"error": "Feld 'commands' fehlt"})
|
|
results = bridge.send_many([str(c) for c in cmds])
|
|
self._send(200, {"results": [
|
|
{"command": c, "response": r, "ok": not str(r).startswith("ERR:")}
|
|
for c, r in results
|
|
]})
|
|
else:
|
|
self._send(404, {"error": "not found"})
|
|
|
|
return Handler
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# CLI
|
|
# --------------------------------------------------------------------------
|
|
|
|
def run_scan():
|
|
found = asyncio.run(ProxiBLE.discover())
|
|
if not found:
|
|
print("kein micro:bit gefunden.")
|
|
return 1
|
|
for name, address in found:
|
|
print(f"{address} {name}")
|
|
return 0
|
|
|
|
|
|
def run_interactive(bridge):
|
|
print("Interaktiv — 'quit' beendet, leere Zeile wiederholt nichts.")
|
|
print("Befehle z.B.: MOVE:FWD:1500 | TURN:LEFT:600 | FACE:HAPPY | DANCE:3")
|
|
while True:
|
|
try:
|
|
line = input("proxi> ").strip()
|
|
except (EOFError, KeyboardInterrupt):
|
|
print()
|
|
return 0
|
|
if not line:
|
|
continue
|
|
if line.lower() in ("quit", "exit", "q"):
|
|
return 0
|
|
print(" ", bridge.send(line))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(
|
|
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("command", nargs="*", help="Befehl(e), direkt gesendet")
|
|
ap.add_argument("--address", "-a", help="BLE-Adresse, überspringt die Suche")
|
|
ap.add_argument("--scan", action="store_true", help="micro:bits suchen und beenden")
|
|
ap.add_argument("--ping", action="store_true", help="PING senden")
|
|
ap.add_argument("--interactive", "-i", action="store_true", help="Eingabeschleife")
|
|
ap.add_argument("--script", help="Datei mit einem Befehl pro Zeile")
|
|
ap.add_argument("--serve", type=int, metavar="PORT", help="HTTP-API starten")
|
|
args = ap.parse_args()
|
|
|
|
if args.scan:
|
|
return run_scan()
|
|
|
|
address = args.address or os.environ.get("PROXI_ADDRESS")
|
|
bridge = Bridge(address)
|
|
bridge.ble.on_event = lambda line: print(f"[proxi] {line}", file=sys.stderr)
|
|
|
|
if args.serve:
|
|
# Erst den Server, dann die Verbindung: die API muss auch dann
|
|
# erreichbar sein, wenn Proxi gerade aus ist.
|
|
server = ThreadingHTTPServer(("0.0.0.0", args.serve), make_handler(bridge))
|
|
bridge.start_autoconnect()
|
|
print(f"HTTP-API auf http://0.0.0.0:{args.serve} "
|
|
f"(POST /command, POST /commands, GET /status)", file=sys.stderr)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
bridge.close()
|
|
return 0
|
|
|
|
# Alle anderen Modi sind interaktiv — da ist sofortiges Scheitern richtig.
|
|
try:
|
|
bridge.connect()
|
|
except Exception as exc:
|
|
print(f"Verbindung fehlgeschlagen: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(f"verbunden mit {bridge.ble.address}", file=sys.stderr)
|
|
|
|
try:
|
|
if args.interactive:
|
|
return run_interactive(bridge)
|
|
|
|
if args.script:
|
|
with open(args.script) as fh:
|
|
for cmd, response in bridge.send_many(fh.readlines()):
|
|
print(f"{cmd:<28} {response}")
|
|
return 0
|
|
|
|
commands = list(args.command)
|
|
if args.ping:
|
|
commands.insert(0, "PING")
|
|
if not commands:
|
|
ap.error("kein Befehl angegeben — siehe --help")
|
|
|
|
failed = False
|
|
for cmd, response in bridge.send_many(commands):
|
|
print(f"{cmd:<28} {response}")
|
|
failed = failed or str(response).startswith("ERR:")
|
|
return 1 if failed else 0
|
|
|
|
except KeyboardInterrupt:
|
|
return 0
|
|
finally:
|
|
bridge.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|