send_at.py

#!/usr/bin/env python3
"""Start command, sleep until specified local time, send input.

Bash analog:

    $ rm -f pipe; mkfifo pipe
    $ at TIMESPEC <<< 'echo -n INPUT >pipe'
    $ COMMAND <pipe

Usage:

    $ python3 -O -msend_at [%Y-%d-%mT]%H:%M:%S.%f input command arg1 arg2 ...

Example:

    $ python3 -O -msend_at 23:59:00.0 'some input' cat

http://stackoverflow.com/questions/20155872/cant-use-named-pipe-from-c-to-communicate-with-shell-script/20175873#20175873
"""
import os
import sys
import time
import warnings
from datetime import datetime
from subprocess import Popen, PIPE

time_info = time.get_clock_info('time')
if time_info.implementation != 'clock_gettime(CLOCK_REALTIME)':
    warnings.warn("Unexpected time implementation {}".format(time_info))

def sleep_until(endtime, *, _time=time.time, _sleep=time.sleep,
                _busyloop_cutoff=40e-3): #  40ms
    while True: # wait, taking possible time adjustments into account
        #NOTE: don't use `lock.acquire(True, timeout)` or derivatives here
        remaining = endtime - _time()
        if remaining < _busyloop_cutoff:
            while endtime > _time():
                pass # busyloop
            return
        _sleep(remaining / 2.)

def send_at(endtime, input, command):
    with Popen(command, stdin=PIPE) as p:
        sleep_until(endtime)
        p.communicate(input)

def parse_localtime(timestr):
    time_format =  "%H:%M:%S.%f" # only time
    if 'T' not in timestr:
        dt = datetime.combine(datetime.now().date(), # local time
                              datetime.strptime(timestr, time_format).time())
    else: # + date
        time_format = "%Y-%m-%dT" + time_format
        dt = datetime.strptime(timestr, time_format)
    return dt

def report_error_and_exit(format, *args):
    print(("{prog}: error: " + format).format(
        prog=os.path.basename(sys.argv[0]), *args), file=sys.stderr)
    sys.exit(2)

def main():
    # parse input timespec
    timestr = sys.argv[1]
    try:
        dt = parse_localtime(timestr)
    except ValueError as e:
        report_error_and_exit(str(e))
    if dt < datetime.now():
        report_error_and_exit("time {} is in the past, do nothing", timestr)
    print("Input will be send at {}".format(dt))
    send_at(dt.timestamp(), os.fsencode(sys.argv[2]), sys.argv[3:])

if __name__ == "__main__":
    main()