You may be interested in this post regarding the use of Python for rules.
It is not associated with a question.
I have several switchable sockets that offer the option of measuring current and power.
I connect these to Apple Home via openHAB using the HomeKit add-on.
As well as showing the switch status (Outlet.OnState characteristic in the HomeKit add-on), Apple HomeKit can display whether the outlet is in use (Outlet.InUseStatus characteristic in the HomeKit add-on).
I set Outlet.InUseStatus via a rule in openHAB:
if current is flowing,
Outlet.InUseStatus = ON;
otherwise,
Outlet.InUseStatus = OFF.
Below are two examples of my sockets from the items configuration.
I use the KNX binding from openHAB to control the sockets, but this is not relevant in this context.
items configuration
Group gCurrentItems "Items für die Strommessung zum Setzen von InUseStatus"
Group g1_8_3__AMI_0416_03 "PA 1.8.3, MDT AMI-0416.03"
Group:Switch:OR(ON,OFF) gOutlet_below_towel_warmer "Steckdose unterm Handtuchwärmer (Accessory)" { homekit="Outlet" }
Switch ch1_8_3___MD__1_M__10_MI__1_O__2__0_R__1_OnState "Steckdose unterm Handtuchwärmer Schalten [%s]" (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer) { homekit="Outlet.OnState", channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__10_MI__1_O__2__0_R__1" }
Switch ch1_8_3___MD__1_M__10_MI__1_O__2__0_R__1_InUseStatus "Steckdose unterm Handtuchwärmer in Nutzung [%s]" (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer) { homekit="Outlet.InUseStatus" }
Switch ch1_8_3___MD__1_M__10_MI__1_O__2__4_R__2 "Steckdose unterm Handtuchwärmer Sperren [%s]" (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer) { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__10_MI__1_O__2__4_R__2" }
Switch ch1_8_3___MD__1_M__10_MI__1_O__2__7_R__3 "Steckdose unterm Handtuchwärmer Status [%s]" (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer) { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__10_MI__1_O__2__7_R__3" }
Number ch1_8_3___MD__1_M__10_MI__1_O__2__12_R__28 "Steckdose unterm Handtuchwärmer Strommesswert [%.1f mA]" (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer, gCurrentItems) { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__10_MI__1_O__2__12_R__28" }
Switch ch1_8_3___MD__1_M__10_MI__1_O__2__17_R__25 "Steckdose unterm Handtuchwärmer Zähler zurücksetzen [%s]" (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer) { homekit="Switchable.OnState", expire="2s, command=OFF", channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__10_MI__1_O__2__17_R__25" }
Number ch1_8_3___MD__1_M__10_MI__1_O__2__18_R__26 "Steckdose unterm Handtuchwärmer Zähler Leistung [%.1f Wh]" <energy> (g1_8_3__AMI_0416_03, gOutlet_below_towel_warmer) ["EnergySensor"] { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__10_MI__1_O__2__18_R__26" }
Group:Switch:OR(ON,OFF) gTowel_warmer "Handtuchwärmer (Accessory)" { homekit="Outlet" }
Switch ch1_8_3___MD__1_M__4_MI__1_O__2__0_R__1_OnState "Handtuchwärmer Schalten [%s]" (g1_8_3__AMI_0416_03, gTowel_warmer) { homekit="Outlet.OnState", channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__4_MI__1_O__2__0_R__1" }
Switch ch1_8_3___MD__1_M__4_MI__1_O__2__0_R__1_InUseStatus "Handtuchwärmer in Nutzung [%s]" (g1_8_3__AMI_0416_03, gTowel_warmer) { homekit="Outlet.InUseStatus" }
Switch ch1_8_3___MD__1_M__4_MI__1_O__2__4_R__2 "Handtuchwärmer Sperren [%s]" (g1_8_3__AMI_0416_03, gTowel_warmer) { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__4_MI__1_O__2__4_R__2" }
Switch ch1_8_3___MD__1_M__4_MI__1_O__2__7_R__3 "Handtuchwärmer Status [%s]" (g1_8_3__AMI_0416_03, gTowel_warmer) { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__4_MI__1_O__2__7_R__3" }
Number ch1_8_3___MD__1_M__4_MI__1_O__2__12_R__28 "Handtuchwärmer Strommesswert [%.1f mA]" (g1_8_3__AMI_0416_03, gTowel_warmer, gCurrentItems) { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__4_MI__1_O__2__12_R__28" }
Switch ch1_8_3___MD__1_M__4_MI__1_O__2__17_R__25 "Handtuchwärmer Zähler zurücksetzen [%s]" (g1_8_3__AMI_0416_03, gTowel_warmer) { homekit="Switchable.OnState", expire="2s, command=OFF", channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__4_MI__1_O__2__17_R__25" }
Number ch1_8_3___MD__1_M__4_MI__1_O__2__18_R__26 "Handtuchwärmer Zähler Leistung [ %.1f Wh]" <energy> (g1_8_3__AMI_0416_03, gTowel_warmer) ["EnergySensor"] { channel="knx:device:mybridge1:knx2oh__1_8_3:ch1_8_3___MD__1_M__4_MI__1_O__2__18_R__26" }
The rule must respond to changes in the values of the items in the ‘gCurrentItems’ group, adjusting the corresponding values of the items with the HomeKit characteristic ‘Outlet.InUseStatus’ as necessary.
The logic of the rule
Here’s how the rule works.
If the value of an item in the ‘gCurrentItems’ group changes:
Find the other groups that the item is still contained in:
Find all items in each group:
Find the items with the HomeKit characteristic ‘homekit=“Outlet.InUseStatus”’, where the group definition contains the homekit-characteristic 'homekit="Outlet"' (There should only be one!):
Set items with 'homekit="Outlet.InUseStatus"' to ON if the value of the item in the 'gCurrentItems' group is greater than 0 (or any threshold value); otherwise, set them to OFF.
On System started event, create a mapping from the triggering item of the gCurrentItems group to the item to be set with the homekit Outlet.InUseStatus characteristic for quick access during operation and initialise the items with the homekit Outlet.InUseStatus characteristic.
Implementation
I like trying new things, so I decided to implement the rule in Python rather than in DSL, JavaScript or Ruby.
Version of Python Scripting add-on
The Python Scripting add-on in openHAB consists of two parts:
- the automation engine (the GraalVM/Jython integration), and the
- helper libraries.
The following versions are used
- The automation engine:
openhab-cli console → bundle:list | grep -i python → "… 5.0.3 …openHAB Add-ons: Bundles: Automation: Python Scripting”; i.e. 5.0.3 - Helper libraries:
grep -i version /etc/openhab/automation/python/lib/openhab/**\_*init*\_**.py → **\_*version*\_** = ‘1.0.15’ # The version string is for backward compatibility with OpenHAB 5.0.0; i.e. 1.0.15
I manually updated the helper libraries by copying the source code from the GitHub repository GitHub - openhab/openhab-python: openHAB Python Library for Python Scripting Automation to the /etc/openhab/automation/python/lib/openhab/ directory.
The following modules and methods have been effective for me
modules
from openhab import logger
from openhab.triggers import when
from openhab import rule, Registry
import scope
methods
<instance of group item> = Registry.getItem(<group_name>)
<list of item instances> = <instance of group item>.getMembers()
<item_name> = <instance of item>.getName()
<list of group_names> = <instance of item>.getGroupNames()
<instance of item> = Registry.getItem(<item_name>)
<instance of metadata> = <instance of item>.getMetadata()
<instance of homekit metadata> = <instance of metadata>.get('homekit')
<homekit value> = <instance of homekit metadata>.getValue()
<homekit configuration> = <instance of homekit metadata>.getConfiguration()
<homekit string> = <instance of homekit metadata>.toString()
<state> = Registry.getItem(<item_name>).getState()
<value> = <state>.floatValue()
<instance of event> = <instance of input>.get("event")
<item_name> = <instance of event>.getItemName()
Registry.getItem(<item_name>).sendCommand(scope.OFF)
Registry.getItem(<item_name>).sendCommand(scope.ON)
The rule
After gathering information about the versions, modules and methods, and noting down the logic in the above form, Copilot generated a usable framework for the desired rule which required minimal reworking.
The rule worked quickly and reliably, just as I had designed it.
The rule code
# file: set_inuse_status.py
from openhab import logger, Registry
from openhab.triggers import when
from openhab import rule
import threading
# --- Configuration ---
G_CURRENT_ITEMS = "gCurrentItems"
INUSE_THRESHOLD = 0.0 # > threshold => ON
# --- Module state ---
_group_to_inuse_item = {} # normalized group name -> Item instance (the InUseStatus item)
_map_lock = threading.Lock()
_map_built = False
# --- Helpers ---
def _norm(name):
return str(name).strip().lower()
def _get_homekit_value(item):
"""
Return the homekit metadata value (e.g. "Outlet.InUseStatus" or "Outlet") or None.
Uses exactly: item.getMetadata().get('homekit').getValue()
"""
md = item.getMetadata()
if md is None:
return None
hk = md.get("homekit")
if hk is None:
return None
val = hk.getValue()
if val is None:
return None
return str(val).strip()
def build_group_to_inuse_map():
"""
Build mapping: normalized group name -> Item instance with homekit='Outlet.InUseStatus',
but only for groups whose group item has homekit='Outlet'.
"""
global _group_to_inuse_item, _map_built
logger.info("build_group_to_inuse_map: starting")
new_map = {}
all_items = Registry.getItems()
for it in list(all_items):
hk_val = _get_homekit_value(it)
if hk_val is None:
continue
if hk_val.strip().lower() != "outlet.inusestatus":
continue
group_names = list(it.getGroupNames())
for g in group_names:
group_item = Registry.getItem(g)
if group_item is None:
continue
group_hk = _get_homekit_value(group_item)
if group_hk is None:
continue
if group_hk.strip().lower() != "outlet":
continue
gn = _norm(g)
new_map[gn] = it
logger.debug(f"Mapped group {gn} -> inuse item {it.getName()} (group homekit={group_hk})")
with _map_lock:
_group_to_inuse_item.clear()
_group_to_inuse_item.update(new_map)
_map_built = True
logger.info(f"build_group_to_inuse_map: finished, mapped {len(_group_to_inuse_item)} groups")
def refresh_group_to_inuse_map():
"""Public refresh function callable from console or other rules."""
build_group_to_inuse_map()
logger.info("refresh_group_to_inuse_map: refreshed mapping")
def _find_inuse_items_for_current_item(current_item):
"""
Return list of InUseStatus items for groups the current_item belongs to.
Only groups that were mapped (i.e. group has homekit='Outlet') are returned.
"""
group_names = list(current_item.getGroupNames())
found = []
with _map_lock:
for g in group_names:
iu = _group_to_inuse_item.get(_norm(g))
if iu:
found.append(iu)
return found
def _state_to_onoff_using_float(state):
"""Use state.floatValue() and convert to 'ON'/'OFF'."""
v = state.floatValue()
return "ON" if v > INUSE_THRESHOLD else "OFF"
# --- Startup: build map and initialize InUseStatus items ---
@rule("Build InUseStatus lookup and initialize at startup")
@when("System started")
def on_system_started(module, input):
build_group_to_inuse_map()
logger.info("on_system_started: initializing InUseStatus items from current measurements")
group = Registry.getItem(G_CURRENT_ITEMS)
if group is None:
logger.info(f"on_system_started: group {G_CURRENT_ITEMS} not found")
return
members = list(group.getMembers())
for m in members:
state = m.getState()
onoff = _state_to_onoff_using_float(state)
inuse_items = _find_inuse_items_for_current_item(m)
if not inuse_items:
logger.debug(f"on_system_started: no InUseStatus item for {m.getName()} (groups: {list(m.getGroupNames())})")
continue
for iu in inuse_items:
if onoff == "ON":
iu.sendCommand("ON")
else:
iu.sendCommand("OFF")
logger.info(f"Initialized {iu.getName()} -> {onoff} (from {m.getName()}={state})")
# --- Rule: reacts to changes in gCurrentItems ---
@rule("Set Outlet.InUseStatus from current measurements")
@when(f"Member of {G_CURRENT_ITEMS} changed")
def on_current_changed(module, input):
# input is a Java HashMap in some environments; extract the event first
event = input.get("event")
if event is None:
return
item_name = event.getItemName()
logger.debug(f"triggering item: {item_name})")
current_item = Registry.getItem(item_name)
if current_item is None:
logger.warning(f"on_current_changed: Registry.getItem({item_name}) returned None")
return
state = current_item.getState()
onoff = _state_to_onoff_using_float(state)
inuse_items = _find_inuse_items_for_current_item(current_item)
if not inuse_items:
logger.debug(f"on_current_changed: no InUseStatus item found for {item_name} (groups: {list(current_item.getGroupNames())})")
return
for iu in inuse_items:
if onoff == "ON":
iu.sendCommand("ON")
else:
iu.sendCommand("OFF")
logger.info(f"Set {iu.getName()} -> {onoff} (triggered by {item_name}={state})")
I can see the desired updates on Outlet.InUseStatus in OpenHAB.
However, the updates to Outlet.InUseStatus are not coming through in my Apple Home app (v. 26.2). This is unfortunate, but it does not seem to be related to the rule.
I covered this issue in Outlet.InUse Characteristic not propagated to Apple Home .
The advantages and benefits
The biggest benefit of the design is that I can now create any Outlet accessories in openHAB without adjusting the rule code. I just need to use the correct group assignments.
The rule only has one hard-wired parameter: the name of the group summarising the current values of the individual Outlets. In my case, this is gCurrentItems.