Add USB support

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- Partial/experimental support for `pybricksdev run usb`.
+
 ### Fixed
 - Fix crash when running `pybricksdev run ble -` (bug introduced in alpha.49).
 

pybricksdev/cli/__init__.py

@@ -171,34 +171,64 @@ def add_parser(self, subparsers: argparse._SubParsersAction):
             )
 
     async def run(self, args: argparse.Namespace):
-        from pybricksdev.ble import find_device
-        from pybricksdev.connections.ev3dev import EV3Connection
-        from pybricksdev.connections.lego import REPLHub
-        from pybricksdev.connections.pybricks import PybricksHub
 
         # Pick the right connection
         if args.conntype == "ssh":
+            from pybricksdev.connections.ev3dev import EV3Connection
+
             # So it's an ev3dev
             if args.name is None:
                 print("--name is required for SSH connections", file=sys.stderr)
                 exit(1)
 
-            hub = EV3Connection()
             device_or_address = socket.gethostbyname(args.name)
+            hub = EV3Connection(device_or_address)
         elif args.conntype == "ble":
+            from pybricksdev.ble import find_device as find_ble
+            from pybricksdev.connections.pybricks import PybricksHubBLE
+
             # It is a Pybricks Hub with BLE. Device name or address is given.
-            hub = PybricksHub()
             print(f"Searching for {args.name or 'any hub with Pybricks service'}...")
-            device_or_address = await find_device(args.name)
-
+            device_or_address = await find_ble(args.name)
+            hub = PybricksHubBLE(device_or_address)
         elif args.conntype == "usb":
-            hub = REPLHub()
-            device_or_address = None
+            from usb.core import find as find_usb
+
+            from pybricksdev.connections.pybricks import PybricksHubUSB
+            from pybricksdev.usb import (
+                LEGO_USB_VID,
+                MINDSTORMS_INVENTOR_USB_PID,
+                SPIKE_ESSENTIAL_USB_PID,
+                SPIKE_PRIME_USB_PID,
+            )
+
+            def is_pybricks_usb(dev):
+                return (
+                    (dev.idVendor == LEGO_USB_VID)
+                    and (
+                        dev.idProduct
+                        in [
+                            SPIKE_PRIME_USB_PID,
+                            SPIKE_ESSENTIAL_USB_PID,
+                            MINDSTORMS_INVENTOR_USB_PID,
+                        ]
+                    )
+                    and dev.product.endswith("Pybricks")
+                )
+
+            device_or_address = find_usb(custom_match=is_pybricks_usb)
+
+            if device_or_address is not None:
+                hub = PybricksHubUSB(device_or_address)
+            else:
+                from pybricksdev.connections.lego import REPLHub
+
+                hub = REPLHub()
         else:
             raise ValueError(f"Unknown connection type: {args.conntype}")
 
         # Connect to the address and run the script
-        await hub.connect(device_or_address)
+        await hub.connect()
         try:
             with _get_script_path(args.file) as script_path:
                 await hub.run(script_path, args.wait)

pybricksdev/connections/ev3dev.py

@@ -18,10 +18,13 @@ class EV3Connection:
     _USER = "robot"
     _PASSWORD = "maker"
 
+    def __init__(self, address: str):
+        self._address = address
+
     def abs_path(self, path):
         return os.path.join(self._HOME, path)
 
-    async def connect(self, address):
+    async def connect(self):
         """Connects to ev3dev using SSH with a known IP address.
 
         Arguments:
@@ -33,9 +36,9 @@ async def connect(self, address):
                 Connect failed.
         """
 
-        print("Connecting to", address, "...", end=" ")
+        print("Connecting to", self._address, "...", end=" ")
         self.client = await asyncssh.connect(
-            address, username=self._USER, password=self._PASSWORD
+            self._address, username=self._USER, password=self._PASSWORD
         )
         print("Connected.", end=" ")
         self.client.sftp = await self.client.start_sftp_client()