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)[source]¶ 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)[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, 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[source]¶ Serialize alarm event and put into storage.
Alarms without a specified timestamp will be assigned a timestamps via
int(round(time.time() * 1000))
Parameters: 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: - 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[source]¶ 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[source]¶ 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[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
-
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 providingActuatorState
asstate
argument and the current actuator value viavalue
argumentIf 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 aDeviceStatus
as thestatus
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