Advanced Usage¶
This guide covers advanced lifx patterns and techniques for building robust LIFX integrations.
Table of Contents¶
- Discovery Methods
- Storing State
- Connection Management
- Concurrency Patterns
- Error Handling
- Device Capabilities
- Custom Effects
- Performance Optimisation
Discovery Methods¶
lifx-async provides two discovery methods with different trade-offs:
UDP Broadcast Discovery¶
The traditional discovery method broadcasts to all devices on the network:
from lifx import discover
async def broadcast_discovery():
async for device in discover(timeout=5.0):
async with device:
color, power, label = await device.get_color()
print(f"Found: {label} ({type(device).__name__})")
Characteristics:
- Sends 1 broadcast + N queries (one per device for type detection)
- Works on any local network
- May miss devices on other subnets
mDNS Discovery¶
mDNS discovery uses DNS-SD to find devices with a single multicast query:
from lifx import discover_mdns
async def mdns_discovery():
async for device in discover_mdns(timeout=5.0):
async with device:
color, power, label = await device.get_color()
print(f"Found: {label} ({type(device).__name__})")
Characteristics:
- Single network query (device type in TXT record)
- Faster discovery with immediate type detection
- Can work across subnets with an mDNS reflector
- Zero dependencies (uses Python stdlib only)
Low-Level mDNS API¶
For raw mDNS data without device instantiation:
from lifx import discover_lifx_services
async def raw_mdns_discovery():
async for record in discover_lifx_services(timeout=5.0):
print(f"Serial: {record.serial}")
print(f"IP: {record.ip}:{record.port}")
print(f"Product ID: {record.product_id}")
print(f"Firmware: {record.firmware}")
Opt-In State Fields on Discovered Devices¶
Discovery functions take no fetch_wifi_info or fetch_ambient_light
argument, so every discovered device starts with both flags off. Set the
matching property before entering the context manager: async with device
is what runs state initialisation, so a flag set inside the body arrives too
late for the first fetch.
from lifx import discover
async def discover_with_signal():
async for device in discover(timeout=5.0):
device.fetch_wifi_info = True # before `async with`
async with device:
print(f"{device.state.label}: {device.state.wifi_info.rssi}")
Setting it inside the body leaves state.wifi_info unpopulated until the next
refresh, which re-fetches the volatile state rather than just the signal:
async for device in discover(timeout=5.0):
async with device:
device.fetch_wifi_info = True
print(device.state.wifi_info.rssi) # None - initialisation already ran
# Re-fetches the volatile state: label, power, colour and the opt-in
# readings, plus zones or tiles. The semi-static fields (firmware
# versions, location, group) stay as initialisation left them.
await device.refresh_state()
print(device.state.wifi_info.rssi) # populated
# For a single reading, skip the flag entirely:
wifi_info = await device.get_wifi_info()
The same ordering applies to discover_mdns(), find_by_serial(),
find_by_ip() and find_by_label() — all of them return devices that have not
been entered yet — and to fetch_ambient_light on lights.
Choosing a Discovery Method¶
| Scenario | Recommended Method |
|---|---|
| General use | discover() or discover_mdns() |
| Fastest discovery | discover_mdns() |
| Cross-subnet (with reflector) | discover_mdns() |
| Maximum compatibility | discover() |
| Raw device data | discover_lifx_services() |
Storing State¶
Device properties return cached values that were last retrieved from the device.
lifx-async automatically populates initial state values when a device is used as an async context manager.
Understanding Stored State¶
All device state properties return cached values or None if not yet fetched:
from lifx import Device
async def check_stored_state():
async with await Device.connect("192.168.1.100") as light:
# Property returns cached value or None
label = light.label
if label:
print(f"Cached label: {label}")
else:
print("No cached label - fetching from device")
label = await light.get_label()
print(f"Label: {label}")
Fetching Fresh Data¶
Use the get_*() methods to always fetch from the device:
async def always_fresh():
async with await Device.connect("192.168.1.100") as light:
# Always fetches from device
# Note: get_color() returns a tuple of (color, power, label)
color, power, label = await light.get_color()
# Get other device info
version = await light.get_version()
# Some properties cache semi-static data
cached_label = light.label # Updated from get_color()
Working with Cached Data¶
Use cached values when available for semi-static data, always fetch volatile state:
async def use_cached_or_fetch():
async with await Device.connect("192.168.1.100") as light:
# Check if we have cached label (semi-static)
label = light.label
if label:
print(f"Using cached label: {label}")
else:
print("No cached label, fetching from device")
label = await light.get_label()
print(f"Fetched label: {label}")
# For volatile state (power, color), always fetch fresh data
# get_color() will only cache the label
color, power, label = await light.get_color()
print(f"Current state of {light.label} - Power: {power}, Color: {color}")
Available Properties¶
Device Properties¶
Device.label- Device name/labelDevice.version- Vendor ID and Product IDDevice.host_firmware- Major and minor host firmware version and build numberDevice.wifi_firmware- Major and minor wifi firmware version and build numberDevice.location- Device location name/labelDevice.group- Device group name/label
Non-State Properties¶
Device.model- Device product model
Light properties¶
Non-State Properties¶
Light.min_kelvin- Lowest supported kelvin valueLight.max_kelvin- Highest supported kelvin value
InfraredLight properties¶
InfraredLight.infrared- Infrared brightness
HevLight properties:¶
HevLight.hev_config- HEV configurationHevLight.hev_result- Last HEV result
MultiZoneLight properties:¶
MultiZoneLight.zone_count- Number of zonesMultiZoneLight.multizone_effect- Either MOVE or OFF
MatrixLight properties:¶
MatrixLight.tile_count- Number of tiles on the chainMatrixLight.device_chain- Details of each tile on the chainMatrixLight.tile_effect- Either MORPH, FLAME, SKY or OFF
CeilingLight properties:¶
CeilingLight.uplight_zone- Zone index of the uplight componentCeilingLight.downlight_zones- Slice representing downlight zonesCeilingLight.uplight_is_on- True if uplight has brightness > 0 (requires recent data)CeilingLight.downlight_is_on- True if any downlight zone has brightness > 0 (requires recent data)
Note: Volatile state properties (power, color, hev_cycle, zones, tile_colors) have been removed. Always use get_*() methods to fetch these values from devices as they change too frequently to benefit from caching.
All cached properties return None if no data has been cached yet, or the cached value if available.
Connection Management¶
Understanding Lazy Connections¶
Each device owns its own connection that opens lazily on first request:
from lifx import Device
async def main():
async with await Device.connect("192.168.1.100") as light:
# Connection opens automatically on first request
await light.set_power(True)
# All subsequent operations reuse the same connection
await light.set_color(Colors.BLUE)
await light.get_label()
# Connection automatically closed when exiting context
Benefits:
- Simple lifecycle: one connection per device
- Lazy opening: connection opens only when needed
- Automatic cleanup on context exit
- Concurrent requests are supported: responses are correlated to each request, so they never mix
Concurrency Patterns¶
Concurrent Requests (Single Device)¶
Send multiple requests concurrently to one device:
import asyncio
from lifx import Device
async def concurrent_operations():
async with await Device.connect("192.168.1.100") as light:
# These execute concurrently!
# get_color() returns (color, power, label)
(color, power, label), version = await asyncio.gather(
light.get_color(),
light.get_version(),
)
print(f"{label}: Power={power}, Color={color}, Firmware={version.firmware}")
Performance Note: Concurrent requests execute with maximum parallelism. However, per the LIFX protocol specification, devices can handle approximately 20 messages per second. When sending many concurrent requests to a single device, consider implementing rate limiting in your application to avoid overwhelming the device.
Multi-Device Control¶
Control multiple devices in parallel:
import asyncio
from lifx import discover, DeviceGroup, Colors
async def multi_device_control():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
# Create different tasks for different devices
tasks = [
group.devices[0].set_color(Colors.RED),
group.devices[1].set_color(Colors.GREEN),
group.devices[2].set_color(Colors.BLUE),
]
# Execute all at once
await asyncio.gather(*tasks)
There is no need to run discovery multiple times on large or lossy networks: a
single discover_devices() call already re-broadcasts its discovery request on
an escalating schedule within the discovery window. If devices are still missed
on slow networks, increase the timeout argument instead.
Error Handling¶
Exception Hierarchy¶
from lifx import (
LifxError, # Base exception
LifxTimeoutError, # Request timeout
LifxConnectionError, # Connection failed
LifxProtocolError, # Invalid protocol response
LifxDeviceNotFoundError,# Device not discovered
LifxNetworkError, # Network issues
LifxUnsupportedCommandError, # Device doesn't support operation
)
Robust Error Handling¶
The library already retransmits each request on an escalating schedule within that request's timeout, so transient packet loss is handled for you. An application-level wrapper like the one below is for retrying whole operations that have genuinely failed (for example, a device that was briefly offline) — not for per-packet reliability.
import asyncio
from lifx import Colors, Device, LifxTimeoutError, LifxConnectionError
async def resilient_control():
max_retries = 3
for attempt in range(max_retries):
try:
async with await Device.connect("192.168.1.100") as light:
await light.set_color(Colors.BLUE)
print("Success!")
return
except LifxTimeoutError:
print(f"Timeout (attempt {attempt + 1}/{max_retries})")
if attempt < max_retries - 1:
await asyncio.sleep(1.0) # Wait before retry
except LifxConnectionError as e:
print(f"Connection failed: {e}")
break # Don't retry connection errors
print("All retries exhausted")
Graceful Degradation¶
from lifx import discover, DeviceGroup, Colors, LifxError
async def best_effort_control():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
results = []
# Try to control all lights, continue on errors
for light in group.lights:
try:
await light.set_color(Colors.GREEN)
results.append((light, "success"))
except LifxError as e:
results.append((light, f"failed: {e}"))
# Report results
for light, status in results:
label = await light.get_label() if status == "success" else "Unknown"
print(f"{label}: {status}")
Device Capabilities¶
Detecting Capabilities¶
Light capabilities are automatically populated:
from lifx import Colors, Device
from lifx.products.registry import ProductCapability
async def check_capabilities():
async with await Device.connect("192.168.1.100") as light:
print(f"Product: {light.model}")
print(f"Capabilities: {light.capabilities}")
# Check specific capabilities
if light.capabilities and light.capabilities.has_capability(ProductCapability.COLOR):
await light.set_color(Colors.BLUE)
if light.capabilities and light.capabilities.has_capability(ProductCapability.MULTIZONE):
print("This is a multizone device!")
if light.capabilities and light.capabilities.has_capability(ProductCapability.INFRARED):
print("Supports infrared!")
Capability-Based Logic¶
from lifx import discover, DeviceGroup, Colors
from lifx.products.registry import ProductCapability
async def capability_aware_control():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
for device in group.devices:
# Color devices
if device.capabilities and device.capabilities.has_capability(ProductCapability.COLOR):
await device.set_color(Colors.PURPLE)
# Multizone devices
if device.capabilities and device.capabilities.has_capability(ProductCapability.MULTIZONE):
await device.set_color_zones(0, 8, Colors.RED)
Custom Effects¶
Creating Smooth Transitions¶
import asyncio
from lifx import HSBK, Device
async def smooth_color_cycle():
async with await Device.connect("192.168.1.100") as light:
hues = [0, 60, 120, 180, 240, 300, 360]
for hue in hues:
color = HSBK(hue=hue, saturation=1.0, brightness=1.0, kelvin=3500)
await light.set_color(color, duration=2.0) # 2 second transition
await asyncio.sleep(2.0)
Synchronised Multi-Device Effects¶
import asyncio
from lifx import discover, DeviceGroup, Colors
async def synchronized_flash():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
# Flash all devices simultaneously
for _ in range(5):
await group.set_color(Colors.RED, duration=0.0)
await asyncio.sleep(0.2)
await group.set_color(Colors.OFF, duration=0.0)
await asyncio.sleep(0.2)
Wave Effect Across Devices¶
import asyncio
from lifx import discover, DeviceGroup, Colors
async def wave_effect():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
# Each device changes colour with a delay
tasks = [
delayed_color_change(device, Colors.BLUE, delay=i * 0.3)
for i, device in enumerate(group.devices)
]
await asyncio.gather(*tasks)
async def delayed_color_change(device, color, delay):
await asyncio.sleep(delay)
await device.set_color(color, duration=1.0)
Performance Optimisation¶
Minimize Network Requests¶
# ❌ Inefficient: Multiple round-trips
async def inefficient():
async with await Device.connect("192.168.1.100") as light:
await light.set_power(True)
await asyncio.sleep(0.1)
await light.set_color(Colors.BLUE)
await asyncio.sleep(0.1)
await light.set_brightness(0.8)
# ✅ Efficient: Set color and brightness together
async def efficient():
async with await Device.connect("192.168.1.100") as light:
await light.set_power(True)
# Set color includes brightness
color = HSBK(hue=240, saturation=1.0, brightness=0.8, kelvin=3500)
await light.set_color(color, duration=0.0)
Batch Operations¶
from lifx import discover, DeviceGroup, Colors
# ❌ Sequential: Takes N * latency
async def sequential():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
for device in group.devices:
await device.set_color(Colors.GREEN)
# ✅ Parallel: Takes ~latency
async def parallel():
devices = []
async for device in discover():
devices.append(device)
group = DeviceGroup(devices)
await group.set_color(Colors.GREEN)
Connection Reuse¶
# ❌ Creates new connection each time
async def no_reuse():
for _ in range(10):
async with await Device.connect("192.168.1.100") as light:
await light.set_brightness(0.5)
# Connection closed here
# ✅ Reuses connection
async def with_reuse():
async with await Device.connect("192.168.1.100") as light:
for _ in range(10):
await light.set_brightness(0.5)
# Connection closed once at end
Fire-and-Forget Mode for Low-Latency One-Shots¶
For sustained streaming (animations, music sync, real-time visualisations), use the Animation layer instead — it paces frame delivery against device acknowledgements internally, so you get high frame rates without per-call latency. The fast=True parameter remains useful for occasional low-latency one-shot updates where no confirmation is needed:
import asyncio
from lifx import HSBK, Device, MultiZoneLight
async def rainbow_sweep():
async with await Device.connect("192.168.1.100") as light:
assert isinstance(light, MultiZoneLight)
zone_count = await light.get_zone_count()
# A short colour sweep at ~20 FPS
for offset in range(0, 360, 5):
colors = [
HSBK(hue=(i * 360 / zone_count + offset) % 360,
saturation=1.0, brightness=1.0, kelvin=3500)
for i in range(zone_count)
]
# Fire-and-forget: no waiting for response
await light.set_extended_color_zones(0, colors, fast=True)
await asyncio.sleep(0.05) # ~20 FPS
When to use fast=True:
- Occasional low-latency one-shot updates where confirmation is unnecessary
- Short bursts of rapid updates, like the sweep above
- Situations where a dropped update is harmless
For sustained high-frame-rate streaming, prefer the Animation Guide.
Trade-offs:
- No confirmation that the device received or applied the colours
- No error detection (timeouts, unsupported commands)
- Best for visual effects where occasional dropped frames are acceptable
Note: MatrixLight.set64() is already fire-and-forget by default, making it ideal for tile animations without any additional parameters.
Next Steps¶
- Troubleshooting Guide — Common issues and solutions
- Network Layer API — Low-level discovery, connection, and transport reference
- Protocol Reference — Protocol layer and packet structures
- API Reference — Complete API documentation