diff --git a/README.md b/README.md index e430648..ce9b764 100644 --- a/README.md +++ b/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. diff --git a/holman/holman.py b/holman/holman.py index 6875c0e..bc55c1a 100644 --- a/holman/holman.py +++ b/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 diff --git a/holmanctl.py b/holmanctl.py index 87152e6..ce01008 100755 --- a/holmanctl.py +++ b/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,8 +92,14 @@ 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): @@ -60,6 +107,17 @@ 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,9 +151,26 @@ 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() @@ -103,30 +178,36 @@ def main(): 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()) diff --git a/tests/conftest.py b/tests/conftest.py index f5dc59c..90da7b1 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_holmanctl.py b/tests/test_holmanctl.py index 5d05aa3..0decfb6 100644 --- a/tests/test_holmanctl.py +++ b/tests/test_holmanctl.py @@ -3,11 +3,28 @@ import holman import holmanctl +MAC = "AA:BB:CC:DD:EE:FF" -def test_connect_failed_stops_manager_and_exits(tap_timer, manager): + +def run_holmanctl(monkeypatch, *args): + monkeypatch.setattr("sys.argv", ["holmanctl", *args]) + return holmanctl.main() + + +def manual_writes(tap_timer): + return next( + c.written + for c in tap_timer.services[0].characteristics + if c.uuid == holman.TapTimer.MANUAL_CHARACTERISTIC_UUID + ) + + +def test_connect_failed_finishes_with_error(tap_timer, manager): listener = holmanctl.TapTimerTestListener(tap_timer) - with pytest.raises(SystemExit): - listener.connect_failed(Exception("boom")) + manager.run() + listener.connect_failed(Exception("boom")) + assert listener.finished + assert listener.exit_code == 1 assert manager.stopped @@ -15,28 +32,97 @@ def test_disconnect_with_auto_reconnect_reconnects(tap_timer, manager): listener = holmanctl.TapTimerTestListener(tap_timer, auto_reconnect=True) listener.disconnect_succeeded() assert tap_timer.connect_calls == 1 - assert not manager.stopped + assert not listener.finished def test_known_lists_tap_timers(monkeypatch, capsys): class KnownManager(holman.TapTimerManager): def __init__(self, adapter_name): super().__init__(adapter_name) - self.aliases["AA:BB:CC:DD:EE:FF"] = "Tap Timer" - self.known_devices = [ - holman.TapTimer(mac_address="AA:BB:CC:DD:EE:FF", manager=self) - ] + self.aliases[MAC] = "Tap Timer" + self.known_devices = [holman.TapTimer(mac_address=MAC, manager=self)] monkeypatch.setattr(holman, "TapTimerManager", KnownManager) - monkeypatch.setattr("sys.argv", ["holmanctl", "--known"]) - holmanctl.main() - assert capsys.readouterr().out == "[AA:BB:CC:DD:EE:FF] Tap Timer\n" + assert run_holmanctl(monkeypatch, "--known") == 0 + assert capsys.readouterr().out == f"[{MAC}] Tap Timer\n" def test_requires_a_command(monkeypatch): - monkeypatch.setattr("sys.argv", ["holmanctl"]) with pytest.raises(SystemExit): - holmanctl.main() + run_holmanctl(monkeypatch) + + +def test_run_starts_tap_then_disconnects(monkeypatch, capsys, simulated_tap_timer): + assert run_holmanctl(monkeypatch, "--connect", MAC, "--run", "10") == 0 + (tap_timer,) = simulated_tap_timer.tap_timers + assert manual_writes(tap_timer) == [bytes([0x01, 0x00, 0x00, 10])] + assert tap_timer.disconnect_calls == 1 + out = capsys.readouterr().out + assert "timer written" in out + assert "Terminate with Ctrl+C" not in out + + +def test_stop_stops_tap_then_disconnects(monkeypatch, simulated_tap_timer): + assert run_holmanctl(monkeypatch, "--connect", MAC, "--stop") == 0 + (tap_timer,) = simulated_tap_timer.tap_timers + assert manual_writes(tap_timer) == [bytes(4)] + assert tap_timer.disconnect_calls == 1 + + +def test_connect_without_action_writes_nothing(monkeypatch, simulated_tap_timer): + assert run_holmanctl(monkeypatch, "--connect", MAC) == 0 + (tap_timer,) = simulated_tap_timer.tap_timers + assert manual_writes(tap_timer) == [] + assert tap_timer.disconnect_calls == 0 + + +def test_run_write_failure_exits_nonzero(monkeypatch, capsys, simulated_tap_timer): + simulated_tap_timer.write_error = Exception("write rejected") + assert run_holmanctl(monkeypatch, "--connect", MAC, "--run", "10") == 1 + (tap_timer,) = simulated_tap_timer.tap_timers + assert tap_timer.disconnect_calls == 1 + assert "timer write failed: write rejected" in capsys.readouterr().out + + +def test_connect_failure_exits_nonzero_without_running_loop( + monkeypatch, manager, simulated_tap_timer +): + # gatt reports some failures synchronously inside connect(), before the + # loop starts; main() must not then block forever in run(). + monkeypatch.setattr(holman, "TapTimerManager", lambda adapter_name: manager) + simulated_tap_timer.connect_error = Exception("Device does not exist") + assert run_holmanctl(monkeypatch, "--connect", MAC, "--run", "10") == 1 + assert not manager.ran + + +@pytest.mark.parametrize("value", ["0", "256", "-5", "ten", "1.5"]) +def test_run_rejects_invalid_minutes(monkeypatch, capsys, value): + with pytest.raises(SystemExit) as excinfo: + run_holmanctl(monkeypatch, "--connect", MAC, "--run", value) + assert excinfo.value.code == 2 + assert "--run" in capsys.readouterr().err + + +@pytest.mark.parametrize("value", ["1", "255"]) +def test_run_accepts_minute_bounds(value): + assert holmanctl.minutes(value) == int(value) + + +@pytest.mark.parametrize( + "args", + [ + ["--auto", MAC, "--run", "10"], + ["--auto", MAC, "--stop"], + ["--discover", "--stop"], + ["--connect", MAC, "--run", "10", "--stop"], + ], +) +def test_run_and_stop_only_with_connect(monkeypatch, manager, args): + monkeypatch.setattr(holman, "TapTimerManager", lambda adapter_name: manager) + with pytest.raises(SystemExit) as excinfo: + run_holmanctl(monkeypatch, *args) + assert excinfo.value.code == 2 + assert not manager.ran def test_public_api(): diff --git a/tests/test_tap_timer.py b/tests/test_tap_timer.py index 00f4e19..8d13226 100644 --- a/tests/test_tap_timer.py +++ b/tests/test_tap_timer.py @@ -101,3 +101,40 @@ def test_connect_failed_notifies_listener(tap_timer, listener): def test_battery_level_not_decoded(tap_timer): assert tap_timer.battery_level() is None + + +def test_write_succeeded_on_manual_characteristic_notifies( + make_holman_service, tap_timer, listener +): + resolve(tap_timer, make_holman_service()) + listener.events.clear() + tap_timer.characteristic_write_value_succeeded(manual_char(tap_timer)) + assert listener.events == ["timer_written"] + + +def test_write_failed_on_manual_characteristic_notifies( + make_holman_service, tap_timer, listener +): + resolve(tap_timer, make_holman_service()) + listener.events.clear() + tap_timer.characteristic_write_value_failed(manual_char(tap_timer), "rejected") + assert listener.events == [("timer_write_failed", "rejected")] + + +def test_write_on_other_characteristic_does_not_notify( + make_holman_service, tap_timer, listener +): + resolve(tap_timer, make_holman_service()) + listener.events.clear() + state = next(c for c in tap_timer.services[0].characteristics if c.uuid != MANUAL) + tap_timer.characteristic_write_value_succeeded(state) + tap_timer.characteristic_write_value_failed(state, "rejected") + assert listener.events == [] + + +def test_base_listener_accepts_write_events(make_holman_service, tap_timer): + # Listeners written before these events existed must keep working. + tap_timer.listener = holman.TapTimerListener() + resolve(tap_timer, make_holman_service()) + tap_timer.characteristic_write_value_succeeded(manual_char(tap_timer)) + tap_timer.characteristic_write_value_failed(manual_char(tap_timer), "rejected")