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, 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, 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)[source]

This class is the 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, 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, 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)[source]

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, 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
add_alarm(device_key: str, reference: str, active: bool, timestamp: Optional[int] = None) → None[source]

Serialize alarm event and put into storage.

Storing alarms without Unix timestamp will result in all sent messages being treated as live and will be assigned a timestamp upon reception, so for a valid history add timestamps to alarms 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[source]

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:
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[source]

Serialize sensor reading and put into storage.

Storing readings without Unix timestamp will result in all sent messages being treated as live readings and will be assigned a timestamp upon reception, so for a valid history add timestamps to readings via int(round(time.time() * 1000))

Parameters:
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[source]

Serialize multiple sensor readings and put into storage.

Storing readings without Unix timestamp will result in all sent messages being treated as live readings and will be assigned a timestamp upon reception, so for a valid history add timestamps to readings via int(round(time.time() * 1000))

Parameters:
Raises:

RuntimeError – Unable to place in storage

connect()[source]

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()[source]

Terminate connection with WolkGateway.

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

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[source]

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 parameter and the current actuator value via value parameter

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[source]

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[source]

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:
remove_device(device_key: str) → None[source]

Remove device from module.

Removes device for subscription topics and lastwill message.

Parameters:device_key (str) – Device identifier