Wolk API

The Wolk class is a wrapper and an API for everything this package has to offer.

class wolk_gateway_module.wolk.Wolk(host: str, port: int, module_name: str, device_status_provider: Callable[[str], wolk_gateway_module.model.device_status.DeviceStatus], actuation_handler: Optional[Callable[[str, str, Union[bool, int, float, str]], None]] = None, actuator_status_provider: Optional[Callable[[str, str], Tuple[wolk_gateway_module.model.actuator_state.ActuatorState, Union[bool, int, float, str]]]] = None, configuration_handler: Optional[Callable[[str, Dict[str, Union[int, float, bool, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]]]], None]] = None, configuration_provider: Optional[Callable[[str], Dict[str, Union[int, float, bool, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]]]]] = None, firmware_handler: Optional[wolk_gateway_module.interface.firmware_handler.FirmwareHandler] = None, connectivity_service: Optional[wolk_gateway_module.connectivity.connectivity_service.ConnectivityService] = None, data_protocol: Optional[wolk_gateway_module.protocol.data_protocol.DataProtocol] = None, firmware_update_protocol: Optional[wolk_gateway_module.protocol.firmware_update_protocol.FirmwareUpdateProtocol] = None, registration_protocol: Optional[wolk_gateway_module.protocol.registration_protocol.RegistrationProtocol] = None, status_protocol: Optional[wolk_gateway_module.protocol.status_protocol.StatusProtocol] = None, outbound_message_queue: Optional[wolk_gateway_module.persistence.outbound_message_queue.OutboundMessageQueue] = None)

Core of this package, tying together all features.

Variables:

• actuation_handler (Optional[Callable[[str, str,str], None]]) – Set new actuator values for your devices
• actuator_status_provider (Optional[Callable[[str, str], Tuple[ActuatorState, Union[bool, int, float, str]]]]) – Get device’s current actuator state
• configuration_handler (Optional[Callable[[str, Dict[str, str]], None]]) – Set new configuration values for your devices
• configuration_provider (Optional[Callable[[str],Dict[]]) – Get device’s current configuration options
• connectivity_service (ConnectivityService) – Service that enables connection to WolkGateway
• data_protocol (DataProtocol) – Parse messages related to device data
• device_status_provider (Callable[[str], DeviceStatus]) – Get device’s current status
• devices (List[Device]) – List of devices added to module
• firmware_handler (Optional[FirmwareHandler]) – Handle commands related to firmware update
• firmware_update_protocol (FirmwareUpdateProtocol) – Parse messages related to firmware update
• host (str) – WolkGateway’s host address
• log (logging.Logger) – Logger instance
• module_name (str) – Name of module used for identification on WolkGateway
• outbound_message_queue (OutboundMessageQueue) – Means of storing messages
• port (int) – WolkGateway’s connectivity port
• registration_protocol (RegistrationProtocol) – Parse messages related to device registration
• status_protocol (StatusProtocol) – Parse messages related to device status

__init__(host: str, port: int, module_name: str, device_status_provider: Callable[[str], wolk_gateway_module.model.device_status.DeviceStatus], actuation_handler: Optional[Callable[[str, str, Union[bool, int, float, str]], None]] = None, actuator_status_provider: Optional[Callable[[str, str], Tuple[wolk_gateway_module.model.actuator_state.ActuatorState, Union[bool, int, float, str]]]] = None, configuration_handler: Optional[Callable[[str, Dict[str, Union[int, float, bool, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]]]], None]] = None, configuration_provider: Optional[Callable[[str], Dict[str, Union[int, float, bool, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]]]]] = None, firmware_handler: Optional[wolk_gateway_module.interface.firmware_handler.FirmwareHandler] = None, connectivity_service: Optional[wolk_gateway_module.connectivity.connectivity_service.ConnectivityService] = None, data_protocol: Optional[wolk_gateway_module.protocol.data_protocol.DataProtocol] = None, firmware_update_protocol: Optional[wolk_gateway_module.protocol.firmware_update_protocol.FirmwareUpdateProtocol] = None, registration_protocol: Optional[wolk_gateway_module.protocol.registration_protocol.RegistrationProtocol] = None, status_protocol: Optional[wolk_gateway_module.protocol.status_protocol.StatusProtocol] = None, outbound_message_queue: Optional[wolk_gateway_module.persistence.outbound_message_queue.OutboundMessageQueue] = None)

Construct an instance ready to communicate with WolkGateway.

Parameters:

• host (str) – Host address of WolkGateway
• port (int) – TCP/IP port of WolkGateway
• module_name (str) – Module identifier used when connecting to gateway
• device_status_provider (Callable[[str], DeviceStatus]) – Provider of device’s current status
• actuation_handler (Optional[Callable[[str, str, str], None]]) – Setter of new device actuator values
• actuator_status_provider (Optional[Callable[[str, str], Tuple[ActuatorState, Union[bool, int, float, str]]]]) – Provider of device’s current actuator status
• configuration_handler (Optional[Callable[[str, Dict[str, Union[bool, int, float, str]]], None]]) – Setter of new device configuration values
• configuration_provider (Optional[Callable[[str], Dict[str, Union[int, float, bool, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]]]]]) – Provider of device’s configuration options
• install_firmware (Optional[Callable[[str, str], None]]) – Handling of firmware installation
• connectivity_service (Optional[ConnectivityService]) – Custom connectivity service implementation
• data_protocol (Optional[DataProtocol]) – Custom data protocol implementation
• firmware_update_protocol (Optional[FirmwareUpdateProtocol]) – Custom firmware update protocol implementation
• registration_protocol (Optional[RegistrationProtocol]) – Custom registration protocol implementation
• status_protocol (Optional[StatusProtocol]) – Custom device status protocol implementation
• outbound_message_queue (Optional[OutboundMessageQueue]) – Custom persistent storage implementation

Raises:

• ValueError – Bad values provided for arguments.
• RuntimeError – An argument is missing its pair.

add_alarm(device_key: str, reference: str, active: bool, timestamp: Optional[int] = None) → None

Serialize alarm event and put into storage.

Alarms without a specified timestamp will be assigned a timestamps via int(round(time.time() * 1000))

Parameters:

• device_key (str) – Device on which the sensor reading occurred
• reference (str) – Alarm reference (unique per device)
• value – Current state of alarm
• timestamp (Optional[int]) – Unix time

Raises:

RuntimeError – Unable to place in storage

add_device(device: wolk_gateway_module.model.device.Device) → None

Add device to module.

Will attempt to send a registration request and update list of subscribed topics.

Parameters:

device (Device) – Device to be added to module

Raises:

• RuntimeError – Unable to store message
• ValueError – Invalid device given

add_sensor_reading(device_key: str, reference: str, value: Union[bool, int, float, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]], timestamp: Optional[int] = None) → None

Serialize sensor reading and put into storage.

Readings without a specified timestamp will be assigned a timestamps via int(round(time.time() * 1000))

Parameters:

• device_key (str) – Device on which the sensor reading occurred
• reference (str) – Sensor reference (unique per device)
• value (Union[bool,int,float,str,Tuple[int, int],Tuple[int, int, int],Tuple[float, float],Tuple[float, float, float],Tuple[str, str],Tuple[str, str, str],]) – Value(s) that the reading yielded
• timestamp (Optional[int]) – Unix time

Raises:

RuntimeError – Unable to place in storage

add_sensor_readings(device_key: str, readings: Dict[str, Union[int, float, bool, str, Tuple[int, int], Tuple[int, int, int], Tuple[float, float], Tuple[float, float, float], Tuple[str, str], Tuple[str, str, str]]], timestamp: Optional[int] = None) → None

Serialize multiple sensor readings and put into storage.

Readings without a specified timestamp will be assigned a timestamps via int(round(time.time() * 1000))

Parameters:

• device_key (str) – Device on which the sensor reading occurred
• readings (Dict[str,Union[bool,int,float,str,Tuple[int, int],Tuple[int, int, int],Tuple[float, float],Tuple[float, float, float],Tuple[str, str],Tuple[str, str, str],]) – dictionary in sensor_reference:value format
• timestamp (Optional[int]) – Unix time

Raises:

RuntimeError – Unable to place in storage

connect() → None

Establish connection with WolkGateway.

Will attempt to publish actuator statuses, configuration options, and current firmware version for all added devices.

Raises:RuntimeError – Error publishing actuator status or configuration

disconnect() → None

Terminate connection with WolkGateway.

publish(device_key: Optional[str] = None) → None

Publish stored messages to WolkGateway.

If device_key parameter is provided, will publish messages only for that specific device.

Parameters:device_key (Optional[str]) – Device for which to publish stored messages

publish_actuator_status(device_key: str, reference: str, state: Optional[wolk_gateway_module.model.actuator_state.ActuatorState] = None, value: Union[bool, int, float, str, None] = None) → None

Publish device actuator status to WolkGateway.

Getting the actuator status is achieved by calling the user’s implementation of actuator_status_provider or optionally an actuator status can be published explicitly by providing ActuatorState as state argument and the current actuator value via value argument

If message is unable to be sent, it will be placed in storage.

If no actuator_status_provider is present, will raise exception.

Parameters:

• device_key (str) – Device on which the sensor reading occurred
• reference (str) – Alarm reference (unique per device)
• state (Optional[ActuatorState]) – Current actuator state for explicitly publishing status
• value (Optional[Union[bool, int, float, str]]) – Current actuator value for explicitly publishing status

Raises:

• ValueError – Provided state is not an instance of ActuatorState
• RuntimeError – Unable to place in storage or no status provider

publish_configuration(device_key: str) → None

Publish device configuration options to WolkGateway.

If message is unable to be sent, it will be placed in storage.

Getting the current configuration is achieved by calling the user’s implementation of configuration_provider.

If no configuration_provider is present, will raise exception.

Parameters:device_key (str) – Device to which the configuration belongs to Raises:RuntimeError – No configuration provider present or no data returned

publish_device_status(device_key: str, status: Optional[wolk_gateway_module.model.device_status.DeviceStatus] = None) → None

Publish current device status to WolkGateway.

Getting the current device status is achieved by calling the user’s provided device_status_provider or a device status can be published explicitly by passing a DeviceStatus as the status parameter.

Parameters:

• device_key (str) – Device to which the status belongs to
• status (Optional[DeviceStatus]) – Current device status

Raises:

• ValueError – status is not of DeviceStatus
• RuntimeError – Failed to publish and store message

remove_device(device_key: str) → None

Remove device from module.

Removes device for subscription topics and lastwill message.

Parameters:device_key (str) – Device identifier