technicalkiwi/lighting-controller
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add segmented_movement pattern with n4 parameter support

Browse Source 
- Add n4 parameter to control server, LED bar receiver, and test script
- Create segmented_movement pattern with alternating forward/backward movement
- Pattern supports n1 (segment length), n2 (spacing), n3 (forward speed), n4 (backward speed)
- Fix test script to send all messages instead of just the first one
- Add segmented_movement to patterns_needing_params for proper parameter transmission
- Pattern intelligently handles all cases: alternating, forward-only, backward-only, or static
- Implements repeating segments with configurable spacing across LED strip
This commit is contained in:
Pi User
2025-10-03 19:54:43 +13:00
parent f4e9f8fff7
commit e4a83e8f0d
7 changed files with 485 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
src/control_server.py
View File

@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""
Control Server for Lighting Controller
Handles lighting control logic and communicates with LED bars via WebSocket.
Handles lighting control logic and communicates with LED bars via SPI or WebSocket.
Receives commands from UI client via WebSocket.
"""
@@ -12,9 +12,10 @@ import logging
import socket
import threading
import time
import argparse
from bar_config import LED_BAR_NAMES, DEFAULT_BAR_SETTINGS
from color_utils import adjust_brightness
from networking import WebSocketClient as SPIClient # SPI transport client
from networking import SPIClient, WebSocketClient
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -36,14 +37,24 @@ PATTERN_NAMES = {
"radiate": "rd",
"sequential_pulse": "sp",
"alternating_phase": "ap",
"segmented_movement": "sm",
}
class LEDController:
"""Handles communication with LED bars via SPI (through ESP32 relay)."""
"""Handles communication with LED bars via SPI or WebSocket."""
def __init__(self, spi_bus: int = 0, spi_device: int = 0, spi_speed_hz: int = 1_000_000):
self.client = SPIClient(bus=spi_bus, device=spi_device, speed_hz=spi_speed_hz)
def __init__(self, transport="spi", **kwargs):
if transport == "spi":
self.client = SPIClient(
bus=kwargs.get('spi_bus', 0),
device=kwargs.get('spi_device', 0),
speed_hz=kwargs.get('spi_speed_hz', 1_000_000)
)
elif transport == "websocket":
self.client = WebSocketClient(uri=kwargs.get('uri', 'ws://192.168.4.1/ws'))
else:
raise ValueError(f"Invalid transport: {transport}. Must be 'spi' or 'websocket'")
@property
def is_connected(self) -> bool:
@@ -84,9 +95,8 @@ class SoundController:
class LightingController:
"""Main lighting control logic."""
def __init__(self):
# SPI defaults: bus 0, CE0, 1MHz; adjust here if needed
self.led_controller = LEDController(spi_bus=0, spi_device=0, spi_speed_hz=1_000_000)
def __init__(self, transport="spi", **transport_kwargs):
self.led_controller = LEDController(transport=transport, **transport_kwargs)
self.sound_controller = SoundController(SOUND_CONTROL_HOST, SOUND_CONTROL_PORT)
# Lighting state
@@ -99,6 +109,7 @@ class LightingController:
self.n1 = 10
self.n2 = 10
self.n3 = 1
self.n4 = 1
self.beat_index = 0
self.beat_sending_enabled = True
@@ -126,6 +137,7 @@ class LightingController:
"n1": self.n1,
"n2": self.n2,
"n3": self.n3,
"n4": self.n4,
"s": self.beat_index % 256,
}
}
@@ -148,7 +160,7 @@ class LightingController:
async def _send_normal_pattern(self):
"""Send normal pattern to all bars."""
patterns_needing_params = ["alternating", "flicker", "n_chase", "rainbow", "radiate"]
patterns_needing_params = ["alternating", "flicker", "n_chase", "rainbow", "radiate", "segmented_movement"]
payload = {
"d": {
@@ -264,6 +276,8 @@ class LightingController:
self.n2 = data["n2"]
if "n3" in data:
self.n3 = data["n3"]
if "n4" in data:
self.n4 = data["n4"]
await self._request_param_update()
elif message_type == "delay_change":
@@ -281,10 +295,11 @@ class LightingController:
class ControlServer:
"""WebSocket server for UI client communication and TCP server for sound."""
def __init__(self):
self.lighting_controller = LightingController()
def __init__(self, transport="spi", enable_heartbeat=False, **transport_kwargs):
self.lighting_controller = LightingController(transport=transport, **transport_kwargs)
self.clients = set()
self.tcp_server = None
self.enable_heartbeat = enable_heartbeat
async def handle_ui_client(self, websocket):
"""Handle UI client WebSocket connection."""
@@ -366,12 +381,30 @@ class ControlServer:
# Connect to LED server
await self.lighting_controller.led_controller.connect()
# Start servers and heartbeat task
await asyncio.gather(
self.start_websocket_server(),
self.start_tcp_server(),
self._heartbeat_loop()
)
# Start servers (optionally include heartbeat)
websocket_task = asyncio.create_task(self._websocket_server_task())
tcp_task = asyncio.create_task(self._tcp_server_task())
tasks = [websocket_task, tcp_task]
if self.enable_heartbeat:
heartbeat_task = asyncio.create_task(self._heartbeat_loop())
tasks.append(heartbeat_task)
await asyncio.gather(*tasks)
async def _websocket_server_task(self):
"""Keep WebSocket server running."""
await self.start_websocket_server()
# Keep the server running indefinitely
while True:
await asyncio.sleep(1)
async def _tcp_server_task(self):
"""Keep TCP server running."""
await self.start_tcp_server()
# Keep the server running indefinitely
while True:
await asyncio.sleep(1)
async def _heartbeat_loop(self):
"""Send periodic heartbeats to keep LED connection alive."""
@@ -394,7 +427,25 @@ class ControlServer:
async def main():
"""Main entry point."""
server = ControlServer()
args = parse_arguments()
transport_kwargs = {}
if args.transport == "spi":
transport_kwargs = {
'spi_bus': args.spi_bus,
'spi_device': args.spi_device,
'spi_speed_hz': args.spi_speed
}
elif args.transport == "websocket":
transport_kwargs = {
'uri': args.uri
}
server = ControlServer(
transport=args.transport,
enable_heartbeat=args.enable_heartbeat,
**transport_kwargs
)
try:
await server.run()
@@ -406,5 +457,61 @@ async def main():
await server.lighting_controller.led_controller.close()
def parse_arguments():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description="Control Server for Lighting Controller")
# Transport selection
transport_group = parser.add_argument_group("Transport Options")
transport_group.add_argument(
"--transport",
choices=["spi", "websocket"],
default="spi",
help="Transport method for LED communication (default: spi)"
)
# Control options
control_group = parser.add_argument_group("Control Options")
control_group.add_argument(
"--enable-heartbeat",
action="store_true",
help="Enable heartbeat system (may cause pattern interruptions)"
)
# SPI options
spi_group = parser.add_argument_group("SPI Options")
spi_group.add_argument(
"--spi-bus",
type=int,
default=0,
help="SPI bus number (default: 0)"
)
spi_group.add_argument(
"--spi-device",
type=int,
default=0,
help="SPI device number (default: 0)"
)
spi_group.add_argument(
"--spi-speed",
type=int,
default=1_000_000,
help="SPI speed in Hz (default: 1000000)"
)
# WebSocket options
ws_group = parser.add_argument_group("WebSocket Options")
ws_group.add_argument(
"--uri",
type=str,
default="ws://192.168.4.1/ws",
help="WebSocket URI for LED communication (default: ws://192.168.4.1/ws)"
)
return parser.parse_args()
if __name__ == "__main__":
asyncio.run(main())
#

57
src/networking.py
View File

@@ -1,6 +1,8 @@
import json
import os
import time
import asyncio
import websockets
try:
import spidev
@@ -8,8 +10,9 @@ except Exception as e:
spidev = None
class WebSocketClient:
def __init__(self, uri=None, *, bus=None, device=None, speed_hz=None):
class SPIClient:
"""SPI transport client."""
def __init__(self, bus=None, device=None, speed_hz=None):
# SPI configuration (defaults can be overridden by args or env)
self.bus = 0 if bus is None else int(bus)
self.device = 0 if device is None else int(device)
@@ -77,3 +80,53 @@ class WebSocketClient:
pass
self.is_connected = False
self.spi = None
class WebSocketClient:
"""WebSocket transport client."""
def __init__(self, uri=None, *, bus=None, device=None, speed_hz=None):
self.uri = uri or "ws://192.168.4.1/ws"
self.websocket = None
self.is_connected = False
async def connect(self):
"""Initializes the WebSocket connection."""
if self.is_connected and self.websocket:
return
try:
self.websocket = await websockets.connect(self.uri)
self.is_connected = True
print(f"WebSocket connected: {self.uri}")
except Exception as e:
print(f"Error opening WebSocket: {e}")
self.is_connected = False
self.websocket = None
async def send_data(self, data):
"""Sends a JSON object over WebSocket."""
if not self.is_connected or not self.websocket:
await self.connect()
if not self.is_connected:
print("WebSocket not connected; cannot send")
return
try:
json_str = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
await self.websocket.send(json_str)
print(f"WebSocket sent: {json_str}")
except Exception as e:
print(f"WebSocket send failed: {e}")
# Attempt simple reopen on next call
self.is_connected = False
self.websocket = None
async def close(self):
"""Closes the WebSocket connection."""
try:
if self.websocket:
await self.websocket.close()
except Exception:
pass
self.is_connected = False
self.websocket = None