Scripting Shelly relay devices in Indigo

Home Automation

This is a proof-of-concept for scripting Shelly relay devices in an Indigo Python action.

I’ve used the Indigo macOS home automation software for many years. It’s a deep, extensible and reliable piece of software. Among the extensible features of the application is its suite of community-supported plugins. There is a plugin for Shelly devices, but it supports only earlier devices and not the current units. As I understand it, the author does not intend to update the plugin. In this post, I’ll show a method for controlling these devices without plugins. The shortcoming here is that the Shelly device doesn’t have a corresponding Indigo device, and everything is handled through action groups and variables.

To temper expectations, I only have access to a Shelly 1PM UL single relay device. The scripts are specific to that device. Everything here is extensible, though, and you can feel free to work with the code to shape it to fit your needs. It’s just a starting point. Note that I’m not currently using authentication on my Shelly relay. If someone wants to hack into my network and turn on my outside lights, have at it! Your use case may be different, of course.

For this device we will provide the ability to turn the device on and off, to toggle it, and to copy certain data (voltage, current, and recent power consumption) into Indigo variables.

Core functionality

Because many installations will have more than one of these devices, core functionality is abstracted into utilities and the class ShellyClient located at /Library/Application Support/Perceptive Automation/Python3-includes/shelly_utils.py. Code located here can be imported and reused in Indigo Python script actions to reduce repetition.

# shelly_utils.py — shared utilities for Shelly Plus (Gen2) via HTTP RPC.

# Explicitly import indigo to access log, variables, etc.
try:
   import indigo  # provided by Indigo scripting runtime
except Exception:
   indigo = None

DEBUG_SHELLY_UTILS = False  # set True to allow the self-test log

def log(msg, is_error=False):
   """
   Prefer Indigo logger; fallback to UTF-8 stdout only if Indigo isn't present.
   Do not swallow Indigo exceptions—surface them.
   """
   if indigo is not None:
      indigo.server.log(str(msg), isError=is_error)
      return

   # Fallback: outside Indigo (e.g., unit tests)
   import sys
   text = str(msg) + "\n"
   try:
      sys.stdout.write(text)
   except UnicodeEncodeError:
      sys.stdout.buffer.write(text.encode("utf-8", "replace"))

# self-test so you can confirm the module actually logs on import
if DEBUG_SHELLY_UTILS:
   try:
      log("[shelly_utils] module loaded; Indigo logging path OK.")
   except Exception as e:
      # if this throws, you'll at least get a Script Error from Indigo
      raise


import requests

def set_var(name: str, value) -> None:
   """Create/update an Indigo variable; booleans become 'true'/'false'."""
   if isinstance(value, bool):
      s = "true" if value else "false"
   elif value is None:
      s = ""
   else:
      s = str(value)
   try:
      if name in indigo.variables:
         indigo.variable.updateValue(name, s)
      else:
         indigo.variable.create(name, value=s)
   except Exception as e:
      log(f"Failed to set variable '{name}': {e}", is_error=True)

class ShellyClient:
   """
   Minimal HTTP RPC client for Shelly Plus 1PM UL (Gen2).
   Exposes high-level helpers that both command and update Indigo variables.
   """

   def __init__(self, ip: str, name: str = "Shelly"):
      self.ip = ip
      self.name = name
      self.base = f"http://{ip}/rpc"

   # ---------- Low-level RPC ----------
   def _get(self, method: str, params: dict | None = None, timeout=5) -> dict:
      r = requests.get(f"{self.base}/{method}", params=params or {}, timeout=timeout)
      r.raise_for_status()
      return r.json()

   def get_status(self, timeout=5) -> dict:
      return self._get("Shelly.GetStatus", timeout=timeout)

   def get_output_state(self, timeout=5) -> bool | None:
      try:
         data = self.get_status(timeout=timeout)
         return bool(data["switch:0"]["output"])
      except Exception as e:
         log(f"{self.name}: status read failed — {e}", is_error=True)
         return None

   # ---------- Internal helpers ----------
   def _update_common_vars_from_status(self, data: dict, prefix: str) -> None:
      sw = data.get("switch:0", {})
      temp_c = (sw.get("temperature") or {}).get("tC")
      voltage = sw.get("voltage")
      current = sw.get("current")
      output  = sw.get("output")

      aenergy = sw.get("aenergy") or {}
      by_minute = aenergy.get("by_minute") or []
      last_min_energy = by_minute[-1] if by_minute else None

      set_var(f"{prefix}Temperature", temp_c)
      set_var(f"{prefix}Voltage", voltage)
      set_var(f"{prefix}Current", current)
      set_var(f"{prefix}InstantaneousEnergy", last_min_energy)
      set_var(f"{prefix}RelayState", output)

   def _command_and_update(self, target_on: bool, prefix: str, timeout=5) -> bool | None:
      """
      Send Switch.Set to target_on, verify via fresh status, update variables,
      and log outcome. Returns final True/False state, or None if unknown/error.
      """
      before = self.get_output_state(timeout=timeout)
      if before is not None:
         log(f"{self.name}: current is {'ON' if before else 'OFF'}; requesting "
             f"{'ON' if target_on else 'OFF'}.")

      # Send command
      try:
         self._get("Switch.Set", params={"id": 0, "on": "true" if target_on else "false"}, timeout=timeout)
      except requests.RequestException as e:
         log(f"{self.name}: Switch.Set failed — {e}", is_error=True)
         if before is not None:
            set_var(f"{prefix}RelayState", before)
         return None

      # Verify and update all variables
      try:
         data = self.get_status(timeout=timeout)
      except requests.RequestException as e:
         log(f"{self.name}: command sent, but verification failed — {e}", is_error=True)
         if before is not None:
            set_var(f"{prefix}RelayState", before)
         return None

      self._update_common_vars_from_status(data, prefix)
      after = bool(data.get("switch:0", {}).get("output"))
      log(f"{self.name}: now {'ON' if after else 'OFF'}; {prefix}RelayState="
          f"{'true' if after else 'false'}.")
      return after

   # ---------- Public high-level helpers ----------
   def update_vars(self, prefix: str, timeout=5) -> None:
      """Fetch status and refresh Indigo variables with the given prefix."""
      try:
         data = self.get_status(timeout=timeout)
      except requests.RequestException as e:
         log(f"{self.name}: status fetch failed — {e}", is_error=True)
         return
      self._update_common_vars_from_status(data, prefix)
      sw = data.get("switch:0", {})
      last_min = (sw.get("aenergy") or {}).get("by_minute", [None])[-1]
      log(f"{self.name}: vars updated (Temp {((sw.get('temperature') or {}).get('tC'))}°C, "
          f"{sw.get('voltage')} V, {sw.get('current')} A, 1-min {last_min}).")

   def toggle_and_update(self, prefix: str, timeout=5) -> bool | None:
      """Toggle relay id 0, then verify and update variables."""
      before = self.get_output_state(timeout=timeout)
      if before is not None:
         log(f"{self.name}: toggling from {'ON' if before else 'OFF'}.")

      try:
         self._get("Switch.Toggle", params={"id": 0}, timeout=timeout)
      except requests.RequestException as e:
         log(f"{self.name}: Switch.Toggle failed — {e}", is_error=True)
         if before is not None:
            set_var(f"{prefix}RelayState", before)
         return None

      try:
         data = self.get_status(timeout=timeout)
      except requests.RequestException as e:
         log(f"{self.name}: toggle sent, but verification failed — {e}", is_error=True)
         if before is not None:
            set_var(f"{prefix}RelayState", before)
         return None

      self._update_common_vars_from_status(data, prefix)
      after = bool(data.get("switch:0", {}).get("output"))
      log(f"{self.name}: toggled to {'ON' if after else 'OFF'}; {prefix}RelayState="
          f"{'true' if after else 'false'}.")
      return after

   def set_on_and_update(self, prefix: str, timeout=5) -> bool | None:
      """Command ON and update variables. Returns final state True/False or None on error."""
      return self._command_and_update(True, prefix, timeout=timeout)

   def set_off_and_update(self, prefix: str, timeout=5) -> bool | None:
      """Command OFF and update variables. Returns final state True/False or None on error."""
      return self._command_and_update(False, prefix, timeout=timeout)

To use the core functionality in an action script, we just need to import shelly_utils and all of the utility functions and ShellyClient class and its instance methods are available to us.

Use in action scripts

To use this shared functionality in an Indigo action, import the utilities, create a client and execute a method on the client.

Turn relay on

from shelly_utils import ShellyClient
ShellyClient("192.168.5.216", "Driveway Lamps").set_on_and_update("drivewayShelly")

This will turn the relay on if it is currently off and will update or create variables for status parameters. These variables begin with the prefix drivewayShelly. By the way, don’t forget to assign an IP reservation for your device.

Turn relay off

from shelly_utils import ShellyClient
ShellyClient("192.168.5.216", "Driveway Lamps").set_off_and_update("drivewayShelly")

Toggle relay

from shelly_utils import ShellyClient
ShellyClient("192.168.5.216", "Driveway Lamps").toggle_and_update("drivewayShelly")

Update variables without changing relay state

from shelly_utils import ShellyClient
ShellyClient("192.168.5.216", "Driveway Lamps").update_vars("drivewayShelly")

Conclusion

This approach avoids the dependency on plugins, applies good code reuse principles, and works! This is obviously just a start and I hope you find it useful.