alarm: Add functions for setting an alarm

An alarm is a callback that triggers at a specific time, rather than at
a specific interval. I build this using GLib.timeout_add_seconds() and
wrapping it with logic to calculate the amount of time until the alarm
should be triggered next.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

emmental/alarm.py

@@ -0,0 +1,48 @@
+# Copyright 2023 (c) Anna Schumaker.
+"""Functions for configuring a callback at a specific time."""
+import datetime
+import math
+from gi.repository import GLib
+
+_GSOURCE_MAPPING = dict()
+_NEXT_ALARM_ID = 1
+
+
+def _calc_seconds(time: datetime.time) -> int:
+    """Calculate the number of seconds until the given time."""
+    now = datetime.datetime.now()
+    then = datetime.datetime.combine(now.date(), time)
+
+    if now >= then:
+        then += datetime.timedelta(days=1)
+
+    return math.ceil((then - now).total_seconds())
+
+
+def __set_alarm(time: datetime.time, func: callable, alarm_id: int) -> None:
+    gsrcid = GLib.timeout_add_seconds(_calc_seconds(time), _do_alarm,
+                                      time, func, alarm_id)
+    _GSOURCE_MAPPING[alarm_id] = gsrcid
+    return alarm_id
+
+
+def _do_alarm(time: datetime.time, func: callable, alarm_id: int) -> bool:
+    """Run an alarm callback."""
+    func()
+    __set_alarm(time, func, alarm_id)
+    return GLib.SOURCE_REMOVE
+
+
+def set_alarm(time: datetime.time, func: callable) -> int:
+    """Register a callback to be called at a specific time."""
+    global _NEXT_ALARM_ID
+
+    res = __set_alarm(time, func, _NEXT_ALARM_ID)
+    _NEXT_ALARM_ID += 1
+    return res
+
+
+def cancel_alarm(alarm_id: int) -> None:
+    """Cancel an alarm."""
+    GLib.source_remove(_GSOURCE_MAPPING[alarm_id])
+    del _GSOURCE_MAPPING[alarm_id]

tests/test_alarm.py

@@ -0,0 +1,72 @@
+# Copyright 2023 (c) Anna Schumaker.
+"""Test our functions for callbacks at a specific time."""
+import datetime
+import unittest.mock
+import emmental.alarm
+from gi.repository import GLib
+
+
+class TestAlarm(unittest.TestCase):
+    """Test case for callbacks at a specific time."""
+
+    def setUp(self):
+        """Set up common variables."""
+        emmental.alarm._GSOURCE_MAPPING.clear()
+        emmental.alarm._NEXT_ALARM_ID = 1
+        self.midnight = datetime.time(hour=0, minute=0, second=0)
+
+    def test_state(self):
+        """Test our global state."""
+        self.assertDictEqual(emmental.alarm._GSOURCE_MAPPING, {})
+        self.assertEqual(emmental.alarm._NEXT_ALARM_ID, 1)
+
+    def test_calc_seconds(self):
+        """Test calculating the seconds until the next alarm."""
+        now = datetime.datetime.now()
+        time = (now + datetime.timedelta(minutes=2)).time()
+        self.assertEqual(emmental.alarm._calc_seconds(time), 120)
+
+        time = (now - datetime.timedelta(minutes=2)).time()
+        self.assertEqual(emmental.alarm._calc_seconds(time), 86280)
+
+    @unittest.mock.patch("gi.repository.GLib.timeout_add_seconds")
+    def test_set_alarm(self, mock_timeout_add: unittest.mock.Mock):
+        """Test setting an alarm."""
+        callback = unittest.mock.Mock()
+        seconds = emmental.alarm._calc_seconds(self.midnight)
+        mock_timeout_add.return_value = 42
+
+        srcid = emmental.alarm.set_alarm(self.midnight, callback)
+        mock_timeout_add.assert_called_with(seconds, emmental.alarm._do_alarm,
+                                            self.midnight, callback, 1)
+        self.assertEqual(srcid, 1)
+        self.assertEqual(emmental.alarm._NEXT_ALARM_ID, 2)
+        self.assertEqual(emmental.alarm._GSOURCE_MAPPING[1], 42)
+
+    @unittest.mock.patch("gi.repository.GLib.source_remove")
+    @unittest.mock.patch("gi.repository.GLib.timeout_add_seconds")
+    def test_cancel_alarm(self, mock_timeout_add: unittest.mock.Mock,
+                          mock_source_remove: unittest.mock.Mock):
+        """Test cancelling an alarm."""
+        callback = unittest.mock.Mock()
+        mock_timeout_add.return_value = 42
+        srcid = emmental.alarm.set_alarm(self.midnight, callback)
+
+        emmental.alarm.cancel_alarm(srcid)
+        mock_source_remove.assert_called_with(42)
+        self.assertNotIn(srcid, emmental.alarm._GSOURCE_MAPPING)
+
+    @unittest.mock.patch("gi.repository.GLib.timeout_add_seconds")
+    def test_do_alarm(self, mock_timeout_add: unittest.mock.Mock):
+        """Test triggering an alarm."""
+        callback = unittest.mock.Mock()
+        seconds = emmental.alarm._calc_seconds(self.midnight)
+        emmental.alarm._GSOURCE_MAPPING[1] = 2
+        mock_timeout_add.return_value = 42
+
+        self.assertEqual(emmental.alarm._do_alarm(self.midnight, callback, 1),
+                         GLib.SOURCE_REMOVE)
+        callback.assert_called()
+        mock_timeout_add.assert_called_with(seconds, emmental.alarm._do_alarm,
+                                            self.midnight, callback, 1)
+        self.assertEqual(emmental.alarm._GSOURCE_MAPPING[1], 42)