How to cancel my light pulsing in my lights python rules

Setup, Configuration and Use Scripts & Rules
,
1
  • Platform information:
    • Hardware: MacOS
    • OS: OsX
    • Java Runtime Environment: Oracle JDK
    • openHAB version: 2.5.8

Hi, I recently started my personal OpenHAB project, at the moment it’s running on raspberry PI with bunch of zwave switchers and sensors (Fibaro ones). So far I implemented just couple of very simple rules to switch off lights when no motion in the room.
However, I do have specific case in one of the bathroom, where motion sensor sometimes is unable to detect movement if somebody is taking a bath :slight_smile: It was pretty annoying so I decided to experiment with more robust rules and switched to python.
My theory of the operation was:

  • Assuming bathroom light is on
  • If motion sensor report no movement (OFF command) then I’m starting timer to wait couple of seconds
  • If that time is elapsed, I wanted to warn the person (if anyone is still at bath) the light is about to be switched off by pulsing the light (since I don’t have a dimmer, just switch actuator) - so he/she could simply wave a hand or just move to keep lights on :slight_smile:

I got to the point where I’m able to pulse that lite once all the timers are done

  • but struggling to keep having lights ON, if motion sensor reports ON (movement reported) during that pulsing period
  • the result is imply light switch going OFF and need to get it ON manually

It seems that I probably overcomplicated couple of things (as I experimented with various concepts…), anyway help or pointers are appreciated :slight_smile:

My configurations are (at the moment it’s just items without thing bindings as I wanted to experiment locally first)
my.items

Switch aKidsBath_Light_Switch "Kids Bathroom - Switch" <light> (gLights)
Switch aKidsBath_Light_Switch_Off_Warn (gSwitchOffWarn)//virtual switch to warn light will go off
Switch vKidsBath_Motion "Kids Bathroom - Motion" <motion> (gMotionSensors) { 
    Static="link"[ 
        actuator = "aKidsBath_Light_Switch", 
        luminance = "vKidsBath_Luminance"
    ]
}
Number vKidsBath_Temperature "Kids Bathroom - Temperature [%.1f °C]" <temperature> (gTemperatureSensors)
Number vKidsBath_Luminance "Kids Bathroom - Luminance [%d lux]" (gLuminanceSensors)

Switch aKajaRoom_Light_Switch "Kaja Room - Switch" <light> (gLights)
Switch aKajaRoom_Light_Switch_Off_Warn (gSwitchOffWarn)

Switch vKajaRoom_Motion "Kaja Room - Motion" <motion> (gMotionSensors) { 
    Static="link"[ 
        actuator = "aKajaRoom_Light_Switch", 
        luminance = "vKajaRoom_Luminance"
    ]
}
Number vKajaRoom_Temperature "KajaRoom - Temperature [%.1f °C]" <temperature> (gTemperatureSensors)
Number vKajaRoom_Luminance "KajaRoom - Luminance [%d lux]" (gLuminanceSensors)

Group gMotionSensors
Group gTemperatureSensors
Group gLuminanceSensors
Group gLights
Group gSwitchOffWarn

demo.sitemap

sitemap demo label="Sitemap Demo" {
    Frame label="Kids Bath" icon="bath" {
        Default item=aKidsBath_Light_Switch label="Kids Bathroom - Switch"
        Default item=vKidsBath_Luminance label="Kids Bathroom - Luminance"
        Default item=vKidsBath_Motion label="Kids Bathroom - Motion"
        Default item=vKidsBath_Temperature label="Kids Bathroom - Temperature"
    }
    Frame label="Kaja Room" icon="room" {
        Default item=aKajaRoom_Light_Switch label="Kaja Room - Switch"
        Default item=vKajaRoom_Luminance label="Kaja Room - Luminance"
        Default item=vKajaRoom_Motion label="Kaja Room - Motion"
        Default item=vKajaRoom_Temperature label="Kaja Room - Temperature"   
    }
}

And the lights.py rules file

from core.rules import rule
from core.triggers import when
from core.log import log_traceback
from core.actions import ScriptExecution
from core.metadata import get_all_namespaces, get_key_value, get_metadata, get_value, set_key_value
from org.joda.time import DateTime

occupancyTimers = {}
offWarnTimers = {}

def get_name(itemName): return get_key_value(itemName, "Config", "name") or itemName

def is_changed_from_undef_or_null(event): return isinstance(event.oldItemState, UnDefType) or event.oldItemState == "NULL"

def get_actuator(itemName): return get_key_value(itemName, "Static", "actuator") or UNDEF

def clean_timer(timers, timerName):
    if timerName in timers:
        timers[timerName].cancel()
        del timers[timerName]
    return

@rule("Lights_NoMotion", description="If no movement, alert with lights and then switch off if still no movement", tags=["lights"])
@when("Member of gMotionSensors changed")
def motion_changed(event):
    light_switch_name = get_actuator(event.itemName)
    timerName = event.itemName

    if event.itemState == ON:
        if items[light_switch_name+"_Off_Warn"] == ON: # If switch off warning is active, stop it
            motion_changed.log.debug("Drop Off Warn state")
            clean_timer(offWarnTimers, light_switch_name)
            events.postUpdate(light_switch_name+"_Off_Warn", "OFF")
            events.sendCommand(light_switch_name, "ON")
        if items[light_switch_name] == OFF:
            motion_changed.log.debug("Switch {} = {}, Motion {} = {}. Do nothing".format(light_switch_name, items[light_switch_name], event.itemName, event.itemState))
            clean_timer(occupancyTimers, timerName)

    if event.itemState == OFF and items[light_switch_name] == ON:
        motion_changed.log.debug("Motion {} is OFF and Switch {} is ON. Go with timers".format(event.itemName, light_switch_name))
        if timerName not in occupancyTimers or occupancyTimers[timerName].hasTerminated():
            motion_changed.log.debug("Scheduling timer for {}".format(timerName))
            occupancyTimers[timerName] = ScriptExecution.createTimer(DateTime.now().plusSeconds(2), lambda aItemName=light_switch_name: events.sendCommand(aItemName+"_Off_Warn", "ON"))
        else:
            motion_changed.log.debug("Re-Scheduling timer for {}".format(timerName))
            occupancyTimers[timerName].reschedule(DateTime.now().plusSeconds(2))


def pulse_warn(timeStarted, warnItemName):
    switch_item_name = warnItemName.split("_Off_Warn", 1)[0]

    events.sendCommand(switch_item_name, "ON" if items[switch_item_name] == OFF else "OFF")
    if DateTime.now().isBefore(timeStarted.plusMillis(1000)):
        offWarnTimers[warnItemName].reschedule(DateTime.now().plusMillis(200))
    else:
        offWarnTimers[warnItemName].cancel()
        del offWarnTimers[warnItemName]
        events.sendCommand(switch_item_name, "ON")
        events.sendCommand(warnItemName, "OFF")

@rule("Light_pulsing", description="Light pulsing", tags=["lights"])
@when("Member of gSwitchOffWarn received command ON")
def pulse_light(event):
    warn_item_name = event.itemName
    if warn_item_name not in offWarnTimers:
        timeStarted = DateTime.now()
        offWarnTimers[warn_item_name] = ScriptExecution.createTimer(DateTime.now(), lambda ts=timeStarted, sw=warn_item_name: pulse_warn(timeStarted, sw))

@rule("End pulsing. Turn of lights", description="Lights off", tags=["lights"])
@when("Member of gSwitchOffWarn received command OFF")
def end_pulse(event):
    switch_item_name = event.itemName.split("_Off_Warn", 1)[0]
    events.sendCommand(switch_item_name, "OFF")

Thanks

2

You can clean up some of your timer management code if you adopt my timer_mgr library which you can find at https://github.com/rkoshak/openhab-rules-tools. Since writing that I’ve saved tons of lines of code and made many of my complicated rules much simpler. It also makes the code much shorter and easy to understand when I can use “5s” instead of Datetime.now().plusSeconds(5).

I’ve also got a debounce library there which might be useful here.

Anyway, looking at your code…

When the motion is ON and the light is OFF already it doesn’t make sense to me that you would cancel the Timer and do nothing. Wouldn’t you want to turn ON the light (if not already ON) in that case?

So we have when the motion sensor goes to OFF and the light is ON we send command ON to the Off_Warn Item (after a few seconds). That triggers the Light_pulsing rule which calls pulse_warn that toggles the light and reschedules if it’s been less than a second. So the light toggles every 200 msec for a second before exiting the looping timer and sending OFF to the switch_item and Off_Warn Item.

If I understand your problem correctly, the problem is when you move again you want to cancel the warning flashing and leave the light ON. And I can see that is not happening with these rules. The way it’s written the only way to exit the looping timer is for the one second to expire and at the end everything is turned OFF. At a minimum the if in pulse_warn should also check the state of the Off_Warn` Item and not reschedule if that Item has become OFF.

However, event there we have a problem because when you command the Item to OFF, the light ends up being set to OFF too, regardless of the state of the motion sensor.

You need an alternative way to tell the looping timer to exit but leave the light ON. I can think of lots of ways to do this but I think the simplest would be to change your motion_changed rule to cancel the Off_Warn timer itself when the ON motion is receivied and then make sure that the light remains or is turned ON. You want to do it from that rule because there could be a race condition where Motion ON is received while the timer is running. And as far as I can tell, the motion Item knows about the light, but the light doesn’t know about the motion Item (though you could add that of course).

Having said all of that, here is how I would probably implement something like this (NOTE this will be untested code that likely contains typos). It’s going to use my rules tools libraries.

Switch vKidsBath_Motion "Kids Bathroom - Motion" <motion> (gMotionSensors) {
    Static="link"[
        actuator = "aKidsBath_Light_Switch",
        liminance = "vKidsBath_Luminace"
    ]
}
Switch vKidsBath_Motion_Raw {
    channel="blah:blah:blah",
    debounce="vKidsBath_Motion"[
        timeout="2s",
        state="OFF"
    ]
}

The above uses the debounce library to wait two seconds before updating vKidsBath_Motion to OFF. This implements the two second delay before starting to flash the lights when the motion sensor goes OFF. Only the OFF state is debounced in this way. ON will immediately be forwarded to vKidsBath_Motion.

from org.joda.time import DateTime
from core.rules import rule
from core.triggers import when
from core.metadata import get_key_value
from core.utils import send_command_if_different
from community.timer_mgr import TimerMgr
from community.looping_timer import LoopingTimer # TimerMgr can't handle looping timers

occupancy_timers = TimerMgr()
off_warn_timers = {}

def get_actuator(itemName):
  return get_key_value(itemName, "Static", "actuator") or None

def pulse_warn(end_time, light):
    """ Called by the looping timer, toggles the given light on and off every 200 msec
    for up to 1 second.
    Arguments:
        started_at: DateTime indicating when the loop was started
        light: the light switch to toggle
    """

    if DateTime.now().isAfter(end_time): # should be moved to Java DateTime before OH 3
        del off_warn_timers[light]
        return None # cancels the looping
    else:
        cmd = "ON" if items[light] == "OFF" else "OFF"
        events.sendCommand(light, cmd)
        return 200 # looping timer will treat integers as milliseconds

@rule("Lights_NoMotion",
           description="If no movement, alert with lights and then switch off is still no movement",
           tags=["lights"])
@when("Member of gMotionSensors changed")
def motion_changed(event):
    light_switch_name = get_actuator(event.itemName)
    timer_name = event.itemName

    # Someone is present
    if event.itemState == ON:

        # cancel the off_warn_timer is it is running
        if timer_name in off_warn_timers:
            off_warn_timers[light_switch_name].cancel()
            del off_warn_timers[light_switch_name]

       # Turn on the light if it isn't already on
       send_command_if_different(light_switch_name, "ON")

        # Schedule/reschedule the OFF timer
        occupancyTimers.check(key=timer_name, when="2s", 
                              function=lambda: events.sendCommand(light_switch_name, "OFF"),
                              reschedule=True)

    # No motion
    else:
    
        # Check for the error case where the off_warn_timer is running when the motion sensor goes off
        if timer_name in off_warn_timers:
            motion_changes.log.warn("Motion sensor went to OFF but there is a warning Timer running! This shouldn't be possible")
            off_warn_timers[timer_name].cancel()
    
        # Create the warning flashing
        off_warn_timers[timer_name] = LoopingTimer(lambda: pulse_warn(DateTime.now().plusSeconds(1)), 
                                                                      light_switch_name)

def scriptUnloaded():
    for t in off_warn_timers:
        t.cancel()
    occupancy_timers.cancelAll()

By eliminating the Off_Warn Item I think it becomes a little simpler. A lot of the book keeping stuff involved with creating and managing the Timers goes away with the use of the libraries.

When the Motion sensor goes ON, if the off warning was flashing that looping timer is canceled and removed. Then turn ON the light if it’s not already ON. Finally, create or reschedule the OFF timer. TimerMgr.check will create and keep track of the timers. The first argument is the key for the Timer, second is how long from now the timer should trigger (supports all sorts of ways to define that), the third is the lambda to call when the timer triggers and reschedule=True will cause TimerMgr to reschedule the timer if it already exists.

When the Motion sensor goes OFF, it actually went off two seconds ago thanks to the debounce. So we can immediately start the looping timer that will flash the light. This creates a LoopingTimer which handles all the book keeping stuff. You pass it a lambda and optionally a when it should first run. We want to run immediately so we only pass the lambda. The lambda needs to return a when (anything supported by TimerMgr like “2s”, DateTimes, integer number of milliseconds) or None when it should stop looping. So all you need to define is the lambda yourself. In this case the lambda checks to see if a second has passed and if so returns None, exiting the loop. If not, toggles the light and returns 200 which reschedules the timer to run the lambda in 200 msecs.

Should motion go back to ON while the off timer is looping, the timer is cancelled which will stop the toggling. No matter what, when motion goes to ON, the light is turned back ON and an off timer created/rescheduled.

Make sure to cancel the timers on unload to avoid errors on a reload of the .py file.

One reason I wrote these libraries and made them available is because it handles so much of the common book keeping that everyone needs to do when dealing with timers like this so all you have to focus on is your specific problem instead of remembering whether or not you deleted the timer at the right point and such.

1 Like
3

This is a hardware solution, but I faced the same problem. Eventually I bought a door sensor, so that if the motion stops, the light does not go out if the door is closed. The door being closed is a pretty good proxy for someone still in the bathroom in our situation. My motion detector cannot tell if someone is in the shower and it was really annoying when the light went out while in the shower. This may or may not work well for you, but it’s helped us.

4

Yes, I was thinking about door sensor. It’s going to be my try in the following iterations :slight_smile:

5

Thanks @rlkoshak - Your answer is far beyond of what I expected to find :slight_smile: Anyway, you gave me a great pointers and ideas. I already started looking at the libs at your github. And I’d say expire and debounce rules are just great, solves a lot of issue. Thanks for pointing it out.
Anyway, I worked on the foundations of your answer and finally got it working perfectly (exactly that way you understood). The nicest thing is that it’s just one rule, so much more easier to understand it.

So here is my code just for a reference.

from org.joda.time import DateTime
from core.rules import rule
from core.triggers import when
from core.metadata import get_key_value
from core.utils import send_command_if_different
from core.log import log_traceback

from community.timer_mgr import TimerMgr
from community.looping_timer import LoopingTimer # TimerMgr can't handle looping timers

# Configurations
LIGHT_OFF_TIMER="30s" # Switch off light after that duration
LIGHT_PULSING_DURATION=2 # Duration of pulsing in seconds
LIGHT_PULSE_PERIOD=200 # Pulsing period in milliseconds

###
occupancy_timers = TimerMgr()
pulsing_timers = {}

def get_actuator(itemName): 
    """ Gets the linked actuator name from the given item. It assumes the actuator name
    is provided in a Static metadata 'actuator'.
    Arguments:
        itemName: Name of the item an actuator link is to be retrieved
    """
    return get_key_value(itemName, "Static", "actuator") or None

def pulse_light(end_time, light):
    """ Called by the looping timer, toggles the given light on and off every 200 msec
    for up to 1 second.
    Arguments:
        end_time: DateTime indicating when the loop should stop
        light: the light switch to toggle
    """
    now = DateTime.now()
    if DateTime.now().isAfter(end_time): # should be moved to Java DateTime before OH 3
        remove_key_value(light, "Mode", "Pulse")
        send_command_if_different(light, "OFF")
        return None # cancels the looping
    else:
        events.sendCommand(light, "ON" if items[light] == OFF else "OFF")
        return LIGHT_PULSE_PERIOD # looping timer will treat integers as milliseconds

@log_traceback
def cancel_pulsing_timer(timerName, log_function=None):
    global pulsing_timers

    if timerName in pulsing_timers:
        if log_function:
            log_function()
        pulsing_timers[timerName].cancel()
        del pulsing_timers[timerName]
    return

def switch_off_light(motion_item_name, light_switch_name):
    # Turn of light only if no motion reported.
    if items[motion_item_name] == OFF:
        send_command_if_different(light_switch_name, "OFF")

@rule("Lights_NoMotion", description="If no movement, alert with lights and then switch off if still no movement", tags=["lights"])
@when("Member of gMotionSensors changed")
def motion_changed(event):
    light_switch_name = get_actuator(event.itemName)
    item_name = event.itemName

    # Someone is present in the room
    if event.itemState == ON:
        # Stop lights pulsing, if any
        cancel_pulsing_timer(item_name)
        #set_key_value(light_switch_name, "Mode", "on")

        # Turn on the light if it isn't already on. Situations it can happen
        # - Someone went to the room with lights OFF
        # - Someone just moved in the room when the lights were pulsing and once canceled it was OFF that time
        send_command_if_different(light_switch_name, "ON")

        # - Someone just went into the room, lights went ON, scheduling timer for X period to switch them OFF
        # - Or, Sensor showed no motion, but person just moved so we need to re-schedule timer to extend ON period.
        occupancy_timers.check(key=item_name, when=LIGHT_OFF_TIMER, 
                              function=lambda mo=item_name, act=light_switch_name : switch_off_light(mo, act),
                              reschedule=True)
    # No motion
    else:
        # Check for the error case where the off_warn_timer is running when the motion sensor goes off
        cancel_pulsing_timer(item_name, lambda: motion_changed.log.warn("Motion sensor went to OFF but there is a warning Timer running! This shouldn't be possible"))
        # Create the warning flashing
        if items[light_switch_name] == ON:
            pulsing_timers[item_name] = LoopingTimer(lambda switch_name=light_switch_name, end_time=DateTime.now().plusSeconds(LIGHT_PULSING_DURATION): pulse_light(end_time, switch_name))

def scriptUnloaded():
    for t in pulsing_timers:
        t.cancel()
    occupancy_timers.cancel_all()

By the way, my next step is to add more robustness to those rules. I’m looking for a things like:

  • Assuring the light will be auto ON only if it’s dark in the room
  • Turning off lights because of the bed time
    But already saw a couple of interesting posts on that forum that might help me.
    Anyway, thank you for your support !
    Marcin
1 Like
6

I did more refactor to the code as it was still having some issues. I didn’t like the fact the occupancy_timers was kind of independent to the pulsing_timers. So, now the operation of my rule is as follows (and the rule code is simpler):

  • If motion goes OFF (after debounce period), start a timer that once expired will start Warning period
  • On Warning period pulse the light and finally switch OFF taking into account current state of the motion (so if motion become ON before starting pulsing do nothing)
  • Once the motion is ON - just turn ON the light and cancel all the timers so the warn period, etc. will be abandoned
    Additionally, I added support for Dimmers. So if the group has mixed Switch’es as well as Dimmers.
  • Switches will keep pulsing
  • While, Dimmers will start dimming down to 0%

Hope it will help anyone.

lights.py

from org.joda.time import DateTime
from core.rules import rule
from core.triggers import when
from core.metadata import get_key_value
from core.utils import send_command_if_different
from core.log import log_traceback
from core.actions import ScriptExecution

from community.timer_mgr import TimerMgr
from community.looping_timer import LoopingTimer # TimerMgr can't handle looping timers
from community.time_utils import to_datetime

# Configurations
LIGHT_OFF_DELAY="15s" # Switch off light after that duration
LIGHT_OFF_WARNING_PERIOD=2000 # Duration of pulsing in milliseconds
LIGHT_OFF_WARNING_STEP_DURATION=200 # Pulsing period in milliseconds for Switches
DIMMER_STEP=10 #Dimm percentage step for off warning duration

# Globals
delayed_lights_off_timers = TimerMgr()
off_warning_timers = {}

# functions
def get_actuator(itemName): 
    """ Gets the linked actuator name from the given item. It assumes the actuator name
    is provided in a Static metadata 'actuator'.
    Arguments:
        itemName: Name of the item an actuator link is to be retrieved
    """
    return get_key_value(itemName, "Static", "actuator") or None

def pulse_light(end_time, light):
    """ Called by the looping timer, toggles the given light on and off every 200 msec until 'end_time' is reached.
    Arguments:
        end_time: DateTime indicating when the loop should stop
        light: the light switch to toggle
    """
    now = DateTime.now()
    if DateTime.now().isAfter(end_time): # should be moved to Java DateTime before OH 3
        send_command_if_different(light, "OFF")
        return None # cancels the looping
    else:
        events.sendCommand(light, "ON" if items[light] == OFF else "OFF")
        return LIGHT_OFF_WARNING_STEP_DURATION # looping timer will treat integers as milliseconds


def dimm_light(end_time, light):
    """ Called by the looping timer, dimms the given light by 5% every 200 msec until 'end_time' is reached.
    Arguments:
        end_time: DateTime indicating when the loop should stop
        light: the light switch to toggle
    """
    now = DateTime.now()
    if DateTime.now().isAfter(end_time): # should be moved to Java DateTime before OH 3
        events.sendCommand(light, "OFF")
        return None # cancels the looping
    else:
        new_state = int(str(ir.getItem(light).state)) - DIMMER_STEP

        if (new_state > 0):
            events.sendCommand(light, str(new_state))
        else:
            events.sendCommand(light, "OFF")
        return LIGHT_OFF_WARNING_STEP_DURATION # looping timer will treat integers as milliseconds

@log_traceback
def cancel_item_off_warning_timer(timerName, log_function=None):
    if timerName in off_warning_timers:
        if log_function:
            log_function()
        off_warning_timers[timerName].cancel()
        del off_warning_timers[timerName]

def cancel_item_all_timers(timerName):
    if delayed_lights_off_timers.has_timer(timerName):
        delayed_lights_off_timers.cancel(timerName)
    cancel_item_off_warning_timer(timerName)

def is_dimmer(light_name):
    return ir.getItem(light_name).type == "Dimmer"

def get_on_off_state(light_name):
    item = ir.getItem(light_name)
    if item.type == "Dimmer":
        return ON if item.state > 0 else OFF
    else:
        return item.state

@log_traceback
def switch_off_light(motion_name, light_name, log):
    # Check for the error case where the off_warn_timer is running when the motion sensor goes off
    cancel_item_off_warning_timer(light_name, lambda item=motion_name: motion_changed.log.warn("Motion sensor {} went to OFF but there is a warning Timer running! This shouldn't be possible".format(item)))

    # If light is still ON and still no motion - warn the lights will go off (pulsing lights, or dimm to low value)
    if get_on_off_state(light_name) == ON and items[motion_name] == OFF:
        off_warning_timers[light_name] = LoopingTimer(
            lambda 
                switch_name=light_name, 
                end_time=to_datetime(LIGHT_OFF_WARNING_PERIOD)
                : dimm_light(end_time, switch_name) if is_dimmer(light_name) else pulse_light(end_time, switch_name))


@rule("Lights_NoMotion", description="If no movement, alert with lights and then switch off if still no movement", tags=["lights"])
@when("Member of gManageableRoomMotionSensors changed")
def motion_changed(event):
    light_item_name = get_actuator(event.itemName)
    motion_item_name = event.itemName

    if not light_item_name:
        motion_changed.log.warn("Actuator link is not defined in the {} motion sensor".format(event.itemName))
        return

    # Someone is present in the room - switch ON light
    if event.itemState == ON:
        # Cancell all timers if exist
        cancel_item_all_timers(light_item_name)

        # Turn on the light if it isn't already on
        send_command_if_different(light_item_name, "ON")

    # No motion - schedule timers to wait some time, then warn the user by pulsing lights or dimm to low value, then switch off if still no motion
    else:
        delayed_lights_off_timers.check(
            key=light_item_name,
            when=LIGHT_OFF_DELAY,
            function=lambda motion=motion_item_name, light=light_item_name, log=motion_changed.log: switch_off_light(motion, light, log)
        )

def scriptUnloaded():
    delayed_lights_off_timers.cancel_all()
    for t in off_warning_timers:
        cancel_item_off_warning_timer(t)