Devices

class aioafero.v1.controllers.device.DeviceController(*args, **kwargs)[source]

Bases: BaseResourcesController[Device]

Top-level / parent devices on bridge.devices (read-only).

ITEM_CLS

alias of Device

__init__(*args, **kwargs)[source]

Initialize instance.

async initialize_elem(afero_device: AferoDevice) → Device[source]

Initialize the element.

Parameters:

afero_device (AferoDevice) – Afero Device that contains the updated states

Returns:

Newly initialized resource

Return type:

Device

async initialize() → None[source]

Initialize controller by fetching all items for this resource type from bridge.

Return type:

None

get_filtered_devices(devices: list[AferoDevice]) → list[AferoDevice][source]

Find parent devices.

Parameters:

devices (list[AferoDevice])

Return type:

list[AferoDevice]

async update_elem(afero_device: AferoDevice) → set[source]

Update the Device with the latest API data.

Parameters:

afero_device (AferoDevice) – Afero Device that contains the updated states

Returns:

States that have been modified

Return type:

set

__contains__(device_id: str) → bool

Return bool if device_id is in items.

Parameters:

device_id (str)

Return type:

bool

__getitem__(device_id: str) → AferoResource

Get item by device_id.

Parameters:

device_id (str)

Return type:

AferoResource

__iter__() → Iterator

Iterate items.

Return type:

Iterator

async emit_to_subscribers(evt_type: EventType, item_id: str, item: AferoResource)

Emit updates to subscribers.

Parameters:

  • evt_type (EventType) – Type of event

  • item_id (str) – ID of the item

  • item (AferoResource) – Item to emit to subscribers

generate_update_dev(device_id: str, states: list[AferoState]) → AferoDevice

Generate update data for the event controller.

Parameters:
Return type:

AferoDevice

get_device(device_id: str) → AferoResource

Return the current model for a tracked device.

Parameters:

device_id (str) – Device ID on this controller (split or parent metadevice ID).

Returns:

Resource model with parsed feature state.

Raises:

DeviceNotFound – If device_id is not tracked by this controller.

Return type:

AferoResource

async initialize_number(func_def: dict, state: AferoState) → tuple[tuple[str, str | None], NumbersFeature] | None

Initialize a number from the provided data.

Parameters:
Return type:

tuple[tuple[str, str | None], NumbersFeature] | None

async initialize_select(functions: list[dict], state: AferoState) → tuple[tuple[str, str | None], SelectFeature] | None

Initialize a select from the provided data.

Parameters:
Return type:

tuple[tuple[str, str | None], SelectFeature] | None

async initialize_sensor(state: AferoState, child_id: str) → AferoSensor | AferoBinarySensor | None

Initialize the sensor.

Parameters:

  • state (AferoState) – State to update

  • child_id (str) – device_id of the parent device

Return type:

AferoSensor | AferoBinarySensor | None

property initialized: bool

Determine if the controller has been initialized.

property items: list[AferoResource]

Return all items for this resource.

async split_sensor_data(state: AferoState) → tuple[Any, str | None]

Split the sensor value and return a tuple of the sensor value and key.

Parameters:

state (AferoState)

Return type:

tuple[Any, str | None]

subscribe(callback: Callable[[EventType, dict | None], None], id_filter: str | tuple[str] | None = None, event_filter: EventType | tuple[EventType] | None = None) → Callable

Subscribe to status changes for this resource type.

Parameters:

  • callback (Callable[[EventType, dict | None], None]) – callback function to call when an event emits.

  • id_filter (str | tuple[str] | None) – Optionally provide resource ID(s) to filter events for.

  • event_filter (EventType | tuple[EventType] | None) – Optionally provide EventType(s) as filter.

Returns:

function to unsubscribe.

Return type:

Callable

property subscribers: dict[str, list[tuple[Callable[[EventType, dict | None], None], tuple[EventType] | None]]]

Get all subscribers aligned to this controller.

async update(device_id: str, obj_in: AferoResource | None = None, states: list[dict] | None = None, send_duplicate_states: bool = False) → ClientResponse | None

Update Afero IoT with the new data.

Parameters:

  • device_id (str) – Afero IoT Device ID

  • obj_in (AferoResource | None) – Afero IoT Resource elements to change

  • states (list[dict] | None) – States to manually set

  • send_duplicate_states (bool) – Send all states, regardless if there’s been a change

Return type:

ClientResponse | None

async update_afero_api(device_id: str, states: list[dict]) → ClientResponse | bool

Update Afero IoT API with the new states.

Parameters:

  • device_id (str) – Afero IoT Device ID

  • states (list[dict]) – States to manually set

Returns:

Response if successful, False otherwise.

Return type:

ClientResponse | bool

async update_number(state: AferoState, cur_item: AferoResource) → str | None

Update the number if its tracked and a change has been detected.

Parameters:

  • state (AferoState) – State to update

  • cur_item (AferoResource) – Current item to update

Returns:

Identifier of the number that was updated or None

Return type:

str | None

async update_select(state: AferoState, cur_item: AferoResource) → str | None

Update the select if its tracked and a change has been detected.

Parameters:

  • state (AferoState) – State to update

  • cur_item (AferoResource) – Current item to update

Returns:

Identifier of the select that was updated or None

Return type:

str | None

async update_sensor(state: AferoState, cur_item: AferoResource) → str | None

Update the sensor if its tracked and a change has been detected.

Parameters:

  • state (AferoState) – State to update

  • cur_item (AferoResource) – Current item to update

Returns:

Identifier of the sensor that was updated or None

Return type:

str | None