API reference

Client

The usual entry point to the library is the create_client() function that creates and sets up a Client instance.

class pyupgw.Client(aws: AwsApi)

Unisenza Plus client

The recommended way to start a client session is with create_client() context manager. Alternatively, create_api() can be used to create the AWS API instance manually before creating the client.

The newly created Client object doesn’t initially know about any devices. populate_devices() needs to be called to fetch them from the server.

Parameters:: aws – The AWS API object user to access the backend service. The authentication needs to be performed before creating the Client object.

async aclose()

Release all resources acquired by the client

When the Client object is used as context manager, this is automatically called on exit.

get_devices() → Iterable[tuple[Gateway, Device]]

Iterate over devices

Returns:: An iterable of tuples (gateway, device) where gateway is the gateway device is connected to. Gateways are also returned as (gateway, gateway).

get_gateways() → list[Gateway]: Get the managed gateways

async populate_devices()

Populate devices from the server

Raises:: ClientError – if populating device data fails

async refresh_all_devices()

Refresh states of all managed devices

This is a convenience function that calls refresh_device_state() for all devices known to the client.

Raises:: ClientError – if the request to get device state fails

async refresh_device_state(gateway: Gateway, device: Device)

Refresh state of a device from the server

The coroutine completes when the server has responded with the new state.

Parameters:

gateway – the gateway the device is connected to
device – the device whose state is refreshed

Raises:

ClientError – if the request to get device state fails

async update_device_state(gateway: Gateway, device: Device, changes: Mapping[str, Any])

Update the state of a device managed by the client

This method will send the changed values to the server. The coroutine will complete after the server has accepted the request, but not yet necessarily applied the changes.

The update itself is asynchronous, and the in-memory attributes of the device models will only be updated after the server has acknowledged that it has applied the changes. Device.subscribe() can be used to subscribe to the updates of the in-memory model.

Parameters:

gateway – the gateway the device is connected to
device – the device whose state is updated
changes – the changes to be published

Raises:

ClientError – if the request to update device state fails

async pyupgw.create_api(username: str, password: str) → AwsApi

Create authenticated AWS API instance

The instance can be used to create Client instance. For a single step creation, use create_client() function instead.

Parameters:

username – the credentials used to log into the cloud service
password – the credentials used to log into the cloud service

Raises:

AuthenticationError – if authentication fails

pyupgw.create_client(username: str, password: str)

Create Unisenza Plus client

This function returns a context manager that initializes and manages resources for a Client instance. It also takes care of authenticating and populating devices from the server.

async with create_client("user@example.com", "password") as client:
   ...  # use client

Parameters:

username – the credentials used to log into the cloud service
password – the credentials used to log into the cloud service

Models

class pyupgw.Device(attributes: AttributesType, dispatch_refresh: Callable[[Device], Awaitable[None]], dispatch_update: Callable[[Device, Mapping[str, Any]], Awaitable[None]])

Bases: Generic[AttributesType]

A managed device

A device object is a dynamic handle for the device data managed by a client. It responds to direct and indirect state changes both from the client and other sources. The instantaneous state of the device can be retrieved using the get_attributes() method.

Parameters:

attributes – the initial attributes
dispatch_refresh – dispatch request to refresh the state of this device
dispatch_update – dispatch request to update the state of this device

get_attributes() → AttributesType: Get the attributes associated with the device

get_device_code() → str: Get device code

get_euid() → str | None: Get device code

get_firmware_version() → str | None: Get firmware version

get_id() → UUID: Get device id

get_model() → str: Get device code

get_name() → str: Get device code

get_type() → DeviceType: Get device type

is_available() → bool

Return True if the device is available, False otherwise

A device is available if the cloud connection is established, and the upstream reports that the device is available.

async refresh()

Refresh the state of the device

This call will delegate to Client.refresh_device_state()

set_attributes(changes: Mapping[str, Any])

Set new value for attributes

Note that this method only changes the attributes in-memory, and the changes are not dispatched to the cloud service.

Parameters:: changes – dictionary containing new values for the attributes

subscribe(callback: Callable[[Device, Mapping[str, Any]], None])

Register a callback to be notified when the state of a device changes

Parameters:: callback – the callback that will be called with the updated device and dictionary of the changed attributes

unsubscribe(callback: Callable[[Device, Mapping[str, Any]], None])

Remove a callback previously registered with subscribe()

Parameters:: callback – the callback to be removed from subscribers

async update(changes: Mapping[str, Any])

Update the state of the device

This call will delegate to Client.update_device_state()

class pyupgw.DeviceAttributes(*, id: UUID, type: DeviceType, device_code: str, model: str, name: str, euid: str | None = None, firmware_version: str | None = None, available: bool = False)

Common device attributes

available: bool: Connection to the device is established

device_code: str: Device code

euid: str | None: EUID

firmware_version: str | None: Firmware version

id: UUID: Device id assigned by the cloud service

model: str: Device model

name: str: Device name

type: DeviceType: Device type

class pyupgw.DeviceType(value)

Enumeration indicating the type of the device

GATEWAY = 'gateway'

HVAC = 'hvac'

class pyupgw.Gateway(attributes: GatewayAttributes, children: Iterable[HvacAttributes], dispatch_refresh: Callable[[Gateway, Device], Awaitable[None]], dispatch_update: Callable[[Gateway, Device, Mapping[str, Any]], Awaitable[None]])

Bases: Device[GatewayAttributes]

A gateway acting between Unisenza IoT devices and the cloud service

Parameters:

attributes – the gateway attributes
children – initial attributes of the children managed by the gateway
dispatch_refresh – dispatch request to refresh the state of this device
dispatch_update – dispatch request to update the state of this device

get_children() → list[HvacDevice]: Get the children of this device

get_ip_address() → str | None: Get the IP address of the gateway

get_mac_address() → str | None: Get the MAC address of the gateway

get_occupant() → Occupant: Get the occupant of the gateway

class pyupgw.GatewayAttributes(type: Literal[DeviceType.GATEWAY], occupant: Occupant, ip_address: str | None = None, mac_address: str | None = None, *, id: UUID, device_code: str, model: str, name: str, euid: str | None = None, firmware_version: str | None = None, available: bool = False)

Bases: DeviceAttributes

Gateway attributes

ip_address: str | None: IP address of the gateway

mac_address: str | None: MAC address of the gateway

occupant: Occupant: Gateway occupant

type: Literal[DeviceType.GATEWAY]: Device type

class pyupgw.HvacAttributes(type: Literal[DeviceType.HVAC], manufacturer: str | None = None, serial_number: str | None = None, system_mode: SystemMode | None = None, running_state: RunningState | None = None, target_temperature: float | None = None, current_temperature: float | None = None, min_temp: float | None = None, max_temp: float | None = None, *, id: UUID, device_code: str, model: str, name: str, euid: str | None = None, firmware_version: str | None = None, available: bool = False)

Bases: DeviceAttributes

Attributes of a HVAC device

current_temperature: float | None: The current temperature as measured by the device

manufacturer: str | None: Device manufacturer

max_temp: float | None: Maximum setpoint temperature

min_temp: float | None: Minimum setpoint temperature

running_state: RunningState | None: The running state (action) of the device

serial_number: str | None: Serial number

system_mode: SystemMode | None: The system mode (state) of the device

target_temperature: float | None: The setpoint temperature

type: Literal[DeviceType.HVAC]: Device type

class pyupgw.HvacDevice(attributes: AttributesType, dispatch_refresh: Callable[[Device], Awaitable[None]], dispatch_update: Callable[[Device, Mapping[str, Any]], Awaitable[None]])

Bases: Device[HvacAttributes]

A HVAC device (smart thermostat)

get_current_temperature() → float | None: Get the current temperature as measured by the device

get_manufacturer() → str | None: Get serial number

get_max_temp() → float | None: Get the maximum setpoint temperature

get_min_temp() → float | None: Get the minimum setpoint temperature

get_running_state() → RunningState | None: Get action the device is performing

get_serial_number() → str | None: Get serial number

get_system_mode() → SystemMode | None: Get mode of the device

get_target_temperature() → float | None: Get the setpoint temperature

async update_system_mode(system_mode: SystemMode): Update the system mode

async update_target_temperature(target_temperature: float): Update the setpoint temperature

class pyupgw.Occupant(id: UUID, identity_id: str)

Occupant of a managed device

id: UUID: Occupant id

identity_id: str: Cognito identity of the occupant

class pyupgw.RunningState(value)

HVAC running state

HEATING = 1

IDLE = 0

class pyupgw.SystemMode(value)

HVAC system mode

HEAT = 4

OFF = 0

Exceptions

exception pyupgw.ClientError: Error in client operation

exception pyupgw.AuthenticationError: Error in authentication