eye service
This commit is contained in:
Alex
415
eye_service_mk2.py
Normal file
415
eye_service_mk2.py
Normal file
@@ -0,0 +1,415 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Vixy Eye Service Mk2 - USB Bridge
|
||||
HTTP API (port 8780) → USB serial → ESP32 Co5300 eyes
|
||||
|
||||
Protocol:
|
||||
S:idle\\n → set state
|
||||
G:127:127\\n → set gaze x:y (0-255)
|
||||
T:1234567\\n → sync animation clock
|
||||
|
||||
Usage:
|
||||
python3 eye_service_mk2.py
|
||||
python3 eye_service_mk2.py --learn # record current USB serials as left/right
|
||||
python3 eye_service_mk2.py --left /dev/tty.usbmodem1 --right /dev/tty.usbmodem2
|
||||
"""
|
||||
|
||||
import serial, serial.tools.list_ports
|
||||
import threading, time, argparse, random, os, json, math, signal, sys
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
|
||||
HTTP_PORT = 8780
|
||||
BAUD = 115200
|
||||
SYNC_INTERVAL = 10.0 # seconds between T: sync commands
|
||||
RECONNECT_INTERVAL = 2.0 # seconds between reconnect attempts
|
||||
SACCADE_MIN_INTERVAL = 15.0
|
||||
SACCADE_MAX_INTERVAL = 30.0
|
||||
SACCADE_DURATION = 0.2
|
||||
|
||||
CONFIG_DIR = os.path.expanduser("~/.vixy")
|
||||
CONFIG_PATH = os.path.join(CONFIG_DIR, "eyes.json")
|
||||
|
||||
VALID_STATES = [
|
||||
"idle","listening","responding","pleasure",
|
||||
"thinking","playful","commanding","love","sleep"
|
||||
]
|
||||
|
||||
# === Shared state ===
|
||||
current_state = "idle"
|
||||
baseline_gaze = [127, 127] # user-commanded gaze (center = 127)
|
||||
state_lock = threading.Lock()
|
||||
running = True
|
||||
|
||||
|
||||
class EyePort:
|
||||
"""One eye's serial connection, identified by USB serial number for stability."""
|
||||
|
||||
def __init__(self, label: str, usb_serial: str = None, device: str = None):
|
||||
self.label = label # "left" or "right"
|
||||
self.usb_serial = usb_serial # stable USB serial number (preferred)
|
||||
self.device = device # /dev path (fallback, unstable)
|
||||
self.ser = None
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def resolve_device(self) -> str | None:
|
||||
"""Find the /dev path for this eye. Prefer USB serial match; fall back to device path."""
|
||||
if self.usb_serial:
|
||||
for p in serial.tools.list_ports.comports():
|
||||
if p.serial_number and p.serial_number == self.usb_serial:
|
||||
return p.device
|
||||
return None
|
||||
return self.device
|
||||
|
||||
def open(self) -> bool:
|
||||
path = self.resolve_device()
|
||||
if not path:
|
||||
return False
|
||||
try:
|
||||
with self.lock:
|
||||
self.ser = serial.Serial(path, BAUD, timeout=1, write_timeout=1)
|
||||
time.sleep(2) # ESP32 USB CDC enumeration
|
||||
print(f"[EYE] {self.label}: opened {path}"
|
||||
+ (f" (USB serial {self.usb_serial})" if self.usb_serial else ""))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[EYE] {self.label}: cannot open {path}: {e}")
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
with self.lock:
|
||||
if self.ser:
|
||||
try:
|
||||
self.ser.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.ser = None
|
||||
|
||||
def write(self, data: bytes) -> bool:
|
||||
with self.lock:
|
||||
if not self.ser or not self.ser.is_open:
|
||||
return False
|
||||
try:
|
||||
self.ser.write(data)
|
||||
self.ser.flush()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[EYE] {self.label}: write error: {e}")
|
||||
try:
|
||||
self.ser.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.ser = None
|
||||
return False
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
with self.lock:
|
||||
return bool(self.ser and self.ser.is_open)
|
||||
|
||||
|
||||
eyes: list[EyePort] = [] # populated in main()
|
||||
|
||||
|
||||
def send_cmd(cmd: str):
|
||||
"""Send a command to all eyes. Holds each port's lock independently."""
|
||||
data = (cmd + "\n").encode()
|
||||
for eye in eyes:
|
||||
eye.write(data)
|
||||
|
||||
def send_state(state: str): send_cmd(f"S:{state}")
|
||||
def send_gaze(x: int, y: int): send_cmd(f"G:{x}:{y}")
|
||||
def send_sync():
|
||||
ms = int(time.time() * 1000) % (2**31)
|
||||
send_cmd(f"T:{ms}")
|
||||
|
||||
|
||||
def sync_loop():
|
||||
"""Periodically resync animation clocks."""
|
||||
while running:
|
||||
time.sleep(SYNC_INTERVAL)
|
||||
if not running:
|
||||
break
|
||||
send_sync()
|
||||
|
||||
|
||||
def reconnect_loop():
|
||||
"""Reopen any eye that's currently disconnected."""
|
||||
while running:
|
||||
time.sleep(RECONNECT_INTERVAL)
|
||||
if not running:
|
||||
break
|
||||
for eye in eyes:
|
||||
if not eye.is_connected():
|
||||
if eye.open():
|
||||
# Restore full state on the reconnected eye
|
||||
with state_lock:
|
||||
state = current_state
|
||||
gx, gy = baseline_gaze
|
||||
send_sync() # cheap, safe to send to both
|
||||
send_state(state)
|
||||
send_gaze(gx, gy)
|
||||
|
||||
|
||||
def saccade_loop():
|
||||
"""Random micro-flicks to keep the eyes feeling alive. Returns to baseline after each."""
|
||||
while running:
|
||||
interval = random.uniform(SACCADE_MIN_INTERVAL, SACCADE_MAX_INTERVAL)
|
||||
# Sleep in small chunks so shutdown is responsive
|
||||
slept = 0.0
|
||||
while slept < interval and running:
|
||||
time.sleep(0.5)
|
||||
slept += 0.5
|
||||
if not running:
|
||||
break
|
||||
|
||||
dx = random.randint(-20, 20)
|
||||
dy = random.randint(-10, 10)
|
||||
with state_lock:
|
||||
gx, gy = baseline_gaze
|
||||
fx = max(0, min(255, gx + dx))
|
||||
fy = max(0, min(255, gy + dy))
|
||||
send_gaze(fx, fy)
|
||||
time.sleep(SACCADE_DURATION)
|
||||
# Return to baseline (re-read in case it changed mid-saccade)
|
||||
with state_lock:
|
||||
gx, gy = baseline_gaze
|
||||
send_gaze(gx, gy)
|
||||
|
||||
|
||||
def doa_to_gaze(doa_angle: float) -> tuple[int, int]:
|
||||
"""Convert DoA angle (0-360) to gaze x:y (0-255)."""
|
||||
rad = math.radians(doa_angle)
|
||||
x = int(127 - 80 * math.sin(rad))
|
||||
y = int(127 - 40 * math.cos(rad))
|
||||
return max(0, min(255, x)), max(0, min(255, y))
|
||||
|
||||
|
||||
# === HTTP API ===
|
||||
class EyeAPIHandler(BaseHTTPRequestHandler):
|
||||
def log_message(self, fmt, *args): pass
|
||||
|
||||
def _json(self, data, status=200):
|
||||
body = json.dumps(data).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_OPTIONS(self):
|
||||
self.send_response(200)
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
||||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||||
self.end_headers()
|
||||
|
||||
def do_GET(self):
|
||||
if self.path == "/health":
|
||||
self._json({
|
||||
"status": "ok",
|
||||
"service": "vixy-eyes-mk2",
|
||||
"eyes": {e.label: e.is_connected() for e in eyes},
|
||||
})
|
||||
elif self.path == "/state":
|
||||
with state_lock:
|
||||
self._json({"state": current_state, "gaze": list(baseline_gaze)})
|
||||
else:
|
||||
self._json({"error": "not found"}, 404)
|
||||
|
||||
def do_POST(self):
|
||||
global current_state
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
data = json.loads(self.rfile.read(length).decode())
|
||||
except Exception as e:
|
||||
self._json({"error": str(e)}, 400)
|
||||
return
|
||||
|
||||
if self.path == "/state":
|
||||
new_state = data.get("state", "").lower()
|
||||
if new_state in VALID_STATES:
|
||||
with state_lock:
|
||||
old, current_state = current_state, new_state
|
||||
save_persistent_state()
|
||||
print(f"[EYE] {old} → {new_state}")
|
||||
send_state(new_state)
|
||||
self._json({"ok": True, "state": new_state})
|
||||
else:
|
||||
self._json({"error": f"invalid state. valid: {VALID_STATES}"}, 400)
|
||||
|
||||
elif self.path == "/gaze":
|
||||
if "doa" in data:
|
||||
x, y = doa_to_gaze(float(data["doa"]))
|
||||
else:
|
||||
x = max(0, min(255, int(data.get("x", 127))))
|
||||
y = max(0, min(255, int(data.get("y", 127))))
|
||||
with state_lock:
|
||||
baseline_gaze[0], baseline_gaze[1] = x, y
|
||||
save_persistent_state()
|
||||
send_gaze(x, y)
|
||||
self._json({"ok": True, "gaze": [x, y]})
|
||||
else:
|
||||
self._json({"error": "not found"}, 404)
|
||||
|
||||
|
||||
# === Port identification ===
|
||||
def list_candidate_ports():
|
||||
"""Return serial ports that look like ESP32 USB CDC."""
|
||||
candidates = []
|
||||
for p in serial.tools.list_ports.comports():
|
||||
d = (p.description or "").lower()
|
||||
if (any(x in d for x in ["cp210", "ch340", "esp32", "usb serial"])
|
||||
or "usbmodem" in p.device
|
||||
or "ttyacm" in p.device.lower()):
|
||||
candidates.append(p)
|
||||
return candidates
|
||||
|
||||
|
||||
def load_eye_config() -> dict:
|
||||
if not os.path.exists(CONFIG_PATH):
|
||||
return {}
|
||||
try:
|
||||
with open(CONFIG_PATH) as f:
|
||||
return json.load(f)
|
||||
except Exception as e:
|
||||
print(f"[EYE] failed to read {CONFIG_PATH}: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
def save_eye_config(cfg: dict):
|
||||
os.makedirs(CONFIG_DIR, exist_ok=True)
|
||||
with open(CONFIG_PATH, "w") as f:
|
||||
json.dump(cfg, f, indent=2)
|
||||
|
||||
|
||||
def learn_eyes():
|
||||
"""Record the currently-connected ESP32 USB serial numbers as left/right."""
|
||||
candidates = list_candidate_ports()
|
||||
if len(candidates) < 2:
|
||||
print(f"[EYE] learn: need 2 candidate ports, found {len(candidates)}")
|
||||
return False
|
||||
# Deterministic assignment: sort by device path so "learn" is repeatable
|
||||
candidates.sort(key=lambda p: p.device)
|
||||
cfg = load_eye_config()
|
||||
cfg.setdefault("eyes", {})
|
||||
cfg["eyes"]["left"] = {"usb_serial": candidates[0].serial_number, "device": candidates[0].device}
|
||||
cfg["eyes"]["right"] = {"usb_serial": candidates[1].serial_number, "device": candidates[1].device}
|
||||
save_eye_config(cfg)
|
||||
print(f"[EYE] learned:")
|
||||
print(f" left = {candidates[0].device} (USB serial {candidates[0].serial_number})")
|
||||
print(f" right = {candidates[1].device} (USB serial {candidates[1].serial_number})")
|
||||
return True
|
||||
|
||||
|
||||
# === Persistent state ===
|
||||
def save_persistent_state():
|
||||
"""Save current_state + baseline_gaze so restarts restore mood/gaze."""
|
||||
cfg = load_eye_config()
|
||||
with state_lock:
|
||||
cfg["last_state"] = current_state
|
||||
cfg["last_gaze"] = list(baseline_gaze)
|
||||
try:
|
||||
save_eye_config(cfg)
|
||||
except Exception as e:
|
||||
print(f"[EYE] failed to save state: {e}")
|
||||
|
||||
|
||||
def load_persistent_state():
|
||||
global current_state
|
||||
cfg = load_eye_config()
|
||||
last_state = cfg.get("last_state")
|
||||
last_gaze = cfg.get("last_gaze")
|
||||
if last_state in VALID_STATES:
|
||||
current_state = last_state
|
||||
if isinstance(last_gaze, list) and len(last_gaze) == 2:
|
||||
baseline_gaze[0], baseline_gaze[1] = int(last_gaze[0]), int(last_gaze[1])
|
||||
|
||||
|
||||
# === Main ===
|
||||
def main():
|
||||
global running
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--left", default=None,
|
||||
help="Override left eye device path (disables USB serial matching)")
|
||||
parser.add_argument("--right", default=None,
|
||||
help="Override right eye device path")
|
||||
parser.add_argument("--port", type=int, default=HTTP_PORT)
|
||||
parser.add_argument("--learn", action="store_true",
|
||||
help="Record currently-connected ESP32 USB serials as left/right and exit")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.learn:
|
||||
sys.exit(0 if learn_eyes() else 1)
|
||||
|
||||
load_persistent_state()
|
||||
|
||||
# Build EyePort objects — prefer CLI overrides, then learned USB serials, then positional auto-detect
|
||||
cfg = load_eye_config().get("eyes", {})
|
||||
|
||||
if args.left and args.right:
|
||||
eyes.append(EyePort("left", device=args.left))
|
||||
eyes.append(EyePort("right", device=args.right))
|
||||
elif cfg.get("left") and cfg.get("right"):
|
||||
eyes.append(EyePort("left", usb_serial=cfg["left"].get("usb_serial"),
|
||||
device=cfg["left"].get("device")))
|
||||
eyes.append(EyePort("right", usb_serial=cfg["right"].get("usb_serial"),
|
||||
device=cfg["right"].get("device")))
|
||||
else:
|
||||
detected = list_candidate_ports()
|
||||
detected.sort(key=lambda p: p.device)
|
||||
print(f"[EYE] no learned config; detected ports: {[p.device for p in detected]}")
|
||||
print(f"[EYE] run with --learn to pin left/right to USB serial numbers")
|
||||
if len(detected) >= 2:
|
||||
eyes.append(EyePort("left", device=detected[0].device))
|
||||
eyes.append(EyePort("right", device=detected[1].device))
|
||||
elif len(detected) == 1:
|
||||
eyes.append(EyePort("left", device=detected[0].device))
|
||||
else:
|
||||
print("[EYE] no ESP32 ports found")
|
||||
sys.exit(1)
|
||||
|
||||
# Open what we can; reconnect loop will retry anything that fails
|
||||
for eye in eyes:
|
||||
eye.open()
|
||||
|
||||
# Restore state on connected eyes
|
||||
send_sync()
|
||||
with state_lock:
|
||||
state = current_state
|
||||
gx, gy = baseline_gaze
|
||||
send_state(state)
|
||||
send_gaze(gx, gy)
|
||||
|
||||
# Background threads
|
||||
threading.Thread(target=sync_loop, daemon=True).start()
|
||||
threading.Thread(target=reconnect_loop, daemon=True).start()
|
||||
threading.Thread(target=saccade_loop, daemon=True).start()
|
||||
|
||||
# Signal handlers for clean shutdown under systemd/k8s
|
||||
def handle_shutdown(sig, frame):
|
||||
global running
|
||||
if not running:
|
||||
return
|
||||
print(f"\n[EYE] signal {sig} received, shutting down")
|
||||
running = False
|
||||
try:
|
||||
server.shutdown()
|
||||
except Exception:
|
||||
pass
|
||||
signal.signal(signal.SIGINT, handle_shutdown)
|
||||
signal.signal(signal.SIGTERM, handle_shutdown)
|
||||
|
||||
server = HTTPServer(("0.0.0.0", args.port), EyeAPIHandler)
|
||||
print(f"[EYE] Vixy Eye Mk2 on port {args.port}")
|
||||
print(f"[EYE] connected: {[e.label for e in eyes if e.is_connected()]}")
|
||||
try:
|
||||
server.serve_forever()
|
||||
finally:
|
||||
running = False
|
||||
for eye in eyes:
|
||||
eye.close()
|
||||
print("[EYE] shutdown complete")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user