277 lines
8.6 KiB
Python
277 lines
8.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Celebright Holiday Lights Controller
|
||
|
|
Based on packet capture analysis of the Celebright protocol.
|
||
|
|
"""
|
||
|
|
import questionary
|
||
|
|
from questionary import Style, Choice
|
||
|
|
import socket
|
||
|
|
import json
|
||
|
|
import asyncio
|
||
|
|
import logging
|
||
|
|
import websockets
|
||
|
|
from typing import Optional, Dict, List, Self
|
||
|
|
import netifaces
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
DISCOVERY_PORT = 49999
|
||
|
|
DISCOVERY_TIMEOUT = 5.0
|
||
|
|
DISCOVERY_MESSAGE = b'App Broadcast Message'
|
||
|
|
WEBSOCKET_ENDPOINT = "/ws"
|
||
|
|
|
||
|
|
log = logging.getLogger("CelebrightController")
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class CelebrightPreset:
|
||
|
|
preset_uuid: str
|
||
|
|
preset_name: str
|
||
|
|
preset_desc: str
|
||
|
|
preset_mode: int
|
||
|
|
brightness: int
|
||
|
|
web_id: int
|
||
|
|
md5: str
|
||
|
|
controller: "CelebrightController"
|
||
|
|
|
||
|
|
async def select(self) -> None:
|
||
|
|
"""Select/load this preset on the controller."""
|
||
|
|
await self.controller.load_preset(self)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class CelebrightController:
|
||
|
|
"""Controller for Celebright holiday lights system."""
|
||
|
|
address: str
|
||
|
|
model_name: str
|
||
|
|
device_id: str
|
||
|
|
firmware: str
|
||
|
|
hardware: int
|
||
|
|
num_leds: int
|
||
|
|
location_id: int
|
||
|
|
timezoneId: int
|
||
|
|
led_strings: list[int]
|
||
|
|
isRGBW: bool
|
||
|
|
colorOrder: str
|
||
|
|
RGBW_type: str
|
||
|
|
bulbType: str
|
||
|
|
_ws: Optional[websockets.ClientConnection] = None
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _discover_broadcast_addresses(cls) -> list[str]:
|
||
|
|
""" Get list of broadcast addresses for all network interfaces. """
|
||
|
|
broadcast_addrs = list()
|
||
|
|
for iface in netifaces.interfaces():
|
||
|
|
addrs = netifaces.ifaddresses(iface)
|
||
|
|
if netifaces.AF_INET in addrs:
|
||
|
|
for addr in addrs[netifaces.AF_INET]:
|
||
|
|
if 'broadcast' in addr:
|
||
|
|
broadcast_addrs.append(addr['broadcast'])
|
||
|
|
broadcast_addrs = sorted(set(broadcast_addrs))
|
||
|
|
log.debug("Discovered System Broadcast addresses: %s", broadcast_addrs)
|
||
|
|
return broadcast_addrs
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _discover_controller(cls, bcast: str) -> Self | None:
|
||
|
|
"""
|
||
|
|
Discover Celebright controller on the network by sending UDP broadcast.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
broadcast_addr: Broadcast address to use (default: 255.255.255.255)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
A list of dictionary containing device information
|
||
|
|
"""
|
||
|
|
|
||
|
|
log.debug("Discovering Celebright controller on: %s", bcast)
|
||
|
|
|
||
|
|
# Create socket for sending broadcast
|
||
|
|
soc = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
|
|
soc.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
||
|
|
soc.bind(('', DISCOVERY_PORT))
|
||
|
|
|
||
|
|
# Send broadcast discovery request ("App Broadcast Message" payload)
|
||
|
|
soc.sendto(DISCOVERY_MESSAGE, (bcast, DISCOVERY_PORT))
|
||
|
|
log.debug(f"Sent Discovery Broadcast: {bcast}:{DISCOVERY_PORT}")
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Wait for response from controller
|
||
|
|
while True:
|
||
|
|
try:
|
||
|
|
data, addr = soc.recvfrom(4096)
|
||
|
|
log.debug(f"Received response from {addr}: {data}")
|
||
|
|
if data:
|
||
|
|
data = json.loads(data.decode('utf-8'))
|
||
|
|
if "model_name" in data:
|
||
|
|
log.info("Discovered Celebright Controller at %s",
|
||
|
|
addr[0])
|
||
|
|
return cls(address=addr[0], **data)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
log.debug("Failed to decode JSON from response: %s", data)
|
||
|
|
|
||
|
|
except socket.timeout:
|
||
|
|
log.debug(f"Controller Timeout on %s", bcast)
|
||
|
|
except Exception as e:
|
||
|
|
log.warning(f"Unhandled Discovery Error: {e}")
|
||
|
|
|
||
|
|
finally:
|
||
|
|
soc.close()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _discover_controllers(cls) -> List[Self]:
|
||
|
|
"""
|
||
|
|
Discover Celebright controllers on the network by sending UDP broadcast.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
A list of dictionary containing device information
|
||
|
|
"""
|
||
|
|
|
||
|
|
controllers = list()
|
||
|
|
for address in cls._discover_broadcast_addresses():
|
||
|
|
result = cls._discover_controller(address)
|
||
|
|
if result:
|
||
|
|
controllers.append(result)
|
||
|
|
|
||
|
|
return controllers
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_discovery(cls) -> list[Self]:
|
||
|
|
"""
|
||
|
|
Factory method to create CelebrightController instances by
|
||
|
|
discovering controllers on the network.
|
||
|
|
"""
|
||
|
|
return cls._discover_controllers()
|
||
|
|
|
||
|
|
async def connect(self) -> None:
|
||
|
|
"""
|
||
|
|
Establish WebSocket connection to the controller.
|
||
|
|
"""
|
||
|
|
ws_url = f"ws://{self.address}{WEBSOCKET_ENDPOINT}"
|
||
|
|
log.debug(f"establishing websocket to {ws_url}...")
|
||
|
|
|
||
|
|
self._ws = await websockets.connect(ws_url)
|
||
|
|
log.info("websocket connection established: %s", ws_url)
|
||
|
|
|
||
|
|
async def send_command(self, topic: str, message: Dict = None) -> None:
|
||
|
|
"""
|
||
|
|
Send a command to the controller via WebSocket.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
topic: Command topic (e.g., "getPresetsAndEventsPaginated")
|
||
|
|
message: Command parameters dictionary (default: {})
|
||
|
|
"""
|
||
|
|
if not self._ws:
|
||
|
|
await self.connect()
|
||
|
|
|
||
|
|
if message is None:
|
||
|
|
message = {}
|
||
|
|
|
||
|
|
command = {"topic": topic, "message": message}
|
||
|
|
|
||
|
|
log.debug(f"sending to '%s': %s", topic, message)
|
||
|
|
await self._ws.send(json.dumps(command))
|
||
|
|
|
||
|
|
async def receive_response(self) -> Optional[Dict]:
|
||
|
|
"""
|
||
|
|
Receive a response from the controller.
|
||
|
|
"""
|
||
|
|
if not self._ws:
|
||
|
|
await self.connect()
|
||
|
|
|
||
|
|
response = await self._ws.recv()
|
||
|
|
data = json.loads(response)
|
||
|
|
log.debug(f"Received Response topic: {data.get('topic')}")
|
||
|
|
return data
|
||
|
|
|
||
|
|
async def get_presets(self) -> list[CelebrightPreset]:
|
||
|
|
"""
|
||
|
|
Get list of available scene presets.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
List of preset dictionaries, or None if error
|
||
|
|
"""
|
||
|
|
log.debug("Requesting preset list...")
|
||
|
|
await self.send_command("getPresetsAndEventsPaginated")
|
||
|
|
presets = list()
|
||
|
|
response = await self.receive_response()
|
||
|
|
if response and response.get('topic') == 'presetsPage':
|
||
|
|
for preset in response.get('message', {}).get('presets', []):
|
||
|
|
presets.append(CelebrightPreset(**preset, controller=self))
|
||
|
|
return presets
|
||
|
|
|
||
|
|
async def load_preset(self, preset: CelebrightPreset) -> None:
|
||
|
|
"""
|
||
|
|
Load/activate a specific scene preset.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
preset_uuid: UUID of the preset to load
|
||
|
|
"""
|
||
|
|
log.info("Loading preset: %s", preset.preset_name)
|
||
|
|
await self.send_command(
|
||
|
|
"loadPreset",
|
||
|
|
{"presetUuid": preset.preset_uuid}
|
||
|
|
)
|
||
|
|
log.info("Setting 12-hour sleep timer...")
|
||
|
|
await self.send_command(
|
||
|
|
"setSleepTimer",
|
||
|
|
{"timerMinutes": 720}
|
||
|
|
)
|
||
|
|
|
||
|
|
async def turn_off(self) -> None:
|
||
|
|
"""Turn off the lights and disable schedule."""
|
||
|
|
log.info("Turning off lights and disabling schedule...")
|
||
|
|
await self.send_command("setTurnOffAndDisableSchedule")
|
||
|
|
|
||
|
|
async def disconnect(self) -> None:
|
||
|
|
"""Close the WebSocket connection."""
|
||
|
|
if self._ws:
|
||
|
|
log.debug("Closing WebSocket connection...")
|
||
|
|
await self._ws.close()
|
||
|
|
self._ws = None
|
||
|
|
log.debug("WebSocket connection closed.")
|
||
|
|
|
||
|
|
async def interact(self):
|
||
|
|
"""Interactive menu for selecting presets or turning off lights."""
|
||
|
|
custom_style = Style([('selected', 'fg:ansicyan bold')])
|
||
|
|
presets = await self.get_presets()
|
||
|
|
choices = [
|
||
|
|
Choice(title=preset.preset_name, value=preset)
|
||
|
|
for preset in presets
|
||
|
|
]
|
||
|
|
choices.append(
|
||
|
|
Choice(title="Turn Off Lights", value="__TURN_OFF__")
|
||
|
|
)
|
||
|
|
choices.append(
|
||
|
|
Choice(title="Exit", value="__EXIT__")
|
||
|
|
)
|
||
|
|
|
||
|
|
while True:
|
||
|
|
selected = await questionary.select(
|
||
|
|
"Select a preset or action:",
|
||
|
|
choices=choices,
|
||
|
|
style=custom_style,
|
||
|
|
use_indicator=True,
|
||
|
|
use_shortcuts=False
|
||
|
|
).ask_async()
|
||
|
|
|
||
|
|
if selected is None:
|
||
|
|
break
|
||
|
|
if selected == "__TURN_OFF__":
|
||
|
|
await self.turn_off()
|
||
|
|
return
|
||
|
|
elif selected == "__EXIT__":
|
||
|
|
return
|
||
|
|
else:
|
||
|
|
preset = selected
|
||
|
|
await preset.select()
|
||
|
|
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
"""Main function demonstrating the Celebright controller usage."""
|
||
|
|
controllers = CelebrightController.from_discovery()
|
||
|
|
controller = controllers[0]
|
||
|
|
await controller.interact()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|