iotlib package

Submodules

iotlib.abstracts module

This module contains abstract base classes for various components of an IoT system.

These classes define the common interfaces for MQTT services, message encoders and decoders, device surrogates, discovery and availability processors, virtual device processors, and devices.

Each class has its own specific methods that need to be implemented in any concrete subclasses.

class iotlib.abstracts.IAvailabilityProcessor[source]

Bases: ABC

Interface for availability processors.

Subclasses must implement the process_availability_update method.

abstract attach(bridge: Surrogate) → None[source]

Attach the processor to a bridge instance.

Parameters:: bridge (Surrogate) – The bridge instance to attach to.
Returns:: None

abstract process_availability_update(availability: bool) → None[source]

Handle an update to the device availability status.

Parameters:: availability (bool) – The new availability status of the device.
Returns:: None

class iotlib.abstracts.ICodec[source]

Bases: ABC

Interface for decoding messages received on MQTT to IoT devices

abstract decode_avail_pl(payload: str) → bool[source]

Decode message received on topic dedicated to availability.

Parameters:: payload (str) – The payload of the message received on the availability topic.
Returns:: True if the decoding is successful, False otherwise.
Return type:: bool

abstract get_availability_topic() → str[source]

Return the availability topic the client must subscribe.

Returns:: The topic dedicated to handle availability messages.
Return type:: str

class iotlib.abstracts.IDiscoveryProcessor[source]

Bases: ABC

Interface for discovery processors.

This interface defines the methods that a discovery processor should implement.

abstract process_discovery_update(devices: list) → None[source]

Process a discovery update.

This method is called when a discovery update is received. It should handle the list of devices and perform any necessary processing.

Parameters:: devices (list) – The list of devices discovered.

class iotlib.abstracts.IEncoder[source]

Bases: ABC

Interface for encoding messages to send on MQTT to IoT devices.

abstract change_state_request(is_on: bool, device_id: int | None) → tuple[str, str][source]

Constructs a change state request for the device.

Parameters:

is_on (bool) – Indicates whether the device should be turned on or off.
device_id (Optional[int]) – The ID of the device. If None, the request is for all devices.

Returns:

A tuple containing the MQTT topic and the payload in JSON format.

Return type:

tuple[str, str]

abstract get_state_request(device_id: int | None) → tuple[str, str][source]

Get the current state request for a device.

Parameters:: device_id (Optional[int]) – The device ID to get the state request for.
Returns:: A tuple containing the state request topic and payload or None if such a request is not accepted.
Return type:: tuple[str, str]

abstract is_pulse_request_allowed(device_id: int | None) → bool[source]

Check if a pulse request is allowed for a device.

Parameters:: device_id (Optional[int]) – The device ID to check if a pulse request is allowed for.
Returns:: True if a pulse request is allowed, False otherwise.
Return type:: bool

class iotlib.abstracts.IMQTTService[source]

Bases: ABC

Interface for the MQTT services used by Surrogate classes

abstract connect(properties: Properties | None = None) → MQTTErrorCode[source]

Connect to a MQTT remote broker.

Parameters:: properties (Optional[mqtt.Properties]) – The properties for the MQTT connection, defaults to None
Returns:: The MQTT error code for the connection attempt.
Return type:: mqtt.MQTTErrorCode

abstract connect_handler_add(handler: Callable) → None[source]

Adds a connect event handler.

Parameters:: handler (Callable) – The callback function to handle the connect event.
Returns:: None

abstract disconnect() → MQTTErrorCode[source]

Disconnect from a MQTT remote broker.

Returns:: The MQTT error code for the disconnection attempt.
Return type:: mqtt.MQTTErrorCode

abstract disconnect_handler_add(handler: Callable) → None[source]

Adds a disconnect event handler.

Parameters:: handler (Callable) – The callback function to handle the disconnect event.
Returns:: None

abstract property mqtt_client: Client

Returns the MQTT client object.

Returns:: The MQTT client object.
Return type:: mqtt.Client

class iotlib.abstracts.IVirtualDevice[source]

Bases: ABC

Interface for virtual devices in the IoT system.

This class defines the common interface for all devices in the system. Subclasses must implement the handle_value and processor_append methods.

abstract handle_value(value) → ResultType[source]

Handles the received value and performs necessary actions.

Parameters:: value (Any) – The received value.
Returns:: The result of handling the value.
Return type:: ResultType

abstract processor_append(processor: IVirtualDeviceProcessor) → None[source]

Appends a VirtualDeviceProcessor to the list of processors.

Parameters:: processor (VirtualDeviceProcessor) – The VirtualDeviceProcessor to append.
Raises:: TypeError – If the processor is not an instance of VirtualDeviceProcessor.
Returns:: None

class iotlib.abstracts.IVirtualDeviceProcessor[source]

Bases: ABC

Interface for virtual device processors.

This class defines the interface for processing updates from virtual devices. Child classes should implement the process_value_update method to handle specific processing logic for the device type.

compatible_with_device(v_dev: any) → bool[source]

Checks if the given virtual device is compatible with this processor.

Parameters:: v_dev (any) – The virtual device to check compatibility with.
Returns:: True if the virtual device is compatible, False otherwise.
Return type:: bool

abstract process_value_update(v_dev: any) → None[source]

Handle an update from a virtual device.

This method is called when a value changes on a virtual device. It should be implemented in child classes to handle specific processing logic for the device type.

Parameters:: v_dev (any) – The virtual device instance.
Returns:: None

class iotlib.abstracts.ResultType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: IntEnum

Enum representing the result types.

Variables:

IGNORE – Represents an ignored result.
SUCCESS – Represents a successful result.
ECHO – Represents a value appearing twice.

ECHO = 1

IGNORE = -1

SUCCESS = 0

class iotlib.abstracts.Surrogate(mqtt_service: IMQTTService, codec: ICodec)[source]

Bases: object

A surrogate class that wraps an MQTT client and codec.

This class acts as a surrogate for devices that use the given MQTT client and codec for communication. It can handle encoding/decoding messages to interact with real devices.

iotlib.bridge module

This module contains the MQTTBridge class which extends the Surrogate class.

The MQTTBridge class is responsible for handling MQTT-specific operations. It uses an MQTT service instance for MQTT operations and a codec instance for encoding and decoding messages.

class iotlib.bridge.MQTTBridge(mqtt_service: IMQTTService, codec: ICodec)[source]

Bases: Surrogate

MQTTBridge is a class that extends the Surrogate class.

This class is responsible for handling MQTT-specific operations.

Variables:

mqtt_service (IMQTTService) – The MQTT service instance.
codec (ICodec) – The codec instance for encoding and decoding messages.

add_availability_processor(processor: IAvailabilityProcessor) → None[source]

Appends an Availability Processor instance to the processor list.

Parameters:: processor (IAvailabilityProcessor) – The Availability Processor instance to append.
Raises:: TypeError – If processor is not an instance of IAvailabilityProcessor.
Returns:: None

property availability: bool

Get the availability status of the bridge.

Returns:: True if the bridge is available, False otherwise.
Return type:: bool

iotlib.client module

This module contains the MQTTClient class which implements the IMQTTService interface.

The MQTTClient class is responsible for handling MQTT-specific operations. It uses the paho.mqtt.client library for MQTT operations and provides methods for connecting, disconnecting, and handling MQTT events.

class iotlib.client.MQTTClient(client_id: str, hostname: str = '127.0.0.1', port: int = 1883, user_name: str | None = None, user_pwd: str | None = None, keepalive: int = 60, tls: bool = False, clean_start: bool = False)[source]

Bases: IMQTTService

MQTTClient is a class that implements the IMQTTService interface.

This class is responsible for handling MQTT-specific operations. It uses the paho.mqtt.client library for MQTT operations and provides methods for connecting, disconnecting, and handling MQTT events.

Variables:

client (mqtt.Client) – The MQTT client instance.
host (str) – The MQTT broker host.
port (int) – The MQTT broker port.
keepalive (int) – The keepalive timeout value for the client.

clean_start: bool = False

client_id: str

connect(properties: Properties | None = None) → MQTTErrorCode[source]

Connect to a MQTT remote broker.

Parameters:: properties (Optional[mqtt.Properties]) – The properties for the MQTT connection, defaults to None
Returns:: The MQTT error code for the connection attempt.
Return type:: mqtt.MQTTErrorCode

connect_handler_add(handler: Callable) → None[source]

Adds a connect event handler.

Parameters:: handler (Callable) – The callback function to handle the connect event.
Returns:: None

property connected: bool

Returns the connection status of the client.

Returns:: True if the client is connected, False otherwise.
Return type:: bool

disconnect() → MQTTErrorCode[source]

Disconnect from a MQTT remote broker.

Returns:: The MQTT error code for the disconnection attempt.
Return type:: mqtt.MQTTErrorCode

disconnect_handler_add(handler: Callable) → None[source]

Adds a disconnect event handler.

Parameters:: handler (Callable) – The callback function to handle the disconnect event.
Returns:: None

hostname: str = '127.0.0.1'

keepalive: int = 60

loop_forever() → MQTTErrorCode[source]: Start a new thread to run the network loop.

property mqtt_client: Client

Returns the MQTT client object.

Returns:: The MQTT client object.
Return type:: mqtt.Client

port: int = 1883

start(properties: Properties | None = None) → MQTTErrorCode[source]

Starts the client MQTT network loop.

Connects the client if not already connected, starts the network loop

Parameters:: properties (Optional[mqtt.Properties]) – The properties to be used for the CONNECT message.
Returns:: The result of calling client.loop_start().
Return type:: mqtt.MQTTErrorCode
Raises:: RuntimeError – If loop_start fails or connection is refused.

property started: bool

Returns the current status of the client.

Returns:: True if the client has started, False otherwise.
Return type:: bool

stop() → MQTTErrorCode[source]

Stops the client MQTT connection.

Stops the async loop if it was not already running. Calls disconnect() to close the client connection.

Returns:: The result of calling disconnect().
Return type:: mqtt.MQTTErrorCode
Raises:: RuntimeError – If loop_stop fails.

tls: bool = False

user_name: str | None = None

user_pwd: str | None = None

class iotlib.client.MQTTClientHelper(*args, **kwargs)[source]

Bases: MQTTClient

This class provides additional helper methods for MQTT operations, not directly related to the IMQTTService interface. Could evolve to a more generic class in the future.

default_message_callback_add(callback: Callable) → None[source]

Adds a callback function to the list of default message callbacks.

Parameters: - callback (Callable): The callback function to be added.

Returns: - None

publish(topic, payload, **kwargs)[source]: Publish a message on a topic.

subscribe(topic, **kwargs)[source]: Subscribes to the specified topic.

subscribe_handler_add(handler: Callable)[source]

Adds a handler function to the list of subscribe event handlers.

Parameters:: handler (Callable) – The handler function to be added.
Returns:: None

iotlib.devconfig module

This module defines the PropertyConfig and ButtonValues enums.

The PropertyConfig enum defines property names, types, and owning nodes for different virtual device types. It allows generic access to property configuration details.

The ButtonValues enum defines string constants for possible button actions like single press, double press, long press etc. Using this enum allows code to refer to button actions through constant values rather than string literals.

class iotlib.devconfig.ButtonValues(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enumeration defining button action values.

This enum defines string constants for possible button actions like single press, double press, long press etc.

Using this enum allows code to refer to button actions through constant values rather than string literals.

Variables:

SINGLE_ACTION – Represents a single button press action.
DOUBLE_ACTION – Represents a double button press action.
LONG_ACTION – Represents a long button press action.
OFF – Represents the button off state.

DOUBLE_ACTION = 'double'

LONG_ACTION = 'long'

OFF = 'off'

SINGLE_ACTION = 'single'

class iotlib.devconfig.PropertyConfig(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Defines property name enums for virtual devices.

This enum defines property names, types, and owning node for different virtual device types. Allows generic access to property configuration details.

Variables:

ALARM_PROPERTY – Represents the ‘alarm.power’ property of type bool.
ADC_PROPERTY – Represents the ‘sensor.voltage’ property of type float.
BUTTON_PROPERTY – Represents the ‘sensor.action’ property of type str.
CONDUCTIVITY_PROPERTY – Represents the ‘sensor.conductivity’ property of type int.
HUMIDITY_PROPERTY – Represents the ‘sensor.humidity’ property of type int.
LIGHT_PROPERTY – Represents the ‘sensor.light’ property of type int.
MOTION_PROPERTY – Represents the ‘sensor.occupancy’ property of type bool.
SWITCH_PROPERTY – Represents the ‘switch.power’ property of type bool.
SWITCH0_PROPERTY – Represents the ‘switch0.power’ property of type bool.
SWITCH1_PROPERTY – Represents the ‘switch1.power’ property of type bool.
TEMPERATURE_PROPERTY – Represents the ‘sensor.temperature’ property of type float.

ADC_PROPERTY = ('sensor.voltage', <class 'float'>)

ALARM_PROPERTY = ('alarm.power', <class 'bool'>)

BUTTON_PROPERTY = ('sensor.action', <class 'str'>)

CONDUCTIVITY_PROPERTY = ('sensor.conductivity', <class 'int'>)

HUMIDITY_PROPERTY = ('sensor.humidity', <class 'int'>)

LIGHT_PROPERTY = ('sensor.light', <class 'int'>)

MOTION_PROPERTY = ('sensor.occupancy', <class 'bool'>)

SWITCH0_PROPERTY = ('switch0.power', <class 'bool'>)

SWITCH1_PROPERTY = ('switch1.power', <class 'bool'>)

SWITCH_PROPERTY = ('switch.power', <class 'bool'>)

TEMPERATURE_PROPERTY = ('sensor.temperature', <class 'float'>)

iotlib.discoverer module

This module defines classes for discovering IoT devices using MQTT.

Discoverer is the base class for device discovery. It uses an MQTT client to discover devices and maintains a list of discovered devices. It also allows adding discovery processors for processing the discovered devices.

ZigbeeDiscoverer and TasmotaDiscoverer are subclasses of Discoverer. They override the base class methods to provide specific implementations for discovering Zigbee and Tasmota devices, respectively.

Device class provides properties for accessing these attributes and methods for getting string representations of the device.

class iotlib.discoverer.Device(address: str, friendly_name: str, model: Model, protocol: Protocol)[source]

Bases: object

Represents a generic device in the system.

This class provides a base for all types of devices. It defines common properties and methods that all devices should have. Specific device types should inherit from this class and add their own unique properties and methods.

Variables:

id (str) – The unique identifier for the device.
name (str) – The human-readable name of the device.
type (str) – The type of the device (e.g., ‘sensor’, ‘actuator’).
status (str) – The current status of the device (e.g., ‘online’, ‘offline’).

property address: str: Returns the address of the device.

property friendly_name: str: Returns the friendly name of the device.

property model: Model: Returns the model of the device.

property protocol: Protocol: Returns the protocol of the device.

class iotlib.discoverer.Discoverer(mqtt_service: IMQTTService)[source]

Bases: object

Discovers devices on the network.

This class provides methods to discover devices on the network using different protocols. It can be used to find devices, get their information, and add them to the system.

Variables:

address (str) – The IP address or hostname of the device.
friendly_name (str) – The friendly name of the device.
model (Model) – The model of the device.
protocol (Protocol) – The protocol used by the device.

add_discovery_processor(processor: IDiscoveryProcessor) → None[source]

Adds a discovery processor to the Discoverer.

Discovery processors are used to process the results of a device discovery operation. This method allows adding custom processors to the Discoverer.

Parameters:: processor (IDiscoveryProcessor) – The discovery processor to be added.

get_devices() → list[Device][source]

Gets the list of discovered devices.

This method returns the list of Device objects that have been discovered by the Discoverer.

Returns:: A list of Device objects.
Return type:: list[Device]

class iotlib.discoverer.TasmotaDiscoverer(mqtt_service: IMQTTService)[source]

Bases: Discoverer

Discovers Tasmota devices on the network.

This class extends the Discoverer class and provides methods to discover Tasmota devices on the network. It uses the Tasmota protocol for device discovery.

Variables:

address (str) – The IP address or hostname of the Tasmota device.
friendly_name (str) – The friendly name of the Tasmota device.
model (Model) – The model of the Tasmota device.
protocol (Protocol) – The protocol used by the Tasmota device, should be Tasmota.

class iotlib.discoverer.UnifiedDiscoverer(mqtt_service: IMQTTService)[source]

Bases: object

A class that unifies the discovery of devices from different protocols.

add_discovery_processor(processor: IDiscoveryProcessor) → None[source]: Appends an Discovery Processor instance to the processor list

get_devices() → list[Device][source]: Returns a list of all devices discovered by all protocol discoverers.

class iotlib.discoverer.ZigbeeDiscoverer(mqtt_service: IMQTTService)[source]

Bases: Discoverer

Discovers Zigbee devices on the network.

This class extends the Discoverer class and provides methods to discover Zigbee devices on the network. It uses the Zigbee protocol for device discovery.

Variables:

address (str) – The IP address or hostname of the Zigbee device.
friendly_name (str) – The friendly name of the Zigbee device.
model (Model) – The model of the Zigbee device.
protocol (Protocol) – The protocol used by the Zigbee device, should be Zigbee.

iotlib.factory module

Module containing factory methods and enums for device models and protocols.

This module provides a factory for creating instances of codecs for different device models and protocols. It also defines enums for representing different device models and c ommunication protocols.

The primary classes are Model and Protocol enums, and the CodecFactory. The Model enum represents different device models, and the Protocol enum represents different communication protocols. The CodecFactory uses these enums to create appropriate codec instances.

Example:

# factory.py code
factory = CodecFactory()
factory.registers(Model.NEO_ALARM, Protocol.ZIGBEE, NeoNasAB02B2)

# User code to discover device model and protocol
import iotlib.factory as factory
codec = factory.create_instance(Model.NEO_ALARM, Protocol.ZIGBEE)

The example above shows how to use the CodecFactory to register a codec for a specific device model and protocol, and create an instance of it.

class iotlib.factory.CodecFactory(*args: Any, **kwargs: Any)[source]

Bases: object

A factory class for creating codec instances.

This class uses the Singleton design pattern to ensure that only one instance of the factory exists. The factory can be used to create instances of different types of codecs.

Variables:: codecs (dict[str, Codec]) – A dictionary mapping codec names to codec instances.

create_instance(model: Model, protocol: Protocol, *args, **kwargs) → ICodec[source]

Creates a new codec instance for the given model and protocol.

This method takes a model, a protocol, and optional arguments, and creates a new codec instance for the given model and protocol using the registered constructor.

Parameters:

model (Model) – The model for which the codec should be created.
protocol (Protocol) – The protocol for which the codec should be created.
args – Additional positional arguments to be passed to the constructor.
kwargs – Additional keyword arguments to be passed to the constructor.

Returns:

A new codec instance for the given model and protocol.

Return type:

ICodec

registers(model: Model, protocol: Protocol, constructor: Callable[[list], ICodec]) → None[source]

Registers a new codec constructor.

This method takes a model, a protocol, and a constructor function, and registers the constructor for creating codecs for the given model and protocol.

Parameters:

model (Model) – The model for which the constructor should be registered.
protocol (Protocol) – The protocol for which the constructor should be registered.
constructor (Callable[[list], ICodec]) – The constructor function to be registered.

class iotlib.factory.Model(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

This enum is used to represent the different models of devices that can be discovered.

Variables:

MIFLORA – Xiaomi Mi Flora plant sensor
NEO_ALARM – Neo NAS-AB02B2 Zigbee Siren
SHELLY_PLUGS – Shelly Plug S WiFi smart plug
SHELLY_UNI – Shelly Uni WiFi relay/dimmer
TUYA_SOIL – Tuya TS0601_soil Zigbee soil moisture sensor
ZB_AIRSENSOR – Sonoff Zigbee air temperature/humidity sensor
ZB_BUTTON – Sonoff SNZB-01 Zigbee wireless button
ZB_MOTION – Sonoff SNZB-03 Zigbee motion sensor
ZB_MINI – Sonoff ZBMINI-L Zigbee wireless switch module
EL_ZBSW02 – eWeLink ZB-SW02 Zigbee wireless switch module

MIFLORA = 'Miflora'

NEO_ALARM = 'NAS-AB02B2'

NONE = 'None'

RING_CAMERA = 'RingCamera'

SHELLY_PLUGS = 'Shelly Plug S'

SHELLY_UNI = 'Shelly Uni'

TUYA_SOIL = 'TS0601_soil'

TUYA_TS0002 = 'TS0002'

UNKNOWN = 'Unknown'

ZB_AIRSENSOR = 'SNZB-02'

ZB_BUTTON = 'SNZB-01'

ZB_MINI = 'ZBMINI-L'

ZB_MOTION = 'SNZB-03'

static from_str(label: str)[source]

Returns the Model enum value corresponding to the given label.

This method takes a label as input and returns the corresponding Model enum value. If the label does not correspond to any Model enum value, it returns None.

Parameters:: label (str) – The label to get the Model enum value for.
Returns:: The Model enum value corresponding to the label, or None if the label does not correspond to any Model enum value.
Return type:: Model

class iotlib.factory.Protocol(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

This enum is used to represent the different communication protocols that can be used by devices.

Variables:

HOMIE – The Homie IoT protocol.
SHELLY – The Shelly smart home devices protocol.
TASMOTA – The Tasmota protocol used by ESP8266/ESP32 boards.
Z2M – The Zigbee2MQTT protocol.
Z2T – The Zigbee2Tasmota protocol.

DEFAULT = 'default'

HOMIE = 'Homie'

RING = 'Ring'

SHELLY = 'Shelly'

TASMOTA = 'Tasmota'

Z2M = 'Zigbee2MQTT'

Z2T = 'Zigbee2Tasmota'

iotlib.processor module

Module containing processor classes for handling virtual devices.

This module provides classes for processing virtual devices. These classes implement the IVirtualDeviceProcessor interface and provide specific functionality for handling virtual devices.

The primary class in this module is the VirtualDeviceLogger. This class logs the actions performed on virtual devices and can be used for debugging and tracking purposes.

Example:

v_switch = Switch()
factory = CodecFactory()
codec = factory.create_instance(model=Model.ZB_MINI,
                                protocol=Protocol.Z2M,
                                device_name='SWITCH_PLUG',
                                v_switch=v_switch)
bridge = iotlib.bridge.MQTTBridge(client, codec)
# Create a logger instance for the virtual switch
logger = VirtualDeviceLogger()
v_switch.processor_append(logger)

The example above shows how to use the VirtualDeviceLogger .

class iotlib.processor.AvailabilityLogger(debug: bool = False)[source]

Bases: IAvailabilityProcessor

Logs availability updates of devices.

This processor logs a message when a device’s availability changes.

attach(bridge: Surrogate) → None[source]

Attach the processor to a bridge instance.

Parameters:: bridge (Surrogate) – The bridge instance to attach to.
Returns:: None

process_availability_update(availability: bool) → None[source]

Handle an update to the device availability status.

Parameters:: availability (bool) – The new availability status of the device.
Returns:: None

class iotlib.processor.AvailabilityPublisher(publish_topic_base: str = None)[source]

Bases: IAvailabilityProcessor

Processes availability updates from MQTTBridge.

This processor handles availability updates from a MQTTBridge instance. It publishes the availability status to a MQTT topic.

attach(bridge: Surrogate) → None[source]

Attach the processor to a bridge instance.

Parameters:: bridge (Surrogate) – The bridge instance to attach to.
Returns:: None

process_availability_update(availability: bool) → None[source]

Handle an update to the device availability status.

Parameters:: availability (bool) – The new availability status of the device.
Returns:: None

class iotlib.processor.ButtonTrigger(mqtt_service: IMQTTService, countdown_long=600)[source]

Bases: IVirtualDeviceProcessor

This processor triggers button actions on virtual devices when their state changes.

compatible_with_device(v_dev: VirtualDevice) → None[source]

Checks if the given virtual device is compatible with this processor.

Parameters:: v_dev (any) – The virtual device to check compatibility with.
Returns:: True if the virtual device is compatible, False otherwise.
Return type:: bool

process_value_update(v_dev: VirtualDevice) → None[source]

Processes a value update from a virtual device.

This method takes a virtual device as input and triggers the appropriate action based on the device’s state.

Parameters:: v_dev (VirtualDevice) – The virtual device whose value has been updated.

Actions:

single press: Start registered switches with default countdown.
double press: Start and stop registered switches for countdown_long.
long press: Stop registered switches.

class iotlib.processor.MotionTrigger(mqtt_service: IMQTTService)[source]

Bases: IVirtualDeviceProcessor

A class that handles motion sensor state changes and triggers registered switches when occupancy is detected.

compatible_with_device(v_dev: VirtualDevice) → None[source]

Checks if the given virtual device is compatible with this processor.

Parameters:: v_dev (any) – The virtual device to check compatibility with.
Returns:: True if the virtual device is compatible, False otherwise.
Return type:: bool

process_value_update(v_dev: VirtualDevice) → None[source]

Process the value update of a virtual device.

Parameters:: v_dev (VirtualDevice) – The virtual device whose value has been updated.

class iotlib.processor.PropertyPublisher(mqtt_service: IMQTTService, publish_topic_base: str | None = None)[source]

Bases: IVirtualDeviceProcessor

A class that publishes property updates to an MQTT broker.

compatible_with_device(v_dev: VirtualDevice) → None[source]

Checks if the given virtual device is compatible with this processor.

Parameters:: v_dev (any) – The virtual device to check compatibility with.
Returns:: True if the virtual device is compatible, False otherwise.
Return type:: bool

process_value_update(v_dev) → None[source]

Publishes the updated value of a virtual device’s property to the MQTT broker.

Parameters:: v_dev (VirtualDevice) – The virtual device whose value has been updated.

class iotlib.processor.VirtualDeviceLogger(debug: bool = False)[source]

Bases: IVirtualDeviceProcessor

Logs updates from virtual devices.

This processor logs a debug message when a virtual device value is updated. It implements the IVirtualDeviceProcessor interface.

Variables:: logger (logging.Logger) – The logger instance used to log debug messages.

compatible_with_device(v_dev: VirtualDevice) → None[source]

Checks if the given virtual device is compatible with this processor.

Parameters:: v_dev (any) – The virtual device to check compatibility with.
Returns:: True if the virtual device is compatible, False otherwise.
Return type:: bool

process_value_update(v_dev: VirtualDevice) → None[source]

Handle an update from a virtual device.

This method is called when a value changes on a virtual device. It should be implemented in child classes to handle specific processing logic for the device type.

Parameters:: v_dev (any) – The virtual device instance.
Returns:: None

iotlib.utils module

class iotlib.utils.InfiniteTimer(seconds, target)[source]

Bases: object

A Timer class that does not stop, unless you want it to. https://stackoverflow.com/questions/12435211/threading-timer-repeat-function-every-n-seconds

cancel()[source]

start()[source]

class iotlib.utils.Singleton(*args, **kwargs)[source]

Bases: type

ref : Python Cookbook Recipes for Mastering Python 3, (David Beazley, Brian K. Jones) Using a Metaclass to Control Instance Creation

iotlib.virtualdev module

Virtual devices serve as an abstraction layer over physical devices, facilitating interoperability across different types of devices. These virtual devices are organized into a hierarchical structure as follows:

Sensors: These virtual devices are responsible for measuring and reporting various environmental conditions. They include:

Temperature Sensor

Humidity Sensor

Light Sensor

Conductivity Sensor

Button Sensor

Motion Sensor

Analog-to-Digital Converter

Operables: These virtual devices are capable of performing certain actions. They include:

Alarm

Switch

Each physical device is associated with one or more virtual devices, which handle the processing and management of data. For instance, an air sensor might be associated with a Temperature Sensor and a Humidity Sensor virtual device, which handle temperature and humidity data respectively. This support is provided by the VirtualDevice class.

class iotlib.virtualdev.ADC(friendly_name=None, quiet_mode=False)[source]

Bases: Sensor

Analog-to-digital converter

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

handle_value(value: float) → list[source]

Handles the received value and performs necessary actions.

Parameters:: value (Any) – The received value.
Returns:: The result of handling the value.
Return type:: ResultType

class iotlib.virtualdev.Alarm(friendly_name: str = None, quiet_mode: bool = False, countdown: int | None = None)[source]

Bases: Operable

Implementation of a virtual Alarm.

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.Button(friendly_name=None)[source]

Bases: Sensor

Button managing 3 types of messages : single, double and long button press

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

property value

Gets the current value of the virtual device.

This method returns the current value of the virtual device.

Returns:: The current value of the virtual device.
Return type:: Any

class iotlib.virtualdev.ConductivitySensor(friendly_name=None, quiet_mode=False)[source]

Bases: Sensor

Conductivity sensor

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.HumiditySensor(friendly_name=None, quiet_mode=False)[source]

Bases: Sensor

Humidity sensor

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.Level(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Represents the level of alarm.

The Level class is an enumeration that defines three levels: LOW, MEDIUM, and HIGH. Each level is associated with a string value.

Variables:

LOW – Represents a low level of alarm.
MEDIUM – Represents a medium level of alarm.
HIGH – Represents a high level of alarm.

HIGH = 'high'

LOW = 'low'

MEDIUM = 'medium'

class iotlib.virtualdev.LightSensor(friendly_name=None, quiet_mode=False)[source]

Bases: Sensor

Light sensor

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.Melodies(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: IntEnum

Enumeration of melodies available for virtual devices.

AMBULANCE = 11

DING_DONG1 = 2

DING_DONG2 = 3

DING_DONG3 = 12

DING_DONG4 = 18

DOG = 10

FIRE = 6

HORN = 7

MELO_01 = 1

MELO_04 = 4

MELO_05 = 5

MELO_08 = 8

MELO_09 = 9

MELO_14 = 14

MELO_15 = 15

PHONE = 13

WAKE_UP1 = 16

WAKE_UP2 = 17

class iotlib.virtualdev.Motion(friendly_name=None, quiet_mode=False)[source]

Bases: Sensor

Virtual button manager

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.Operable(friendly_name: str = None, quiet_mode: bool = False, countdown: int | None = None)[source]

Bases: VirtualDevice

Root implementation of a virtual Operable device (Switch or Alarm)

property countdown: int | None

Gets the countdown value.

Returns:: The countdown value of the virtual device.
Return type:: Optional[int]

property device_id: int | None

Gets the device ID.

Returns:: The device ID of the virtual device.
Return type:: int | None

trigger_change_state(mqtt_service: IMQTTService, is_on: bool, on_time: int | None = None) → None[source]

Triggers a change in the state of a device.

This method sends a request message to the device to change the current state of the device.

Parameters:

mqtt_service (IMQTTService) – The client instance used for communication.
is_on (bool) – The new state of the device (True for ON, False for OFF).
on_time (int | None, optional) – The duration for which the device should remain ON, in seconds. If None, the device will remain ON indefinitely, defaults to None.

Raises:

TypeError – If mqtt_service is not an instance of MQTTService.

trigger_get_state(mqtt_service: IMQTTService, device_id: str = None) → None[source]

Triggers a state request to the device.

Parameters:

mqtt_service (IMQTTService) – The client instance used for communication.
device_id (str, optional) – Optional device ID to retrieve state for, defaults to None.

Raises:

TypeError – If mqtt_service is not an instance of MQTTService.

trigger_start(mqtt_service: IMQTTService, on_time: int | None = None) → bool[source]

Ask the device to start.

This method triggers the device to start. If the device is already in the “on” state, no action is required and the method returns False. Otherwise, the method requests the device to turn on by calling the trigger_change_state method with the is_on parameter set to True.

Parameters:

mqtt_service (IMQTTService) – The client object used to communicate with the device.
on_time (int | None, optional) – The time duration for which the device should remain on. If None, the device will remain ON indefinitely, defaults to None.

Returns:

Returns True if the switch state is OFF when the method is called.

Return type:

bool

Raises:

TypeError – If mqtt_service is not an instance of MQTTService.

trigger_stop(mqtt_service: IMQTTService) → bool[source]

Ask the device to stop.

This method is used to send a request to the device to stop its operation. If the switch state is currently ON, it will send a request to turn it OFF via MQTT.

Parameters:: mqtt_service (IMQTTService) – The client object used to communicate with the device.
Returns:: Returns True if the switch state is ON when the method is called, indicating that a request to turn it OFF has been sent.
Return type:: bool
Raises:: TypeError – If mqtt_service is not an instance of MQTTService.

class iotlib.virtualdev.Sensor(friendly_name=None, quiet_mode=False)[source]

Bases: VirtualDevice

Sensor virtual devices

This virtual sensor device manages sensor values that can be read by observer devices.

add_observer(device: Operable) → None[source]

Add an observer device to be notified of sensor value changes.

This method adds an Operable device to the list of observer devices.

Parameters:: device (Operable) – The Operable device to be added as an observer.

get_sensor_observers() → list[VirtualDevice][source]

Get the list of observer devices.

This method returns the list of Operable device objects that are observing the sensor values.

Returns:: The list of Operable device objects that are observing the sensor values.
Return type:: list[VirtualDevice]

class iotlib.virtualdev.Switch(*argc, **kwargs)[source]

Bases: Operable

Implementation of a virtual Switch

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.Switch0(*args, **kwargs)[source]

Bases: Switch

Virtual switch #0 of a multi channel device

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.Switch1(*args, **kwargs)[source]

Bases: Switch

Virtual switch #1 of a multi channel device

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

class iotlib.virtualdev.TemperatureSensor(friendly_name=None, quiet_mode=False)[source]

Bases: Sensor

Temperature sensor

get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

handle_value(value: float) → list[source]

Handles the received value and performs necessary actions.

Parameters:: value (Any) – The received value.
Returns:: The result of handling the value.
Return type:: ResultType

class iotlib.virtualdev.VirtualDevice(friendly_name: str, quiet_mode: bool)[source]

Bases: IVirtualDevice

Virtual devices serve as an abstraction layer over physical devices, facilitating interoperability across different types of devices.

abstract get_property() → str[source]

Gets the property of the device.

This method returns the property of the device.

Returns:: The property of the device.
Return type:: str

handle_value(value) → ResultType[source]

Handles the received value and performs necessary actions.

Parameters:: value (Any) – The received value.
Returns:: The result of handling the value.
Return type:: ResultType

processor_append(processor: IVirtualDeviceProcessor) → None[source]

Appends a VirtualDeviceProcessor to the list of processors.

Parameters:: processor (VirtualDeviceProcessor) – The VirtualDeviceProcessor to append.
Raises:: TypeError – If the processor is not an instance of VirtualDeviceProcessor.
Returns:: None

set_encoder(encoder: IEncoder) → None[source]

Sets the encoder for the device.

This method sets the encoder for the device. It validates the type of the encoder before setting it.

Parameters:: encoder (IEncoder) – The encoder to set.
Raises:: TypeError – If the encoder is not of the expected type.

property value

Gets the current value of the virtual device.

This method returns the current value of the virtual device.

Returns:: The current value of the virtual device.
Return type:: Any

Module contents

This is the initialization module for the iotlib package.