16406482466

Committed 21 Jul 2025 01:23AM UTC coverage: 51.043% (+0.3%) from 50.769%

Build # 16406482466

Build Type

push

github

Committed by

dlech

Commit Message

cli: add NXT and EV3 to USB match for run command

Add the NXT and EV3 USB product IDs to the list of supported devices
for the `run` command in the CLI.

Run Details

67 of 273 branches covered (24.54%)

Branch coverage included in aggregate %.

1989 of 3755 relevant lines covered (52.97%)

0.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.23

pybricksdev/cli/__init__.py

# SPDX-License-Identifier: MIT
# Copyright (c) 2019-2024 The Pybricks Authors

"""Command line wrapper around pybricksdev library."""

import argparse
import asyncio
import contextlib
import logging
import os
import socket
import sys
from abc import ABC, abstractmethod
from os import PathLike, path
from tempfile import NamedTemporaryFile
from typing import ContextManager, TextIO

import argcomplete
from argcomplete.completers import FilesCompleter

from pybricksdev import __name__ as MODULE_NAME
from pybricksdev import __version__ as MODULE_VERSION

PROG_NAME = (
    f"{path.basename(sys.executable)} -m {MODULE_NAME}"
    if sys.argv[0].endswith("__main__.py")
    else path.basename(sys.argv[0])
)


class Tool(ABC):
    """Common base class for tool implementations."""

    @abstractmethod
    def add_parser(self, subparsers: argparse._SubParsersAction):
        """
        Overriding methods must at least do the following::

            parser = subparsers.add_parser('tool', ...)
            parser.tool = self

        Then additional arguments can be added using the ``parser`` object.
        """
        pass

    @abstractmethod
    async def run(self, args: argparse.Namespace):
        """
        Overriding methods should provide an implementation to run the tool.
        """
        pass


def _get_script_path(file: TextIO) -> ContextManager[PathLike]:
    """
    Gets the path to a script on the file system.

    If the file is ``sys.stdin``, the contents are copied to a temporary file
    and the path to the temporary file is returned. Otherwise, the file is closed
    and the path is returned.

    The context manager will delete the temporary file, if applicable.
    """
    if file is sys.stdin:
        # Have to close the temp file so that mpy-cross can read it, so we
        # create our own context manager to delete the file when we are done
        # using it.

        @contextlib.contextmanager
        def temp_context():
            try:
                with NamedTemporaryFile(suffix=".py", delete=False) as temp:
                    temp.write(file.buffer.read())

                yield temp.name
            finally:
                try:
                    os.remove(temp.name)
                except NameError:
                    # if NamedTemporaryFile() throws, temp is not defined
                    pass
                except OSError:
                    # file was already deleted or other strangeness
                    pass

        return temp_context()

    file.close()
    return contextlib.nullcontext(file.name)


class Compile(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "compile",
            help="compile a Pybricks program without running it",
        )
        parser.add_argument(
            "file",
            metavar="<file>",
            help="path to a MicroPython script or `-` for stdin",
            type=argparse.FileType(),
        )
        parser.add_argument(
            "--abi",
            metavar="<n>",
            help="the MPY ABI version, one of %(choices)s (default: %(default)s)",
            default=6,
            choices=[5, 6],
            type=int,
        )
        parser.tool = self

    async def run(self, args: argparse.Namespace):
        from pybricksdev.compile import compile_multi_file, print_mpy

        with _get_script_path(args.file) as script_path:
            mpy = await compile_multi_file(script_path, args.abi)
        print_mpy(mpy)


class Run(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "run",
            help="run a Pybricks program",
        )
        parser.tool = self
        parser.add_argument(
            "conntype",
            metavar="<connection type>",
            help="connection type: %(choices)s",
            choices=["ble", "usb", "ssh"],
        )
        parser.add_argument(
            "file",
            metavar="<file>",
            help="path to a MicroPython script or `-` for stdin",
            type=argparse.FileType(),
        )
        parser.add_argument(
            "-n",
            "--name",
            metavar="<name>",
            required=False,
            help="hostname or IP address for SSH connection; "
            "Bluetooth device name or Bluetooth address for BLE connection; "
            "serial port name for USB connection",
        )

        if hasattr(argparse, "BooleanOptionalAction"):
            # BooleanOptionalAction requires Python 3.9
            parser.add_argument(
                "--wait",
                help="wait for the program to complete before disconnecting",
                action=argparse.BooleanOptionalAction,
                default=True,
            )
        else:
            parser.add_argument(
                "--wait",
                help="wait for the program to complete before disconnecting (default)",
                action="store_true",
                default=True,
            )
            parser.add_argument(
                "--no-wait",
                help="disconnect as soon as program is done downloading",
                action="store_false",
                dest="wait",
            )

    async def run(self, args: argparse.Namespace):

        # 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)

            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.
            print(f"Searching for {args.name or 'any hub with Pybricks service'}...")
            device_or_address = await find_ble(args.name)
            hub = PybricksHubBLE(device_or_address)
        elif args.conntype == "usb":
            from usb.core import find as find_usb

            from pybricksdev.connections.pybricks import PybricksHubUSB
            from pybricksdev.usb import (
                EV3_USB_PID,
                LEGO_USB_VID,
                MINDSTORMS_INVENTOR_USB_PID,
                NXT_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 [
                            NXT_USB_PID,
                            EV3_USB_PID,
                            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()
        try:
            with _get_script_path(args.file) as script_path:
                await hub.run(script_path, args.wait)
        finally:
            await hub.disconnect()


class Download(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "download",
            help="upload a Pybricks program without running it",
        )
        parser.tool = self
        parser.add_argument(
            "conntype",
            metavar="<connection type>",
            help="connection type: %(choices)s",
            choices=["ble", "usb", "ssh"],
        )
        parser.add_argument(
            "file",
            metavar="<file>",
            help="path to a MicroPython script or `-` for stdin",
            type=argparse.FileType(),
        )
        parser.add_argument(
            "-n",
            "--name",
            metavar="<name>",
            required=False,
            help="hostname or IP address for SSH connection; "
            "Bluetooth device name or Bluetooth address for BLE connection; "
            "serial port name for USB connection",
        )

    async def run(self, args: argparse.Namespace):
        # 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)

            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.
            print(f"Searching for {args.name or 'any hub with Pybricks service'}...")
            device_or_address = await find_ble(args.name)
            hub = PybricksHubBLE(device_or_address)
        elif args.conntype == "usb":
            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 upload the script without running it
        await hub.connect()
        try:
            with _get_script_path(args.file) as script_path:
                await hub.download(script_path)
        finally:
            await hub.disconnect()


class Flash(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "flash", help="flash firmware on a LEGO Powered Up device"
        )
        parser.tool = self

        parser.add_argument(
            "firmware",
            metavar="<firmware-file>",
            type=argparse.FileType(mode="rb"),
            help="the firmware .zip file",
        ).completer = FilesCompleter(allowednames=(".zip",))

        parser.add_argument(
            "-n", "--name", metavar="<name>", type=str, help="a custom name for the hub"
        )

    def run(self, args: argparse.Namespace):
        from pybricksdev.cli.flash import flash_firmware

        return flash_firmware(args.firmware, args.name)


class DFUBackup(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser("backup", help="backup firmware using DFU")
        parser.tool = self
        parser.add_argument(
            "firmware",
            metavar="<firmware-file>",
            type=argparse.FileType(mode="wb"),
            help="the firmware .bin file",
        ).completer = FilesCompleter(allowednames=(".bin",))

    async def run(self, args: argparse.Namespace):
        from pybricksdev.dfu import backup_dfu

        backup_dfu(args.firmware)


class DFURestore(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "restore",
            help="restore firmware using DFU",
        )
        parser.tool = self
        parser.add_argument(
            "firmware",
            metavar="<firmware-file>",
            type=argparse.FileType(mode="rb"),
            help="the firmware .bin file",
        ).completer = FilesCompleter(allowednames=(".bin",))

    async def run(self, args: argparse.Namespace):
        from pybricksdev.dfu import restore_dfu

        restore_dfu(args.firmware)


class DFU(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        self.parser = subparsers.add_parser(
            "dfu",
            help="use DFU to backup or restore firmware",
        )
        self.parser.tool = self
        self.subparsers = self.parser.add_subparsers(
            metavar="<action>", dest="action", help="the action to perform"
        )

        for tool in DFUBackup(), DFURestore():
            tool.add_parser(self.subparsers)

    def run(self, args: argparse.Namespace):
        if args.action not in self.subparsers.choices:
            self.parser.error(
                f'Missing name of action: {"|".join(self.subparsers.choices.keys())}'
            )

        return self.subparsers.choices[args.action].tool.run(args)


class OADFlash(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "flash",
            help="update firmware on a LEGO Powered Up device using TI OAD",
        )
        parser.tool = self
        parser.add_argument(
            "firmware",
            metavar="<firmware-file>",
            type=argparse.FileType(mode="rb"),
            help="the firmware .oda file",
        ).completer = FilesCompleter(allowednames=(".oda",))

    async def run(self, args: argparse.Namespace):
        from pybricksdev.cli.oad import flash_oad_image

        await flash_oad_image(args.firmware)


class OADInfo(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "info",
            help="get information about firmware on a LEGO Powered Up device using TI OAD",
        )
        parser.tool = self

    async def run(self, args: argparse.Namespace):
        from pybricksdev.cli.oad import dump_oad_info

        await dump_oad_info()


class OAD(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        self.parser = subparsers.add_parser(
            "oad",
            help="update firmware on a LEGO Powered Up device using TI OAD",
        )
        self.parser.tool = self
        self.subparsers = self.parser.add_subparsers(
            metavar="<action>", dest="action", help="the action to perform"
        )

        for tool in OADFlash(), OADInfo():
            tool.add_parser(self.subparsers)

    def run(self, args: argparse.Namespace):
        if args.action not in self.subparsers.choices:
            self.parser.error(
                f'Missing name of action: {"|".join(self.subparsers.choices.keys())}'
            )

        return self.subparsers.choices[args.action].tool.run(args)


class LWP3Repl(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser(
            "repl",
            help="interactive REPL for sending and receiving LWP3 messages",
        )
        parser.tool = self

    def run(self, args: argparse.Namespace):
        from pybricksdev.cli.lwp3.repl import repl, setup_repl_logging

        setup_repl_logging()
        return repl()


class LWP3(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        self.parser = subparsers.add_parser(
            "lwp3", help="interact with devices using LWP3"
        )
        self.parser.tool = self
        self.subparsers = self.parser.add_subparsers(
            metavar="<lwp3-tool>", dest="lwp3_tool", help="the tool to run"
        )

        for tool in (LWP3Repl(),):
            tool.add_parser(self.subparsers)

    def run(self, args: argparse.Namespace):
        if args.lwp3_tool not in self.subparsers.choices:
            self.parser.error(
                f'Missing name of tool: {"|".join(self.subparsers.choices.keys())}'
            )

        return self.subparsers.choices[args.lwp3_tool].tool.run(args)


class Udev(Tool):
    def add_parser(self, subparsers: argparse._SubParsersAction):
        parser = subparsers.add_parser("udev", help="print udev rules to stdout")
        parser.tool = self

    async def run(self, args: argparse.Namespace):
        from importlib.resources import read_text

        from pybricksdev import resources

        print(read_text(resources, resources.UDEV_RULES))


def main():
    """Runs ``pybricksdev`` command line interface."""

    if sys.platform == "win32":
        # Hack around bad side-effects of pythoncom on Windows
        try:
            from bleak_winrt._winrt import MTA, init_apartment
        except ImportError:
            from winrt._winrt import MTA, init_apartment

        init_apartment(MTA)

    # Provide main description and help.
    parser = argparse.ArgumentParser(
        prog=PROG_NAME,
        description="Utilities for Pybricks developers.",
        epilog="Run `%(prog)s <tool> --help` for tool-specific arguments.",
    )

    parser.add_argument(
        "-v", "--version", action="version", version=f"{MODULE_NAME} v{MODULE_VERSION}"
    )
    parser.add_argument(
        "-d", "--debug", action="store_true", help="enable debug logging"
    )

    subparsers = parser.add_subparsers(
        metavar="<tool>",
        dest="tool",
        help="the tool to use",
    )

    for tool in Compile(), Run(), Download(), Flash(), DFU(), OAD(), LWP3(), Udev():
        tool.add_parser(subparsers)

    argcomplete.autocomplete(parser)
    args = parser.parse_args()

    logging.basicConfig(
        format="%(asctime)s: %(levelname)s: %(name)s: %(message)s",
        level=logging.DEBUG if args.debug else logging.WARNING,
    )

    if not args.tool:
        parser.error(f'Missing name of tool: {"|".join(subparsers.choices.keys())}')

    asyncio.run(subparsers.choices[args.tool].tool.run(args))

1	# SPDX-License-Identifier: MIT
2	# Copyright (c) 2019-2024 The Pybricks Authors
3
4	"""Command line wrapper around pybricksdev library."""
5
6	import argparse	1✔
7	import asyncio	1✔
8	import contextlib	1✔
9	import logging	1✔
10	import os	1✔
11	import socket	1✔
12	import sys	1✔
13	from abc import ABC, abstractmethod	1✔
14	from os import PathLike, path	1✔
15	from tempfile import NamedTemporaryFile	1✔
16	from typing import ContextManager, TextIO	1✔
17
18	import argcomplete	1✔
19	from argcomplete.completers import FilesCompleter	1✔
20
21	from pybricksdev import __name__ as MODULE_NAME	1✔
22	from pybricksdev import __version__ as MODULE_VERSION	1✔
23
24	PROG_NAME = (	1✔
25	f"{path.basename(sys.executable)} -m {MODULE_NAME}"
26	if sys.argv[0].endswith("__main__.py")
27	else path.basename(sys.argv[0])
28	)
29
30
31	class Tool(ABC):	1✔
32	"""Common base class for tool implementations."""
33
34	@abstractmethod	1✔
35	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
36	"""
37	Overriding methods must at least do the following::
38
39	parser = subparsers.add_parser('tool', ...)
40	parser.tool = self
41
42	Then additional arguments can be added using the ``parser`` object.
43	"""
44	pass	×
45
46	@abstractmethod	1✔
47	async def run(self, args: argparse.Namespace):	1✔
48	"""
49	Overriding methods should provide an implementation to run the tool.
50	"""
51	pass	×
52
53
54	def _get_script_path(file: TextIO) -> ContextManager[PathLike]:	1✔
55	"""
56	Gets the path to a script on the file system.
57
58	If the file is ``sys.stdin``, the contents are copied to a temporary file
59	and the path to the temporary file is returned. Otherwise, the file is closed
60	and the path is returned.
61
62	The context manager will delete the temporary file, if applicable.
63	"""
64	if file is sys.stdin:	1✔
65	# Have to close the temp file so that mpy-cross can read it, so we
66	# create our own context manager to delete the file when we are done
67	# using it.
68
69	@contextlib.contextmanager	×
70	def temp_context():	×
71	try:	×
72	with NamedTemporaryFile(suffix=".py", delete=False) as temp:	×
73	temp.write(file.buffer.read())	×
74
75	yield temp.name	×
76	finally:
77	try:	×
78	os.remove(temp.name)	×
79	except NameError:	×
80	# if NamedTemporaryFile() throws, temp is not defined
81	pass	×
82	except OSError:	×
83	# file was already deleted or other strangeness
84	pass	×
85
86	return temp_context()	×
87
88	file.close()	1✔
89	return contextlib.nullcontext(file.name)	1✔
90
91
92	class Compile(Tool):	1✔
93	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
94	parser = subparsers.add_parser(	×
95	"compile",
96	help="compile a Pybricks program without running it",
97	)
98	parser.add_argument(	×
99	"file",
100	metavar="<file>",
101	help="path to a MicroPython script or `-` for stdin",
102	type=argparse.FileType(),
103	)
104	parser.add_argument(	×
105	"--abi",
106	metavar="<n>",
107	help="the MPY ABI version, one of %(choices)s (default: %(default)s)",
108	default=6,
109	choices=[5, 6],
110	type=int,
111	)
112	parser.tool = self	×
113
114	async def run(self, args: argparse.Namespace):	1✔
115	from pybricksdev.compile import compile_multi_file, print_mpy	×
116
117	with _get_script_path(args.file) as script_path:	×
118	mpy = await compile_multi_file(script_path, args.abi)	×
119	print_mpy(mpy)	×
120
121
122	class Run(Tool):	1✔
123	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
124	parser = subparsers.add_parser(	×
125	"run",
126	help="run a Pybricks program",
127	)
128	parser.tool = self	×
129	parser.add_argument(	×
130	"conntype",
131	metavar="<connection type>",
132	help="connection type: %(choices)s",
133	choices=["ble", "usb", "ssh"],
134	)
135	parser.add_argument(	×
136	"file",
137	metavar="<file>",
138	help="path to a MicroPython script or `-` for stdin",
139	type=argparse.FileType(),
140	)
141	parser.add_argument(	×
142	"-n",
143	"--name",
144	metavar="<name>",
145	required=False,
146	help="hostname or IP address for SSH connection; "
147	"Bluetooth device name or Bluetooth address for BLE connection; "
148	"serial port name for USB connection",
149	)
150
151	if hasattr(argparse, "BooleanOptionalAction"):	×
152	# BooleanOptionalAction requires Python 3.9
153	parser.add_argument(	×
154	"--wait",
155	help="wait for the program to complete before disconnecting",
156	action=argparse.BooleanOptionalAction,
157	default=True,
158	)
159	else:
160	parser.add_argument(	×
161	"--wait",
162	help="wait for the program to complete before disconnecting (default)",
163	action="store_true",
164	default=True,
165	)
166	parser.add_argument(	×
167	"--no-wait",
168	help="disconnect as soon as program is done downloading",
169	action="store_false",
170	dest="wait",
171	)
172
173	async def run(self, args: argparse.Namespace):	1✔
174
175	# Pick the right connection
176	if args.conntype == "ssh":	×
177	from pybricksdev.connections.ev3dev import EV3Connection	×
178
179	# So it's an ev3dev
180	if args.name is None:	×
181	print("--name is required for SSH connections", file=sys.stderr)	×
182	exit(1)	×
183
184	device_or_address = socket.gethostbyname(args.name)	×
185	hub = EV3Connection(device_or_address)	×
186	elif args.conntype == "ble":	×
187	from pybricksdev.ble import find_device as find_ble	×
188	from pybricksdev.connections.pybricks import PybricksHubBLE	×
189
190	# It is a Pybricks Hub with BLE. Device name or address is given.
191	print(f"Searching for {args.name or 'any hub with Pybricks service'}...")	×
192	device_or_address = await find_ble(args.name)	×
193	hub = PybricksHubBLE(device_or_address)	×
194	elif args.conntype == "usb":	×
195	from usb.core import find as find_usb	×
196
197	from pybricksdev.connections.pybricks import PybricksHubUSB	×
198	from pybricksdev.usb import (	×
199	EV3_USB_PID,
200	LEGO_USB_VID,
201	MINDSTORMS_INVENTOR_USB_PID,
202	NXT_USB_PID,
203	SPIKE_ESSENTIAL_USB_PID,
204	SPIKE_PRIME_USB_PID,
205	)
206
207	def is_pybricks_usb(dev):	×
208	return (	×
209	(dev.idVendor == LEGO_USB_VID)
210	and (
211	dev.idProduct
212	in [
213	NXT_USB_PID,
214	EV3_USB_PID,
215	SPIKE_PRIME_USB_PID,
216	SPIKE_ESSENTIAL_USB_PID,
217	MINDSTORMS_INVENTOR_USB_PID,
218	]
219	)
220	and dev.product.endswith("Pybricks")
221	)
222
223	device_or_address = find_usb(custom_match=is_pybricks_usb)	×
224
225	if device_or_address is not None:	×
226	hub = PybricksHubUSB(device_or_address)	×
227	else:
228	from pybricksdev.connections.lego import REPLHub	×
229
230	hub = REPLHub()	×
231	else:
232	raise ValueError(f"Unknown connection type: {args.conntype}")	×
233
234	# Connect to the address and run the script
235	await hub.connect()	×
236	try:	×
237	with _get_script_path(args.file) as script_path:	×
238	await hub.run(script_path, args.wait)	×
239	finally:
240	await hub.disconnect()	×
241
242
243	class Download(Tool):	1✔
244	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
245	parser = subparsers.add_parser(	1✔
246	"download",
247	help="upload a Pybricks program without running it",
248	)
249	parser.tool = self	1✔
250	parser.add_argument(	1✔
251	"conntype",
252	metavar="<connection type>",
253	help="connection type: %(choices)s",
254	choices=["ble", "usb", "ssh"],
255	)
256	parser.add_argument(	1✔
257	"file",
258	metavar="<file>",
259	help="path to a MicroPython script or `-` for stdin",
260	type=argparse.FileType(),
261	)
262	parser.add_argument(	1✔
263	"-n",
264	"--name",
265	metavar="<name>",
266	required=False,
267	help="hostname or IP address for SSH connection; "
268	"Bluetooth device name or Bluetooth address for BLE connection; "
269	"serial port name for USB connection",
270	)
271
272	async def run(self, args: argparse.Namespace):	1✔
273	# Pick the right connection
274	if args.conntype == "ssh":	1✔
275	from pybricksdev.connections.ev3dev import EV3Connection	1✔
276
277	# So it's an ev3dev
278	if args.name is None:	1✔
279	print("--name is required for SSH connections", file=sys.stderr)	1✔
280	exit(1)	1✔
281
282	device_or_address = socket.gethostbyname(args.name)	1✔
283	hub = EV3Connection(device_or_address)	1✔
284	elif args.conntype == "ble":	1✔
285	from pybricksdev.ble import find_device as find_ble	1✔
286	from pybricksdev.connections.pybricks import PybricksHubBLE	1✔
287
288	# It is a Pybricks Hub with BLE. Device name or address is given.
289	print(f"Searching for {args.name or 'any hub with Pybricks service'}...")	1✔
290	device_or_address = await find_ble(args.name)	1✔
291	hub = PybricksHubBLE(device_or_address)	1✔
292	elif args.conntype == "usb":	1✔
293	from usb.core import find as find_usb	1✔
294
295	from pybricksdev.connections.pybricks import PybricksHubUSB	1✔
296	from pybricksdev.usb import (	1✔
297	LEGO_USB_VID,
298	MINDSTORMS_INVENTOR_USB_PID,
299	SPIKE_ESSENTIAL_USB_PID,
300	SPIKE_PRIME_USB_PID,
301	)
302
303	def is_pybricks_usb(dev):	1✔
304	return (	×
305	(dev.idVendor == LEGO_USB_VID)
306	and (
307	dev.idProduct
308	in [
309	SPIKE_PRIME_USB_PID,
310	SPIKE_ESSENTIAL_USB_PID,
311	MINDSTORMS_INVENTOR_USB_PID,
312	]
313	)
314	and dev.product.endswith("Pybricks")
315	)
316
317	device_or_address = find_usb(custom_match=is_pybricks_usb)	1✔
318
319	if device_or_address is not None:	1✔
320	hub = PybricksHubUSB(device_or_address)	1✔
321	else:
322	from pybricksdev.connections.lego import REPLHub	×
323
324	hub = REPLHub()	×
325	else:
326	raise ValueError(f"Unknown connection type: {args.conntype}")	×
327
328	# Connect to the address and upload the script without running it
329	await hub.connect()	1✔
330	try:	1✔
331	with _get_script_path(args.file) as script_path:	1✔
332	await hub.download(script_path)	1✔
333	finally:
334	await hub.disconnect()	1✔
335
336
337	class Flash(Tool):	1✔
338	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
339	parser = subparsers.add_parser(	×
340	"flash", help="flash firmware on a LEGO Powered Up device"
341	)
342	parser.tool = self	×
343
344	parser.add_argument(	×
345	"firmware",
346	metavar="<firmware-file>",
347	type=argparse.FileType(mode="rb"),
348	help="the firmware .zip file",
349	).completer = FilesCompleter(allowednames=(".zip",))
350
351	parser.add_argument(	×
352	"-n", "--name", metavar="<name>", type=str, help="a custom name for the hub"
353	)
354
355	def run(self, args: argparse.Namespace):	1✔
356	from pybricksdev.cli.flash import flash_firmware	×
357
358	return flash_firmware(args.firmware, args.name)	×
359
360
361	class DFUBackup(Tool):	1✔
362	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
363	parser = subparsers.add_parser("backup", help="backup firmware using DFU")	×
364	parser.tool = self	×
365	parser.add_argument(	×
366	"firmware",
367	metavar="<firmware-file>",
368	type=argparse.FileType(mode="wb"),
369	help="the firmware .bin file",
370	).completer = FilesCompleter(allowednames=(".bin",))
371
372	async def run(self, args: argparse.Namespace):	1✔
373	from pybricksdev.dfu import backup_dfu	×
374
375	backup_dfu(args.firmware)	×
376
377
378	class DFURestore(Tool):	1✔
379	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
380	parser = subparsers.add_parser(	×
381	"restore",
382	help="restore firmware using DFU",
383	)
384	parser.tool = self	×
385	parser.add_argument(	×
386	"firmware",
387	metavar="<firmware-file>",
388	type=argparse.FileType(mode="rb"),
389	help="the firmware .bin file",
390	).completer = FilesCompleter(allowednames=(".bin",))
391
392	async def run(self, args: argparse.Namespace):	1✔
393	from pybricksdev.dfu import restore_dfu	×
394
395	restore_dfu(args.firmware)	×
396
397
398	class DFU(Tool):	1✔
399	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
400	self.parser = subparsers.add_parser(	×
401	"dfu",
402	help="use DFU to backup or restore firmware",
403	)
404	self.parser.tool = self	×
405	self.subparsers = self.parser.add_subparsers(	×
406	metavar="<action>", dest="action", help="the action to perform"
407	)
408
409	for tool in DFUBackup(), DFURestore():	×
410	tool.add_parser(self.subparsers)	×
411
412	def run(self, args: argparse.Namespace):	1✔
413	if args.action not in self.subparsers.choices:	×
414	self.parser.error(	×
415	f'Missing name of action: {"\|".join(self.subparsers.choices.keys())}'
416	)
417
418	return self.subparsers.choices[args.action].tool.run(args)	×
419
420
421	class OADFlash(Tool):	1✔
422	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
423	parser = subparsers.add_parser(	×
424	"flash",
425	help="update firmware on a LEGO Powered Up device using TI OAD",
426	)
427	parser.tool = self	×
428	parser.add_argument(	×
429	"firmware",
430	metavar="<firmware-file>",
431	type=argparse.FileType(mode="rb"),
432	help="the firmware .oda file",
433	).completer = FilesCompleter(allowednames=(".oda",))
434
435	async def run(self, args: argparse.Namespace):	1✔
436	from pybricksdev.cli.oad import flash_oad_image	×
437
438	await flash_oad_image(args.firmware)	×
439
440
441	class OADInfo(Tool):	1✔
442	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
443	parser = subparsers.add_parser(	×
444	"info",
445	help="get information about firmware on a LEGO Powered Up device using TI OAD",
446	)
447	parser.tool = self	×
448
449	async def run(self, args: argparse.Namespace):	1✔
450	from pybricksdev.cli.oad import dump_oad_info	×
451
452	await dump_oad_info()	×
453
454
455	class OAD(Tool):	1✔
456	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
457	self.parser = subparsers.add_parser(	×
458	"oad",
459	help="update firmware on a LEGO Powered Up device using TI OAD",
460	)
461	self.parser.tool = self	×
462	self.subparsers = self.parser.add_subparsers(	×
463	metavar="<action>", dest="action", help="the action to perform"
464	)
465
466	for tool in OADFlash(), OADInfo():	×
467	tool.add_parser(self.subparsers)	×
468
469	def run(self, args: argparse.Namespace):	1✔
470	if args.action not in self.subparsers.choices:	×
471	self.parser.error(	×
472	f'Missing name of action: {"\|".join(self.subparsers.choices.keys())}'
473	)
474
475	return self.subparsers.choices[args.action].tool.run(args)	×
476
477
478	class LWP3Repl(Tool):	1✔
479	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
480	parser = subparsers.add_parser(	×
481	"repl",
482	help="interactive REPL for sending and receiving LWP3 messages",
483	)
484	parser.tool = self	×
485
486	def run(self, args: argparse.Namespace):	1✔
487	from pybricksdev.cli.lwp3.repl import repl, setup_repl_logging	×
488
489	setup_repl_logging()	×
490	return repl()	×
491
492
493	class LWP3(Tool):	1✔
494	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
495	self.parser = subparsers.add_parser(	×
496	"lwp3", help="interact with devices using LWP3"
497	)
498	self.parser.tool = self	×
499	self.subparsers = self.parser.add_subparsers(	×
500	metavar="<lwp3-tool>", dest="lwp3_tool", help="the tool to run"
501	)
502
503	for tool in (LWP3Repl(),):	×
504	tool.add_parser(self.subparsers)	×
505
506	def run(self, args: argparse.Namespace):	1✔
507	if args.lwp3_tool not in self.subparsers.choices:	×
508	self.parser.error(	×
509	f'Missing name of tool: {"\|".join(self.subparsers.choices.keys())}'
510	)
511
512	return self.subparsers.choices[args.lwp3_tool].tool.run(args)	×
513
514
515	class Udev(Tool):	1✔
516	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
517	parser = subparsers.add_parser("udev", help="print udev rules to stdout")	×
518	parser.tool = self	×
519
520	async def run(self, args: argparse.Namespace):	1✔
521	from importlib.resources import read_text	×
522
523	from pybricksdev import resources	×
524
525	print(read_text(resources, resources.UDEV_RULES))	×
526
527
528	def main():	1✔
529	"""Runs ``pybricksdev`` command line interface."""
530
531	if sys.platform == "win32":	×
532	# Hack around bad side-effects of pythoncom on Windows
533	try:	×
534	from bleak_winrt._winrt import MTA, init_apartment	×
535	except ImportError:	×
536	from winrt._winrt import MTA, init_apartment	×
537
538	init_apartment(MTA)	×
539
540	# Provide main description and help.
541	parser = argparse.ArgumentParser(	×
542	prog=PROG_NAME,
543	description="Utilities for Pybricks developers.",
544	epilog="Run `%(prog)s <tool> --help` for tool-specific arguments.",
545	)
546
547	parser.add_argument(	×
548	"-v", "--version", action="version", version=f"{MODULE_NAME} v{MODULE_VERSION}"
549	)
550	parser.add_argument(	×
551	"-d", "--debug", action="store_true", help="enable debug logging"
552	)
553
554	subparsers = parser.add_subparsers(	×
555	metavar="<tool>",
556	dest="tool",
557	help="the tool to use",
558	)
559
560	for tool in Compile(), Run(), Download(), Flash(), DFU(), OAD(), LWP3(), Udev():	×
561	tool.add_parser(subparsers)	×
562
563	argcomplete.autocomplete(parser)	×
564	args = parser.parse_args()	×
565
566	logging.basicConfig(	×
567	format="%(asctime)s: %(levelname)s: %(name)s: %(message)s",
568	level=logging.DEBUG if args.debug else logging.WARNING,
569	)
570
571	if not args.tool:	×
572	parser.error(f'Missing name of tool: {"\|".join(subparsers.choices.keys())}')	×
573
574	asyncio.run(subparsers.choices[args.tool].tool.run(args))	×

pybricks / pybricksdev / 16406482466

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous