Automatically discover and verify ACE 2 on generic USB adapters

.github/workflows/release.yml

@@ -54,6 +54,7 @@ jobs:
           bash multiace/managed/tests/test_managed_guards.sh
           bash multiace/managed/tests/test_updater_asset_selection.sh
           python -m unittest discover -s multiace/managed/tests -v
+          python -m unittest -v test_ace_protocol_v2_probe
 
           python multiace/managed/build_package.py \
             --output "$RUNNER_TEMP/multiace-managed-check.tar.gz"

README.md

@@ -257,6 +257,28 @@ Snapmaker U1 USB Port - HUB
 
 > **Note:** VCC (5V) is not connected in the USB cable - only data lines. Each ACE Pro is powered by its own external power supply.
 
+#### Automatic USB-RS485 adapter discovery (experimental)
+
+multiACE automatically identifies ACE 2 units on generic USB serial adapters.
+No adapter VID/PID or discovery setting is required. ACE 2 support and generic
+USB probing are enabled by default; existing explicit `false` settings are
+respected. Existing `v2_extra_usb_ids` entries remain supported as manual overrides.
+
+The scan runs while the printer is idle, excludes serial paths configured in
+other Klipper sections and ports reported open by visible Linux processes,
+and uses only the V2 `DISCOVER_DEVICE` and `GET_INFO` queries. A candidate must
+pass CRC, command and sequence checks and return a non-zero UID plus ACE 2
+firmware identity. Each scan probes at most one new port, rotates through the
+candidates, and backs off after failures. These queries do not move filament
+or change settings on an ACE 2.
+
+Port ownership checks are best-effort and serial exclusivity is advisory.
+Automatic discovery sends protocol bytes to eligible USB serial ports;
+unrelated devices may interpret those bytes differently. Set
+`v2_probe_generic_usb: false` in `[ace]` to disable generic probing if needed.
+Host serial permissions must still allow Klipper to open the device.
+Hardware validation of the automatic flow is pending.
+
 
 ### PTFE Tube Splitters
 

multiace/config/extended/ace.cfg

@@ -1,18 +1,25 @@
 [save_variables]
 filename: /home/lava/printer_data/config/extended/multiace/ace_vars.cfg
 [ace]
+# Expected number of physical ACE units; change this for multiple units.
 ace_device_count: 1
 # Maximum seconds to wait for expected ACE units during startup (default 20.0s).
 # Set to 0.0 to disable the wait loop on single-extruder setups.
 #startup_wait: 20.0
-enable_ace_v2: true
+# ACE 2 support is enabled by default. Uncomment only to disable it.
+#enable_ace_v2: false
+
+# Keep ACE 2 units before ACE Pro (V1) units; the code default is USB order.
 v2_order: first
 
-# Extra USB vid:pid accepted as an ACE 2 - only for a generic USB-RS485
-# adapter in place of the genuine CH343 cable. Leave unset otherwise
-# (CH340 is a generic chip and would match unrelated devices).
+# Generic USB-RS485 ACE 2 adapters are discovered and verified automatically.
+# No USB device type or VID/PID is needed. Existing explicit overrides remain
+# supported; use these only when deliberately bypassing generic verification.
 #v2_extra_usb_ids: 1a86:7523
 
+# Optional escape hatch: disable automatic generic USB probing.
+#v2_probe_generic_usb: false
+
 
 update_url_base: https://postapocalyptic-diy.com/software/updates
 

multiace/klipper/extras/ace.py

@@ -1259,7 +1259,7 @@ def _parse_idx_list(key):
         self._homing_flag_path = config.get(
             'homing_flag_path', '/tmp/multiace_homing_active')
 
-        self._enable_ace_v2 = config.getboolean('enable_ace_v2', False)
+        self._enable_ace_v2 = config.getboolean('enable_ace_v2', True)
 
         self._v2_order = config.getchoice('v2_order',
                                           {'usb': 'usb', 'first': 'first',
@@ -1275,9 +1275,19 @@ def _parse_idx_list(key):
         self._v2_extra_usb_ids = self._parse_v2_extra_usb_ids(
             config.get('v2_extra_usb_ids', ''))
         AceProtocolV2.EXTRA_USB_IDS = self._v2_extra_usb_ids
+        self._v2_probe_generic_usb = config.getboolean(
+            'v2_probe_generic_usb', True)
+        self._v2_generic_probe_cursor = None
+        self._v2_generic_probe_cache = {}
+        self._v2_generic_probe_in_progress = set()
+        self._configured_serial_paths = self._read_configured_serial_paths(
+            config)
         if self._v2_extra_usb_ids:
             logging.info('[multiACE] V2 extra USB IDs (opt-in): %s' % (
                 ', '.join('%s:%s' % p for p in self._v2_extra_usb_ids)))
+        if self._v2_probe_generic_usb:
+            logging.info('[multiACE] V2 generic USB probing enabled; '
+                         'only verified ACE 2 devices will be accepted')
 
         # V2 print-time feed-assist mode:
         #   'constant' (default) - on extrusion just keep the ACE feed
@@ -2299,6 +2309,150 @@ def _parse_v2_extra_usb_ids(self, raw):
                     % tok)
         return tuple(pairs)
 
+    def _read_configured_serial_paths(self, config):
+        """Capture configured serial devices so generic scans avoid them."""
+        paths = set()
+        self._configured_serial_paths_valid = True
+        try:
+            sections = config.get_prefix_sections('')
+            if not sections:
+                raise ValueError('configuration section inventory is empty')
+        except Exception:
+            self._configured_serial_paths_valid = False
+            logging.warning('[multiACE] Automatic USB probing suspended: '
+                            'cannot enumerate configured serial ports')
+            return paths
+        own_section = getattr(config, 'get_name', lambda: 'ace')()
+        for section in sections:
+            try:
+                if section.get_name().lower() == own_section.lower():
+                    continue
+                path = (section.get('serial', None) or '').strip()
+            except Exception:
+                self._configured_serial_paths_valid = False
+                continue
+            if path.startswith('/dev/'):
+                paths.add(path)
+        return paths
+
+    def _is_printer_idle_for_generic_probe(self):
+        """Allow broad identification exchanges only while idle."""
+        if not self._configured_serial_paths_valid:
+            return False
+        try:
+            stats = self.printer.lookup_object('print_stats', None)
+            if stats is None:
+                return False
+            state = stats.get_status(self.reactor.monotonic()).get('state')
+            return state in ('standby', 'complete', 'cancelled', 'error')
+        except Exception:
+            return False
+
+    def _configured_serial_path_set(self):
+        return {os.path.realpath(path)
+                for path in self._configured_serial_paths}
+
+    def _probe_v2_generic_candidate(self, path, active_paths, probe_budget):
+        real_path = os.path.realpath(path)
+        if real_path in active_paths:
+            return True
+
+        if real_path in self._configured_serial_path_set():
+            logging.info('[multiACE] V2 probe skipped path=%s '
+                         'category=configured_serial_port', path)
+            return False
+
+        if not self._is_printer_idle_for_generic_probe():
+            return False
+
+        real_dev = os.path.basename(real_path)
+        vendor, product = AceProtocolV2._read_usb_ids(real_dev)
+        cache_key = (real_path, vendor, product)
+        now = time.monotonic()
+        cached = self._v2_generic_probe_cache.get(cache_key)
+        if cached is not None:
+            cached_at, verified, result = cached
+            # Keep positive identity results short-lived so unplug/replug on
+            # the same adapter path is re-identified on the next scan.
+            cache_ttl = (0.25 if verified else
+                         5.0 if result.get('category') in (
+                             'port_in_use', 'permission_denied', 'device_disconnected')
+                         else 30.0)
+            if now - cached_at < cache_ttl:
+                return verified
+
+        if cache_key in self._v2_generic_probe_in_progress:
+            return False
+
+        if AceProtocolV2.serial_port_in_use(path):
+            result = {
+                'ok': False,
+                'category': 'port_in_use',
+                'error': 'path is open by a visible process',
+            }
+            self._v2_generic_probe_cache[cache_key] = (
+                now, False, result)
+            self._usb_log.info(
+                'PROBE skip path=%s reason=already-open', path)
+            logging.info('[multiACE] V2 probe skipped path=%s '
+                         'category=port_in_use', path)
+            return False
+
+        if probe_budget[0] <= 0:
+            return False
+        probe_budget[0] -= 1
+        self._v2_generic_probe_cursor = real_path
+        probe_start = time.monotonic()
+        probe_done = threading.Event()
+        probe_result = {'result': None}
+        self._v2_generic_probe_in_progress.add(cache_key)
+
+        def _probe_worker():
+            try:
+                probe_result['result'] = AceProtocolV2.probe_device(
+                    path, query_timeout=0.35)
+            except Exception as e:
+                probe_result['result'] = {'ok': False, 'error': str(e)}
+            finally:
+                self._v2_generic_probe_in_progress.discard(cache_key)
+                probe_done.set()
+
+        threading.Thread(target=_probe_worker, daemon=True,
+                         name='multiace-v2-probe').start()
+        probe_deadline = self.reactor.monotonic() + 1.5
+        while not probe_done.is_set():
+            now_reactor = self.reactor.monotonic()
+            if now_reactor >= probe_deadline:
+                break
+            self.reactor.pause(min(probe_deadline, now_reactor + 0.025))
+        result = probe_result['result']
+        if result is None:
+            result = {
+                'ok': False,
+                'category': 'probe_timeout',
+                'error': 'probe exceeded 1.5s deadline',
+            }
+        verified = bool(result.get('ok'))
+        self._v2_generic_probe_cache[
+            cache_key] = (time.monotonic(), verified, result)
+        elapsed_ms = (time.monotonic() - probe_start) * 1000.0
+        if verified:
+            uid = '-'.join('%08x' % part for part in result['uid'])
+            self._usb_log.info(
+                'PROBE verified path=%s model=%s firmware=%s uid=%s '
+                'time=%.1fms', path, result['model'], result['firmware'],
+                uid, elapsed_ms)
+        else:
+            self._usb_log.info(
+                'PROBE rejected path=%s category=%s reason=%s time=%.1fms',
+                path, result.get('category', 'no_identity'),
+                result.get('error', 'no identity response'), elapsed_ms)
+            logging.info(
+                '[multiACE] V2 probe rejected path=%s category=%s reason=%s',
+                path, result.get('category', 'no_identity'),
+                result.get('error', 'no identity response'))
+        return verified
+
     def _scan_ace_devices(self, context='unknown'):
         scan_start = time.monotonic()
         self._usb_stats['scans'] += 1
@@ -2307,16 +2461,84 @@ def _scan_ace_devices(self, context='unknown'):
 
         active_protocols = KNOWN_PROTOCOLS if self._enable_ace_v2 \
             else tuple(p for p in KNOWN_PROTOCOLS if p is not AceProtocolV2)
+        discovered_by_protocol = {
+            protocol_cls: protocol_cls.discover()
+            for protocol_cls in active_protocols
+        }
+        probe_budget = [1]
+        active_paths = set()
+        for idx, ser in self._serials.items():
+            if (ser is not None and getattr(ser, 'is_open', False)
+                    and idx < len(self._ace_devices)):
+                active_paths.add(os.path.realpath(self._ace_devices[idx]))
+        configured_serial_paths = self._configured_serial_path_set()
+
         for protocol_cls in active_protocols:
-            for path in protocol_cls.discover():
+            for path in discovered_by_protocol[protocol_cls]:
                 if path in ace_devices:
                     continue
+                real_path = os.path.realpath(path)
+                if real_path in configured_serial_paths:
+                    logging.info('[multiACE] Discovery skipped configured '
+                                 'serial path %s', path)
+                    continue
+                if protocol_cls is AceProtocolV2:
+                    if real_path not in active_paths:
+                        official = AceProtocolV2.is_official_usb_path(path)
+                        real_dev = os.path.basename(real_path)
+                        usb_ids = AceProtocolV2._read_usb_ids(real_dev)
+                        explicitly_configured = (
+                            usb_ids in self._v2_extra_usb_ids)
+                        if official or explicitly_configured:
+                            if AceProtocolV2.serial_port_in_use(path):
+                                logging.info(
+                                    '[multiACE] V2 discovery skipped '
+                                    'path=%s category=port_in_use', path)
+                                continue
+                        elif (not self._v2_probe_generic_usb
+                              or not self._probe_v2_generic_candidate(
+                                  path, active_paths, probe_budget)):
+                            continue
                 self._ace_path_protocol[path] = protocol_cls
                 ace_devices.append(path)
                 real_dev = os.path.basename(os.path.realpath(path))
                 logging.info('[multiACE] Found device %s (%s) protocol=%s' % (
                     path, real_dev, protocol_cls.NAME))
 
+        if (self._enable_ace_v2 and self._v2_probe_generic_usb
+                and self._is_printer_idle_for_generic_probe()):
+            known_paths = set(discovered_by_protocol.get(
+                AceProtocolV1, ()))
+            known_paths.update(discovered_by_protocol.get(
+                AceProtocolV2, ()))
+            known_real_paths = {os.path.realpath(path)
+                                for path in known_paths.union(ace_devices)}
+            candidates = AceProtocolV2.discover_usb_serial_candidates()
+            # Resume after the last attempted port. A non-responsive adapter
+            # must not monopolize the one-probe budget on successive scans.
+            candidate_paths = [os.path.realpath(path) for path in candidates]
+            if self._v2_generic_probe_cursor in candidate_paths:
+                offset = candidate_paths.index(self._v2_generic_probe_cursor) + 1
+                candidates = candidates[offset:] + candidates[:offset]
+            for path in candidates:
+                real_path = os.path.realpath(path)
+                if real_path in known_real_paths:
+                    continue
+                if real_path in configured_serial_paths:
+                    logging.info('[multiACE] Generic V2 probe skipped '
+                                 'configured serial path %s', path)
+                    continue
+                if not self._probe_v2_generic_candidate(
+                        path, active_paths, probe_budget):
+                    continue
+                self._ace_path_protocol[path] = AceProtocolV2
+                ace_devices.append(path)
+                known_real_paths.add(real_path)
+                real_dev = os.path.basename(real_path)
+                logging.info(
+                    '[multiACE] Verified generic ACE 2 candidate %s (%s)' % (
+                        path, real_dev))
+
         ace_devices.sort(key=self._ace_path_sort_key)
 
         scan_ms = (time.monotonic() - scan_start) * 1000