BX2 MK1: exact alias BX2 and two-outlet start on f006 byte 1

README.md

@@ -2,6 +2,12 @@
 [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/).
 
+**Supported / discovered devices:**
+- Confirmed on BX2 Dual Outlet Bluetooth Tap Timer MK1 (SKU CO3112) only. Newer BX2 not tested.
+- Default discovery allowlist is exact advertised names `Tap Timer` and `BX2`.
+- BX2 MK1: two-zone start via f006 byte 1 (zone 1 = tap index 0, zone 2 = tap index 1). Session unlock on c001 (AE 8E) when that characteristic exists.
+- Extra exact names: `HOLMAN_ACCEPTED_ALIASES` or constructor `accepted_aliases`. Replacement service UUID list: `HOLMAN_SERVICE_UUIDS` or constructor `service_uuids`.
+
 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.
 
 ## Prerequisites

holman/holman.py

@@ -1,24 +1,88 @@
 """
 Module for managing Holman Bluetooth tap timers.
 """
+import os
+
 import gatt
 
 
+_DEFAULT_ALIASES = ('Tap Timer', 'BX2')
+
+
+def _get_default_aliases():
+    """
+    Tap Timer / BX2 allowlist, plus HOLMAN_ACCEPTED_ALIASES extras.
+
+    If another Holman or compatible timer appears with a different
+    advertised name, it can be added via env without a code change.
+    """
+    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 _get_default_service_uuids():
+    '''
+    CO3015 / CO3012 / CO3011 unless HOLMAN_SERVICE_UUIDS is set.
+
+    When the env var is set, the comma-separated value fully replaces
+    the hardcoded defaults (not extras). Unset or blank = defaults.
+
+    If another Holman or compatible timer appears with a different
+    vendor service UUID, it can be matched via env without a code change.
+    '''
+    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 tuple(TapTimer.SERVICE_UUIDS)
+
+
+
 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.
+        if accepted_aliases is None:
+            accepted_aliases = _get_default_aliases()
+        self.accepted_aliases = tuple(accepted_aliases)
+        if service_uuids is None:
+            service_uuids = _get_default_service_uuids()
+        self.service_uuids = tuple(u.lower() for u in service_uuids)
         self.listener = None
         self.discovered_tap_timers = {}
+        super().__init__(adapter_name)
 
     def tap_timers(self):
         """
@@ -33,12 +97,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 +151,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'
+    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 +173,24 @@ 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)
+        if service_uuids is None:
+            service_uuids = _get_default_service_uuids()
+        self.service_uuids = tuple(u.lower() for u in 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 +229,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 +261,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,19 +290,36 @@ def name(self):
         """The name of the tap timer."""
         return str(self.alias())
 
-    def start(self, runtime=1):
+    def _unlock(self):
         """
-        Turn on the tap for ```runtime``` minutes.
+        Session unlock so the tap will accept f006 start/stop writes.
 
-        :param runtime: the number of minutes to run the tap
+        BX1/BX2 expose GATT c001; writing AE 8E unlocks the session for
+        subsequent f006 manual control. If c001 is missing, no-op.
         """
-        runtime = 255 if runtime > 255 else runtime
+        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.
+
+        f006 start is ``[0x01, tap_index, 0x00, minutes]``. Byte 0 is on
+        (0x01). Byte 1 is the tap index: 0x00 = zone 1, 0x01 = zone 2
+        (hex ``010100NN`` for zone 2). Zone 1 matches the original SDK ON
+        payload ``01 00 00 <mins>``.
+        '''
+        runtime = 255 if runtime > 255 else max(1, int(runtime))
+        zone = max(1, min(int(zone), 2))
+        tap = 0x00 if zone == 1 else 0x01
+        self._unlock()
         if self._manual_characteristic:
-            value = bytes([0x01, 0x00, 0x00, runtime])
+            value = bytes([0x01, tap, 0x00, runtime])
             self._manual_characteristic.write_value(value)
 
     def stop(self):
         """Turn off the tap."""
+        self._unlock()
         if self._manual_characteristic:
             value = bytes([0x00, 0x00, 0x00, 0x00])
             self._manual_characteristic.write_value(value)

holmanctl.py

@@ -90,6 +90,18 @@ 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="Zone 1 (tap index 0) or 2 (tap index 1). Default 1")
     args = arg_parser.parse_args()
 
     global tap_timer_manager
@@ -114,6 +126,28 @@ 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()
+                print(f"starting zone={args.zone} minutes={args.minutes}")
+                self.tap_timer.start(runtime=args.minutes, zone=args.zone)
+
+        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: