Move all animations to Pico firmware, Pi becomes HTTP-serial bridge

Pico runs all 9 mood animations and jaw modes locally instead of
receiving per-LED serial commands from the Pi. New protocol: bare
integers for fire-and-forget jaw amplitude, "mood X" / "jaw X" for
mode switches. Cuts light_service.py from 409 to 193 lines.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Alex
2026-01-31 21:34:03 -06:00
parent 135ab1138c
commit c74371a24a
4 changed files with 442 additions and 349 deletions

View File

@@ -8,14 +8,15 @@ LED strip control for head-vixy (Vixy's robotic head) via Raspberry Pi Pico 2.
┌─────────────────┐ USB Serial ┌──────────────────┐ ┌─────────────────┐ USB Serial ┌──────────────────┐
│ Raspberry Pi 5 │◄──────────────────►│ Pico 2 │ │ Raspberry Pi 5 │◄──────────────────►│ Pico 2 │
│ (light_service)│ /dev/ttyACM0 │ (main.py) │ │ (light_service)│ /dev/ttyACM0 │ (main.py) │
│ Port 8781 │ │ │ Port 8781 │ │ Animation Engine
└─────────────────┘ │ GP0 → Jaw LEDs └─────────────────┘ │
│ GP1Mood LEDs │ HTTP-to-serial │ GP0Jaw LEDs
bridge │ GP1 → Mood LEDs │
└──────────────────┘ └──────────────────┘
``` ```
- **Raspberry Pi 5** - Runs Python service, HTTP API, animation logic - **Raspberry Pi 5** - Thin HTTP-to-serial bridge, forwards mode commands
- **Raspberry Pi Pico 2** - Controls LED strips via PIO, receives serial commands - **Raspberry Pi Pico 2** - Runs all LED animations locally via PIO
- **Mood strip**: 56x WS2812B LEDs on GP1 - **Mood strip**: 56x WS2812B LEDs on GP1
- **Jaw strip**: 14x WS2812B LEDs on GP0 (index 0 damaged, use 1-13) - **Jaw strip**: 14x WS2812B LEDs on GP0 (index 0 damaged, use 1-13)
@@ -28,11 +29,12 @@ The BBB PRU approach burned 4 boards. Pico 2 is $5, runs MicroPython, has hardwa
``` ```
head-lights/ head-lights/
├── README.md # This file ├── README.md # This file
├── light_service.py # Pi service (animations + HTTP API) ├── light_service.py # Pi service (HTTP-to-serial bridge)
├── requirements.txt # Python deps (pyserial) ├── requirements.txt # Python deps (pyserial)
├── vixy-lights.service # systemd unit file ├── vixy-lights.service # systemd unit file
├── docs/plans/ # Design documents
└── firmware/ └── firmware/
└── main.py # Pico 2 MicroPython firmware └── main.py # Pico 2 MicroPython firmware (animation engine)
``` ```
## States ## States
@@ -49,7 +51,6 @@ head-lights/
| love | Soft pink 💕 | Gentle breathing | Tender/affectionate | | love | Soft pink 💕 | Gentle breathing | Tender/affectionate |
| sleep | Dim blue-gray | Very slow, dim | Low power/resting | | sleep | Dim blue-gray | Very slow, dim | Low power/resting |
## API Endpoints ## API Endpoints
``` ```
@@ -57,30 +58,26 @@ GET /health - Service health (includes serial connection status)
GET /state - Current light state + jaw level GET /state - Current light state + jaw level
POST /state - Set state: {"state": "listening"} POST /state - Set state: {"state": "listening"}
POST /jaw/level - Set jaw level: {"level": 0-100} POST /jaw/level - Set jaw level: {"level": 0-100}
POST /jaw/mode - Set jaw mode: {"mode": "talking"}
``` ```
**Port: 8781** **Port: 8781**
## Pico Serial Protocol ## Pico Serial Protocol
Commands sent over USB serial at 115200 baud: All animations run on the Pico. The Pi sends mode commands over USB serial at 115200 baud:
``` | Command | Example | Response | Description |
<strip> <index> <r> <g> <b> - Set single LED |---------|---------|----------|-------------|
<strip> -1 - Trigger strip update (show) | `<integer>` | `72` | *(none)* | Jaw amplitude 0-100, fire-and-forget |
<strip> clear - Clear strip | `mood <mode>` | `mood thinking` | `OK` | Switch mood animation |
<strip> fill <r> <g> <b> - Fill entire strip (mood only) | `jaw <mode>` | `jaw talking` | `OK` | Switch jaw animation mode |
```
Strip IDs: 0 = Jaw, 1 = Mood **Mood modes:** idle, listening, responding, pleasure, thinking, playful, commanding, love, sleep, off
Examples: **Jaw modes:** idle, talking, off
```
1 0 255 0 0 # Set mood LED 0 to red Bare integers are the fast path for jaw amplitude during speech — minimal bytes, no response wait.
1 -1 # Show mood strip
0 clear # Clear jaw strip
1 fill 0 255 255 # Fill mood with cyan
```
## Installation ## Installation
@@ -122,7 +119,7 @@ curl http://head-vixy.local:8781/health
# Get current state # Get current state
curl http://head-vixy.local:8781/state curl http://head-vixy.local:8781/state
# Set state # Set mood state
curl -X POST http://head-vixy.local:8781/state \ curl -X POST http://head-vixy.local:8781/state \
-H "Content-Type: application/json" \ -H "Content-Type: application/json" \
-d '{"state": "thinking"}' -d '{"state": "thinking"}'
@@ -131,6 +128,11 @@ curl -X POST http://head-vixy.local:8781/state \
curl -X POST http://head-vixy.local:8781/jaw/level \ curl -X POST http://head-vixy.local:8781/jaw/level \
-H "Content-Type: application/json" \ -H "Content-Type: application/json" \
-d '{"level": 75}' -d '{"level": 75}'
# Set jaw mode
curl -X POST http://head-vixy.local:8781/jaw/mode \
-H "Content-Type: application/json" \
-d '{"mode": "talking"}'
``` ```
## Unified Head Control ## Unified Head Control

View File

@@ -0,0 +1,41 @@
# Pico-Side Animations - Phase 1
Move all animation logic from Pi's `light_service.py` to Pico's `firmware/main.py`.
Pi becomes a thin HTTP-to-serial bridge.
## New Serial Protocol
| Input | Example | Response | Description |
|-------|---------|----------|-------------|
| `<int>` | `72` | *(none)* | Jaw amplitude 0-100, fire-and-forget |
| `mood <mode>` | `mood thinking` | `OK` | Switch mood animation |
| `jaw <mode>` | `jaw talking` | `OK` | Switch jaw animation |
Mood modes: idle, listening, responding, pleasure, thinking, playful, commanding, love, sleep, off
Jaw modes: idle, talking, off
Bare integers are jaw amplitude (fast path, no response). This replaces per-LED serial flooding.
## Pico Firmware Architecture
Non-blocking main loop:
1. Check serial (non-blocking poll)
2. If data: integer -> jaw amplitude; string -> mode command
3. Tick mood animation (based on mood_mode + time)
4. Tick jaw animation (based on jaw_mode or amplitude)
5. ~10-20ms frame pacing
All 9 mood animations ported from light_service.py using `time.ticks_ms()`.
Jaw: "level" mode maps 0-100 to 13 LEDs, "talking" mode oscillates autonomously.
## Pi light_service.py Changes
Remove: all animation logic, mood_buffer, per-LED helpers, main animation loop.
Keep: HTTP API, serial connection management.
POST /state -> sends `mood <state>\n`, waits for OK.
POST /jaw/level -> sends `<level>\n`, fire-and-forget (no readline).
## Files Changed
- `firmware/main.py` - Full rewrite with animation engine
- `light_service.py` - Strip down to HTTP-to-serial bridge

View File

@@ -1,31 +1,43 @@
""" """
Pico 2 LED Controller for Vixy's Head Pico 2 LED Controller for Vixy's Head - Animation Engine
Dual WS2812 strip control via USB serial. Runs all animations locally. Pi sends mode commands, Pico does the rest.
- Strip 0 (Jaw): GP0, 14 LEDs (index 0 damaged)
Hardware:
- Strip 0 (Jaw): GP0, 14 LEDs (index 0 damaged, use 1-13)
- Strip 1 (Mood): GP1, 56 LEDs - Strip 1 (Mood): GP1, 56 LEDs
Protocol: Protocol:
"0 index r g b" - Set jaw LED <integer> - Jaw amplitude 0-100 (fire-and-forget, no response)
"1 index r g b" - Set mood LED mood <mode> - Set mood animation (responds OK)
"0 -1" - Trigger jaw output jaw <mode> - Set jaw animation (responds OK)
"1 -1" - Trigger mood output
"0 clear" - Clear jaw Mood modes: idle, listening, responding, pleasure, thinking, playful, commanding, love, sleep, off
"1 clear" - Clear mood Jaw modes: idle, talking, off
"1 fill r g b" - Fill mood strip
""" """
import sys import sys
import array import array
import math
import time import time
from machine import Pin from machine import Pin
import rp2 import rp2
try:
from urandom import getrandbits
except ImportError:
from random import getrandbits
# Configuration # Configuration
JAW_PIN = 0 JAW_PIN = 0
MOOD_PIN = 1 MOOD_PIN = 1
JAW_LEDS = 14 JAW_LEDS = 14
MOOD_LEDS = 56 MOOD_LEDS = 56
JAW_OFFSET = 1 # Skip index 0 (damaged)
JAW_USABLE = JAW_LEDS - JAW_OFFSET # 13
FRAME_MS = 16 # ~60fps target
# WS2812 PIO program - 800kHz timing # WS2812 PIO program - 800kHz timing
@rp2.asm_pio(sideset_init=rp2.PIO.OUT_LOW, out_shiftdir=rp2.PIO.SHIFT_LEFT, autopull=True, pull_thresh=24) @rp2.asm_pio(sideset_init=rp2.PIO.OUT_LOW, out_shiftdir=rp2.PIO.SHIFT_LEFT, autopull=True, pull_thresh=24)
@@ -72,63 +84,317 @@ class WS2812Strip:
# Initialize strips # Initialize strips
jaw = WS2812Strip(JAW_PIN, JAW_LEDS, 0) jaw = WS2812Strip(JAW_PIN, JAW_LEDS, 0)
mood = WS2812Strip(MOOD_PIN, MOOD_LEDS, 1) mood = WS2812Strip(MOOD_PIN, MOOD_LEDS, 1)
strips = [jaw, mood]
# Clear both on startup
jaw.clear() jaw.clear()
jaw.show() jaw.show()
mood.clear() mood.clear()
mood.show() mood.show()
# === Color definitions ===
# (r, g, b) tuples
MOOD_COLORS = {
"idle": (0, 255, 255),
"listening": (160, 255, 255),
"responding": (100, 220, 255),
"pleasure": (180, 150, 255),
"thinking": (255, 180, 50),
"playful": (255, 130, 90),
"commanding": (220, 50, 120),
"love": (255, 160, 180),
"sleep": (20, 22, 35),
}
JAW_COLOR = (0, 200, 255)
# === Animation state ===
mood_mode = "idle"
jaw_mode = "off"
jaw_amplitude = 0 # 0-100, set by bare integer commands
# Per-animation persistent state
larson_pos = 0
larson_dir = 1
wave_offset = 0.0
# Sparkle buffer: store per-LED brightness as fixed-point (0-255)
sparkle_buf = array.array("B", [0] * MOOD_LEDS)
# === Helpers ===
def scale_color_grb(r, g, b, factor_256):
"""Scale RGB and return GRB packed int. factor_256 is 0-256 (fixed-point for 0.0-1.0)."""
rr = (r * factor_256) >> 8
gg = (g * factor_256) >> 8
bb = (b * factor_256) >> 8
return (gg << 16) | (rr << 8) | bb
def pack_grb(r, g, b):
return (g << 16) | (r << 8) | b
# Precompute a sine table (64 entries, values 0-255 representing 0.0-1.0)
SINE_TABLE_SIZE = 64
_sine_table = array.array("B", [0] * SINE_TABLE_SIZE)
for _i in range(SINE_TABLE_SIZE):
# sin maps to 0..1 range: (sin(x) + 1) / 2
_v = (math.sin(2.0 * math.pi * _i / SINE_TABLE_SIZE) + 1.0) / 2.0
_sine_table[_i] = int(_v * 255)
def sine_lookup(phase_256):
"""Look up sine value from table. phase_256 is 0-255 mapping to 0-2pi. Returns 0-255."""
idx = (phase_256 * SINE_TABLE_SIZE) >> 8
return _sine_table[idx % SINE_TABLE_SIZE]
# === Mood animation ticks ===
def tick_pulse(now_ms, color, speed_mhz, min_b, max_b):
"""Solid pulse animation. speed_mhz is milli-Hz (500 = 0.5 Hz)."""
# phase = (now_ms * speed_mhz / 1000) mod 256
phase = ((now_ms * speed_mhz) // 1000) & 0xFF
raw = sine_lookup(phase) # 0-255
# Map to min_b..max_b range (both 0-255 scale)
brightness = min_b + ((max_b - min_b) * raw >> 8)
r, g, b = color
packed = scale_color_grb(r, g, b, brightness)
ar = mood.ar
for i in range(MOOD_LEDS):
ar[i] = packed
mood.show()
def tick_wave(color):
"""Traveling wave animation."""
global wave_offset
r, g, b = color
ar = mood.ar
for i in range(MOOD_LEDS):
# Wave phase for this LED
phase = int(((i - wave_offset) / 15.0) * 256) & 0xFF
raw = sine_lookup(phase) # 0-255
# Map to 0.2-1.0 range: 51 + (205 * raw / 255)
brightness = 51 + ((204 * raw) >> 8)
ar[i] = scale_color_grb(r, g, b, brightness)
mood.show()
wave_offset += 0.5
def tick_larson(color):
"""Larson scanner (bouncing dot with trail)."""
global larson_pos, larson_dir
r, g, b = color
ar = mood.ar
# Clear
for i in range(MOOD_LEDS):
ar[i] = 0
# Bright point
if 0 <= larson_pos < MOOD_LEDS:
ar[larson_pos] = pack_grb(r, g, b)
# Trail (8 LEDs behind)
for t in range(1, 9):
idx = larson_pos - larson_dir * t
if 0 <= idx < MOOD_LEDS:
# cosine fade: cos((t/8) * pi/2) mapped to 0-256
fade = _sine_table[(SINE_TABLE_SIZE // 4 - (t * SINE_TABLE_SIZE // 32)) % SINE_TABLE_SIZE]
ar[idx] = scale_color_grb(r, g, b, fade)
mood.show()
larson_pos += larson_dir
if larson_pos >= MOOD_LEDS - 1:
larson_pos = MOOD_LEDS - 1
larson_dir = -1
elif larson_pos <= 0:
larson_pos = 0
larson_dir = 1
def tick_sparkle(color):
"""Random sparkle with decay."""
r, g, b = color
ar = mood.ar
buf = sparkle_buf
for i in range(MOOD_LEDS):
# Decay existing brightness by ~15%: multiply by 217/256
v = (buf[i] * 217) >> 8
# Random spawn: ~15% chance = getrandbits check
if getrandbits(3) == 0: # 1/8 = 12.5%, close enough to 15%
v = 255
buf[i] = v
ar[i] = scale_color_grb(r, g, b, v)
mood.show()
def tick_mood(now_ms):
"""Run one frame of the current mood animation."""
mode = mood_mode
if mode == "off":
return
color = MOOD_COLORS.get(mode)
if color is None:
return
if mode == "idle":
tick_pulse(now_ms, color, 500, 51, 204) # 0.5 Hz, 20-80%
elif mode == "listening":
tick_pulse(now_ms, color, 2000, 153, 255) # 2.0 Hz, 60-100%
elif mode == "responding":
tick_wave(color)
elif mode == "pleasure":
tick_pulse(now_ms, color, 800, 77, 230) # 0.8 Hz, 30-90%
elif mode == "thinking":
tick_larson(color)
elif mode == "playful":
tick_sparkle(color)
elif mode == "commanding":
tick_pulse(now_ms, color, 3000, 128, 255) # 3.0 Hz, 50-100%
elif mode == "love":
tick_pulse(now_ms, color, 600, 102, 230) # 0.6 Hz, 40-90%
elif mode == "sleep":
tick_pulse(now_ms, color, 200, 26, 77) # 0.2 Hz, 10-30%
# === Jaw animation ticks ===
def jaw_set_amplitude(amp):
"""Set jaw LEDs based on amplitude 0-100."""
num_lit = (amp * JAW_USABLE) // 100
r, g, b = JAW_COLOR
on = pack_grb(r, g, b)
ar = jaw.ar
ar[0] = 0 # Damaged LED always off
for i in range(JAW_OFFSET, JAW_LEDS):
if i - JAW_OFFSET < num_lit:
ar[i] = on
else:
ar[i] = 0
jaw.show()
def tick_jaw(now_ms):
"""Run one frame of jaw animation."""
mode = jaw_mode
if mode == "level":
jaw_set_amplitude(jaw_amplitude)
elif mode == "talking":
# Autonomous talking: oscillate with some randomness
phase = ((now_ms * 6000) // 1000) & 0xFF # 6 Hz base
base = sine_lookup(phase) # 0-255
# Add randomness: +/- 30
jitter = (getrandbits(6) - 32) # -32 to 31
val = base + jitter
if val < 0:
val = 0
elif val > 255:
val = 255
amp = (val * 100) >> 8 # Scale to 0-100
jaw_set_amplitude(amp)
elif mode == "idle":
# Subtle dim glow
ar = jaw.ar
ar[0] = 0
dim = scale_color_grb(JAW_COLOR[0], JAW_COLOR[1], JAW_COLOR[2], 15)
for i in range(JAW_OFFSET, JAW_LEDS):
ar[i] = dim
jaw.show()
elif mode == "off":
pass # Already cleared on mode switch
# === Command processing ===
def process_command(line):
"""Parse incoming command. Returns True if response needed."""
global mood_mode, jaw_mode, jaw_amplitude
line = line.strip()
if not line:
return
# Fast path: bare integer = jaw amplitude
try:
val = int(line)
jaw_amplitude = max(0, min(100, val))
jaw_mode = "level"
return # Fire-and-forget, no response
except ValueError:
pass
parts = line.split()
cmd = parts[0]
if cmd == "mood" and len(parts) >= 2:
new_mode = parts[1]
if new_mode == "off":
mood_mode = "off"
mood.clear()
mood.show()
elif new_mode in MOOD_COLORS:
mood_mode = new_mode
# Reset animation state on mode switch
_reset_mood_state()
else:
print("ERR: unknown mood")
return
print("OK")
elif cmd == "jaw" and len(parts) >= 2:
new_mode = parts[1]
if new_mode in ("idle", "talking", "off"):
jaw_mode = new_mode
if new_mode == "off":
jaw.clear()
jaw.show()
print("OK")
else:
print("ERR: unknown jaw mode")
else:
print("ERR: unknown command")
def _reset_mood_state():
"""Reset per-animation state on mode switch."""
global larson_pos, larson_dir, wave_offset
larson_pos = 0
larson_dir = 1
wave_offset = 0.0
for i in range(MOOD_LEDS):
sparkle_buf[i] = 0
# === Main loop ===
print("Pico LED Controller ready") print("Pico LED Controller ready")
print(f"Jaw: GP{JAW_PIN}, {JAW_LEDS} LEDs") print(f"Jaw: GP{JAW_PIN}, {JAW_LEDS} LEDs")
print(f"Mood: GP{MOOD_PIN}, {MOOD_LEDS} LEDs") print(f"Mood: GP{MOOD_PIN}, {MOOD_LEDS} LEDs")
# Non-blocking serial setup
import select as _sel
_poller = _sel.poll()
_poller.register(sys.stdin, _sel.POLLIN)
_read_buf = ""
def process_command(line):
"""Parse and execute a command."""
parts = line.strip().split()
if len(parts) < 2:
return
try:
strip_id = int(parts[0])
if strip_id not in (0, 1):
return
strip = strips[strip_id]
cmd = parts[1]
if cmd == "clear":
strip.clear()
strip.show()
print("OK")
elif cmd == "-1":
strip.show()
print("OK")
elif cmd == "fill" and len(parts) >= 5:
r, g, b = int(parts[2]), int(parts[3]), int(parts[4])
strip.fill(r, g, b)
strip.show()
print("OK")
elif len(parts) >= 5:
index = int(cmd)
r, g, b = int(parts[2]), int(parts[3]), int(parts[4])
strip.set_led(index, r, g, b)
print("OK")
except (ValueError, IndexError) as e:
print(f"ERR: {e}")
# Main loop - blocking readline is fine for this use case
while True: while True:
try: now = time.ticks_ms()
line = sys.stdin.readline()
if line: # Drain all available serial data (non-blocking)
process_command(line) while _poller.poll(0):
except Exception as e: ch = sys.stdin.read(1)
print(f"ERR: {e}") if ch == '\n':
process_command(_read_buf)
_read_buf = ""
else:
_read_buf += ch
# Tick animations
tick_mood(now)
tick_jaw(now)
# Frame pacing
elapsed = time.ticks_diff(time.ticks_ms(), now)
if elapsed < FRAME_MS:
time.sleep_ms(FRAME_MS - elapsed)

View File

@@ -1,37 +1,26 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Vixy Light Service - Day 91 (Pico Edition) Vixy Light Service - HTTP-to-Serial Bridge
Controllable LED strips for head-vixy via Pico 2 USB serial
Thin bridge between HTTP API and Pico 2 animation engine.
All animations run on the Pico. This service just forwards mode commands.
Hardware: Hardware:
- Raspberry Pi Pico 2 connected via USB (/dev/ttyACM0) - Raspberry Pi Pico 2 connected via USB (/dev/ttyACM0)
- Mood strip: 56 LEDs on GP1 (strip ID 1) - Pico runs all LED animations locally
- Jaw strip: 14 LEDs on GP0 (strip ID 0), skip index 0, use 1-13
States match the eye service for unified control:
- idle: Slow cyan breathing pulse
- listening: Bright cyan gentle pulse
- responding: Blue-cyan traveling wave
- pleasure: Soft purple slow pulse 💜
- thinking: Amber/gold larson scanner
- playful: Warm coral bouncy sparkles 🦊
- commanding: Deep magenta strong pulse 😈
- love: Soft pink gentle breathing 💕
- sleep: Very dim blue-gray, nearly off
API: API:
GET /health - Service health GET /health - Service health
GET /state - Current state GET /state - Current state
POST /state - Set state {"state": "listening"} POST /state - Set state {"state": "listening"}
POST /jaw/level - Set jaw level {"level": 0-100} POST /jaw/level - Set jaw level {"level": 0-100}
POST /jaw/mode - Set jaw mode {"mode": "talking"}
""" """
import time import time
import math
import threading import threading
import json import json
import signal import signal
import random
from http.server import HTTPServer, BaseHTTPRequestHandler from http.server import HTTPServer, BaseHTTPRequestHandler
import serial import serial
@@ -40,18 +29,12 @@ HTTP_PORT = 8781
SERIAL_PORT = "/dev/ttyACM0" SERIAL_PORT = "/dev/ttyACM0"
SERIAL_BAUD = 115200 SERIAL_BAUD = 115200
# LED counts (must match Pico firmware)
MOOD_LEDS = 56
JAW_LEDS = 14
JAW_OFFSET = 1 # Skip first LED (damaged)
JAW_USABLE = JAW_LEDS - JAW_OFFSET # 13 usable LEDs
VALID_STATES = ["idle", "listening", "responding", "pleasure", "thinking", "playful", "commanding", "love", "sleep"] VALID_STATES = ["idle", "listening", "responding", "pleasure", "thinking", "playful", "commanding", "love", "sleep"]
VALID_JAW_MODES = ["idle", "talking", "off"]
# === Shared State === # === Shared State ===
current_state = "idle" current_state = "idle"
jaw_level = 0 # 0-100 jaw_level = 0
state_lock = threading.Lock()
serial_lock = threading.Lock() serial_lock = threading.Lock()
running = True running = True
ser = None ser = None
@@ -61,8 +44,7 @@ def init_serial():
global ser global ser
try: try:
ser = serial.Serial(SERIAL_PORT, SERIAL_BAUD, timeout=1) ser = serial.Serial(SERIAL_PORT, SERIAL_BAUD, timeout=1)
time.sleep(0.5) # Let Pico settle time.sleep(0.5)
# Clear any startup messages
ser.read_all() ser.read_all()
print(f"[LIGHT] Connected to Pico on {SERIAL_PORT}") print(f"[LIGHT] Connected to Pico on {SERIAL_PORT}")
return True return True
@@ -70,8 +52,8 @@ def init_serial():
print(f"[LIGHT] Serial error: {e}") print(f"[LIGHT] Serial error: {e}")
return False return False
def send_command(cmd): def send_mode_command(cmd):
"""Send command to Pico, return True if OK.""" """Send a mode command to Pico, wait for OK response."""
global ser global ser
with serial_lock: with serial_lock:
try: try:
@@ -86,91 +68,21 @@ def send_command(cmd):
ser = None ser = None
return False return False
def mood_fill(r, g, b): def send_amplitude(level):
"""Fill mood strip with color.""" """Send bare integer for jaw amplitude. Fire-and-forget."""
return send_command(f"1 fill {r} {g} {b}") global ser
def mood_set(index, r, g, b):
"""Set single mood LED."""
if 0 <= index < MOOD_LEDS:
send_command(f"1 {index} {r} {g} {b}")
def mood_show():
"""Trigger mood strip update."""
return send_command("1 -1")
def mood_clear():
"""Clear mood strip."""
return send_command("1 clear")
def jaw_fill(r, g, b):
"""Fill jaw strip (skipping damaged first LED)."""
with serial_lock: with serial_lock:
try: try:
if ser is None or not ser.is_open: if ser is None or not ser.is_open:
if not init_serial(): if not init_serial():
return False return False
# Set index 0 to off (damaged) ser.write(f"{level}\n".encode())
ser.write(f"0 0 0 0 0\n".encode()) return True
ser.readline()
# Fill usable LEDs
for i in range(JAW_OFFSET, JAW_LEDS):
ser.write(f"0 {i} {r} {g} {b}\n".encode())
ser.readline()
ser.write("0 -1\n".encode())
return ser.readline().decode().strip() == "OK"
except Exception as e: except Exception as e:
print(f"[LIGHT] Jaw fill error: {e}") print(f"[LIGHT] Serial error: {e}")
ser = None
return False return False
def jaw_set_level(level, color):
"""Set jaw LEDs based on level 0-100."""
num_lit = int((level / 100) * JAW_USABLE)
r, g, b = color
with serial_lock:
try:
if ser is None or not ser.is_open:
if not init_serial():
return False
# Index 0 always off
ser.write(f"0 0 0 0 0\n".encode())
ser.readline()
# Set LEDs based on level
for i in range(JAW_OFFSET, JAW_LEDS):
if i - JAW_OFFSET < num_lit:
ser.write(f"0 {i} {r} {g} {b}\n".encode())
else:
ser.write(f"0 {i} 0 0 0\n".encode())
ser.readline()
ser.write("0 -1\n".encode())
return ser.readline().decode().strip() == "OK"
except Exception as e:
print(f"[LIGHT] Jaw level error: {e}")
return False
def jaw_clear():
"""Clear jaw strip."""
return send_command("0 clear")
# === Color Utilities ===
def scale_color(color, factor):
"""Scale RGB color by a factor (0-1)."""
return tuple(max(0, min(255, int(c * factor))) for c in color)
# === State Colors ===
STATE_COLORS = {
"idle": (0, 255, 255), # Cyan
"listening": (160, 255, 255), # Bright cyan
"responding": (100, 220, 255),# Blue-cyan
"pleasure": (180, 150, 255), # Soft purple
"thinking": (255, 180, 50), # Amber/gold
"playful": (255, 130, 90), # Warm coral
"commanding": (220, 50, 120), # Deep magenta
"love": (255, 160, 180), # Soft pink
"sleep": (20, 22, 35), # Very dim blue-gray
}
# === HTTP API Handler === # === HTTP API Handler ===
class LightAPIHandler(BaseHTTPRequestHandler): class LightAPIHandler(BaseHTTPRequestHandler):
def log_message(self, format, *args): def log_message(self, format, *args):
@@ -184,7 +96,6 @@ class LightAPIHandler(BaseHTTPRequestHandler):
self.wfile.write(json.dumps(data).encode()) self.wfile.write(json.dumps(data).encode())
def do_GET(self): def do_GET(self):
global current_state, jaw_level
if self.path == '/health': if self.path == '/health':
self._send_json({ self._send_json({
"status": "ok", "status": "ok",
@@ -192,7 +103,6 @@ class LightAPIHandler(BaseHTTPRequestHandler):
"serial": ser is not None and ser.is_open "serial": ser is not None and ser.is_open
}) })
elif self.path == '/state': elif self.path == '/state':
with state_lock:
self._send_json({"state": current_state, "jaw_level": jaw_level}) self._send_json({"state": current_state, "jaw_level": jaw_level})
else: else:
self._send_json({"error": "Not found"}, 404) self._send_json({"error": "Not found"}, 404)
@@ -207,9 +117,9 @@ class LightAPIHandler(BaseHTTPRequestHandler):
if self.path == '/state': if self.path == '/state':
new_state = data.get('state', '').lower() new_state = data.get('state', '').lower()
if new_state in VALID_STATES: if new_state in VALID_STATES:
with state_lock:
old_state = current_state old_state = current_state
current_state = new_state current_state = new_state
send_mode_command(f"mood {new_state}")
print(f"[LIGHT] State: {old_state} -> {new_state}") print(f"[LIGHT] State: {old_state} -> {new_state}")
self._send_json({"success": True, "state": new_state}) self._send_json({"success": True, "state": new_state})
else: else:
@@ -218,11 +128,18 @@ class LightAPIHandler(BaseHTTPRequestHandler):
elif self.path == '/jaw/level': elif self.path == '/jaw/level':
level = int(data.get('level', 0)) level = int(data.get('level', 0))
level = max(0, min(100, level)) level = max(0, min(100, level))
with state_lock:
jaw_level = level jaw_level = level
print(f"[LIGHT] Jaw level: {level}%") send_amplitude(level)
self._send_json({"success": True, "jaw_level": level}) self._send_json({"success": True, "jaw_level": level})
elif self.path == '/jaw/mode':
mode = data.get('mode', '').lower()
if mode in VALID_JAW_MODES:
send_mode_command(f"jaw {mode}")
self._send_json({"success": True, "jaw_mode": mode})
else:
self._send_json({"error": f"Invalid jaw mode. Valid: {VALID_JAW_MODES}"}, 400)
else: else:
self._send_json({"error": "Not found"}, 404) self._send_json({"error": "Not found"}, 404)
except Exception as e: except Exception as e:
@@ -235,12 +152,6 @@ class LightAPIHandler(BaseHTTPRequestHandler):
self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.send_header('Access-Control-Allow-Headers', 'Content-Type')
self.end_headers() self.end_headers()
def run_http_server():
server = HTTPServer(('0.0.0.0', HTTP_PORT), LightAPIHandler)
print(f"[LIGHT] HTTP API listening on port {HTTP_PORT}")
while running:
server.handle_request()
# === Signal Handler === # === Signal Handler ===
def signal_handler(sig, frame): def signal_handler(sig, frame):
global running global running
@@ -250,156 +161,29 @@ def signal_handler(sig, frame):
signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGTERM, signal_handler)
# === Main ===
# === Animation Helpers ===
def solid_pulse(speed, min_b=0.1, max_b=1.0):
"""Return current brightness factor for smooth pulse."""
t = (math.sin(time.time() * speed) + 1) / 2
return min_b + (max_b - min_b) * t
# === Mood Animation Buffer (for complex animations) ===
mood_buffer = [(0, 0, 0)] * MOOD_LEDS
def mood_buffer_show():
"""Send entire mood buffer to Pico."""
with serial_lock:
try:
if ser is None or not ser.is_open:
if not init_serial():
return
for i, (r, g, b) in enumerate(mood_buffer):
ser.write(f"1 {i} {r} {g} {b}\n".encode())
ser.readline()
ser.write("1 -1\n".encode())
ser.readline()
except Exception as e:
print(f"[LIGHT] Buffer show error: {e}")
def mood_buffer_fill(color):
"""Fill buffer with solid color."""
for i in range(MOOD_LEDS):
mood_buffer[i] = color
def mood_buffer_set(index, color):
"""Set single buffer pixel."""
if 0 <= index < MOOD_LEDS:
mood_buffer[index] = color
# === Main Animation Loop ===
def main(): def main():
global current_state, jaw_level, running global running
print("[LIGHT] Vixy Light Service (Pico Edition) starting... 🦊") print("[LIGHT] Vixy Light Service starting...")
if not init_serial(): if not init_serial():
print("[LIGHT] Warning: Could not connect to Pico, will retry...") print("[LIGHT] Warning: Could not connect to Pico, will retry...")
# Start HTTP server thread # Set initial mood on Pico
http_thread = threading.Thread(target=run_http_server, daemon=True) send_mode_command("mood idle")
http_thread.start()
# Animation state server = HTTPServer(('0.0.0.0', HTTP_PORT), LightAPIHandler)
larson_pos = 0 print(f"[LIGHT] HTTP API listening on port {HTTP_PORT}")
larson_dir = 1 print("[LIGHT] Service ready")
wave_offset = 0
last_jaw_level = -1
print("[LIGHT] Service ready 🦊")
try: try:
while running: while running:
with state_lock: server.handle_request()
state = current_state
jaw = jaw_level
color = STATE_COLORS.get(state, (255, 255, 255))
# Update jaw if changed
if jaw != last_jaw_level:
jaw_set_level(jaw, (0, 200, 255)) # Cyan jaw
last_jaw_level = jaw
if state == "idle":
brightness = solid_pulse(0.5, 0.2, 0.8)
c = scale_color(color, brightness)
mood_fill(*c)
time.sleep(0.05)
elif state == "listening":
brightness = solid_pulse(2.0, 0.6, 1.0)
c = scale_color(color, brightness)
mood_fill(*c)
time.sleep(0.05)
elif state == "responding":
# Traveling wave
for i in range(MOOD_LEDS):
wave = (math.sin(2 * math.pi * (i - wave_offset) / 15) + 1) / 2
mood_buffer[i] = scale_color(color, wave * 0.8 + 0.2)
mood_buffer_show()
wave_offset += 0.5
time.sleep(0.03)
elif state == "pleasure":
brightness = solid_pulse(0.8, 0.3, 0.9)
c = scale_color(color, brightness)
mood_fill(*c)
time.sleep(0.05)
elif state == "thinking":
# Larson scanner
mood_buffer_fill((0, 0, 0))
if 0 <= larson_pos < MOOD_LEDS:
mood_buffer[larson_pos] = color
# Trail
for t in range(1, 9):
idx = larson_pos - larson_dir * t
if 0 <= idx < MOOD_LEDS:
fade = math.cos((t / 8) * math.pi / 2)
mood_buffer[idx] = scale_color(color, fade)
mood_buffer_show()
larson_pos += larson_dir
if larson_pos >= MOOD_LEDS - 1 or larson_pos <= 0:
larson_dir *= -1
time.sleep(0.04)
elif state == "playful":
# Sparkle effect
for i in range(MOOD_LEDS):
mood_buffer[i] = scale_color(mood_buffer[i], 0.85)
for i in range(MOOD_LEDS):
if random.random() < 0.15:
mood_buffer[i] = color
mood_buffer_show()
time.sleep(0.05)
elif state == "commanding":
brightness = solid_pulse(3.0, 0.5, 1.0)
c = scale_color(color, brightness)
mood_fill(*c)
time.sleep(0.03)
elif state == "love":
brightness = solid_pulse(0.6, 0.4, 0.9)
c = scale_color(color, brightness)
mood_fill(*c)
time.sleep(0.05)
elif state == "sleep":
brightness = solid_pulse(0.2, 0.1, 0.3)
c = scale_color(color, brightness)
mood_fill(*c)
time.sleep(0.1)
else:
mood_fill(50, 50, 50)
time.sleep(0.1)
finally: finally:
print("[LIGHT] Cleaning up LEDs...") print("[LIGHT] Cleaning up...")
mood_clear() send_mode_command("mood off")
jaw_clear() send_mode_command("jaw off")
if ser: if ser:
ser.close() ser.close()
print("[LIGHT] Shutdown complete") print("[LIGHT] Shutdown complete")