How do I schedule a one-time script to run x minutes from now? (alternative for 'at')

Question

(Background: I'd like to control a light source with a motion sensor. Light should turn off x minutes after last detected motion. The framework is in place, scheduling is what remains to be done.)

Currently, when motion is detected the light gets turned on and a job to turn it off in 'now + x minutes' is scheduled. Whenever motion is detected during the x minutes the job gets removed from the queue and a new one is set up, extending effectively the time the light stays on.

I tried the "at" command but job handling is quite clunky. Whenever a job is removed from the queue an email gets sent. I looked at the Python crontab module but it would need much additional programming (handling relative time, removing old cronjobs, etc.) and seems to be slower.

What are my alternatives (bash, python, perl)?

-- Edit: My python skills are at beginner level, here's what I put together:

#!/usr/bin/env python2.7
# based on http://raspi.tv/2013/how-to-use-interrupts-with-python-on-the-raspberry-pi-and-rpi-gpio-part-2
# more than 160 seconds without activity are required to re-trigger action

import time
from subprocess import call
import os
import RPi.GPIO as GPIO

PIR = 9 # data pin of PIR sensor (in)
LED = 7 # positive pin of LED (out)
timestamp = '/home/pi/events/motiontime' # file to store last motion detection time (in epoch)
SOUND = '/home/pi/events/sounds/Hello.wav' # reaction sound

# GPIO setup
GPIO.setmode(GPIO.BCM)
GPIO.setup(PIR,GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
GPIO.setup(LED,GPIO.OUT)

# function which gets called when motion is reported (sensor includes own delay-until-hot again
# and sensibility settings
def my_callback(channel):
        now = time.time() # store current epoch time in variable 'now'
        f = open(timestamp, "r") 
        then = float(f.readline()) # read last detection time from file
        difference = now - then # calculate time that has passed
    call(['/home/pi/bin/kitchenlights.sh', '-1']) # turn light on
    call(['/home/pi/bin/lighttimer.sh']) # schedule at job to turn lights off
    if difference > 160: # if more than 160 seconds without activity have passed then...
                GPIO.output(LED, True) # turn on LED
                if not os.path.isfile("/home/pi/events/muted"): # check if system is muted, else
                        call(['/usr/bin/mplayer', '-really-quiet', '-noconsolecontrols', SOUND]) # play sound
                GPIO.output(LED, False) # turn of LED
                f = open(timestamp, "w") 
                f.write(repr(now)) # update timestamp
                f.close()
        else: # when less than 160 seconds have passed do nothing and
                f = open(timestamp, "w")
                f.write(repr(now)) # update timestamp (thus increasing the interval of silence)
                f.close()

GPIO.add_event_detect(PIR, GPIO.RISING,callback=my_callback,bouncetime=100)  # add rising edge detection on a channel

while True:
        time.sleep(0.2)
        pass

Now that questions come in I think I could put a countdown in the while loop, right? How would that work?

Answer 1

I would approach this with the threading module. To do this, you'd set up the following thread class:

class CounterThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.count = 0
        self.start()

    def run(self):
        while self.count < COUNTLIMIT:
            time.sleep(0.1)            
            self.count += 0.1

        #Call function to turn off light here                
        return

    def newSig(self):
        self.count = 0

This is a thread which everytime it recieves a new signal (the thread's newSig function is called), the counter restarts. If the COUNTLIMIT is reached (how long you want to wait in seconds), then you call the function to turn off the light.

Here's how you'd incorporate this into your code:

import threading
from subprocess import call
import os
import time
import RPi.GPIO as GPIO

PIR = 9 # data pin of PIR sensor (in)
LED = 7 # positive pin of LED (out)
SOUND = '/home/pi/events/sounds/Hello.wav' # reaction sound

COUNTLIMIT = 160
countThread = None
WATCHTIME = 600 #Run for 10 minutes

# GPIO setup
GPIO.setmode(GPIO.BCM)
GPIO.setup(PIR,GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
GPIO.setup(LED,GPIO.OUT)

#------------------------------------------------------------

class CounterThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.count = 0
        self.start()

    def run(self):
        call(['/home/pi/bin/kitchenlights.sh', '-1'])     # turn light on
        while self.count < COUNTLIMIT:
            time.sleep(0.1)            
            self.count += 0.1

        call(['/home/pi/bin/kitchenlights.sh', '-0']) 
        threadKiller()
        return

    def newSig(self):
        self.count = 0

#------------------------------------------------------------

def my_callback(channel):
    '''function which gets called when motion is reported (sensor includes own delay-until-hot again and sensibility settings'''

    global countThread

    try: 
        countThread.newSig()
    except:
        countThread = CounterThread() 

#------------------------------------------------------------

def threadKiller():
    global countThread
    countThread = None

#------------------------------------------------------------

def main():
    GPIO.add_event_detect(PIR, GPIO.RISING,callback=my_callback,bouncetime=100)  # add rising edge detection on a channel
    t = 0
    while t < WATCHTIME:
        t += 0.1
        time.sleep(0.1)

#------------------------------------------------------------

if __name__ == "__main__": main()

I don't have any way to test this, so please let me know if there is anything that breaks. Since you said you're new to Python I made a few formatting changes to make your code a bit prettier. These things are generally considered to be good form, but are optional. However, you need to be careful about indents, because as you have them in your question, your code should not run (it will throw an IndentError)

Hope this helps