device package

Submodules

device.crypto module

device.crypto.certificate_is_valid(crypto_store_dir: str) → bool

Utility function that checks the certificate validity

Parameters:

crypto_store_dir (str) – Path to the folder where crypto information are stored

Returns:

True if the certificate is valid, False otherwise.

Return type:

bool

device.crypto.device_has_certificate(crypto_store_dir: str) → bool

Utility function that checks if a certificate is present for the device

Parameters:

crypto_store_dir (str) – Path to the folder where crypto information is stored

Returns:

True if the certificate is present, False otherwise

Return type:

bool

device.crypto.generate_csr(realm: str, device_id: str, crypto_store_dir: str) → bytes

Utility function that generate the csr for the device

Parameters:

  • realm (str) – The Astarte realm where the device will be registered

  • device_id (str) – The device ID

  • crypto_store_dir (str) – Path to the folder where crypto information is stored

Returns:

The device certificate signing request file

Return type:

bytes

device.crypto.import_device_certificate(client_crt: str, crypto_store_dir: str) → None

device.device module

class device.device.Device(device_id: str, realm: str, credentials_secret: str, pairing_base_url: str, persistency_dir: str, loop: asyncio.AbstractEventLoop | None = None, ignore_ssl_errors: bool = False)

Bases: object

Basic class to define an Astarte Device.

Device represents an Astarte Device. It is the base class used for managing the Device lifecycle and data. Users should instantiate a Device with the right credentials and connect it to the configured instance to start working with it.

Threading and Concurrency

This SDK uses paho-mqtt under the hood to provide Transport connectivity. As such, it is bound by paho-mqtt’s behaviors in terms of threading. When a Device connects, a new thread is spawned and an event loop is run there to manage all the connection events.

This SDK spares the user from this detail - on the other hand, when configuring callbacks, threading has to be taken into account. When creating a Device, it is possible to specify an asyncio.loop() to automatically manage this detail. When a loop is specified, all callbacks will be called in the context of that loop, guaranteeing thread-safety and making sure that the user does not have to take any further action beyond consuming the callback.

When a loop is not specified, callbacks are invoked just as standard Python functions. This inevitably means that the user will have to take into account the fact that the callback will be invoked in the Thread of the MQTT connection. In particular, blocking the execution of that thread might cause deadlocks and, in general, malfunctions in the SDK. For this reason, the usage of asyncio is strongly recommended.

on_connected

A function that will be invoked everytime the device successfully connects.

Type:

Callable[[Device], None]

on_disconnected

A function that will be invoked everytime the device disconnects. The int parameter bears the disconnect reason.

Type:

Callable[[Device, int], None]

on_data_received

A function that will be invoked everytime data is received from Astarte. Parameters are the Device itself, the Interface name, the Interface path, and the payload. The payload will reflect the type defined in the Interface.

Type:

Callable[[Device, string, string, object], None]

add_interface(interface_definition: dict) → None

Adds an Interface to the Device

This will add an Interface definition to the Device. It has to be called before connect(), as it will be used for building the Device Introspection.

Parameters:

interface_definition (dict) – An Astarte Interface definition in the form of a Python dictionary. Usually obtained by using json.loads() on an Interface file.

add_interface_from_file(interface_file: Path)

Adds an interface to the device

This will add an interface definition to the device. It has to be called before connect(), as it will be used for building the device introspection.

Parameters:

interface_file (Path) – An absolute path to an Astarte interface json file.

Raises:

  • FileNotFoundError – If specified file does not exists.

  • TypeError – If speficied file is not a .json file.

add_interfaces_from_dir(interfaces_dir: Path)

Adds a series of interfaces to the device

This will add all the interfaces contained in the provided folder to the device. It has to be called before connect(), as it will be used for building the device introspection.

Parameters:

interfaces_dir (Path) – An absolute path to an a folder containing some Astarte interface .json files.

Raises:

  • FileNotFoundError – If specified directory does not exists.

  • NotADirectoryError – If speficied directory exists but it’s not a directory.

connect() → None

Connects the Device asynchronously.

When calling connect, a new connection thread is spawned and the Device will start a connection routine. The function might return before the Device connects: you want to use the on_connected callback to ensure you are notified upon connection.

In case the Device gets disconnected unexpectedly, it will try to reconnect indefinitely until disconnect() is called.

disconnect() → None

Disconnects the Device asynchronously.

When calling disconnect, the connection thread is requested to terminate the disconnection, and the thread is stopped when the disconnection happens. The function might return before the Device connects: you want to use the on_disconnected callback to ensure you are notified upon connection. When doing so, check the return code parameter: if it is 0, it means the disconnection happened following an explicit disconnection request.

get_device_id() → str

Returns the Device ID of the Device.

Returns:

The Id of the device

Return type:

str

is_connected() → bool

Returns whether the Device is currently connected.

Returns:

The device connection status.

Return type:

bool

remove_interface(interface_name: str) → None

Removes an Interface from the Device

Removes an Interface definition from the Device. It has to be called before connect(), as it will be used for building the Device Introspection.

Parameters:

interface_name (str) – The name of an Interface previously added with add_interface().

send(interface_name: str, interface_path: str, payload: object, timestamp: datetime | None = None) → None

Sends an individual message to an interface.

Parameters:

  • interface_name (str) – The name of an the Interface to send data to.

  • interface_path (str) – The path on the Interface to send data to.

  • payload (object) – The value to be sent. The type should be compatible to the one specified in the interface path.

  • timestamp (datetime, optional) – If sending a Datastream with explicit_timestamp, you can specify a datetime object which will be registered as the timestamp for the value.

Raises:

TypeError – If the interface or the payload are not compatible.

send_aggregate(interface_name: str, interface_path: str, payload: dict, timestamp: datetime | None = None) → None

Sends an aggregate message to an interface

Parameters:

  • interface_name (str) – The name of the Interface to send data to.

  • interface_path (str) – The endpoint to send the data to

  • payload (dict) – A dictionary containing the path:value map for the aggregate.

  • timestamp (datetime, optional) – If the Datastream has explicit_timestamp, you can specify a datetime object which will be registered as the timestamp for the value.

Raises:

TypeError – If the interface or the payload are not compatible.

unset_property(interface_name: str, interface_path: str) → None

Unset the specified property on an interface.

Parameters:

  • interface_name (str) – The name of the Interface where the property to unset is located.

  • interface_path (str) – The path on the Interface to unset.

Raises:

TypeError – If the interface is of type datastream.

device.exceptions module

exception device.exceptions.APIError(body)

Bases: AstarteError

Exception raised when Astarte APIs throw unhandled errors.

body -- the body of the API reply, which may carry further details
exception device.exceptions.AstarteError

Bases: Exception

Base class for Astarte Errors.

exception device.exceptions.AuthorizationError(body)

Bases: AstarteError

Exception raised when Astarte APIs refuse authentication.

body -- the body of the API reply, which may carry further details
exception device.exceptions.DeviceAlreadyRegisteredError

Bases: AstarteError

Exception raised in case a Device being registered has already been registered.

device.interface module

class device.interface.Interface(interface_definition: dict)

Bases: object

Class that represent an Interface definition

Interfaces are a core concept of Astarte which defines how data is exchanged between Astarte and its peers. They are not to be intended as OOP interfaces, but rather as the following definition:

In Astarte each interface has an owner, can represent either a continuous data stream or a snapshot of a set of properties, and can be either aggregated into an object or be an independent set of individual members.

name

Interface name

Type:

str

version_major

Interface version major number

Type:

int

version_minor

Interface version minor number

Type:

int

type

Interface type

Type:

str

ownership

Interface ownership

Type:

str

aggregation

Interface aggregation policy

Type:

str

mappings

Interface mapping list

Type:

Mapping

get_mapping(endpoint) → Mapping | None

Retrieve the Mapping with the given endpoint from the Interface :param endpoint: The Mapping endpoint :type endpoint: str

Returns:

The Mapping if found, None otherwise

Return type:

Mapping or None

is_aggregation_object() → bool

Check if the current Interface is a datastream with aggregation object :returns: True if aggregation: object :rtype: bool

is_server_owned() → bool

Check the Interface ownership :returns: True if ownership: server :rtype: bool

is_type_properties()

Check the Interface type :returns: True if type: properties :rtype: bool

validate(path: str, payload, timestamp: datetime) → tuple[bool, str]

Interface Data validation

Parameters:

  • path (str) – Data endpoint in interface

  • payload (object) – Data to validate

  • timestamp (datetime or None) – Timestamp associated to the payload

Returns:

  • bool – Success of the validation operation

  • str – Error message if success is False

device.introspection module

class device.introspection.Introspection

Bases: object

Class that represent the introspection of a device.

The introspection is the list od interfaces that the device declares to the server it is compatible with.

In any given time a device can have a single interface with a given name, multiple interfaces with the same name but different major/minor are not supported.

add_interface(interface_definition: dict) → None

Adds an Interface to the Introspection

This will add an Interface definition to the Device. It has to be called before connect(), as it will be used for building the Device Introspection.

Parameters:

interface_definition (dict) – An Astarte Interface definition in the form of a Python dictionary. Usually obtained by using json.loads() on an Interface file.

get_all_interfaces() → list[Interface]

Retrieve all the list of all Interfaces in Device’s Introspection

Returns:

The list of all Interfaces in the Introspection

Return type:

list

get_all_server_owned_interfaces() → list[Interface]

Retrieve all the list of all Interfaces in Device’s Introspection with server ownership

Returns:

The list of all Interfaces in the Introspection that have ownership “server”

Return type:

list

get_interface(interface_name: str) → Interface | None

Retrieve an Interface definition from the Introspection

Parameters:

interface_name (str) – The name of an Interface previously added with add_interface().

Returns:

the Interface definition if found in the Introspection, None otherwise

Return type:

Interface or None

remove_interface(interface_name: str) → None

Removes an Interface from the Introspection

Removes an Interface definition from the Device. It has to be called before connect(), as it will be used for building the Device Introspection.

Parameters:

interface_name (str) – The name of an Interface previously added with add_interface().

device.mapping module

device.mapping.MapType

Mapping type to python type mapping

alias of Union[int, float, str, bytes, bool, datetime, List[int], List[float], List[str], List[bytes], List[bool], List[datetime]]

class device.mapping.Mapping(mapping_definition: dict, interface_type: str)

Bases: object

Class that represent a data Mapping Mappings are designed around REST controller semantics: each mapping describes an endpoint which is resolved to a path, it is strongly typed, and can have additional options. Just like in REST controllers, Endpoints can be parametrized to build REST-like collection and trees. Parameters are identified by %{parameterName}, with each endpoint supporting any number of parameters (see Limitations).

endpoint

Path of the Mapping

Type:

str

type

Type of the Mapping (see notes)

Type:

str

explicit_timestamp

Flag that defines if the Mapping requires a timestamp associated to the Payload before send

Type:

bool

reliability

Reliability level of the Mapping (see notes)

Notes

Supported data types

The following types are supported:

  • double: A double-precision floating-point number as specified by binary64, by the IEEE

754 standard (NaNs and other non-numerical values are not supported). * integer: A signed 32 bit integer. * boolean: Either true or false, adhering to JSON boolean type. * longinteger: A signed 64-bit integer (please note that longinteger is represented as a string by default in JSON-based APIs.). * string: An UTF-8 string, at most 65536 bytes long. * binaryblob: An arbitrary sequence of any byte that should be shorter than 64 KiB. ( binaryblob is represented as a base64 string by default in JSON-based APIs.). * datetime: A UTC timestamp, internally represented as milliseconds since 1st Jan 1970 using a signed 64 bits integer. (datetime is represented as an ISO 8601 string by default in JSON based APIs.) * doublearray, integerarray, booleanarray, longintegerarray, stringarray, binaryblobarray, datetimearray: A list of values, represented as a JSON Array. Arrays can have up to 1024 items and each item must respect the limits of its scalar type (i.e. each string in a stringarray must be at most 65535 bytes long, each binary blob in a binaryblobarray must be shorter than 64 KiB.)

Quality of Service

Data messages QoS is chosen according to mapping settings, such as reliability. Properties are always published using QoS 2.

INTERFACE TYPE

RELIABILITY

QOS

properties

always unique

2

datastream

unreliable

0

datastream

guaranteed

1

datastream

unique

2
validate(payload: MapType, timestamp: datetime) → tuple[bool, str]

Mapping data validation

Parameters:

  • payload (MapType) – Data to validate

  • timestamp (datetime or None) – Timestamp associated to the payload

Returns:

  • bool – Success of the validation operation

  • str – Error message if success is False

device.mapping.type_strings = {'binaryblob': <class 'bytes'>, 'binaryblobarray': <class 'list'>, 'boolean': <class 'bool'>, 'booleanarray': <class 'list'>, 'datetime': <class 'datetime.datetime'>, 'datetimearray': <class 'list'>, 'double': <class 'float'>, 'doublearray': <class 'list'>, 'integer': <class 'int'>, 'integerarray': <class 'list'>, 'longinteger': <class 'int'>, 'longintegerarray': <class 'list'>, 'string': <class 'str'>, 'stringarray': <class 'list'>}

Mapping quality of service

device.pairing_handler module

device.pairing_handler.generate_device_id(namespace: UUID, unique_data: str) → str

Deterministically generate a device Id based on UUID namespace identifier and unique data.

Parameters:

  • namespace (UUID) – UUID namespace of the device_id

  • unique_data (str) – device unique data used to generate the device_id

Returns:

the generated device Id, using the standard Astarte Device ID encoding (base64 urlencoding without padding).

Return type:

str

device.pairing_handler.generate_random_device_id() → str

Quick way to generate a device Id. Pay attention that each time this value is different and the use in production is discouraged.

Returns:

the generated device Id, using the standard Astarte Device ID encoding (base64 urlencoding without padding).

Return type:

str

device.pairing_handler.obtain_device_certificate(device_id: str, realm: str, credentials_secret: str, pairing_base_url: str, crypto_store_dir: str, ignore_ssl_errors: bool) → None

Utility function that gets a device certificate from Astarte based on a locally generated csr

Parameters:

  • device_id (str) – The device ID

  • realm (str) – The Astarte realm where the device is registered

  • credentials_secret (str) – The credentials secret for the device in the given realm

  • pairing_base_url (str) – The base URL for the Astarte pairing APIs

  • crypto_store_dir (str) – Path to the folder where crypto information are stored

  • ignore_ssl_errors (str) – Set to True to ignore SSL errors

Raises:
device.pairing_handler.obtain_device_transport_information(device_id: str, realm: str, credentials_secret: str, pairing_base_url: str, ignore_ssl_errors: bool) → dict

Utility function that requests the device transport information to Astarte

Parameters:

  • device_id (str) – The device ID

  • realm (str) – The Astarte realm where the device is registered

  • credentials_secret (str) – The credentials secret for the device in the given realm

  • pairing_base_url (str) – The base URL for the Astarte pairing APIs

  • ignore_ssl_errors (str) – Set to True to ignore SSL errors

Returns:

The device transport information

Return type:

dict

Raises:
device.pairing_handler.register_device_with_jwt_token(device_id: str, realm: str, jwt_token: str, pairing_base_url: str, ignore_ssl_errors: bool = False) → str

Registers a Device against an Astarte instance/realm with a JWT Token

Returns the Credentials secret for the Device

Parameters:

  • device_id (str) – The Device ID to register.

  • realm (str) – The Realm in which to register the Device.

  • jwt_token (str) – A JWT Token to Authenticate against Pairing API. The token must have access to Pairing API and to the agent API paths.

  • pairing_base_url (str) – The Base URL of Pairing API of the Astarte Instance the Device will be registered in.

  • ignore_ssl_errors (bool) – Useful if you’re registering a device into a test instance of Astarte with self-signed certificates. It is not recommended to leave this true in production. Defaults to false, if true SSL errors will be ignored when registering a device.

Returns:

The credentials secret obtained after the registration

Return type:

str

device.pairing_handler.register_device_with_private_key(device_id: str, realm: str, private_key_file: str, pairing_base_url: str, ignore_ssl_errors: bool) → str

Registers a Device against an Astarte instance/realm with a Private Key

Returns the Credentials secret for the Device

Parameters:

  • device_id (str) – The Device ID to register.

  • realm (str) – The Realm in which to register the Device.

  • private_key_file (str) – Path to the Private Key file for the Realm. It will be used to Authenticate against Pairing API.

  • pairing_base_url (str) – The Base URL of Pairing API of the Astarte Instance the Device will be registered in.

  • ignore_ssl_errors (str) – Set to True to ignore SSL errors

Returns:

The credentials secret obtained after the registration

Return type:

str

Module contents