diff --git a/.gitignore b/.gitignore index dd2b7e6..3ecd5cb 100644 --- a/.gitignore +++ b/.gitignore @@ -102,3 +102,6 @@ ENV/ # Visual Studio Code .vscode/ + +# bluez +bluez diff --git a/README.md b/README.md index 3fa413f..bf8653a 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # Holman Python SDK -[Holman CO3015](https://www.holmanindustries.com.au/products/bluetooth-tap-timer-co3015/) -and [BTX1](https://www.holmanindustries.com.au/products/btx1-tap-mounted-smart-valve/) are Bluetooth tap timers made by [Holman](https://www.holmanindustries.com.au/). +[Holman CO3015](https://www.holmanindustries.com.au/products/bluetooth-tap-timer-co3015/), +[BTX1](https://www.holmanindustries.com.au/products/btx1-tap-mounted-smart-valve/), +and [BTX2](https://www.holmanindustries.com.au/products/btx2-dual-outlet-bluetooth-tap-timer/) are Bluetooth tap timers made by [Holman](https://www.holmanindustries.com.au/). The Holman Python SDK for Linux allows you to integrate your Holman(s) into any type of Linux application or script that can execute Python code. @@ -97,6 +98,38 @@ manager.start_discovery() manager.run() ``` +### Configuring accepted device aliases + +By default the SDK accepts devices whose advertised alias is exactly `Tap Timer` (upstream BX1) or `BX2` (this device). Match is membership (`if alias not in self.accepted_aliases: return None`); an empty alias does not match. Pass `accepted_aliases` or set `HOLMAN_ACCEPTED_ALIASES` (comma-separated) to add another exact name, not a prefix. + +```python +manager = holman.TapTimerManager( + adapter_name='hci0', + accepted_aliases=('Tap Timer', 'BX2', 'BX3')) +``` + +```bash +export HOLMAN_ACCEPTED_ALIASES="BX3" +sudo holmanctl --discover +``` + +### Configuring accepted BLE service UUIDs + +By default discovery **and** connect-time service pick use the same three known Holman vendor services: CO3015 (`0a75f000-f9ad-467a-e564-3c19163ad543`), CO3012 (`aacaebbb-af4b-baf3-7361-989ffeb0b129`, some BTX2), and CO3011 (`c521f000-0d70-4d4f-8e43-40d84c50ab38`, BTX1 / the on-air BX2 UUID). Named constants stay on `TapTimer`. + +Set `HOLMAN_SERVICE_UUIDS` to a comma-separated list to **replace** that filter (not add to it). Unset = hardcoded defaults; set = exactly those UUIDs. Or pass `service_uuids` to `TapTimerManager` (constructor wins over env). + +```python +manager = holman.TapTimerManager( + adapter_name='hci0', + service_uuids=('c521f000-0d70-4d4f-8e43-40d84c50ab38',)) +``` + +```bash +export HOLMAN_SERVICE_UUIDS="c521f000-0d70-4d4f-8e43-40d84c50ab38" +sudo holmanctl --discover +``` + ### Connecting to a Holman tap timer and receiving user input events Once `TapTimerManager` has discovered a Holman tap timer you can use the `TapTimer` object(s) that you retrieved from `TapTimerManager.tap_timers()` to connect to it. Alternatively you can create a new instance of `TapTimer` using the name of your Bluetooth adapter (typically `hci0`) and Holman's MAC address. @@ -121,7 +154,30 @@ As with Holman tap timer discovery, remember to start the Bluetooth event loop w ### Start the tap running -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. +Once a Holman tap timer is connected you can start the tap with `TapTimer.start(runtime=1)`. Pass a runtime in minutes. On dual-outlet BX2/BTX2 units, pass `zone=1` (Sprinkler, tap 0, `[0x01, 0x00, 0x00, mins]`) or `zone=2` (Hose, tap 1, `[0x01, 0x01, 0x00, mins]`). Byte 1 is the outlet. + +```python +tap_timer.start(runtime=5, zone=1) +tap_timer.stop() +``` + +From the CLI: + +``` +sudo holmanctl --start AA:BB:CC:DD:EE:FF --minutes 5 --zone 1 +sudo holmanctl --stop AA:BB:CC:DD:EE:FF +sudo holmanctl --debug --start AA:BB:CC:DD:EE:FF --minutes 5 --zone 2 +``` + +`--debug` logs the f006 write as hex (`01010005` for Hose, 5 minutes). Unknown `--zone` values are refused. + +Payload helpers are pure functions (`holman.payload.manual_payload`) and have unit tests that do not need Bluetooth: + +``` +python3 -m unittest discover -s tests -v +``` + +See [docs/BX2.md](docs/BX2.md) for the BX2 GATT notes (zones, unlock, what not to read) and [docs/homeassistant-example.md](docs/homeassistant-example.md) for a generic Home Assistant / Lovelace sketch (Sprinkler / Hose; no site addresses). ## Support diff --git a/docs/BX2.md b/docs/BX2.md new file mode 100644 index 0000000..fede51e --- /dev/null +++ b/docs/BX2.md @@ -0,0 +1,91 @@ +# Holman BX2 notes + +Protocol notes for a Holman **BX2** (dual outlet, advertised name `BX2`) with this SDK. No device addresses or credentials belong in this tree. + +`AE 8E` is a shared session unlock also used by public BX1 ESPHome configs, not a per-device secret. + +## Identity + +- Advertised alias is exactly `BX2` (bluetoothctl Name/Alias and Bleak `local_name`). Not `Holman BX2` or `BTX2`. Discovery matches exact names `Tap Timer` (BX1) and `BX2`. +- This BX2 advertises vendor service `c521f000-0d70-4d4f-8e43-40d84c50ab38` (already labelled BTX1 / CO3011). Discovery and connect-time service pick use the same list: CO3015, CO3012, and CO3011 by default (the on-air BX2 UUID is CO3011). Set `HOLMAN_SERVICE_UUIDS` to a comma-separated list to replace that filter, or pass `service_uuids` to `TapTimerManager`. Another BTX2 UUID (`aacaebbb-...`) stays in the default list for other units. +- Manufacturer company id `0x0374`. BLE address type is **random**. + +## GATT (safe) + +| UUID | Role | +| --- | --- | +| `c521f000-...` | Vendor service | +| `0000c001-...` | Write. Session unlock `AE 8E`. | +| `0000f006-...` | Write. Manual start/stop. | +| `0000f004-...` | Read. 12-byte state. Last byte `01` means running **when the official app started the valve**. | + +`start()` / `stop()` send the unlock when `c001` is present. + +## Manual payload + +4 bytes on `f006`: + + [0x01, tap, 0x00, minutes] + +| Tap | Name | Start write | Hex | +| --- | --- | --- | --- | +| 0 | Sprinkler | `[0x01, 0x00, 0x00, minutes]` | `010000NN` | +| 1 | Hose | `[0x01, 0x01, 0x00, minutes]` | `010100NN` | +| — | Stop | `[0x00, 0x00, 0x00, 0x00]` | `00000000` | + +- Byte 0 is on/off (`0x01` start, `0x00` stop). +- Byte 1 is the outlet: tap `0x00` = Sprinkler, tap `0x01` = Hose. +- `minutes` is `1...255`. +- Stop is all-off (both outlets). +- Sprinkler matches the original single-outlet SDK ON payload `01 00 00 `. BX1 stays compatible if callers leave the default tap 0. + +A 10-byte ESPHome-style pad (`01 00 00 mins` + six zeros) is accepted if written **without** response. A 10-byte write **with** response returned ATT `0x0e` and dropped the link. Prefer the 4-byte form. + +If a 4-byte write **with** response fails (ATT `0x0e`), retry **without** response. That is common while a run is already active. + +## Dual outlet behaviour + +- The timer can run **one outlet at a time**. Starting Hose while Sprinkler is open is not a second concurrent valve. +- To switch outlets: write stop (`00 00 00 00`), then start the other tap. A start-while-running write with response often errors; stop first. +- There is no extra reset characteristic required after a manual run. Stop is the all-zero `f006` write. + +## What not to do + +- **Do not read `0000e002-...`.** That drops the connection. +- Reading `f004` without a prior `c001` unlock can also return ATT `0x0e` and drop the link. +- First LE connect often fails with `le-connection-abort-by-local` / "failed to discover services, device disconnected". Retry. Two clients racing the same adapter makes this worse. +- `f004` last byte is **not a reliable water-is-flowing flag** after an SDK write. The official app sets it to `01`. A 4-byte start can open the valve while last byte stays `00`. Treat a successful write as optimistic; confirm physically if it matters. +- A successful GATT write can still look dry if that outlet is blocked. Confirm water, not only BLE ACKs. + +## Other characteristics + +Seen on the same service, not required for manual run: + +- `f003` — identity blob (MAC bytes reversed), readable without unlock. +- `f005`, `e001`, `c002` — read/write. Unlock may be required. Not needed for start/stop. +- `46a60001-ca26-425a-9bc6-d917829d2906` — write + notify. Untouched. + +## App pairing vs session unlock + +BlueZ `Paired`/`Bonded` can stay **no**. The Holman app still talks to the timer. Multiple phones can start a **manual** run at the same time. `AE 8E` is a session unlock, not exclusive SMP pairing. + +The printed manual's "one smartphone" line is about **scheduling ownership**, not a hard lock on manual GATT writes. The physical dial can still disable onboard schedules; that does not block these manual `f006` writes. + +## Suggested PR surface + +1. Accept exact aliases `Tap Timer` and `BX2` (optional `HOLMAN_ACCEPTED_ALIASES` adds more exact names). +2. Discover and connect with the same service UUID list (defaults include the on-air BX2 UUID CO3011 / `c521f000-...`). `HOLMAN_SERVICE_UUIDS` or `service_uuids=` fully replaces the list when set. +3. Unlock `c001` with `AE 8E` when the characteristic exists. +4. `start(runtime, zone=1)` writes `[0x01, tap, 0, mins]` (tap `0x00` Sprinkler / `0x01` Hose); `stop()` writes zeros. +5. CLI `--start` / `--stop` / `--minutes` / `--zone`. +6. README mention of BTX2 / BX2 and a link here. + +Leave Home Assistant bindings, retries, and site addresses out of the SDK. + +## CLI + +``` +holmanctl --discover +holmanctl --start AA:BB:CC:DD:EE:FF --minutes 2 --zone 1 +holmanctl --stop AA:BB:CC:DD:EE:FF +``` diff --git a/docs/homeassistant-example.md b/docs/homeassistant-example.md new file mode 100644 index 0000000..9d7cd01 --- /dev/null +++ b/docs/homeassistant-example.md @@ -0,0 +1,67 @@ +# Home Assistant / Lovelace example (generic) + +Sprinkler = tap 0. Hose = tap 1. One outlet at a time. No site IPs or MACs. + +f006 start bytes used by this SDK: + +| Tap | Name | Start | Hex | +| --- | --- | --- | --- | +| 0 | Sprinkler | `[0x01, 0x00, 0x00, minutes]` | `010000NN` | +| 1 | Hose | `[0x01, 0x01, 0x00, minutes]` | `010100NN` | +| — | Stop | `[0x00, 0x00, 0x00, 0x00]` | `00000000` | + +## Package snippet + +```yaml +holman_bt: + - mac: AA:BB:CC:DD:EE:FF + name: Holman BX2 + default_runtime: 5 +``` + +## Lovelace cards + +Entities are named from the device name plus the tap +(`switch.holman_bx2_sprinkler`, `switch.holman_bx2_hose`). + +```yaml +type: vertical-stack +cards: + - type: entities + title: Holman BX2 + entities: + - entity: number.holman_bx2_runtime + name: Minutes + - entity: switch.holman_bx2_sprinkler + name: Sprinkler + icon: mdi:sprinkler + - entity: switch.holman_bx2_hose + name: Hose + icon: mdi:hose + - type: horizontal-stack + cards: + - type: button + name: Start Sprinkler + icon: mdi:sprinkler + tap_action: + action: call-service + service: holman_bt.start + data: + zone: 1 + minutes: 5 + - type: button + name: Start Hose + icon: mdi:hose + tap_action: + action: call-service + service: holman_bt.start + data: + zone: 2 + minutes: 5 + - type: button + name: Stop + icon: mdi:water-off + tap_action: + action: call-service + service: holman_bt.stop +``` diff --git a/holman/__init__.py b/holman/__init__.py index 1456ceb..f5fa261 100644 --- a/holman/__init__.py +++ b/holman/__init__.py @@ -1 +1,3 @@ from .holman import TapTimerManager, TapTimerManagerListener, TapTimer, TapTimerListener +from .payload import clamp_runtime, manual_payload, tap_for_zone, tap_name +from .aliases import get_default_aliases, alias_accepted, get_default_service_uuids diff --git a/holman/aliases.py b/holman/aliases.py new file mode 100644 index 0000000..6b0fc35 --- /dev/null +++ b/holman/aliases.py @@ -0,0 +1,78 @@ +''' +Discovery allowlists: exact advertised names and vendor service UUIDs. +''' + +import os + +DEFAULT_ALIASES = ('Tap Timer', 'BX2') + +HOLMAN_CO3015_SERVICE_UUID = '0a75f000-f9ad-467a-e564-3c19163ad543' +HOLMAN_CO3012_SERVICE_UUID = 'aacaebbb-af4b-baf3-7361-989ffeb0b129' # some BTX2 +HOLMAN_CO3011_SERVICE_UUID = 'c521f000-0d70-4d4f-8e43-40d84c50ab38' # BTX1 / BX2 +DEFAULT_SERVICE_UUIDS = ( + HOLMAN_CO3015_SERVICE_UUID, + HOLMAN_CO3012_SERVICE_UUID, + HOLMAN_CO3011_SERVICE_UUID, +) + + +def get_default_aliases(): + ''' + Exact ``Tap Timer`` (BX1) and ``BX2`` by default. Env adds more exact names. + ''' + aliases = list(DEFAULT_ALIASES) + env = os.environ.get('HOLMAN_ACCEPTED_ALIASES') + if env: + for name in env.split(','): + name = name.strip() + if name and name not in aliases: + aliases.append(name) + return tuple(aliases) + + +def alias_accepted(alias, accepted_aliases=None): + ''' + True when ``alias`` is a non-empty exact member of the allowlist. + ''' + if accepted_aliases is None: + accepted_aliases = get_default_aliases() + if not alias: + return False + return alias in accepted_aliases + + +def get_default_service_uuids(): + ''' + Default Holman vendor services (CO3015, CO3012, CO3011). + + If ``HOLMAN_SERVICE_UUIDS`` is set, that comma-separated list fully + replaces the defaults (not extras). Unset or blank = defaults. + ''' + env = os.environ.get('HOLMAN_SERVICE_UUIDS') + if env and env.strip(): + uuids = [] + for raw in env.split(','): + raw = raw.strip().lower() + if raw and raw not in uuids: + uuids.append(raw) + if uuids: + return tuple(uuids) + return DEFAULT_SERVICE_UUIDS + + +def resolve_aliases(accepted_aliases=None): + ''' + Constructor list wins; otherwise ``get_default_aliases()`` (env extras). + ''' + if accepted_aliases is not None: + return tuple(accepted_aliases) + return get_default_aliases() + + +def resolve_service_uuids(service_uuids=None): + ''' + Constructor list wins (lowercased); otherwise ``get_default_service_uuids()``. + ''' + if service_uuids is not None: + return tuple(u.lower() for u in service_uuids) + return get_default_service_uuids() diff --git a/holman/holman.py b/holman/holman.py index 1d287d1..011148d 100644 --- a/holman/holman.py +++ b/holman/holman.py @@ -1,24 +1,47 @@ """ Module for managing Holman Bluetooth tap timers. """ +import logging + import gatt +from .aliases import resolve_aliases, resolve_service_uuids +from .payload import manual_payload, tap_name + +_LOGGER = logging.getLogger(__name__) + class TapTimerManager(gatt.DeviceManager): """ Entry point for managing and discovering Holman ``TapTimer``s. """ - def __init__(self, adapter_name='hci0'): + def __init__(self, adapter_name='hci0', accepted_aliases=None, + service_uuids=None): """ Instantiates a ``TapTimerManager`` :param adapter_name: name of Bluetooth adapter used by this tap timer manager + :param accepted_aliases: exact advertised names to accept during + discovery. Defaults to ``('Tap Timer', 'BX2')`` + plus extra names from ``HOLMAN_ACCEPTED_ALIASES`` + (comma-separated exact strings). + :param service_uuids: vendor service UUIDs for discovery and + connect-time service pick. Defaults to + CO3015 / CO3012 / CO3011. When + ``HOLMAN_SERVICE_UUIDS`` is set, that + comma-separated list fully replaces the + defaults. Constructor wins over env. """ - super().__init__(adapter_name) + # DeviceManager.__init__ calls update_devices() -> make_device(), + # which reads accepted_aliases and service_uuids. + # Constructor list wins over HOLMAN_ACCEPTED_ALIASES / HOLMAN_SERVICE_UUIDS. + self.accepted_aliases = resolve_aliases(accepted_aliases) + self.service_uuids = resolve_service_uuids(service_uuids) self.listener = None self.discovered_tap_timers = {} + super().__init__(adapter_name) def tap_timers(self): """ @@ -33,12 +56,13 @@ def start_discovery(self, service_uuids=None): Assign a `TapTimerManagerListener` to the `listener` attribute to collect discovered Holmans. """ - super().start_discovery(service_uuids=TapTimer.SERVICE_UUIDS) + super().start_discovery(service_uuids=list(self.service_uuids)) def make_device(self, mac_address): device = gatt.Device( mac_address=mac_address, manager=self, managed=False) - if device.alias() != 'Tap Timer': + alias = device.alias() or '' + if alias not in self.accepted_aliases: return None return TapTimer(mac_address=mac_address, manager=self) @@ -86,17 +110,20 @@ class TapTimer(gatt.Device): :param listener: instance of ``TapTimerListener`` that will be notified with all events """ - HOLMAN_CO3015_SERVICE_UUID = '0a75f000-f9ad-467a-e564-3c19163ad543' - HOLMAN_CO3011_SERVICE_UUID = 'c521f000-0d70-4d4f-8e43-40d84c50ab38' # model BTX1 + HOLMAN_CO3012_SERVICE_UUID = 'aacaebbb-af4b-baf3-7361-989ffeb0b129' # some BTX2 + HOLMAN_CO3011_SERVICE_UUID = 'c521f000-0d70-4d4f-8e43-40d84c50ab38' # BTX1 / BX2 STATE_CHARACTERISTIC_UUID = '0000f004-0000-1000-8000-00805f9b34fb' MANUAL_CHARACTERISTIC_UUID = '0000f006-0000-1000-8000-00805f9b34fb' + AUTH_CHARACTERISTIC_UUID = '0000c001-0000-1000-8000-00805f9b34fb' + AUTH_PAYLOAD = bytes((0xAE, 0x8E)) SERVICE_UUIDS = [ HOLMAN_CO3015_SERVICE_UUID, + HOLMAN_CO3012_SERVICE_UUID, HOLMAN_CO3011_SERVICE_UUID] - def __init__(self, mac_address, manager): + def __init__(self, mac_address, manager, service_uuids=None): """ Create an instance with given Bluetooth adapter name and MAC address. @@ -105,13 +132,22 @@ def __init__(self, mac_address, manager): format: ``AA:BB:CC:DD:EE:FF`` :param manager: reference to the `TapTimerManager` that manages this tap timer + :param service_uuids: vendor service UUIDs for connect-time + service pick. Defaults to the manager + list, else the same resolved defaults / + ``HOLMAN_SERVICE_UUIDS`` replace list. """ super().__init__(mac_address=mac_address, manager=manager) + if service_uuids is None: + service_uuids = getattr(manager, 'service_uuids', None) + self.service_uuids = resolve_service_uuids(service_uuids) + self.listener = None self._battery_level = None self._manual_characteristic = None self._state_characteristic = None + self._auth_characteristic = None self._state = bytes([0]) def connect(self): @@ -150,9 +186,10 @@ def disconnect_succeeded(self): def services_resolved(self): super().services_resolved() + accepted = {uuid.lower() for uuid in self.service_uuids} holman_service = next(( service for service in self.services - if service.uuid in self.SERVICE_UUIDS), None) + if service.uuid.lower() in accepted), None) if holman_service is None: if self.listener: # TODO: Use proper exception subclass @@ -181,6 +218,9 @@ def services_resolved(self): "Holman GATT characteristic %s missing", self.STATE_CHARACTERISTIC_UUID)) return + self._auth_characteristic = next(( + char for char in holman_service.characteristics + if char.uuid == self.AUTH_CHARACTERISTIC_UUID), None) self._refresh_state() # TODO: Only fire connected event when we read the firmware @@ -207,22 +247,38 @@ def name(self): """The name of the tap timer.""" return str(self.alias()) - def start(self, runtime=1): - """ - Turn on the tap for ```runtime``` minutes. - - :param runtime: the number of minutes to run the tap - """ - runtime = 255 if runtime > 255 else runtime - if self._manual_characteristic: - value = bytes([0x01, 0x00, 0x00, runtime]) - self._manual_characteristic.write_value(value) + def _unlock(self): + """Best-effort BX1/BX2 session unlock on c001 (AE 8E).""" + if self._auth_characteristic: + self._auth_characteristic.write_value(self.AUTH_PAYLOAD) + + def start(self, runtime=1, zone=1): + ''' + Turn on the tap for ``runtime`` minutes. + + Builds the same 4-byte f006 payload as ``manual_payload``: + Sprinkler ``[0x01, 0x00, 0x00, mins]``, Hose ``[0x01, 0x01, 0x00, mins]``. + Unknown zones raise ``ValueError`` (fail closed). Runtime is + clamped to 1..255. Missing f006 raises ``RuntimeError``. + ''' + value = manual_payload(True, runtime, zone) + _LOGGER.debug( + 'f006 write {} zone={} name={}'.format( + value.hex(), zone, tap_name(zone)), + ) + self._unlock() + if not self._manual_characteristic: + raise RuntimeError('Holman GATT characteristic f006 missing') + self._manual_characteristic.write_value(value) def stop(self): - """Turn off the tap.""" - if self._manual_characteristic: - value = bytes([0x00, 0x00, 0x00, 0x00]) - self._manual_characteristic.write_value(value) + '''Turn off the tap (all-zero f006).''' + value = manual_payload(False) + _LOGGER.debug('f006 write {} stop'.format(value.hex())) + self._unlock() + if not self._manual_characteristic: + raise RuntimeError('Holman GATT characteristic f006 missing') + self._manual_characteristic.write_value(value) def characteristic_write_value_succeeded(self, characteristic): self._refresh_state() diff --git a/holman/payload.py b/holman/payload.py new file mode 100644 index 0000000..04d7364 --- /dev/null +++ b/holman/payload.py @@ -0,0 +1,59 @@ +''' +f006 manual payloads. No BLE imports. + +Byte 0 is on/off. Byte 1 is the outlet. Sprinkler is the original SDK +start [0x01, 0x00, 0x00, minutes]. Hose is [0x01, 0x01, 0x00, minutes] +(hex 010100NN). +''' + +CMD_ON = 0x01 +CMD_OFF = 0x00 +PAD_BYTE = 0x00 +TAP_SPRINKLER = 0x00 +TAP_HOSE = 0x01 +TAP_BY_ZONE = {1: TAP_SPRINKLER, 2: TAP_HOSE} +ZONE_NAMES = {1: 'Sprinkler', 2: 'Hose'} +ZONES = (1, 2) +DEFAULT_ZONE = 1 +RUNTIME_MIN = 1 +RUNTIME_MAX = 255 + + +def clamp_runtime(minutes): + return max(RUNTIME_MIN, min(int(minutes), RUNTIME_MAX)) + + +def tap_for_zone(zone): + ''' + Map zone 1/2 to the f006 tap byte. Unknown zones fail closed. + ''' + try: + key = int(zone) + except (TypeError, ValueError) as err: + raise ValueError( + 'unknown Holman zone %r; expected %s' % (zone, ZONES) + ) from err + try: + return TAP_BY_ZONE[key] + except KeyError as err: + raise ValueError( + 'unknown Holman zone %r; expected %s' % (zone, ZONES) + ) from err + + +def tap_name(zone): + try: + return ZONE_NAMES[int(zone)] + except (KeyError, TypeError, ValueError): + return 'zone %s' % zone + + +def manual_payload(on, minutes=5, zone=DEFAULT_ZONE): + ''' + 4-byte f006 write: stop is all-zero; start is [CMD_ON, tap, PAD, mins]. + ''' + if not on: + return bytes([CMD_OFF, CMD_OFF, CMD_OFF, CMD_OFF]) + tap = tap_for_zone(zone) + mins = clamp_runtime(minutes) + return bytes([CMD_ON, tap, PAD_BYTE, mins]) diff --git a/holmanctl.py b/holmanctl.py old mode 100755 new mode 100644 index 55b17a1..0efd2d6 --- a/holmanctl.py +++ b/holmanctl.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 +import logging import sys from argparse import ArgumentParser import holman @@ -90,7 +91,24 @@ def main(): metavar='address', type=str, help="Disconnect a Holman tap timer with a given MAC address") + arg_commands_group.add_argument( + '--start', + metavar='address', + type=str, + help="Connect and start a zone for --minutes (default zone 1)") + arg_commands_group.add_argument( + '--stop', + metavar='address', + type=str, + help="Connect and send all-off") + arg_parser.add_argument('--minutes', type=int, default=2, help="Minutes for --start") + arg_parser.add_argument('--zone', type=int, default=1, help="Outlet 1 Sprinkler or 2 Hose (BX2). Default 1") + arg_parser.add_argument( + '--debug', action='store_true', + help='Log f006 write hex and BLE debug to stderr') args = arg_parser.parse_args() + if args.debug: + logging.basicConfig(level=logging.DEBUG) global tap_timer_manager tap_timer_manager = holman.TapTimerManager(adapter_name=args.adapter) @@ -114,6 +132,37 @@ def main(): tap_timer = holman.TapTimer(mac_address=args.disconnect, manager=tap_timer_manager) tap_timer.disconnect() return + elif args.start: + tap_timer = holman.TapTimer(mac_address=args.start, manager=tap_timer_manager) + + class StartListener(TapTimerTestListener): + def connect_succeeded(self): + super().connect_succeeded() + from holman.payload import manual_payload, tap_name + try: + payload = manual_payload(True, args.minutes, args.zone) + print( + 'starting zone={} ({}) minutes={} payload={}'.format( + args.zone, tap_name(args.zone), args.minutes, payload.hex())) + self.tap_timer.start(runtime=args.minutes, zone=args.zone) + except ValueError as err: + print('start refused: {}'.format(err)) + tap_timer_manager.stop() + sys.exit(2) + + tap_timer.listener = StartListener(tap_timer=tap_timer) + tap_timer.connect() + elif args.stop: + tap_timer = holman.TapTimer(mac_address=args.stop, manager=tap_timer_manager) + + class StopListener(TapTimerTestListener): + def connect_succeeded(self): + super().connect_succeeded() + print("stopping") + self.tap_timer.stop() + + tap_timer.listener = StopListener(tap_timer=tap_timer) + tap_timer.connect() print("Terminate with Ctrl+C") try: diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_payload.py b/tests/test_payload.py new file mode 100644 index 0000000..7d3e07f --- /dev/null +++ b/tests/test_payload.py @@ -0,0 +1,242 @@ +"Shipped f006 payload builder — import the real function, not a copy." + +import importlib.util +import os +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] / 'holman' + + +def _load(name, filename): + spec = importlib.util.spec_from_file_location(name, ROOT / filename) + mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod + spec.loader.exec_module(mod) + return mod + + +payload = _load('holman.payload', 'payload.py') +aliases = _load('holman.aliases', 'aliases.py') +manual_payload = payload.manual_payload +tap_for_zone = payload.tap_for_zone +tap_name = payload.tap_name +clamp_runtime = payload.clamp_runtime +get_default_aliases = aliases.get_default_aliases +alias_accepted = aliases.alias_accepted +get_default_service_uuids = aliases.get_default_service_uuids +resolve_aliases = aliases.resolve_aliases +resolve_service_uuids = aliases.resolve_service_uuids +DEFAULT_SERVICE_UUIDS = aliases.DEFAULT_SERVICE_UUIDS +HOLMAN_CO3015_SERVICE_UUID = aliases.HOLMAN_CO3015_SERVICE_UUID +HOLMAN_CO3012_SERVICE_UUID = aliases.HOLMAN_CO3012_SERVICE_UUID +HOLMAN_CO3011_SERVICE_UUID = aliases.HOLMAN_CO3011_SERVICE_UUID +UNKNOWN_SERVICE_UUID = '00000000-0000-0000-0000-000000000000' + + +class TestManualPayload(unittest.TestCase): + def test_start_sprinkler(self): + self.assertEqual(list(manual_payload(True, 3, 1)), [0x01, 0x00, 0x00, 3]) + + def test_start_hose(self): + self.assertEqual(list(manual_payload(True, 1, 2)), [0x01, 0x01, 0x00, 1]) + + def test_hose_payload_bytes(self): + self.assertEqual(list(manual_payload(True, 5, 2)), [0x01, 0x01, 0x00, 5]) + + def test_stop_all_zero(self): + self.assertEqual(list(manual_payload(False, 10, 2)), [0x00, 0x00, 0x00, 0x00]) + self.assertEqual(list(manual_payload(False, 0, 0)), [0x00, 0x00, 0x00, 0x00]) + + def test_minutes_clamped(self): + self.assertEqual(list(manual_payload(True, 0, 2))[3], 1) + self.assertEqual(list(manual_payload(True, 999, 1))[3], 255) + self.assertEqual(clamp_runtime(0), 1) + self.assertEqual(clamp_runtime(999), 255) + + def test_invalid_zone_fails_closed(self): + with self.assertRaises(ValueError): + manual_payload(True, 5, 3) + with self.assertRaises(ValueError): + manual_payload(True, 5, 0) + with self.assertRaises(ValueError): + tap_for_zone(9) + with self.assertRaises(ValueError): + tap_for_zone('hose') + + def test_tap_helpers(self): + self.assertEqual(tap_for_zone(1), 0x00) + self.assertEqual(tap_for_zone(2), 0x01) + self.assertEqual(tap_name(1), 'Sprinkler') + self.assertEqual(tap_name(2), 'Hose') + + +class TestAcceptedAliases(unittest.TestCase): + def _clear_alias_env(self): + return os.environ.pop('HOLMAN_ACCEPTED_ALIASES', None) + + def _restore_alias_env(self, aliases): + if aliases is not None: + os.environ['HOLMAN_ACCEPTED_ALIASES'] = aliases + + def test_default_exact_tap_timer_and_bx2(self): + old_aliases = self._clear_alias_env() + try: + self.assertEqual(get_default_aliases(), ('Tap Timer', 'BX2')) + finally: + self._restore_alias_env(old_aliases) + + def test_env_adds_exact_alias(self): + old_aliases = self._clear_alias_env() + os.environ['HOLMAN_ACCEPTED_ALIASES'] = 'BX3' + try: + self.assertEqual(get_default_aliases(), ('Tap Timer', 'BX2', 'BX3')) + finally: + os.environ.pop('HOLMAN_ACCEPTED_ALIASES', None) + self._restore_alias_env(old_aliases) + + def test_unknown_names_fail_closed(self): + old_aliases = self._clear_alias_env() + try: + accepted = get_default_aliases() + self.assertTrue(alias_accepted('Tap Timer', accepted)) + self.assertTrue(alias_accepted('BX2', accepted)) + for name in ('', 'Tap', 'BX', 'Holman BX2', 'BTX2', 'BX21', 'Tap Timer '): + self.assertFalse(alias_accepted(name, accepted)) + self.assertNotIn('Tap', accepted) + self.assertNotIn('BX', accepted) + self.assertNotIn('', accepted) + finally: + self._restore_alias_env(old_aliases) + + +class TestServiceUuids(unittest.TestCase): + def _clear_uuid_env(self): + return os.environ.pop('HOLMAN_SERVICE_UUIDS', None) + + def _restore_uuid_env(self, value): + if value is not None: + os.environ['HOLMAN_SERVICE_UUIDS'] = value + + def test_default_includes_co3011(self): + old = self._clear_uuid_env() + try: + uuids = get_default_service_uuids() + self.assertEqual(uuids, DEFAULT_SERVICE_UUIDS) + self.assertIn(HOLMAN_CO3011_SERVICE_UUID, uuids) + self.assertIn(HOLMAN_CO3015_SERVICE_UUID, uuids) + self.assertIn(HOLMAN_CO3012_SERVICE_UUID, uuids) + finally: + self._restore_uuid_env(old) + + def test_env_replaces_list(self): + old = self._clear_uuid_env() + os.environ['HOLMAN_SERVICE_UUIDS'] = ( + UNKNOWN_SERVICE_UUID + ', ' + HOLMAN_CO3011_SERVICE_UUID) + try: + uuids = get_default_service_uuids() + self.assertEqual( + uuids, (UNKNOWN_SERVICE_UUID, HOLMAN_CO3011_SERVICE_UUID)) + self.assertNotIn(HOLMAN_CO3015_SERVICE_UUID, uuids) + self.assertNotIn(HOLMAN_CO3012_SERVICE_UUID, uuids) + finally: + os.environ.pop('HOLMAN_SERVICE_UUIDS', None) + self._restore_uuid_env(old) + + def test_unknown_uuid_not_in_default_unless_env_set(self): + old = self._clear_uuid_env() + try: + self.assertNotIn(UNKNOWN_SERVICE_UUID, get_default_service_uuids()) + os.environ['HOLMAN_SERVICE_UUIDS'] = UNKNOWN_SERVICE_UUID + self.assertEqual(get_default_service_uuids(), (UNKNOWN_SERVICE_UUID,)) + self.assertNotIn(HOLMAN_CO3011_SERVICE_UUID, get_default_service_uuids()) + finally: + os.environ.pop('HOLMAN_SERVICE_UUIDS', None) + self._restore_uuid_env(old) + + +class TestResolveAliasesAndUuids(unittest.TestCase): + def _clear_envs(self): + return ( + os.environ.pop('HOLMAN_ACCEPTED_ALIASES', None), + os.environ.pop('HOLMAN_SERVICE_UUIDS', None), + ) + + def _restore_envs(self, aliases, uuids): + if aliases is not None: + os.environ['HOLMAN_ACCEPTED_ALIASES'] = aliases + if uuids is not None: + os.environ['HOLMAN_SERVICE_UUIDS'] = uuids + + def test_unset_env_resolve_defaults(self): + old_aliases, old_uuids = self._clear_envs() + try: + self.assertEqual(resolve_aliases(), ('Tap Timer', 'BX2')) + self.assertEqual(resolve_aliases(None), ('Tap Timer', 'BX2')) + uuids = resolve_service_uuids() + self.assertEqual(uuids, DEFAULT_SERVICE_UUIDS) + self.assertIn(HOLMAN_CO3011_SERVICE_UUID, uuids) + finally: + self._restore_envs(old_aliases, old_uuids) + + def test_env_appends_exact_alias_via_resolve(self): + old_aliases, old_uuids = self._clear_envs() + os.environ['HOLMAN_ACCEPTED_ALIASES'] = 'BX3' + try: + self.assertEqual(resolve_aliases(), ('Tap Timer', 'BX2', 'BX3')) + finally: + os.environ.pop('HOLMAN_ACCEPTED_ALIASES', None) + self._restore_envs(old_aliases, old_uuids) + + def test_env_full_replaces_service_uuids_via_resolve(self): + old_aliases, old_uuids = self._clear_envs() + os.environ['HOLMAN_SERVICE_UUIDS'] = UNKNOWN_SERVICE_UUID + try: + uuids = resolve_service_uuids() + self.assertEqual(uuids, (UNKNOWN_SERVICE_UUID,)) + self.assertNotIn(HOLMAN_CO3015_SERVICE_UUID, uuids) + self.assertNotIn(HOLMAN_CO3012_SERVICE_UUID, uuids) + self.assertNotIn(HOLMAN_CO3011_SERVICE_UUID, uuids) + finally: + os.environ.pop('HOLMAN_SERVICE_UUIDS', None) + self._restore_envs(old_aliases, old_uuids) + + def test_resolve_aliases_constructor_wins_over_env(self): + old_aliases, old_uuids = self._clear_envs() + os.environ['HOLMAN_ACCEPTED_ALIASES'] = 'Nope' + try: + self.assertEqual(resolve_aliases(('BX2',)), ('BX2',)) + self.assertNotIn('Nope', resolve_aliases(('BX2',))) + finally: + os.environ.pop('HOLMAN_ACCEPTED_ALIASES', None) + self._restore_envs(old_aliases, old_uuids) + + def test_resolve_service_uuids_constructor_wins_over_env(self): + old_aliases, old_uuids = self._clear_envs() + os.environ['HOLMAN_SERVICE_UUIDS'] = UNKNOWN_SERVICE_UUID + try: + self.assertEqual( + resolve_service_uuids((HOLMAN_CO3011_SERVICE_UUID,)), + (HOLMAN_CO3011_SERVICE_UUID,)) + self.assertNotIn( + UNKNOWN_SERVICE_UUID, + resolve_service_uuids((HOLMAN_CO3011_SERVICE_UUID,))) + finally: + os.environ.pop('HOLMAN_SERVICE_UUIDS', None) + self._restore_envs(old_aliases, old_uuids) + + def test_device_data_bx2_and_co3011(self): + old_aliases, old_uuids = self._clear_envs() + try: + self.assertTrue(alias_accepted('BX2')) + self.assertEqual( + HOLMAN_CO3011_SERVICE_UUID, + 'c521f000-0d70-4d4f-8e43-40d84c50ab38') + self.assertIn(HOLMAN_CO3011_SERVICE_UUID, resolve_service_uuids()) + finally: + self._restore_envs(old_aliases, old_uuids) + + +if __name__ == '__main__': + unittest.main()