Script to activate an alarm at a time in the future

Question

I am making a flashlight plugin for macOS.

Here is the main code, plugin.py, the method results gets called first and is passed a dictionary like this for example:

{ "~arguments": "3h30m" }

And I try to parse the time span given and activate a macOS alert later.

Anyway, I'd appreciate any pointers at all, I'm pretty new to python:

import datetime
import json
import os
import re
import subprocess
import threading
import time
import unittest

stop_audio = False

def seconds_to_text(seconds):
    """Return the user-friendly version of the time duration specified by seconds.

    Outputs should resemble:
        "3 hours and 30 minutes"
        "20 minutes"
        "1 minute and 30 seconds"
        "10 hours, 30 minutes and 10 seconds"
    """
    # Special case because it's faster this way
    if seconds == 0:
        return "0 seconds"
    # Need the hours, minutes and seconds individually for putting into string
    hours = seconds // (60 * 60)
    hours = int(hours)
    seconds %= 60 * 60
    minutes = seconds // 60
    minutes = int(minutes)
    seconds %= 60
    seconds = int(seconds)

    formatted_text = ""
    if hours > 0:
        formatted_text += str(hours) + " " + ("hour", "hours")[hours > 1]
    if minutes > 0:
        if formatted_text.count(" ") > 0:
            formatted_text += (" and ", ", ")[seconds > 0]
        formatted_text += str(minutes) + " " + ("minute", "minutes")[minutes > 1]
    if seconds > 0:
        if formatted_text.count(" ") > 0:
            formatted_text += " and "
        formatted_text += str(seconds) + " " + ("second", "seconds")[seconds > 1]
    return formatted_text

def parse_time_span(time_string):
    """Convert an inputted string representing a timespan, like 3h30m15s, into a duration in seconds."""
    pattern = re.compile(r"^(?:(?P<hours>\d+)h)?(?:(?P<minutes>\d+)m)?(?:(?P<seconds>\d+)s)?$")
    (hours, minutes, seconds) = pattern.match(time_string).groups()
    hours = 0 if hours is None else int(hours)
    minutes = 0 if minutes is None else int(minutes)
    seconds = 0 if seconds is None else int(seconds)
    total_seconds = datetime.timedelta(hours=hours, minutes=minutes, seconds=seconds).total_seconds()
    return round(total_seconds)

def parse_absolute_time(time_string):
    """Convert an inputted string like '7:30PM' or '22:00' into the number of seconds from now until
    that time. If the time is earlier in the day than the current time, take the number of seconds
    until that time occurrs tomorrow.
    """
    # As there are so many possible input formats, "19:30", "10", "6:00AM", etc. I thought a sensible
    # way to parse the inputs would be to use a dictionary which pairs patterns with parsing rules.
    time = None
    formats = {
        "^\d{1,2}$": "%H",
        "^\d{1,2}(AM|PM)$": "%I%p",
        "^\d{1,2}:\d{2}$": "%H:%M",
        "^\d{1,2}:\d{2}(AM|PM)$": "%I:%M%p"
        }
    for key, value in formats.items():
        if re.match(key, time_string, re.IGNORECASE):
            time = datetime.datetime.strptime(time_string, value).time()
    if time is None:
        # need to let the caller know that the time wasn't in a recognised format
        raise ValueError
    time = datetime.datetime.combine(datetime.datetime.today().date(), time)
    if datetime.datetime.now() > time:
        # it's likely the user wants to set an alarm for tomorrow
        time = time + datetime.timedelta(days = 1)
    total_seconds = (time - datetime.datetime.now()).total_seconds()
    return round(total_seconds)

def show_alert(message="Flashlight alarm"):
    """Display a macOS dialog."""
    message = json.dumps(str(message))
    os.system("osascript dialog.scpt {0}".format(message))

stop_sound = False
def play_alarm(file_name = "beep.wav"):
    """Repeat the sound specified to mimic an alarm."""
    while not stop_sound:
        process = subprocess.Popen(["afplay", file_name])
        while not stop_sound:
            print("stop_sound",stop_sound)
            print("process.poll()", process.poll())
            print("process.poll is not None", process.poll() is not None)
            if process.poll() is not None:
                break
            time.sleep(0.1)
        if stop_sound:
            process.kill()

def alert_after_timeout(timeout, message):
    """After timeout seconds, show an alert and play the alarm sound."""
    global stop_sound
    time.sleep(timeout)
    process = None
    thread = threading.Thread(target=play_alarm)
    thread.start()
    # show_alert is synchronous, it must be closed before the script continues
    show_alert(message)

    stop_sound = True
    thread.join()

def results(fields, original_query):
    arguments = fields["~arguments"].split(" ")
    time = arguments[0]
    message = " ".join(arguments[1:])
    with open("results.html") as html:
        # which input format is the user trying to use?
        pattern = re.compile(r"^(?:(?P<hours>\d+)h)?(?:(?P<minutes>\d+)m)?(?:(?P<seconds>\d+)s)?$")
        if pattern.match(time):
            seconds = 0
            try:
                seconds = parse_time_span(time)
            except AttributeError:
                return {
                    "title": "Don't understand.",
                    "run_args": [],
                    "html": "Make sure your input is formatted properly.",
                    "webview_transparent_background": True
                    }
            return {
                "title": "%s in %s" % (message or "Alarm", seconds_to_text(seconds)),
                "run_args": [seconds, message or "%s alarm" % seconds_to_text(seconds)],
                "html": html.read(),
                "webview_transparent_background": True
                }
        else:
            try:
                return {
                    "title": "Set an alarm for %s" % time,
                    "run_args": [parse_absolute_time(time), message or "%s alarm" % (time)],
                    "html": html.read(),
                    "webview_transparent_background": True
                    }
            except ValueError:
                return {
                    "title": "Don't understand.",
                    "run_args": [],
                    "html": "Make sure your input is formatted properly.",
                    "webview_transparent_background": True
                    }

def run(seconds, message):
    alert_after_timeout(seconds, message)

class TestParsingAndFormattingFunctions(unittest.TestCase):
    """Test that the functions which parse strings into times and format times as strings are all working."""

    def test_parse_time_span(self):
        """Make sure parse_time_span properly converts a string, formatted like 3h30m30s, into a time duration."""
        # Testing for normal data
        self.assertEqual(parse_time_span("3h30m"), 12600)
        self.assertEqual(parse_time_span("8h30m"), 30600)
        self.assertEqual(parse_time_span("1m15s"), 75)
        self.assertEqual(parse_time_span("20m"), 1200)
        # Testing extreme data
        self.assertEqual(parse_time_span("23h59m59s"), 86399)
        self.assertEqual(parse_time_span("0h1m0s"), 60)
        self.assertEqual(parse_time_span("60s"), 60)
        # Testing abnormal data, these should all error
        with self.assertRaises(AttributeError):
            parse_time_span("five o-clock")
        with self.assertRaises(AttributeError):
            parse_time_span("1.5s")
        with self.assertRaises(AttributeError):
            parse_time_span("25")

    def test_seconds_to_text(self):
        """Make sure seconds_to_text formats a string into the correct human-readable structure."""
        # Testing with normal inputs
        self.assertEqual(seconds_to_text(18000), "5 hours")
        self.assertEqual(seconds_to_text(12600), "3 hours and 30 minutes")
        self.assertEqual(seconds_to_text(1200), "20 minutes")
        self.assertEqual(seconds_to_text(60), "1 minute")
        # Testing with extreme inputs
        self.assertEqual(seconds_to_text(0), "0 seconds")
        self.assertEqual(seconds_to_text(86399), "23 hours, 59 minutes and 59 seconds")
        # Testing with invalid inputs
        with self.assertRaises(TypeError):
            seconds_to_text("What's a string doing here?")

Answer 1

Since you're using time already you may want to change seconds_to_text to use it too. As an example of a simple HH:MM:SS format you could use something like:

>>> time.strftime('%H:%M:%S', time.gmtime(((3 * 60 + 5) * 60) + 2))
'03:05:02'

Not the format that you want, but one you may want to consider. However it shows that you can create a simple time object just from seconds. Which can drastically simplify your code. By using dir we can check if there is an interface to get the; hours, minuets, and seconds, and use them.

>>> dir(time.gmtime(((3 * 60 + 5) * 60) + 2))
['__add__', '__class__', '__contains__', '__delattr__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__getitem__', '__getslice__', '__gt__', '__hash__', '__init__', '__le__', '__len__', '__lt__', '__mul__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__rmul__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', 'n_fields', 'n_sequence_fields', 'n_unnamed_fields', 'tm_hour', 'tm_isdst', 'tm_mday', 'tm_min', 'tm_mon', 'tm_sec', 'tm_wday', 'tm_yday', 'tm_year']
>>> t = time.gmtime(((3 * 60 + 5) * 60) + 2);(t.tm_hour, t.tm_min, t.tm_sec)
(3, 5, 2)

Finally it makes more sense to add an "s" if; hours, minuets, or seconds, are greater than one, rather than spelling both; hour, and hours. And you shouldn't bother with premature optimizations such as if seconds == 0:. If it is a bottle neck you would want to increase the performance of the function as a whole, not the special case.

---

I'd then go on to further to change how your functions work.

First you'd want to separate your functions out a little more. You should want a single function that can read any format of input. After this you then change it to relative time or absolute time. You want your functions to be as simple as possible, so changing them to seconds doesn't make sense. There's two ways you can come at changing input to time:

1. Have a very ridged format, so you say only times are allowed in the format HH:MM:SS.
2. Allow almost anything as input. Use a regex that dynamically changes the input to be able to extract the data.

For example this function could be something like:

def text_to_time(text):
    match = re.match("^(?i)(\d{1,2})(?:(\D+?)(\d{1,2})?)?(?:(\D+?)(\d{1,2})?)?(AM|PM)?$", text)
    if match is None:
        raise ValueError("Unknown format for input {!r}".format(text))
    groups = list(match.groups(''))
    for i, f in ((0, '%H'), (2, '%M'), (4, '%S'), (5, '%p')):
        if groups[i]:
            groups[i] = f
    if groups[5] == '%p':
        groups[0] = '%I'
    return datetime.datetime.strptime(text, ''.join(groups)).time()

After this you'd want to change it to absolute and relative time. This makes them both really simple functions. And is pretty much what you were doing in those functions before without the extraction of the time.

def relative_time(time_object):
    t = time_object
    return datetime.timedelta(hours=t.hour, minutes=t.minute, seconds=t.second)


def absolute_time(time_object):
    time = datetime.datetime.combine(datetime.datetime.today().date(), time_object)
    if datetime.datetime.now() > time:
        # it's likely the user wants to set an alarm for tomorrow
        time += datetime.timedelta(days=1)
    return time - datetime.datetime.now()

The final big change I'd make is how you format your 'pretty time'. Rather than repeating yourself, you should want to create a function to; format time, and join a list of words. First formatting the time just joins the time with the name, and an 's' if there is more than one. Or it returns an empty string, if there is nothing. After this you want to join the list of formatted times, if we could we would use ''.join, however as we need the "and" we can't. To do this you can use itertools.zip_longest and itertools.chain.from_iterable. This is as we can create a list of separators, and change the last to the word we want to use. After this we then join them together. This can get you:

def format_time(time, name, plural='s'):
    if time == 0:
        return ''
    return '{} {}'.format(time, name + (plural if time > 1 else ''))


def join_list(items, sep=', ', conn=' and '):
    joins = [sep] * (len(items) - 1)
    if joins:
        joins[-1] = conn
    chain = itertools.chain.from_iterable
    zip = itertools.zip_longest
    return ''.join(chain(zip(items, joins, fillvalue='')))


def seconds_to_text(seconds):
    # As empty output is not wanted.
    if seconds == 0:
        return '0 seconds'
    time_object = time.gmtime(seconds)
    times = [
        format_time(time_object.tm_hour, 'hour'),
        format_time(time_object.tm_min, 'minute'),
        format_time(time_object.tm_sec, 'second')
    ]
    return join_list([t for t in times if t])

Finally you'd need to fix results to work with the above, and so I'd make a couple of changes. You want to have one try so that if the single conversion fails, you return the same stuff. You also want to open the file for as little as possible, as all you're doing is reading it flat out. Finally using the same regex as you did before you want to just merge the above functions into it, which can get you something like:

def results(fields, original_query):
    arguments = fields["~arguments"].split(" ")
    time = arguments[0]
    message = " ".join(arguments[1:])
    try:
        time_object = text_to_time(text)
    except ValueError:
        return {
            "title": "Don't understand.",
            "run_args": [],
            "html": "Make sure your input is formatted properly.",
            "webview_transparent_background": True
        }
    with open("results.html") as html:
        html = html.read()
    # which input format is the user trying to use?
    pattern = re.compile(r"^(?:(?P<hours>\d+)h)?(?:(?P<minutes>\d+)m)?(?:(?P<seconds>\d+)s)?$")
    if pattern.match(time):
        seconds = round(relative_time(time_object).to_seconds())
        time_text = seconds_to_text(seconds)
        return {
            "title": "%s in %s" % (message or "Alarm", time_text),
            "run_args": [
                seconds,
                message or "%s alarm" % time_text
            ],
            "html": html,
            "webview_transparent_background": True
        }
    else:
        return {
            "title": "Set an alarm for %s" % time,
            "run_args": [
                round(absolute_time(time_object).to_seconds()),
                message or "%s alarm" % (time)
            ],
            "html": html,
            "webview_transparent_background": True
        }

Answer 2

global stop_sound

This is bad practice and calls for a better encapsulation.

So, let's analyze what this flag is used for: it stops an infinite loop within a function that runs inside a threading.Thread. If you’d used Thread subclassing, you could have used an attribute instead. Subclassing Thread also gives you the ability to stop it from the creator's thread:

class AlarmThread(threading.Thread):
    def __init__(self, filename='beep.wav'):
        super().__init__()
        self.filename = filename
        self.ongoing = None

    def run(self):
        self.ongoing = True
        while self.ongoing:
            self.process = subprocess.Popen(["afplay", self.filename])
            self.process.wait()

    def stop(self):
        if self.ongoing is not None:
            self.ongoing = False
            self.process.kill()

This implementation is incomplete as it is subject to race conditions (if the stop executes entirely when the AlarmThread is between the while and the Popen call). Let's use some locking mechanism to fix it, the core idea being to make manipulation of the ongoing flag and the sound process an atomic operation:

class AlarmThread(threading.Thread):
    def __init__(self, filename='beep.wav', player="afplay"):
        super().__init__()
        self.command = [player, filename]
        self.ongoing = True
        self.lock = threading.Lock()

    def run(self):
        while True:
            with self.lock:
                if not self.ongoing:
                    return
                self.process = subprocess.Popen(self.command)
            self.process.wait()

    def stop(self):
        with self.lock:
            self.ongoing = False
            try:
                self.process.kill()
            except AttributeError:
                # In case the thread is stopped before being started
                pass

Usage being:

def alert_after_timeout(timeout, message):
    """After timeout seconds, show an alert and play the alarm sound."""
    time.sleep(timeout)
    thread = AlarmThread()
    thread.start()
    # show_alert is synchronous, it must be closed before the script continues
    show_alert(message)

    thread.stop()
    thread.join()

---

Now the title of the question got me thinking that you’d use a threading.Timer to wait for the given amount of time before runing your notification thingy. You don't. Here is how I’d do (besides, it avoid declaring a method that does nothing more than calling another with the exact same arguments):

def alert(message):
    """show an alert and play the alarm sound."""
    alarm = AlarmThread()
    alarm.start()

    subprocess.run(['osascript', 'dialog.scpt', json.dumps(str(message))])

    alarm.stop()
    alarm.join()

def run(seconds, message):
    timer = threading.Timer(seconds, alert, args=(message,))
    timer.start()
    timer.join()

I’m also using subprocess.run here instead of os.system for consistency with the rest of the code (and also for personal preference). I'm also a bit sceptical about the json.dumps(str(message)) part… the aim of json.dumps is already to convert to a string, why call str on top of that?