feat(holmanctl): add --run and --stop for one-shot tap control

README.md

@@ -75,6 +75,13 @@ sudo holmanctl --connect AA:BB:CC:DD:EE:FF # Replace the MAC address with your H
 sudo holmanctl --help # To list all available commands
 ```
 
+To start or stop the tap from a script or cron job, add `--run` or `--stop` to `--connect`. `holmanctl` connects, writes the command, disconnects and exits: with status 0 if the tap timer accepted the command, or 1 if connecting or writing failed.
+
+```
+sudo holmanctl --connect AA:BB:CC:DD:EE:FF --run 10 # Run the tap for 10 minutes (1-255)
+sudo holmanctl --connect AA:BB:CC:DD:EE:FF --stop   # Stop the tap
+```
+
 ## SDK Usage
 
 ### Discovering nearby Holman tap timers
@@ -126,6 +133,8 @@ As with Holman tap timer discovery, remember to start the Bluetooth event loop w
 
 Once a Holman tap timer is connected you can start the tap with `TapTimer.start(runtime=1)`.  Pass this a runtime (in minutes) for how long to run the tap.
 
+Stop it with `TapTimer.stop()`. Both write to the tap timer asynchronously; your listener's `timer_written()` is called once the command has been accepted, or `timer_write_failed(error)` if it couldn't be written.
+
 ## Development
 
 The project uses [uv](https://docs.astral.sh/uv/) for environment and dependency management.

holman/holman.py

@@ -244,9 +244,13 @@ def stop(self):
 
     def characteristic_write_value_succeeded(self, characteristic):
         self._refresh_state()
+        if self.listener and characteristic == self._manual_characteristic:
+            self.listener.timer_written()
 
     def characteristic_write_value_failed(self, characteristic, error):
         self._refresh_state()
+        if self.listener and characteristic == self._manual_characteristic:
+            self.listener.timer_write_failed(error)
 
 
 class TapTimerListener:
@@ -269,3 +273,11 @@ def started_disconnecting(self):
 
     def disconnect_succeeded(self):
         pass
+
+    def timer_written(self):
+        """Called when a ``start()`` or ``stop()`` reached the tap timer."""
+        pass
+
+    def timer_write_failed(self, error):
+        """Called when a ``start()`` or ``stop()`` could not be written."""
+        pass

holmanctl.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 import sys
-from argparse import ArgumentParser
+from argparse import ArgumentParser, ArgumentTypeError
 
 import holman
 
@@ -29,19 +29,60 @@ def started_disconnecting(self):
     def disconnect_succeeded(self):
         self.print("disconnected")
 
+    def timer_written(self):
+        self.print("timer written")
+
+    def timer_write_failed(self, error):
+        self.print("timer write failed: " + str(error))
+
     def print(self, string):
         print("Holman tap timer " + self.tap_timer.mac_address + " " + string)
 
 
 class TapTimerTestListener(TapTimerPrintListener):
-    def __init__(self, tap_timer, auto_reconnect=False):
+    """
+    Drives ``holmanctl``: optionally starts or stops the tap once connected,
+    then disconnects and stops the event loop.
+
+    ``exit_code`` is 0 on success and 1 if connecting or writing failed.
+    ``finished`` is set once there is nothing left to wait for.
+
+    :param run_minutes: start the tap for this many minutes once connected
+    :param stop_tap: stop the tap once connected
+    """
+
+    def __init__(
+        self, tap_timer, auto_reconnect=False, run_minutes=None, stop_tap=False
+    ):
         super().__init__(tap_timer)
         self.auto_reconnect = auto_reconnect
+        self.run_minutes = run_minutes
+        self.stop_tap = stop_tap
+        self.exit_code = 0
+        self.finished = False
+
+    def connect_succeeded(self):
+        super().connect_succeeded()
+        if self.run_minutes is not None:
+            self.tap_timer.start(runtime=self.run_minutes)
+        elif self.stop_tap:
+            self.tap_timer.stop()
 
     def connect_failed(self, error):
         super().connect_failed(error)
-        self.tap_timer.manager.stop()
-        sys.exit(0)
+        self.exit_code = 1
+        self._finish()
+
+    def timer_written(self):
+        super().timer_written()
+        # Disconnect rather than just exiting: BlueZ would otherwise keep
+        # the connection open after this process is gone.
+        self.tap_timer.disconnect()
+
+    def timer_write_failed(self, error):
+        super().timer_write_failed(error)
+        self.exit_code = 1
+        self.tap_timer.disconnect()
 
     def disconnect_succeeded(self):
         super().disconnect_succeeded()
@@ -51,15 +92,32 @@ def disconnect_succeeded(self):
             print("Disconnected, reconnecting...")
             self.tap_timer.connect()
         else:
-            self.tap_timer.manager.stop()
-            sys.exit(0)
+            self._finish()
+
+    def _finish(self):
+        # gatt can report a failure synchronously from inside connect(),
+        # before the event loop runs; stop() is a no-op then, so main()
+        # also checks ``finished`` before calling run().
+        self.finished = True
+        self.tap_timer.manager.stop()
 
 
 class TapTimerManagerPrintListener(holman.TapTimerManagerListener):
     def tap_timer_discovered(self, tap_timer):
         print("Discovered Holman tap timer", tap_timer.mac_address)
 
 
+def minutes(value):
+    """argparse type for a tap runtime: a whole number of minutes, 1-255."""
+    try:
+        runtime = int(value)
+    except ValueError:
+        raise ArgumentTypeError(f"invalid number of minutes: {value!r}") from None
+    if not 1 <= runtime <= 255:
+        raise ArgumentTypeError(f"must be between 1 and 255 minutes, got {runtime}")
+    return runtime
+
+
 def main():
     arg_parser = ArgumentParser(description="Holman Tap Timer Demo")
     arg_parser.add_argument(
@@ -93,40 +151,63 @@ def main():
         type=str,
         help="Disconnect a Holman tap timer with a given MAC address",
     )
+    arg_action_group = arg_parser.add_mutually_exclusive_group()
+    arg_action_group.add_argument(
+        "--run",
+        metavar="minutes",
+        type=minutes,
+        help="With --connect: run the tap for 1-255 minutes, then exit",
+    )
+    arg_action_group.add_argument(
+        "--stop",
+        action="store_true",
+        help="With --connect: stop the tap, then exit",
+    )
     args = arg_parser.parse_args()
 
+    one_shot = args.run is not None or args.stop
+    if one_shot and not args.connect:
+        arg_parser.error("--run and --stop require --connect")
+
     tap_timer_manager = holman.TapTimerManager(adapter_name=args.adapter)
+    listener = None
 
     if args.discover:
         tap_timer_manager.listener = TapTimerManagerPrintListener()
         tap_timer_manager.start_discovery()
     if args.known:
         for tap_timer in tap_timer_manager.tap_timers():
             print(f"[{tap_timer.mac_address}] {tap_timer.alias()}")
-        return
+        return 0
     elif args.connect:
         tap_timer = holman.TapTimer(mac_address=args.connect, manager=tap_timer_manager)
-        tap_timer.listener = TapTimerTestListener(tap_timer=tap_timer)
+        listener = TapTimerTestListener(
+            tap_timer=tap_timer, run_minutes=args.run, stop_tap=args.stop
+        )
+        tap_timer.listener = listener
         tap_timer.connect()
     elif args.auto:
         tap_timer = holman.TapTimer(mac_address=args.auto, manager=tap_timer_manager)
-        tap_timer.listener = TapTimerTestListener(
-            tap_timer=tap_timer, auto_reconnect=True
-        )
+        listener = TapTimerTestListener(tap_timer=tap_timer, auto_reconnect=True)
+        tap_timer.listener = listener
         tap_timer.connect()
     elif args.disconnect:
         tap_timer = holman.TapTimer(
             mac_address=args.disconnect, manager=tap_timer_manager
         )
         tap_timer.disconnect()
-        return
+        return 0
 
-    print("Terminate with Ctrl+C")
+    if listener and listener.finished:
+        return listener.exit_code
+    if not one_shot:
+        print("Terminate with Ctrl+C")
     try:
         tap_timer_manager.run()
     except KeyboardInterrupt:
         pass
+    return listener.exit_code if listener else 0
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())

tests/conftest.py

@@ -19,6 +19,7 @@ def __init__(self, adapter_name):
         self.known_devices = []
         self.discovery_uuids = None
         self.stopped = False
+        self.ran = False
 
     def devices(self):
         return self.known_devices
@@ -30,7 +31,12 @@ def device_discovered(self, device):
         pass
 
     def stop(self):
-        self.stopped = True
+        # Like gatt, stop() only quits a loop that is already running.
+        if self.ran:
+            self.stopped = True
+
+    def run(self):
+        self.ran = True
 
 
 class FakeDevice:
@@ -106,6 +112,12 @@ def started_disconnecting(self):
     def disconnect_succeeded(self):
         self.events.append("disconnect_succeeded")
 
+    def timer_written(self):
+        self.events.append("timer_written")
+
+    def timer_write_failed(self, error):
+        self.events.append(("timer_write_failed", str(error)))
+
 
 @pytest.fixture
 def manager():
@@ -149,3 +161,52 @@ def make(
         return FakeService(service_uuid, chars)
 
     return make
+
+
+@pytest.fixture
+def simulated_tap_timer(monkeypatch, make_holman_service):
+    """
+    Make every ``TapTimer`` behave like a reachable device: ``connect()``
+    resolves a Holman service, writes to it report back, and
+    ``disconnect()`` completes.
+
+    Set ``connect_error`` or ``write_error`` on the returned object to
+    simulate those failures; ``tap_timers`` collects connected instances.
+    """
+
+    class Simulation:
+        connect_error = None
+        write_error = None
+        tap_timers = []
+
+    simulation = Simulation()
+
+    def connect(device):
+        device.connect_calls += 1
+        simulation.tap_timers.append(device)
+        if simulation.connect_error is not None:
+            device.connect_failed(simulation.connect_error)
+            return
+        service = make_holman_service()
+        for char in service.characteristics:
+            char.write_value = _reporting_write(device, char)
+        device.services = [service]
+        device.services_resolved()
+
+    def _reporting_write(device, char):
+        def write_value(value):
+            char.written.append(value)
+            if simulation.write_error is not None:
+                device.characteristic_write_value_failed(char, simulation.write_error)
+            else:
+                device.characteristic_write_value_succeeded(char)
+
+        return write_value
+
+    def disconnect(device):
+        device.disconnect_calls += 1
+        device.disconnect_succeeded()
+
+    monkeypatch.setattr(FakeDevice, "connect", connect)
+    monkeypatch.setattr(FakeDevice, "disconnect", disconnect)
+    return simulation