15313256588

Committed 29 May 2025 12:15AM UTC coverage: 50.59% (+7.8%) from 42.821%

Build # 15313256588

Build Type

push

github

Committed by

web-flow

Commit Message

cli: Add a new command for downloading program without running it.

Sometimes, it's useful to upload the program to hubs without running it, especially when hubs are in complex environments and require manual start for safety and convenience purposes.

Run Details

67 of 275 branches covered (24.36%)

Branch coverage included in aggregate %.

1948 of 3708 relevant lines covered (52.54%)

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 (
                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()
        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	LEGO_USB_VID,
200	MINDSTORMS_INVENTOR_USB_PID,
201	SPIKE_ESSENTIAL_USB_PID,
202	SPIKE_PRIME_USB_PID,
203	)
204
205	def is_pybricks_usb(dev):	×
206	return (	×
207	(dev.idVendor == LEGO_USB_VID)
208	and (
209	dev.idProduct
210	in [
211	SPIKE_PRIME_USB_PID,
212	SPIKE_ESSENTIAL_USB_PID,
213	MINDSTORMS_INVENTOR_USB_PID,
214	]
215	)
216	and dev.product.endswith("Pybricks")
217	)
218
219	device_or_address = find_usb(custom_match=is_pybricks_usb)	×
220
221	if device_or_address is not None:	×
222	hub = PybricksHubUSB(device_or_address)	×
223	else:
224	from pybricksdev.connections.lego import REPLHub	×
225
226	hub = REPLHub()	×
227	else:
228	raise ValueError(f"Unknown connection type: {args.conntype}")	×
229
230	# Connect to the address and run the script
231	await hub.connect()	×
232	try:	×
233	with _get_script_path(args.file) as script_path:	×
234	await hub.run(script_path, args.wait)	×
235	finally:
236	await hub.disconnect()	×
237
238
239	class Download(Tool):	1✔
240	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
241	parser = subparsers.add_parser(	1✔
242	"download",
243	help="upload a Pybricks program without running it",
244	)
245	parser.tool = self	1✔
246	parser.add_argument(	1✔
247	"conntype",
248	metavar="<connection type>",
249	help="connection type: %(choices)s",
250	choices=["ble", "usb", "ssh"],
251	)
252	parser.add_argument(	1✔
253	"file",
254	metavar="<file>",
255	help="path to a MicroPython script or `-` for stdin",
256	type=argparse.FileType(),
257	)
258	parser.add_argument(	1✔
259	"-n",
260	"--name",
261	metavar="<name>",
262	required=False,
263	help="hostname or IP address for SSH connection; "
264	"Bluetooth device name or Bluetooth address for BLE connection; "
265	"serial port name for USB connection",
266	)
267
268	async def run(self, args: argparse.Namespace):	1✔
269	# Pick the right connection
270	if args.conntype == "ssh":	1✔
271	from pybricksdev.connections.ev3dev import EV3Connection	1✔
272
273	# So it's an ev3dev
274	if args.name is None:	1✔
275	print("--name is required for SSH connections", file=sys.stderr)	1✔
276	exit(1)	1✔
277
278	device_or_address = socket.gethostbyname(args.name)	1✔
279	hub = EV3Connection(device_or_address)	1✔
280	elif args.conntype == "ble":	1✔
281	from pybricksdev.ble import find_device as find_ble	1✔
282	from pybricksdev.connections.pybricks import PybricksHubBLE	1✔
283
284	# It is a Pybricks Hub with BLE. Device name or address is given.
285	print(f"Searching for {args.name or 'any hub with Pybricks service'}...")	1✔
286	device_or_address = await find_ble(args.name)	1✔
287	hub = PybricksHubBLE(device_or_address)	1✔
288	elif args.conntype == "usb":	1✔
289	from usb.core import find as find_usb	1✔
290
291	from pybricksdev.connections.pybricks import PybricksHubUSB	1✔
292	from pybricksdev.usb import (	1✔
293	LEGO_USB_VID,
294	MINDSTORMS_INVENTOR_USB_PID,
295	SPIKE_ESSENTIAL_USB_PID,
296	SPIKE_PRIME_USB_PID,
297	)
298
299	def is_pybricks_usb(dev):	1✔
300	return (	×
301	(dev.idVendor == LEGO_USB_VID)
302	and (
303	dev.idProduct
304	in [
305	SPIKE_PRIME_USB_PID,
306	SPIKE_ESSENTIAL_USB_PID,
307	MINDSTORMS_INVENTOR_USB_PID,
308	]
309	)
310	and dev.product.endswith("Pybricks")
311	)
312
313	device_or_address = find_usb(custom_match=is_pybricks_usb)	1✔
314
315	if device_or_address is not None:	1✔
316	hub = PybricksHubUSB(device_or_address)	1✔
317	else:
318	from pybricksdev.connections.lego import REPLHub	×
319
320	hub = REPLHub()	×
321	else:
322	raise ValueError(f"Unknown connection type: {args.conntype}")	×
323
324	# Connect to the address and upload the script without running it
325	await hub.connect()	1✔
326	try:	1✔
327	with _get_script_path(args.file) as script_path:	1✔
328	await hub.download(script_path)	1✔
329	finally:
330	await hub.disconnect()	1✔
331
332
333	class Flash(Tool):	1✔
334	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
335	parser = subparsers.add_parser(	×
336	"flash", help="flash firmware on a LEGO Powered Up device"
337	)
338	parser.tool = self	×
339
340	parser.add_argument(	×
341	"firmware",
342	metavar="<firmware-file>",
343	type=argparse.FileType(mode="rb"),
344	help="the firmware .zip file",
345	).completer = FilesCompleter(allowednames=(".zip",))
346
347	parser.add_argument(	×
348	"-n", "--name", metavar="<name>", type=str, help="a custom name for the hub"
349	)
350
351	def run(self, args: argparse.Namespace):	1✔
352	from pybricksdev.cli.flash import flash_firmware	×
353
354	return flash_firmware(args.firmware, args.name)	×
355
356
357	class DFUBackup(Tool):	1✔
358	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
359	parser = subparsers.add_parser("backup", help="backup firmware using DFU")	×
360	parser.tool = self	×
361	parser.add_argument(	×
362	"firmware",
363	metavar="<firmware-file>",
364	type=argparse.FileType(mode="wb"),
365	help="the firmware .bin file",
366	).completer = FilesCompleter(allowednames=(".bin",))
367
368	async def run(self, args: argparse.Namespace):	1✔
369	from pybricksdev.dfu import backup_dfu	×
370
371	backup_dfu(args.firmware)	×
372
373
374	class DFURestore(Tool):	1✔
375	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
376	parser = subparsers.add_parser(	×
377	"restore",
378	help="restore firmware using DFU",
379	)
380	parser.tool = self	×
381	parser.add_argument(	×
382	"firmware",
383	metavar="<firmware-file>",
384	type=argparse.FileType(mode="rb"),
385	help="the firmware .bin file",
386	).completer = FilesCompleter(allowednames=(".bin",))
387
388	async def run(self, args: argparse.Namespace):	1✔
389	from pybricksdev.dfu import restore_dfu	×
390
391	restore_dfu(args.firmware)	×
392
393
394	class DFU(Tool):	1✔
395	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
396	self.parser = subparsers.add_parser(	×
397	"dfu",
398	help="use DFU to backup or restore firmware",
399	)
400	self.parser.tool = self	×
401	self.subparsers = self.parser.add_subparsers(	×
402	metavar="<action>", dest="action", help="the action to perform"
403	)
404
405	for tool in DFUBackup(), DFURestore():	×
406	tool.add_parser(self.subparsers)	×
407
408	def run(self, args: argparse.Namespace):	1✔
409	if args.action not in self.subparsers.choices:	×
410	self.parser.error(	×
411	f'Missing name of action: {"\|".join(self.subparsers.choices.keys())}'
412	)
413
414	return self.subparsers.choices[args.action].tool.run(args)	×
415
416
417	class OADFlash(Tool):	1✔
418	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
419	parser = subparsers.add_parser(	×
420	"flash",
421	help="update firmware on a LEGO Powered Up device using TI OAD",
422	)
423	parser.tool = self	×
424	parser.add_argument(	×
425	"firmware",
426	metavar="<firmware-file>",
427	type=argparse.FileType(mode="rb"),
428	help="the firmware .oda file",
429	).completer = FilesCompleter(allowednames=(".oda",))
430
431	async def run(self, args: argparse.Namespace):	1✔
432	from pybricksdev.cli.oad import flash_oad_image	×
433
434	await flash_oad_image(args.firmware)	×
435
436
437	class OADInfo(Tool):	1✔
438	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
439	parser = subparsers.add_parser(	×
440	"info",
441	help="get information about firmware on a LEGO Powered Up device using TI OAD",
442	)
443	parser.tool = self	×
444
445	async def run(self, args: argparse.Namespace):	1✔
446	from pybricksdev.cli.oad import dump_oad_info	×
447
448	await dump_oad_info()	×
449
450
451	class OAD(Tool):	1✔
452	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
453	self.parser = subparsers.add_parser(	×
454	"oad",
455	help="update firmware on a LEGO Powered Up device using TI OAD",
456	)
457	self.parser.tool = self	×
458	self.subparsers = self.parser.add_subparsers(	×
459	metavar="<action>", dest="action", help="the action to perform"
460	)
461
462	for tool in OADFlash(), OADInfo():	×
463	tool.add_parser(self.subparsers)	×
464
465	def run(self, args: argparse.Namespace):	1✔
466	if args.action not in self.subparsers.choices:	×
467	self.parser.error(	×
468	f'Missing name of action: {"\|".join(self.subparsers.choices.keys())}'
469	)
470
471	return self.subparsers.choices[args.action].tool.run(args)	×
472
473
474	class LWP3Repl(Tool):	1✔
475	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
476	parser = subparsers.add_parser(	×
477	"repl",
478	help="interactive REPL for sending and receiving LWP3 messages",
479	)
480	parser.tool = self	×
481
482	def run(self, args: argparse.Namespace):	1✔
483	from pybricksdev.cli.lwp3.repl import repl, setup_repl_logging	×
484
485	setup_repl_logging()	×
486	return repl()	×
487
488
489	class LWP3(Tool):	1✔
490	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
491	self.parser = subparsers.add_parser(	×
492	"lwp3", help="interact with devices using LWP3"
493	)
494	self.parser.tool = self	×
495	self.subparsers = self.parser.add_subparsers(	×
496	metavar="<lwp3-tool>", dest="lwp3_tool", help="the tool to run"
497	)
498
499	for tool in (LWP3Repl(),):	×
500	tool.add_parser(self.subparsers)	×
501
502	def run(self, args: argparse.Namespace):	1✔
503	if args.lwp3_tool not in self.subparsers.choices:	×
504	self.parser.error(	×
505	f'Missing name of tool: {"\|".join(self.subparsers.choices.keys())}'
506	)
507
508	return self.subparsers.choices[args.lwp3_tool].tool.run(args)	×
509
510
511	class Udev(Tool):	1✔
512	def add_parser(self, subparsers: argparse._SubParsersAction):	1✔
513	parser = subparsers.add_parser("udev", help="print udev rules to stdout")	×
514	parser.tool = self	×
515
516	async def run(self, args: argparse.Namespace):	1✔
517	from importlib.resources import read_text	×
518
519	from pybricksdev import resources	×
520
521	print(read_text(resources, resources.UDEV_RULES))	×
522
523
524	def main():	1✔
525	"""Runs ``pybricksdev`` command line interface."""
526
527	if sys.platform == "win32":	×
528	# Hack around bad side-effects of pythoncom on Windows
529	try:	×
530	from bleak_winrt._winrt import MTA, init_apartment	×
531	except ImportError:	×
532	from winrt._winrt import MTA, init_apartment	×
533
534	init_apartment(MTA)	×
535
536	# Provide main description and help.
537	parser = argparse.ArgumentParser(	×
538	prog=PROG_NAME,
539	description="Utilities for Pybricks developers.",
540	epilog="Run `%(prog)s <tool> --help` for tool-specific arguments.",
541	)
542
543	parser.add_argument(	×
544	"-v", "--version", action="version", version=f"{MODULE_NAME} v{MODULE_VERSION}"
545	)
546	parser.add_argument(	×
547	"-d", "--debug", action="store_true", help="enable debug logging"
548	)
549
550	subparsers = parser.add_subparsers(	×
551	metavar="<tool>",
552	dest="tool",
553	help="the tool to use",
554	)
555
556	for tool in Compile(), Run(), Download(), Flash(), DFU(), OAD(), LWP3(), Udev():	×
557	tool.add_parser(subparsers)	×
558
559	argcomplete.autocomplete(parser)	×
560	args = parser.parse_args()	×
561
562	logging.basicConfig(	×
563	format="%(asctime)s: %(levelname)s: %(name)s: %(message)s",
564	level=logging.DEBUG if args.debug else logging.WARNING,
565	)
566
567	if not args.tool:	×
568	parser.error(f'Missing name of tool: {"\|".join(subparsers.choices.keys())}')	×
569
570	asyncio.run(subparsers.choices[args.tool].tool.run(args))	×

pybricks / pybricksdev / 15313256588

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