EmulatedLifxServer¶

The EmulatedLifxServer class manages the UDP server and routes packets to emulated devices.

Overview¶

The server:

Listens on a UDP socket for LIFX protocol packets
Parses packet headers to determine routing
Forwards packets to appropriate devices
Sends response packets back to clients
Supports both targeted and broadcast packets

API Reference¶

EmulatedLifxServer ¶

EmulatedLifxServer(
    devices: list[EmulatedLifxDevice],
    device_manager: IDeviceManager,
    bind_address: str = "127.0.0.1",
    port: int = LIFX_UDP_PORT,
    track_activity: bool = True,
    storage=None,
    activity_observer: ActivityObserver | None = None,
    scenario_manager: HierarchicalScenarioManager | None = None,
    persist_scenarios: bool = False,
    scenario_storage: IScenarioStorageBackend | None = None,
)

UDP server that simulates LIFX devices

METHOD	DESCRIPTION
`start`	Start the server
`stop`	Stop the server

Source code in packages/lifx-emulator-core/src/lifx_emulator/server.py

def __init__(
    self,
    devices: list[EmulatedLifxDevice],
    device_manager: IDeviceManager,
    bind_address: str = "127.0.0.1",
    port: int = LIFX_UDP_PORT,
    track_activity: bool = True,
    storage=None,
    activity_observer: ActivityObserver | None = None,
    scenario_manager: HierarchicalScenarioManager | None = None,
    persist_scenarios: bool = False,
    scenario_storage: IScenarioStorageBackend | None = None,
):
    # Device manager (required dependency injection)
    self._device_manager = device_manager
    self.bind_address = bind_address
    self.port = port
    self.transport = None
    self.storage = storage

    # Scenario storage backend (optional - only needed for persistence)
    self.scenario_persistence: IScenarioStorageBackend | None = None
    if persist_scenarios:
        if scenario_storage is None:
            raise ValueError(
                "scenario_storage is required when persist_scenarios=True"
            )
        if scenario_manager is None:
            raise ValueError(
                "scenario_manager is required when persist_scenarios=True "
                "(must be pre-loaded from storage before server initialization)"
            )
        self.scenario_persistence = scenario_storage

    # Scenario manager (shared across all devices for runtime updates)
    self.scenario_manager = scenario_manager or HierarchicalScenarioManager()

    # Add initial devices to the device manager
    for device in devices:
        # Update device port to match server port
        device.state.port = self.port
        self._device_manager.add_device(device, self.scenario_manager)

    # Activity observer - defaults to ActivityLogger if track_activity=True
    if activity_observer is not None:
        self.activity_observer = activity_observer
    elif track_activity:
        self.activity_observer = ActivityLogger(max_events=100)
    else:
        self.activity_observer = NullObserver()

    # Statistics tracking
    self.start_time = time.time()
    self.packets_received = 0
    self.packets_sent = 0
    self.packets_received_by_type: dict[int, int] = defaultdict(int)
    self.packets_sent_by_type: dict[int, int] = defaultdict(int)
    self.error_count = 0

Functions¶

start `async` ¶

start()

Start the server

Source code in packages/lifx-emulator-core/src/lifx_emulator/server.py

async def start(self):
    """Start the server"""
    loop = asyncio.get_running_loop()
    self.transport, _ = await loop.create_datagram_endpoint(
        lambda: self.LifxProtocol(self), local_addr=(self.bind_address, self.port)
    )

stop `async` ¶

stop()

Stop the server

Source code in packages/lifx-emulator-core/src/lifx_emulator/server.py

async def stop(self):
    """Stop the server"""
    if self.transport:
        self.transport.close()

Usage¶

Basic Server¶

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

# Create devices
devices = [create_color_light("d073d5000001")]

# Set up repository and manager (required)
device_repository = DeviceRepository()
device_manager = DeviceManager(device_repository)

# Create server
server = EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700)

# Start server
await server.start()

# ... do work ...

# Stop server
await server.stop()

Context Manager¶

The recommended way to use the server:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

devices = [create_color_light("d073d5000001")]
device_manager = DeviceManager(DeviceRepository())

async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    # Server automatically starts
    # Your test code here
    pass
# Server automatically stops

Multiple Devices¶

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import (
    create_color_light,
    create_multizone_light,
    create_tile_device,
)
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

devices = [
    create_color_light("d073d5000001"),
    create_multizone_light("d073d8000001", zone_count=16),
    create_tile_device("d073d9000001", tile_count=5),
]

device_manager = DeviceManager(DeviceRepository())

async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    # All devices are discoverable and controllable
    pass

Parameters¶

`devices`¶

List of EmulatedLifxDevice instances to emulate.

Type: list[EmulatedLifxDevice]

`device_manager`¶

The device manager handling device lifecycle and packet routing.

Type: DeviceManager

Required: Yes

Notes: - The server delegates all device management operations to this manager - Must be created with a DeviceRepository instance - This is a required parameter in v2.0.0+

Example:

from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

device_repository = DeviceRepository()
device_manager = DeviceManager(device_repository)

`bind_address`¶

IP address to bind the UDP server to.

Type: str

Examples: - "0.0.0.0" - Listen on all interfaces - "127.0.0.1" - Localhost only - "192.168.1.100" - Specific interface

`port`¶

UDP port to listen on.

Type: int

Default: 56700 (standard LIFX port)

`track_activity`¶

Enable packet activity tracking for the HTTP API dashboard.

Type: bool

Default: True

Notes: - When enabled, the server tracks recent packet activity for the API dashboard - Disable to reduce memory usage in production or CI environments - Activity tracking is independent of packet logging (controlled by --verbose CLI flag)

Example:

# Disable activity tracking
server = EmulatedLifxServer(
    devices,
    "127.0.0.1",
    56700,
    track_activity=False
)

`storage`¶

Optional persistent storage backend for device state.

Type: DevicePersistenceAsyncFile | None

Default: None

Notes: - When provided, device state changes are automatically saved asynchronously - Allows device state to persist across emulator restarts - Must be used with devices created with the same storage instance - See Persistent Storage Guide for details

Example:

from lifx_emulator.devices import DevicePersistenceAsyncFile
from lifx_emulator.factories import create_color_light

storage = DevicePersistenceAsyncFile()
device = create_color_light("d073d5000001", storage=storage)

device_manager = DeviceManager(DeviceRepository())

# Create server with storage support
server = EmulatedLifxServer(
    [device],
    device_manager,
    "127.0.0.1",
    56700,
    storage=storage
)

`activity_observer`¶

Optional observer for packet activity events.

Type: ActivityObserver | None

Default: None

Notes: - Implement ActivityObserver protocol to receive packet events - Useful for custom activity tracking or metrics collection - Receives events for all packets sent and received

`scenario_manager`¶

Optional scenario manager for test scenario configuration.

Type: HierarchicalScenarioManager | None

Default: None

Notes: - When provided, enables runtime scenario management via REST API - Supports device-specific, type-specific, location-based, group-based, and global scenarios - Scenarios control packet dropping, delays, malformed responses, etc. - See Testing Scenarios Guide for detailed examples

Example:

from lifx_emulator.scenarios import HierarchicalScenarioManager
from lifx_emulator.devices import DeviceManager
from lifx_emulator.repositories import DeviceRepository

manager = HierarchicalScenarioManager()
device_manager = DeviceManager(DeviceRepository())

server = EmulatedLifxServer(
    devices,
    device_manager,
    "127.0.0.1",
    56700,
    scenario_manager=manager
)

# Now scenario management API is available

`persist_scenarios`¶

Enable persistent storage of scenario configurations.

Type: bool

Default: False

Notes: - When enabled, scenario configurations are saved to ~/.lifx-emulator/scenarios.json - Scenarios are restored from disk on startup - Requires scenario_manager to be provided - Ignored if scenario_manager is None

Example:

# Enable both state and scenario persistence
device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(
    devices,
    device_manager,
    "127.0.0.1",
    56700,
    storage=storage,
    scenario_manager=manager,
    persist_scenarios=True
)

Methods¶

Lifecycle Methods¶

`async start()`¶

Start the UDP server and begin accepting connections.

Notes: - Call this method before sending any packets to the emulator - Binds to the configured address and port - Logs server startup information - Required if not using context manager

Example:

device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700)
await server.start()
try:
    # Use server
    pass
finally:
    await server.stop()

`async stop()`¶

Stop the UDP server and clean up resources.

Notes: - Gracefully closes the UDP endpoint - Cleans up internal state - Safe to call multiple times - Automatically called by context manager

Context Manager Protocol¶

The server implements the async context manager protocol for clean resource management:

`async aenter()`¶

Enter async context manager - automatically calls start().

`async aexit()`¶

Exit async context manager - automatically calls stop().

Recommended Usage:

device_manager = DeviceManager(DeviceRepository())
async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    # Server is automatically started
    # Perform your tests
    pass
# Server is automatically stopped

Utility Methods¶

`get_uptime_ns()`¶

Get the server uptime in nanoseconds since startup.

Returns: int - Nanoseconds elapsed since server started

Notes: - Returns 0 if server hasn't started yet - Useful for performance testing and benchmarking - Uses monotonic clock for accurate timing

Example:

device_manager = DeviceManager(DeviceRepository())
async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    await asyncio.sleep(1)
    uptime_ns = server.get_uptime_ns()
    uptime_ms = uptime_ns / 1_000_000
    print(f"Server uptime: {uptime_ms:.2f}ms")

`invalidate_scenario_cache()`¶

Clear the internal scenario precedence cache.

Notes: - Normally called automatically after scenario updates via API - Only needed if modifying scenario manager state outside of API - Safe to call - no side effects - Non-blocking operation

Example:

from lifx_emulator.scenarios import HierarchicalScenarioManager, ScenarioConfig

manager = HierarchicalScenarioManager()
device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(
    devices, device_manager, "127.0.0.1", 56700, scenario_manager=manager
)

async with server:
    # Update scenarios externally
    manager.set_global_scenario(ScenarioConfig(...))
    # Invalidate cache to apply changes immediately
    server.invalidate_scenario_cache()

Packet Routing¶

Broadcast Packets¶

Packets with tagged=True or target=000000000000 are forwarded to all devices:

# GetService broadcasts are answered by all devices
# Client discovers all emulated devices

Targeted Packets¶

Packets with a specific target serial are routed to that device:

# LightSetColor for d073d5000001 goes only to that device
# Other devices don't see the packet

Unknown Targets¶

Packets for unknown serial addresses are silently dropped:

# Packet for d073d5999999 (not in server) is ignored
# No error or response generated

Response Handling¶

The server handles responses automatically:

Device processes packet and returns response(s)
Server packs response packets to bytes
Server sends responses back to source address
Multiple responses (e.g., multizone StateMultiZone) are sent sequentially

Concurrency¶

The server uses asyncio for concurrent operation:

# Multiple clients can send packets concurrently
# Each device processes packets independently
# Responses are sent asynchronously

Error Handling¶

The server handles errors gracefully:

Invalid packets are logged and dropped
Device exceptions are caught and logged
Server continues running despite errors
Malformed headers don't crash the server

Lifecycle¶

Startup¶

Create UDP endpoint
Bind to address and port
Start receiving packets
Log server start

Runtime¶

Receive packet bytes
Parse header
Route to device(s)
Get responses
Send responses

Shutdown¶

Stop accepting packets
Close UDP endpoint
Clean up resources
Log server stop

Testing Integration¶

pytest-asyncio¶

import pytest
from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

@pytest.fixture
def device_manager():
    return DeviceManager(DeviceRepository())

@pytest.fixture
async def lifx_server(device_manager):
    device = create_color_light("d073d5000001")
    server = EmulatedLifxServer([device], device_manager, "127.0.0.1", 56700)

    async with server:
        yield server

@pytest.mark.asyncio
async def test_discovery(lifx_server):
    # Test code using the server
    pass

Module-Scoped Fixture¶

For faster tests, use module scope:

@pytest.fixture(scope="module")
def device_manager():
    return DeviceManager(DeviceRepository())

@pytest.fixture(scope="module")
async def lifx_server(device_manager):
    devices = [create_color_light(f"d073d500000{i}") for i in range(5)]
    server = EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700)

    async with server:
        yield server

Next Steps¶

Device API - EmulatedLifxDevice documentation
Factory Functions - Device creation
Integration Testing Tutorial - Integration test examples

EmulatedLifxServer¶

Overview¶

API Reference¶

EmulatedLifxServer ¶

Functions¶

start async ¶

stop async ¶

Usage¶

Basic Server¶

Context Manager¶

Multiple Devices¶

Parameters¶

devices¶

device_manager¶

bind_address¶

port¶

track_activity¶

storage¶

activity_observer¶

scenario_manager¶

persist_scenarios¶

Methods¶

Lifecycle Methods¶

async start()¶

async stop()¶

Context Manager Protocol¶

async __aenter__()¶

async __aexit__()¶

Utility Methods¶

get_uptime_ns()¶

invalidate_scenario_cache()¶

Packet Routing¶

Broadcast Packets¶

Targeted Packets¶

Unknown Targets¶

Response Handling¶

Concurrency¶

Error Handling¶

Lifecycle¶

Startup¶

Runtime¶

Shutdown¶

Testing Integration¶

pytest-asyncio¶

Module-Scoped Fixture¶

Next Steps¶

start `async` ¶

stop `async` ¶

`devices`¶

`device_manager`¶

`bind_address`¶

`port`¶

`track_activity`¶

`storage`¶

`activity_observer`¶

`scenario_manager`¶

`persist_scenarios`¶

`async start()`¶

`async stop()`¶

`async aenter()`¶

`async aexit()`¶

`get_uptime_ns()`¶

`invalidate_scenario_cache()`¶