# FlyBrain 昆蟲遙控站：在電腦上用 WiFi（UDP）掃描、控制好幾隻 StickC 六足昆蟲。
#
# 昆蟲那一端跑 bug_client.py（UIFlow 積木版）。這支程式有兩個部分：
# 1. BugNet 類別：掃描、記住每隻昆蟲、送動作指令。之後要接 FlyBrain 的果蠅神經系統時，
# 只要 import 這個類別，讓大腦決定動作後呼叫 go() 就好。
# 2. 一個本機網頁：每隻昆蟲一張卡片，按按鈕就能手動控制。
#
# 用法（電腦和昆蟲要連同一個 WiFi）：
# python firmware/m5stickc_hexapod/bug_server.py
# 然後用瀏覽器打開 http://127.0.0.1:8090 ，按「掃描」。
from __future__ import annotations
import argparse
import asyncio
import socket
import time
from aiohttp import web
BUG_PORT = 4210 # 昆蟲在這個 port 等訊息（和 bug_client.py 的 UDP_PORT 一樣）
LISTEN_PORT = 4211 # 電腦在這個 port 收昆蟲的回覆
# 昆蟲所在網段的廣播位址。我們的昆蟲連在 192.168.0.x，電腦在 192.168.1.x（不同網段），
# 要廣播到「昆蟲那一段」路由器才會轉送過去；網段不同時，在網頁上或用 --broadcast 改。
BUG_BROADCAST = "192.168.0.255"
MOTIONS = ["forward", "backward", "left", "right", "uturn"]
# 找出這台電腦在區域網路上的 IP。
# 做法：對外面開一個 UDP「連線」（UDP 的 connect 不會真的送出封包），
# 作業系統就會選好要走哪張網卡，再問它用的是哪個 IP。
# 輸出：例如 "192.168.1.10"；沒有網路時回傳 "127.0.0.1"。
def lan_ip() -> str:
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
probe.connect(("10.255.255.255", 1))
return probe.getsockname()[0]
except OSError:
return "127.0.0.1"
finally:
probe.close()
# 算出電腦自己那一段的「子網路廣播位址」：把 IP 最後一段換成 255，例如 192.168.1.102 -> 192.168.1.255。
# 為什麼不用 255.255.255.255：那是「全部廣播」，實測昆蟲收不到；子網路廣播會照路由送出去。
# 這裡假設常見的子網路遮罩 255.255.255.0（IP 前三段相同就在同一個網段）。
def subnet_broadcast(ip: str) -> str:
parts = ip.split(".")
return parts[0] + "." + parts[1] + "." + parts[2] + ".255"
class BugNet:
# 管理所有昆蟲的物件。
# 輸入：listen_port 電腦收回覆用的 port（0 表示讓系統隨便挑一個空的，測試時用）；
# bug_port 昆蟲等訊息的 port；my_ip 告訴昆蟲回覆要送到哪個 IP（預設自動找）；
# broadcasts 掃描時廣播到哪些位址（預設：昆蟲那一段 BUG_BROADCAST，加上電腦自己那一段）。
def __init__(self, listen_port: int = LISTEN_PORT, bug_port: int = BUG_PORT, my_ip: str | None = None,
broadcasts: list[str] | None = None):
self.bug_port = bug_port
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 允許送「廣播」：一次送給同一個 WiFi 裡的所有裝置，掃描時要用。
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
self.sock.bind(("0.0.0.0", listen_port))
self.listen_port = self.sock.getsockname()[1]
if my_ip is None:
my_ip = lan_ip()
self.my_ip = my_ip
if broadcasts is None:
broadcasts = [BUG_BROADCAST]
own = subnet_broadcast(my_ip)
if own not in broadcasts:
broadcasts.append(own)
self.broadcasts = broadcasts
# 找到的昆蟲：名字 -> {"address": (ip, port), "state": 最後的狀態, "last_seen": 時間}
self.bugs: dict[str, dict] = {}
# 把一行文字送到某個地址。
def _send_to(self, text: str, address: tuple[str, int]) -> None:
self.sock.sendto(text.encode(), address)
# 在 seconds 秒之內，把收到的回覆全部讀進來，更新每隻昆蟲的狀態。
# 輸出：這段時間收到的回覆清單，每一筆是 (地址, 文字)。
def collect_replies(self, seconds: float) -> list[tuple[tuple[str, int], str]]:
replies = []
deadline = time.monotonic() + seconds
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
break
self.sock.settimeout(remaining)
try:
data, address = self.sock.recvfrom(256)
except (socket.timeout, OSError):
break
text = data.decode(errors="replace").strip()
replies.append((address, text))
self._remember(address, text)
return replies
# 根據一則回覆更新昆蟲清單。
# HERE,名字,IP 是掃描的回覆：把這隻昆蟲加進清單（地址用封包實際的來源，最可靠）。
# 其他回覆（OK／PONG／ERR）就更新那隻昆蟲最後的狀態。
def _remember(self, address: tuple[str, int], text: str) -> None:
parts = text.split(",")
now = time.time()
if parts[0] == "HERE" and len(parts) >= 2:
name = parts[1]
self.bugs[name] = {"address": address, "state": "ready", "last_seen": now}
return
for info in self.bugs.values():
if info["address"] == address:
info["state"] = text
info["last_seen"] = now
# 掃描：送出 HELLO，等昆蟲回報。
# 輸入：seconds 等多久；targets 要問的地址清單，預設是 broadcasts 裡的每個廣播位址。
# 輸出：目前認識的所有昆蟲名字（排好順序）。
# UDP 可能掉封包，所以 HELLO 連送 3 次；同一隻回了好幾次也只記一筆。
def scan(self, seconds: float = 1.0, targets: list[tuple[str, int]] | None = None) -> list[str]:
if targets is None:
targets = []
for address in self.broadcasts:
targets.append((address, self.bug_port))
hello = "HELLO," + self.my_ip + "," + str(self.listen_port)
for attempt in range(3):
for target in targets:
self._send_to(hello, target)
self.collect_replies(seconds / 3)
return sorted(self.bugs)
# 把一行指令送給某一隻昆蟲；name 是 "all" 時送給每一隻。
# 輸出：實際送到的昆蟲名字清單（名字不認識時是空清單）。
def send(self, name: str, text: str) -> list[str]:
sent = []
for bug_name, info in self.bugs.items():
if name == "all" or name == bug_name:
self._send_to(text, info["address"])
sent.append(bug_name)
return sent
# 叫昆蟲做一個動作，例如 go("bug-3f2a", "forward")。
def go(self, name: str, motion: str) -> list[str]:
if motion not in MOTIONS:
raise ValueError("unknown motion: " + motion)
return self.send(name, "GO," + motion)
# 緊急停止。
def stop(self, name: str = "all") -> list[str]:
return self.send(name, "STOP")
# 問狀態，回覆會在 collect_replies() 時更新到 bugs 裡。
def ping(self, name: str = "all") -> list[str]:
return self.send(name, "PING")
# 給網頁用的清單：每隻昆蟲的名字、IP、狀態、幾秒前有消息。
def summary(self) -> list[dict]:
rows = []
now = time.time()
for name in sorted(self.bugs):
info = self.bugs[name]
rows.append({"name": name, "ip": info["address"][0], "state": info["state"],
"seen_s": round(now - info["last_seen"], 1)})
return rows
def close(self) -> None:
self.sock.close()
# ---------------- 本機網頁 ----------------
PAGE = """
FlyBrain 昆蟲遙控站
🪲 FlyBrain 昆蟲遙控站
電腦和昆蟲連同一個 WiFi。先按「掃描」，找到的每一隻昆蟲會出現一張卡片。
"""
# 檢查一段文字是不是 IPv4 位址，例如 "192.168.0.255"：四段、每段是 0～255 的數字。
def is_ipv4(text: str) -> bool:
parts = text.split(".")
if len(parts) != 4:
return False
for part in parts:
if not part.isdigit() or int(part) > 255:
return False
return True
# 建立網頁伺服器。BugNet 的方法會等網路回覆，所以丟到背景執行緒跑，網頁才不會卡住。
def make_web_app(net: BugNet) -> web.Application:
async def page(request: web.Request) -> web.Response:
return web.Response(text=PAGE, content_type="text/html", charset="utf-8")
async def bugs(request: web.Request) -> web.Response:
return web.json_response({"bugs": net.summary(), "broadcasts": net.broadcasts, "my_ip": net.my_ip})
async def scan(request: web.Request) -> web.Response:
body = await request.json()
# 網頁輸入框裡的廣播位址（逗號分開）；格式不對的就不收，全部不對時沿用原本的設定。
typed = []
for piece in str(body.get("broadcasts", "")).split(","):
piece = piece.strip()
if is_ipv4(piece):
typed.append(piece)
if typed:
net.broadcasts = typed
names = await asyncio.to_thread(net.scan)
return web.json_response({"bugs": net.summary(), "message": "找到 " + str(len(names)) + " 隻",
"broadcasts": net.broadcasts, "my_ip": net.my_ip})
async def go(request: web.Request) -> web.Response:
body = await request.json()
motion = str(body.get("motion", ""))
if motion not in MOTIONS:
return web.json_response({"bugs": net.summary(), "message": "不認識的動作"}, status=400)
sent = net.go(str(body.get("bug", "all")), motion)
await asyncio.to_thread(net.collect_replies, 0.3)
return web.json_response({"bugs": net.summary(), "message": "已送給 " + str(len(sent)) + " 隻"})
async def stop(request: web.Request) -> web.Response:
body = await request.json()
sent = net.stop(str(body.get("bug", "all")))
await asyncio.to_thread(net.collect_replies, 0.3)
return web.json_response({"bugs": net.summary(), "message": "停止：" + str(len(sent)) + " 隻"})
app = web.Application()
app.router.add_get("/", page)
app.router.add_get("/api/bugs", bugs)
app.router.add_post("/api/bugs", bugs)
app.router.add_post("/api/scan", scan)
app.router.add_post("/api/go", go)
app.router.add_post("/api/stop", stop)
return app
def main() -> None:
parser = argparse.ArgumentParser(description="FlyBrain 昆蟲遙控站")
parser.add_argument("--port", type=int, default=8090, help="網頁的 port")
parser.add_argument("--broadcast", action="append", default=None,
help="掃描的廣播位址，可以給好幾次，例如 --broadcast 192.168.0.255")
args = parser.parse_args()
net = BugNet(broadcasts=args.broadcast)
print("電腦 IP：" + net.my_ip + "，收回覆的 port：" + str(net.listen_port))
print("掃描廣播到：" + ", ".join(net.broadcasts))
print("用瀏覽器打開 http://127.0.0.1:" + str(args.port))
web.run_app(make_web_app(net), host="127.0.0.1", port=args.port, print=None)
if __name__ == "__main__":
main()