Persistent Storage Guide¶

import asyncio

from lifx_emulator import EmulatedLifxServer, create_color_light
from lifx_emulator.devices import DevicePersistenceAsyncFile, DeviceManager
from lifx_emulator.repositories import DeviceRepository


async def main():
    # Create storage handler (uses ~/.lifx-emulator by default)
    storage = DevicePersistenceAsyncFile()

    # Create device with storage; saved state for this serial is restored
    device = create_color_light("d073d5000001", storage=storage)

    # Direct state assignments are not saved automatically, so queue a save
    device.state.label = "My Light"
    device.state.color.hue = 21845  # 120 degrees
    await storage.save_device_state(device.state)

    # Start server; packets that change state (SetColor, SetLabel, ...) are
    # saved automatically from here on
    server = EmulatedLifxServer(
        [device], DeviceManager(DeviceRepository()), "127.0.0.1", 56700
    )
    async with server:
        await asyncio.sleep(60)

    # Drain the device's pending saves and flush everything to disk
    await device.close()
    await storage.shutdown()


asyncio.run(main())

from lifx_emulator.devices import DevicePersistenceAsyncFile

# Use custom directory
storage = DevicePersistenceAsyncFile("/var/lib/lifx-emulator")

# Now state files will be stored in /var/lib/lifx-emulator/

import asyncio

from lifx_emulator import create_color_light
from lifx_emulator.devices import DevicePersistenceAsyncFile


async def main():
    # First session - create state and save it
    storage = DevicePersistenceAsyncFile()
    device1 = create_color_light("d073d5000001", storage=storage)
    device1.state.label = "Kitchen Light"
    device1.state.color.hue = 10923  # Orange
    await storage.save_device_state(device1.state)  # Queue async save
    await storage.shutdown()  # Flush pending writes to disk

    # Later session - state is restored
    storage = DevicePersistenceAsyncFile()
    device2 = create_color_light("d073d5000001", storage=storage)
    assert device2.state.label == "Kitchen Light"
    assert device2.state.color.hue == 10923


asyncio.run(main())

import asyncio

from lifx_emulator import create_color_light
from lifx_emulator.devices import DevicePersistenceAsyncFile
from lifx_emulator.protocol.header import LifxHeader
from lifx_emulator.protocol.packets import Device


async def main():
    storage = DevicePersistenceAsyncFile()
    device = create_color_light("d073d5000001", storage=storage)

    # A SetLabel packet changes state, so a save is queued automatically
    header = LifxHeader(
        source=1,
        target=device.state.get_target_bytes(),
        sequence=1,
        pkt_type=Device.SetLabel.PKT_TYPE,
    )
    device.process_packet(header, Device.SetLabel(label="Kitchen Light"))

    # Drain the device's pending saves, then flush storage to disk
    await device.close()
    await storage.shutdown()
    print(storage.load_device_state("d073d5000001")["label"])  # Kitchen Light


asyncio.run(main())

from lifx_emulator import create_color_light, create_multizone_light
from lifx_emulator.devices import DevicePersistenceAsyncFile

storage = DevicePersistenceAsyncFile()

# Create multiple devices - each maintains its own state file
devices = [
    create_color_light("d073d5000001", storage=storage),
    create_color_light("d073d5000002", storage=storage),
    create_multizone_light("d073d5800001", storage=storage),
]

# All state is independently persisted and restored

import asyncio

from lifx_emulator.devices import DevicePersistenceAsyncFile


async def main():
    storage = DevicePersistenceAsyncFile()

    # List all saved devices
    print(storage.list_devices())

    # Delete saved state for one device (async; True if a file was removed)
    removed = await storage.delete_device_state("d073d5000001")

    # Delete all saved state (synchronous; returns the number deleted)
    count = storage.delete_all_device_states()


asyncio.run(main())

import asyncio

from lifx_emulator import EmulatedLifxServer, create_color_light
from lifx_emulator.devices import DevicePersistenceAsyncFile, DeviceManager
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.scenarios import HierarchicalScenarioManager, ScenarioConfig


async def main():
    storage = DevicePersistenceAsyncFile()
    manager = HierarchicalScenarioManager()
    manager.set_device_scenario(
        "d073d5000001",
        ScenarioConfig(response_delays={107: 0.5}),  # Delay StateColor (107) replies by 500ms
    )

    device = create_color_light(
        "d073d5000001", storage=storage, scenario_manager=manager
    )
    server = EmulatedLifxServer(
        [device],
        DeviceManager(DeviceRepository()),
        "127.0.0.1",
        56700,
        scenario_manager=manager,
    )

    async with server:
        await asyncio.sleep(60)

    # Device state persists across restarts; the scenario does not
    await device.close()
    await storage.shutdown()


asyncio.run(main())

import asyncio

from lifx_emulator.scenarios import ScenarioConfig, ScenarioPersistenceAsyncFile


async def main():
    scenario_storage = ScenarioPersistenceAsyncFile()  # ~/.lifx-emulator

    # Load saved scenarios (an empty manager if none are saved)
    manager = await scenario_storage.load()
    manager.set_global_scenario(ScenarioConfig(drop_packets={101: 0.3}))

    # Write ~/.lifx-emulator/scenarios.json
    await scenario_storage.save(manager)


asyncio.run(main())

from lifx_emulator.devices import DevicePersistenceAsyncFile

storage = DevicePersistenceAsyncFile()
count = storage.delete_all_device_states()
print(f"Deleted {count} device states")