Usage examples

This page walks you through the typical workflow for integrating with a CAME Domotic server: connecting, discovering what the server offers, fetching and controlling devices, and monitoring real-time changes. For a minimal “hello world” example, see Getting started.

Note

The examples below assume the library is installed (pip install aiocamedomotic). Unless otherwise noted, all code runs inside an async with block:

import asyncio

from aiocamedomotic import CameDomoticAPI
from aiocamedomotic.models import (
    AnalogSensorType, DeviceType, DigitalInputStatus, LightStatus,
    LightType, OpeningStatus, RelayStatus, ScenarioStatus,
    ServerFeature, ThermoZoneFanSpeed,
    ThermoZoneMode, ThermoZoneSeason, ThermoZoneStatus,
    Timer, TimerTimeSlot, TimerUpdate,
    DeviceUpdate, LightUpdate, OpeningUpdate, RelayUpdate,
    ThermoZoneUpdate, ScenarioUpdate, DigitalInputUpdate, PlantUpdate,
)
from aiocamedomotic.errors import (
    CameDomoticError,
    CameDomoticServerNotFoundError,
    CameDomoticAuthError,
    CameDomoticServerTimeoutError,
    CameDomoticServerError,
)

async with await CameDomoticAPI.async_create(
    "192.168.x.x", "username", "password"
) as api:
    ...

Connecting to the server

Creating the API client

Create a CameDomoticAPI instance with the async factory method. The async with statement ensures resources are cleaned up automatically:

import asyncio
from aiocamedomotic import CameDomoticAPI

async def main():
    async with await CameDomoticAPI.async_create(
        "192.168.x.x", "username", "password"
    ) as api:
        server_info = await api.async_get_server_info()
        print(f"Connected to server: {server_info.keycode}")

asyncio.run(main())

Note

The session is not authenticated at creation time. The library authenticates lazily on the first real API call, like async_get_server_info(). If the credentials are invalid, a CameDomoticAuthError will be raised at that point.

Using an existing HTTP session

If you already have an aiohttp.ClientSession (e.g. in Home Assistant), pass it via the websession parameter:

async with await CameDomoticAPI.async_create(
    "192.168.x.x", "username", "password",
    websession=my_existing_session
) as api:
    ...

Handling connection errors

The library raises specific exceptions for different failure scenarios:

from aiocamedomotic import CameDomoticAPI
from aiocamedomotic.errors import (
    CameDomoticServerNotFoundError,
    CameDomoticAuthError,
    CameDomoticServerTimeoutError,
    CameDomoticServerError,
)

async def main():
    try:
        async with await CameDomoticAPI.async_create(
            "192.168.x.x", "username", "password"
        ) as api:
            lights = await api.async_get_lights()
    except CameDomoticServerNotFoundError:
        print("Server not reachable. Check the IP address.")
    except CameDomoticAuthError:
        print("Authentication failed. Check your credentials.")
    except CameDomoticServerTimeoutError:
        print("Request timed out. The server may be busy, retry later.")
    except CameDomoticServerError as err:
        print(f"Server error: {err}")

The exception hierarchy is:

• CameDomoticError — base class

  • CameDomoticServerNotFoundError — host unreachable (transient)

  • CameDomoticAuthError — bad credentials or too many sessions

  • CameDomoticServerError — other server errors

    • CameDomoticServerTimeoutError — request timeout (transient, retryable)

Server configuration

Server information

Retrieve the server properties with async_get_server_info(). The keycode property serves as a unique identifier for the server:

server_info = await api.async_get_server_info()

print(f"Keycode: {server_info.keycode}")
print(f"Software version: {server_info.swver}")
print(f"Server type: {server_info.type}")
print(f"Board type: {server_info.board}")
print(f"Serial number: {server_info.serial}")

Example output:

Keycode: 0000FFFF9999AAAA
Software version: 1.2.3
Server type: 0
Board type: 3
Serial number: 0011ffee

Connectivity check

Use async_ping() to verify the server is reachable and measure round-trip latency:

try:
    latency_ms = await api.async_ping()
    print(f"Server responded in {latency_ms:.1f} ms")
except CameDomoticServerNotFoundError:
    print("Server is unreachable")
except CameDomoticServerTimeoutError:
    print("Server timed out")

Available features

The features property on ServerInfo lists the capabilities configured on the server. These are the functional blocks you would see in the official CAME Domotic mobile app (lights, openings, scenarios, etc.):

server_info = await api.async_get_server_info()

for feature in server_info.features:
    print(f"Feature: {feature}")

Example output:

Feature: lights
Feature: openings
Feature: thermoregulation
Feature: scenarios
Feature: digitalin
Feature: analogin
Feature: energy
Feature: loadsctrl

The features property returns plain strings whose known values are defined in ServerFeature. You can compare entries against enum members to decide which device APIs to call:

from aiocamedomotic.models import ServerFeature

if ServerFeature.LIGHTS in server_info.features:
    lights = await api.async_get_lights()

if ServerFeature.OPENINGS in server_info.features:
    openings = await api.async_get_openings()

if ServerFeature.RELAYS in server_info.features:
    relays = await api.async_get_relays()

if ServerFeature.THERMOREGULATION in server_info.features:
    zones = await api.async_get_thermo_zones()
    sensors = await api.async_get_analog_sensors()

if ServerFeature.SCENARIOS in server_info.features:
    scenarios = await api.async_get_scenarios()

if ServerFeature.DIGITALIN in server_info.features:
    digital_inputs = await api.async_get_digital_inputs()

if ServerFeature.ANALOGIN in server_info.features:
    analog_inputs = await api.async_get_analog_inputs()

if ServerFeature.TIMERS in server_info.features:
    timers = await api.async_get_timers()

Floors and rooms

Retrieve the building topology to understand how devices are organized. async_get_topology() merges data from multiple server endpoints and nested device list commands, ensuring that floors and rooms are discovered even on servers where some endpoints return empty:

topology = await api.async_get_topology()

for floor in topology.floors:
    print(f"Floor {floor.id}: {floor.name}")
    for room in floor.rooms:
        print(f"  Room {room.id}: {room.name}")

Example output:

Floor 0: Ground Floor
  Room 1: Living Room
  Room 2: Kitchen
Floor 1: First Floor
  Room 3: Bedroom
  Room 4: Bathroom

Working with devices

All device types follow the same pattern: fetch the list, find a specific device, and control it. Lights are shown in full detail below; the other device types use the same approach with their own properties and methods.

Lights

Fetching and inspecting lights:

lights = await api.async_get_lights()

for light in lights:
    print(
        f"ID: {light.act_id}, Name: {light.name}, "
        f"Status: {light.status}, Type: {light.type}"
    )

Example output:

ID: 1, Name: Living Room Chandelier, Status: LightStatus.ON, Type: LightType.STEP_STEP
ID: 2, Name: Hallway Night Light, Status: LightStatus.OFF, Type: LightType.DIMMER
ID: 3, Name: RGB Strip, Status: LightStatus.ON, Type: LightType.RGB

Finding a specific light:

# By ID
chandelier = next((l for l in lights if l.act_id == 1), None)

# By name
hallway = next((l for l in lights if l.name == "Hallway Night Light"), None)

Controlling lights:

from aiocamedomotic.models import LightStatus

# Simple on/off (STEP_STEP lights)
if chandelier:
    await chandelier.async_set_status(LightStatus.ON)
    await chandelier.async_set_status(LightStatus.OFF)

# Dimmable lights: set brightness (0-100)
if hallway:
    await hallway.async_set_status(LightStatus.ON, brightness=50)
    await hallway.async_set_status(LightStatus.ON, brightness=100)

# RGB lights: set color as [R, G, B] (each 0-255)
rgb_strip = next((l for l in lights if l.type == LightType.RGB), None)
if rgb_strip:
    await rgb_strip.async_set_status(LightStatus.ON, rgb=[255, 0, 0])
    await rgb_strip.async_set_status(LightStatus.ON, brightness=75, rgb=[0, 128, 255])

Note

The brightness parameter is silently ignored for non-dimmable (STEP_STEP) lights. The rgb parameter is silently ignored for non-RGB lights.

Openings

Openings represent shutters, awnings, and similar motorized covers. They support opening, closing, stopping, and slat tilting (open/close) for covers with adjustable slats (e.g., venetian blinds).

import asyncio
from aiocamedomotic.models import OpeningStatus

openings = await api.async_get_openings()

for opening in openings:
    print(f"ID: {opening.open_act_id}, Name: {opening.name}, Status: {opening.status}")

# Control an opening
shutter = next((o for o in openings if o.open_act_id == 10), None)
if shutter:
    await shutter.async_set_status(OpeningStatus.OPENING)
    await asyncio.sleep(5)
    await shutter.async_set_status(OpeningStatus.STOPPED)
    await asyncio.sleep(5)
    await shutter.async_set_status(OpeningStatus.CLOSING)

    # Tilt slats (for covers with adjustable slats)
    await asyncio.sleep(5)
    await shutter.async_set_status(OpeningStatus.SLAT_OPEN)
    await asyncio.sleep(5)
    await shutter.async_set_status(OpeningStatus.SLAT_CLOSE)

Scenarios

Scenarios are pre-configured automation sequences. They can only be activated (fire-and-forget); there is no bidirectional status control.

scenarios = await api.async_get_scenarios()

for scenario in scenarios:
    print(f"ID: {scenario.id}, Name: {scenario.name}, Status: {scenario.scenario_status}")

# Activate a scenario
good_morning = next((s for s in scenarios if s.name == "Good morning"), None)
if good_morning:
    await good_morning.async_activate()

Thermoregulation zones

zones = await api.async_get_thermo_zones()

for zone in zones:
    print(
        f"ID: {zone.act_id}, Name: {zone.name}, "
        f"Temperature: {zone.temperature}°C, "
        f"Setpoint: {zone.set_point}°C, "
        f"Mode: {zone.mode}, Season: {zone.season}"
    )

Example output:

ID: 1, Name: Living Room, Temperature: 20.0°C, Setpoint: 21.5°C, Mode: ThermoZoneMode.AUTO, Season: ThermoZoneSeason.WINTER
ID: 52, Name: Bedroom, Temperature: 19.5°C, Setpoint: 20.0°C, Mode: ThermoZoneMode.MANUAL, Season: ThermoZoneSeason.WINTER

Controlling thermoregulation zones:

from aiocamedomotic.models import ThermoZoneFanSpeed, ThermoZoneMode, ThermoZoneSeason

# Set target temperature (keeps current mode).
# Only effective when the zone is in MANUAL mode; in AUTO or other modes
# the server silently discards the new setpoint without returning an error.
zone = zones[0]
await zone.async_set_temperature(22.0)

# To guarantee a setpoint change regardless of the current mode, switch to
# MANUAL and set the temperature in a single call:
await zone.async_set_config(mode=ThermoZoneMode.MANUAL, set_point=22.0)

# Change operating mode (keeps current temperature)
await zone.async_set_mode(ThermoZoneMode.MANUAL)

# Full configuration with fan speed
await zone.async_set_config(
    mode=ThermoZoneMode.MANUAL,
    set_point=21.5,
    fan_speed=ThermoZoneFanSpeed.MEDIUM,
)

# Set fan speed (keeps current mode and temperature)
await zone.async_set_fan_speed(ThermoZoneFanSpeed.SLOW)

# Change global season for all zones (plant-level command — season cannot
# be changed per zone)
await api.async_set_thermo_season(ThermoZoneSeason.WINTER)

Warning

Setting the season to PLANT_OFF forces all zones to OFF.

When the season is set to PLANT_OFF, the CAME server automatically switches every thermoregulation zone to ThermoZoneMode.OFF. Reverting the season back to WINTER or SUMMER does not restore the previous zone modes — each zone stays OFF until its mode is changed manually (e.g. via async_set_mode() or async_set_config()).

If your application needs to restore zone operation after re-enabling a season, you must track each zone’s previous mode yourself and re-apply it after changing the season.

Note

Temperature values are returned as floats in degrees Celsius.

The fan_speed parameter in async_set_config is optional; when provided, the extended_infos flag is set automatically.

Season can only be changed at the plant level via async_set_thermo_season().

Analog sensors

Analog sensors provide top-level readings (temperature, humidity, pressure) from the thermoregulation system. Each sensor carries an AnalogSensorType that identifies the kind of measurement it represents.

Fetching and inspecting sensors:

from aiocamedomotic.models import AnalogSensorType

sensors = await api.async_get_analog_sensors()

for sensor in sensors:
    print(
        f"Name: {sensor.name}, Type: {sensor.sensor_type}, "
        f"Value: {sensor.value}, Unit: {sensor.unit}"
    )

Example output:

Name: Outdoor Temperature, Type: AnalogSensorType.TEMPERATURE, Value: 21.5, Unit: C
Name: Indoor Humidity, Type: AnalogSensorType.HUMIDITY, Value: 55, Unit: %
Name: Barometric Pressure, Type: AnalogSensorType.PRESSURE, Value: 1013, Unit: hPa

Filtering by sensor type:

# Get only temperature sensors
temp_sensors = [
    s for s in sensors if s.sensor_type == AnalogSensorType.TEMPERATURE
]
for s in temp_sensors:
    print(f"{s.name}: {s.value}°{s.unit}")

# Find a specific sensor by ID
outdoor = next((s for s in sensors if s.act_id == 100), None)

Digital inputs (binary sensors)

Digital inputs are read-only binary sensors such as physical buttons or contact sensors. They report their state (ACTIVE/IDLE) but cannot be controlled remotely. ACTIVE means the input is triggered (e.g. a button is being pressed); IDLE means the input is in its normal resting state.

from aiocamedomotic.models import DigitalInputStatus

digital_inputs = await api.async_get_digital_inputs()

for di in digital_inputs:
    print(
        f"ID: {di.act_id}, Name: {di.name}, "
        f"Status: {di.status}, Address: {di.addr}"
    )

Example output:

ID: 0, Name: digitalin_PvGCT, Status: DigitalInputStatus.UNKNOWN, Address: 200
ID: 1, Name: digitalin_BuTbB, Status: DigitalInputStatus.IDLE, Address: 201

Finding a specific digital input:

# By ID
button = next((di for di in digital_inputs if di.act_id == 1), None)

# By name
sensor = next((di for di in digital_inputs if di.name == "Front door button"), None)

Note

Some digital inputs do not report a status until their first state change. In that case, status returns DigitalInputStatus.UNKNOWN.

Analog inputs (standalone sensors)

Analog inputs are read-only standalone sensors exposed via the analogin feature. They provide a numeric reading and a unit of measurement and cannot be controlled remotely.

Note

These sensors are independent of the thermoregulation system’s AnalogSensor. The same physical sensor may appear in both endpoints.

if ServerFeature.ANALOGIN in server_info.features:
    analog_inputs = await api.async_get_analog_inputs()

    for ai in analog_inputs:
        print(
            f"ID: {ai.act_id}, Name: {ai.name}, "
            f"Value: {ai.value}, Unit: {ai.unit}"
        )

Example output:

ID: 89, Name: Hygrometer, Value: 47.0, Unit: %
ID: 90, Name: Outdoor Thermometer, Value: 21.5, Unit: C
ID: 91, Name: Barometer, Value: 1013.0, Unit: hPa

Finding a specific analog input:

# By ID
thermo = next((ai for ai in analog_inputs if ai.act_id == 90), None)

# By name
hygro = next((ai for ai in analog_inputs if ai.name == "Hygrometer"), None)

Relays

Relays are simple on/off switches that can be controlled remotely.

Fetching and inspecting relays:

from aiocamedomotic.models import RelayStatus

relays = await api.async_get_relays()

for relay in relays:
    print(f"ID: {relay.act_id}, Name: {relay.name}, Status: {relay.status}")

Example output:

ID: 31, Name: Garden Pump, Status: RelayStatus.ON
ID: 32, Name: Gate Motor, Status: RelayStatus.OFF

Finding a specific relay:

# By ID
pump = next((r for r in relays if r.act_id == 31), None)

# By name
gate = next((r for r in relays if r.name == "Gate Motor"), None)

Controlling relays:

if pump:
    await pump.async_set_status(RelayStatus.ON)
    await asyncio.sleep(5)
    await pump.async_set_status(RelayStatus.OFF)

Timers

Timers are scheduling entities that define time-based activation windows for associated devices. Each timer has an enabled/disabled state, a day-of-week schedule, and up to 4 time slots. Timers support remote control: you can enable/disable them, toggle individual days, and configure the timetable.

Fetching and inspecting timers:

timers = await api.async_get_timers()

for timer in timers:
    print(
        f"ID: {timer.id}, Name: {timer.name}, "
        f"Enabled: {timer.enabled}, "
        f"Days: {timer.active_days}"
    )

    for slot in timer.timetable:
        print(
            f"  Slot {slot.index}: "
            f"start={slot.start_hour:02d}:{slot.start_min:02d}:{slot.start_sec:02d}"
        )

Example output:

ID: 163, Name: Test timer, Enabled: True, Days: ['Monday', 'Wednesday', 'Friday']
  Slot 0: start=10:00:00
ID: 164, Name: Timer 2, Enabled: True, Days: ['Tuesday', 'Thursday', 'Sunday']
  Slot 1: start=12:00:00
  Slot 2: start=11:00:00

Finding a specific timer:

# By ID
my_timer = next((t for t in timers if t.id == 163), None)

# By name
irrigation = next((t for t in timers if t.name == "Irrigation"), None)

Understanding the days bitmask

The days property is a 7-bit integer bitmask where each bit represents a day of the week. Bit 0 is Monday, bit 6 is Sunday:

Bit:   6    5    4    3    2    1    0
Day:  Sun  Sat  Fri  Thu  Wed  Tue  Mon

Common values:

• 1 — Monday only

• 15 — Monday through Thursday (1+2+4+8)

• 31 — Monday through Friday (weekdays)

• 96 — Saturday and Sunday (weekend)

• 127 — every day

The active_days property returns a human-readable list, and is_active_on_day() checks a specific day:

timer = timers[0]

print(timer.days)                    # 21
print(timer.active_days)             # ['Monday', 'Wednesday', 'Friday']
print(timer.is_active_on_day(0))     # True  (Monday)
print(timer.is_active_on_day(1))     # False (Tuesday)

Understanding the timetable

Each timer has up to 4 time slots (indices 0–3). The timetable property returns a list of TimerTimeSlot objects for the slots that are currently configured. Empty slots are simply absent from the list.

Each TimerTimeSlot exposes:

• index — the slot position (0–3)

• start_hour, start_min, start_sec — the activation start time

• stop_hour, stop_min, stop_sec — the stop time (None on some firmware versions)

• active — whether the slot is individually active (None on some firmware versions)

for slot in timer.timetable:
    start = f"{slot.start_hour:02d}:{slot.start_min:02d}:{slot.start_sec:02d}"
    if slot.stop_hour is not None:
        stop = f"{slot.stop_hour:02d}:{slot.stop_min:02d}:{slot.stop_sec:02d}"
    else:
        stop = "N/A"
    print(f"  Slot {slot.index}: {start} → {stop}")

Example output:

Slot 0: 10:00:00 → 18:30:00
Slot 2: 22:00:00 → N/A

Note

The stop and active fields mayb be not present in the server response. The corresponding properties return None in that case. Your code should handle both cases.

Enabling and disabling timers

Toggle a timer’s global enabled state:

timer = timers[0]

# Disable the timer
await timer.async_disable()
print(timer.enabled)  # False

# Re-enable the timer
await timer.async_enable()
print(timer.enabled)  # True

Toggling days of the week

Add or remove individual days from the timer’s schedule. The day parameter is a zero-based index: 0 = Monday, 6 = Sunday.

# Enable Sunday (day index 6)
await timer.async_enable_day(6)
print(timer.active_days)  # [..., 'Sunday']

# Disable Friday (day index 4)
await timer.async_disable_day(4)
print(timer.is_active_on_day(4))  # False

Setting the timetable

Use async_set_timetable() to configure all 4 time slots at once. Pass a list of exactly 4 entries — each is either a (hour, minute, second) tuple for an active slot, or None for an empty slot:

# Set slot 0 to 06:30:00, slot 2 to 22:00:00, leave slots 1 and 3 empty
await timer.async_set_timetable([
    (6, 30, 0),     # slot 0
    None,           # slot 1 (empty)
    (22, 0, 0),     # slot 2
    None,           # slot 3 (empty)
])

# Verify
for slot in timer.timetable:
    print(f"  Slot {slot.index}: {slot.start_hour:02d}:{slot.start_min:02d}")

# Clear all slots
await timer.async_set_timetable([None, None, None, None])

Important

async_set_timetable() always sends all 4 slots to the server. To keep existing slots unchanged, read the current timetable first and merge your changes:

# Read current slots into a 4-element list
current: list[tuple[int, int, int] | None] = [None, None, None, None]
for slot in timer.timetable:
    current[slot.index] = (slot.start_hour, slot.start_min, slot.start_sec)

# Modify only slot 3
current[3] = (14, 30, 0)

# Send the merged timetable
await timer.async_set_timetable(current)

Complete timer example

This example fetches timers, prints their configuration, toggles the enabled state, adds a day, sets a time slot, and reverts everything:

import asyncio
from aiocamedomotic import CameDomoticAPI

async def main():
    async with await CameDomoticAPI.async_create(
        "192.168.x.x", "username", "password"
    ) as api:
        timers = await api.async_get_timers()
        if not timers:
            print("No timers found")
            return

        timer = timers[0]
        print(f"Timer: {timer.name} (ID: {timer.id})")
        print(f"  Enabled: {timer.enabled}")
        print(f"  Days: {timer.active_days}")
        print(f"  Slots: {len(timer.timetable)}")

        # Save original state
        was_enabled = timer.enabled
        had_sunday = timer.is_active_on_day(6)

        # Toggle enabled
        if was_enabled:
            await timer.async_disable()
        else:
            await timer.async_enable()
        print(f"  Enabled toggled to: {timer.enabled}")

        # Toggle Sunday
        if had_sunday:
            await timer.async_disable_day(6)
        else:
            await timer.async_enable_day(6)
        print(f"  Days now: {timer.active_days}")

        # Add a time slot
        current: list[tuple[int, int, int] | None] = [None] * 4
        for slot in timer.timetable:
            current[slot.index] = (
                slot.start_hour, slot.start_min, slot.start_sec
            )
        current[3] = (14, 30, 0)
        await timer.async_set_timetable(current)
        print(f"  Slots after adding slot 3: {len(timer.timetable)}")

        # Revert everything
        current[3] = None
        await timer.async_set_timetable(current)
        if had_sunday:
            await timer.async_enable_day(6)
        else:
            await timer.async_disable_day(6)
        if was_enabled:
            await timer.async_enable()
        else:
            await timer.async_disable()
        print("  Reverted to original state")

asyncio.run(main())

Monitoring real-time updates

The CAME Domotic server supports long polling for real-time status updates. When you call async_get_updates(), the request blocks on the server until one or more device state changes are detected, then returns an UpdateList containing all pending updates. This is the recommended mechanism for monitoring devices in real time without repeatedly fetching full device lists.

Basic polling loop

A typical polling loop continuously calls async_get_updates() and processes each batch of updates as it arrives:

import asyncio
from aiocamedomotic.errors import CameDomoticServerTimeoutError

async with await CameDomoticAPI.async_create(
    "192.168.x.x", "username", "password"
) as api:
    while True:
        try:
            updates = await api.async_get_updates(timeout=120)
        except CameDomoticServerTimeoutError:
            # Long poll timed out with no updates; simply retry
            continue

        for update in updates.get_typed_updates():
            print(f"[{update.device_type}] {update.name} (ID: {update.device_id})")

        # Brief pause to avoid tight looping on rapid server responses
        await asyncio.sleep(1)

Example output:

[DeviceType.LIGHT] Living Room Chandelier (ID: 1)
[DeviceType.OPENING] Bedroom Shutter (ID: 10)
[DeviceType.THERMO_ZONE] Living Room (ID: 1)

Note

The asyncio.sleep(1) acts as a safety throttle in case the server returns immediately (e.g. errors or rapid update bursts). If you need to run the polling loop alongside other logic, wrap it in asyncio.create_task().

Configuring timeouts

All API methods use a default timeout of 30 seconds, configurable via the command_timeout parameter when creating the API instance:

async with await CameDomoticAPI.async_create(
    "192.168.x.x", "username", "password", command_timeout=15
) as api:
    lights = await api.async_get_lights()  # uses 15s timeout

async_get_updates() is the only method that accepts its own timeout parameter, allowing it to use a different timeout than the rest of the API. This is because async_get_updates() uses long polling — the server holds the connection open until updates are available, which can take much longer than a regular command round-trip. A longer timeout (e.g. 60–120 seconds) is strongly recommended to avoid premature disconnections:

# Instance-level timeout is 15s for regular commands,
# but async_get_updates uses its own 120s timeout
updates = await api.async_get_updates(timeout=120)

Note

If no timeout is passed to async_get_updates(), it falls back to the instance-level command_timeout (default: 30s). For most real-time monitoring use cases, a timeout of 60–120 seconds is recommended.

Typed updates and filtering

The UpdateList supports iteration over raw dicts for backward compatibility. For typed update objects with convenient properties, use get_typed_updates() or filter by device type with get_typed_by_device_type():

updates = await api.async_get_updates()

# Filter by device type
light_updates = updates.get_typed_by_device_type(DeviceType.LIGHT)
for light in light_updates:
    print(
        f"Light '{light.name}': status={light.status}, "
        f"type={light.light_type}, brightness={light.perc}%"
    )

# Dispatch by update type using isinstance
for update in updates.get_typed_updates():
    if isinstance(update, LightUpdate):
        print(f"Light '{update.name}': {update.status.name}, brightness={update.perc}%")

    elif isinstance(update, OpeningUpdate):
        print(f"Opening '{update.name}': {update.status.name}")

    elif isinstance(update, ThermoZoneUpdate):
        print(
            f"Thermo '{update.name}': {update.temperature}°C, "
            f"setpoint={update.set_point}°C, mode={update.mode.name}"
        )

    elif isinstance(update, ScenarioUpdate):
        print(f"Scenario '{update.name}': {update.scenario_status.name}")

    elif isinstance(update, DigitalInputUpdate):
        print(f"Input '{update.name}': status={update.status}, addr={update.addr}")

    elif isinstance(update, RelayUpdate):
        print(f"Relay '{update.name}': {update.status.name}")

    elif isinstance(update, TimerUpdate):
        print(
            f"Timer '{update.name}': enabled={update.enabled}, "
            f"days={update.days}, slots={len(update.timetable)}"
        )

    elif isinstance(update, PlantUpdate):
        print("Plant configuration changed, re-fetching devices...")

Handling plant updates

A plant_update_ind signals that the device configuration on the server has changed (e.g. devices were added, removed, or reconfigured). When this happens, all locally cached device lists must be discarded and re-fetched:

updates = await api.async_get_updates()

if updates.has_plant_update:
    lights = await api.async_get_lights()
    openings = await api.async_get_openings()
    relays = await api.async_get_relays()
    scenarios = await api.async_get_scenarios()
    timers = await api.async_get_timers()
    thermo_zones = await api.async_get_thermo_zones()
    sensors = await api.async_get_analog_sensors()
    digital_inputs = await api.async_get_digital_inputs()

Note

Plant updates are relatively rare. They typically occur when an installer modifies the system configuration. Failing to handle them may result in stale device data or missing newly added devices.

Timer status updates

When a timer is modified (enabled/disabled, day toggled, timetable changed), the server sends a timer_info_ind status update containing the full current state of the affected timer. This happens regardless of whether the change was made through this library, the CAME app, or the physical panel.

The update payload mirrors the timer list response — it includes name, id, enabled, days, bars (the number of timetable slots reported by the server), and the complete timetable array. The library parses this into a TimerUpdate object.

Applying timer updates to cached objects:

If you maintain a local cache of Timer objects, you can update them when a timer_info_ind arrives:

# Assume `timers_cache` is a dict mapping timer ID → Timer object
timers_cache = {t.id: t for t in await api.async_get_timers()}

while True:
    try:
        updates = await api.async_get_updates(timeout=120)
    except CameDomoticServerTimeoutError:
        continue

    for update in updates.get_typed_updates():
        if isinstance(update, TimerUpdate):
            cached = timers_cache.get(update.device_id)
            if cached:
                # Replace the raw_data with the fresh state from the
                # server — this updates all properties automatically
                cached.raw_data.update(update.raw_data)
                print(
                    f"Timer '{cached.name}' updated: "
                    f"enabled={cached.enabled}, "
                    f"days={cached.active_days}, "
                    f"slots={len(cached.timetable)}"
                )

    await asyncio.sleep(1)

Sequence of updates during a typical control session:

When you run a series of timer commands, the server sends one timer_info_ind for each change. For example, disabling a timer, then enabling Sunday, then adding a time slot, produces three consecutive updates:

timer_info_ind: enabled=0, days=15, timetable=[{index: 0, start: 10:00:00}]
timer_info_ind: enabled=1, days=79, timetable=[{index: 0, start: 10:00:00}]
timer_info_ind: enabled=1, days=79, timetable=[{index: 0, ...}, {index: 3, start: 14:30:00}]

Each update is a complete snapshot of the timer’s state — not a delta. You can safely overwrite the cached timer data with the update payload without needing to merge changes.

Note

The timer_info_ind indication name was confirmed from real server traffic (firmware 3.0.1). A legacy variant timer_update_ind is also handled for firmware compatibility.

Advanced topics

Checking authentication status

Session management is automatic and transparent. If you need to inspect the session status for any reason, use is_session_valid() on the auth attribute:

if api.auth.is_session_valid():
    print("Session is authenticated and valid.")
else:
    print("No valid session — it will be renewed automatically on the next call.")

Note

You rarely need to call this. The library handles reauthentication automatically whenever the session expires.

Device autodiscovery

The library exposes known MAC address OUI prefixes for CAME Domotic devices via CAME_MAC_PREFIXES. Combined with async_is_came_endpoint(), this allows identifying CAME ETI/Domo servers on the local network without credentials:

from aiocamedomotic import CAME_MAC_PREFIXES, async_is_came_endpoint

async def async_discover_came_server(host: str, mac_address: str) -> bool:
    """Check if a network device is a CAME Domotic server.

    Args:
        host: IP address or hostname of the device.
        mac_address: MAC address in colon-separated uppercase hex
            format, e.g. "00:1C:B2:AA:BB:CC".
    """
    # Step 1: Quick MAC prefix check (prefixes use "AA:BB:CC" format)
    mac_upper = mac_address.upper()
    if not any(mac_upper.startswith(prefix) for prefix in CAME_MAC_PREFIXES):
        return False

    # Step 2: Verify the device exposes a valid CAME API endpoint
    return await async_is_came_endpoint(host)

Note

CAME_MAC_PREFIXES contains prefixes in colon-separated uppercase hex format (e.g. "00:1C:B2"). Make sure the MAC address you pass uses the same format ("00:1C:B2:AA:BB:CC") before comparing. Other common representations such as 001cb2aabbcc, 00-1C-B2-AA-BB-CC, or lowercase 00:1c:b2:aa:bb:cc will not match directly — normalize to uppercase colon-separated format first, as shown in the example above.

Debugging with traffic logging

The library includes a built-in traffic logger that records every HTTP request and response exchanged with the CAME server. Sensitive data (passwords, session tokens, server identifiers) is automatically anonymized, so the output is safe to share publicly — for example, when reporting issues on GitHub.

The traffic logger uses the aiocamedomotic.traffic logger name (a child of the main aiocamedomotic logger). Since it is a child logger, it inherits the handler and formatter already configured by the library — you only need to lower its level to DEBUG:

import logging

# Enable traffic logging (anonymized request/response payloads)
logging.getLogger("aiocamedomotic.traffic").setLevel(logging.DEBUG)

That single line is all you need. The traffic log messages will appear alongside the normal library output, using the same format.

Example output:

2025-03-15 10:23:01.123 DEBUG (MainThread) [aiocamedomotic.traffic] HTTP POST http://192.168.1.***/domo/ [status=200, 42.5ms]
--> {"sl_cmd":"sl_registration_req","sl_login":"ad***","sl_pwd":"***"}
<-- {"sl_client_id":"504***","sl_keep_alive_timeout_sec":900,"sl_data_ack_reason":0}

The following fields are redacted automatically:

• sl_pwd, sl_new_pwd → fully replaced with ***

• sl_login → first 2 characters preserved (e.g. ad***)

• sl_client_id, client → first 3 characters preserved (e.g. 504***)

• keycode → first 8 characters preserved (e.g. 61305E97********)

• serial → first 3 characters preserved (e.g. 037***)

• Camera URIs (uri, uri_still) → embedded credentials redacted

• Host IP in the URL → last octet masked (e.g. 192.168.1.***)

• Usernames inside sl_users_list items → partially masked

Note

The traffic logger level is independent from the main library logger. Setting aiocamedomotic to WARNING and aiocamedomotic.traffic to DEBUG is a valid configuration — you will see only the HTTP traffic, without the library’s internal debug messages.

Tip

To write the traffic log to a file for later analysis, add a dedicated FileHandler. Use propagate = False if you want the traffic to go only to the file and not to the console:

import logging

traffic_logger = logging.getLogger("aiocamedomotic.traffic")
traffic_logger.setLevel(logging.DEBUG)
traffic_logger.addHandler(logging.FileHandler("came_traffic.log"))
traffic_logger.propagate = False  # file only, no console output

See also

For full API details, see the API Reference.