Shelly webhooks

SCSmartDevices supports webhooks for Shelly smart switches. When enabled, a webhook server is started to listen for webhook events posted by a Shelly device. For example, your application can be immediately notified when an input switch on a Shelly smart switch is turned on or off.

The SupportedWebhooks attrbute of a device object lists the webhook events that each device supports (if any). See this page for documentation: https://shelly-api-docs.shelly.cloud/gen2/ComponentsAndServices/Webhook#webhookcreate

To use webhooks you must:

1. Properly configure the ShellyWebhooks section of the SCSmartDevices configuration block.
2. Have your client app running on a system that accepts inbound http connections on the IP address and port configured in the ShellyWebhooks section.
3. Be using a Shelly device (typically Gen 3 or later) that supports wbhooks.
4. Add the Webhooks key to a device's input of output configuration so that webhook handlers are installed for that component.

Here's an example application:

  """Example of using the SmartDevice control to handle webhooks."""
  import platform
  import pprint
  import sys
  import threading
  import time

  from sc_foundation import SCLogger
  from switch_init import switch_init

  from sc_smart_device import SCSmartDevice

  # Test a Shelly switch
  device_identity = "Spello Test"  # Change this to the name of your device in the configuration file

  # Note: Webhooks are not supported on Tasmota devices, so no Tasmota test section for this example.


  def test_webhooks(logger: SCLogger, smart_switch_control: SCSmartDevice, wake_event: threading.Event):
      """Test function for webhooks SmartDevices control."""
      loop_delay = 2
      loop_count = 0
      max_loops = 20

      logger.log_message(f"\n\n\nTesting webhook functionality for device: {device_identity}", "summary")

      # Get the device
      try:
          device = smart_switch_control.get_device(device_identity)
          device_status = smart_switch_control.get_device_status(device)
          if device_status:
              logger.log_message(f"Device {device_identity} is online.", "summary")
          else:
              logger.log_message(f"Device {device_identity} is offline or not found.", "error")
      except RuntimeError as e:
          logger.log_message(f"Error getting status for device {device_identity}: {e}", "error")
          sys.exit(1)
      except TimeoutError as e:
          logger.log_message(f"Timeout error getting status for device {device_identity}: {e}", "error")
      else:
          logger.log_message(f"{device_identity} initial status:\n {smart_switch_control.print_device_status(device_identity)}", "detailed")

          while loop_count < max_loops:
              logger.log_message(f"Waiting for webhook events... (Loop {loop_count + 1}/{max_loops})", "detailed")

              if wake_event.is_set():
                  # We were woken by a webhook call
                  while True:
                      event = smart_switch_control.pull_webhook_event()
                      if not event:
                          break

                      event_name = event.get("Event")
                      event_device = event.get("Device", {}).get("Name")
                      event_component = event.get("Component", {}).get("Name")
                      logger.log_message(f"Received webhook event: Name: {event_name}, Device: {event_device}, Component: {event_component}", "detailed")
                      event_str = pprint.pformat(event, indent=2)
                      logger.log_message(f"\nWebhook event detail: {event_str}\n", "debug")
                  wake_event.clear()

              time.sleep(loop_delay)
              loop_count += 1


  def main():
      """Main function to run the example code."""
      wake_event = threading.Event()

      print(f"Hello from switch_webhooks running on {platform.system()}")

      # Initialize the configuration manager, logger, and SmartDevices control
      try:
          _config, logger, smart_switch_control = switch_init(wake_event=wake_event)
      except RuntimeError as e:
          print(f"Initialization error: {e}", file=sys.stderr)
          sys.exit(1)

      test_webhooks(logger, smart_switch_control, wake_event)


  if __name__ == "__main__":
      main()

Pulling specific events from the queue

Incoming webhook events are held in an internal FIFO queue. Calling pull_webhook_event() with no arguments returns and removes the oldest queued event (or None when the queue is empty).

If your app only cares about events for a particular device or component, you can pass one or more optional filters. pull_webhook_event() then returns the oldest event for which every supplied filter matches, removing just that event and leaving the rest of the queue untouched. Unsupplied filters are ignored, and None is returned when no queued event matches.

Argument	Type	Matches when
device_id	int \| None	event["Device"]["ID"] equals it
device_name	str \| None	event["Device"]["Name"] equals it
component_id	int \| None	event["Component"]["ID"] equals it
component_name	str \| None	event["Component"]["Name"] equals it

# Oldest event of any kind (unchanged, backward-compatible behaviour)
event = smart_switch_control.pull_webhook_event()

# Oldest event for the device named "Boiler"
event = smart_switch_control.pull_webhook_event(device_name="Boiler")

# Oldest event for one specific output of one specific device (AND)
event = smart_switch_control.pull_webhook_event(device_id=1, component_name="Output 1")

The same filtered signature is available on SCSmartDevice, SmartDeviceWorker and the underlying ShellyProvider.

Tasmota devices

Tasmota ESP32 devices don't support webhooks, but they do support signalling to a client app via Matter (MQTT) events. This will be supported in a later version of this package.