16817720475

Committed 07 Aug 2025 11:01PM UTC coverage: 53.891%. Remained the same

Build # 16817720475

Build Type

push

github

Committed by

dlech

Commit Message

LICENSE: bump copyright year to 2025

Run Details

63 of 241 branches covered (26.14%)

Branch coverage included in aggregate %.

1959 of 3511 relevant lines covered (55.8%)

0.56 hits per line

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

44.93

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 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, encoding="utf-8"
                ) 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(encoding="utf-8"),
        )
        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"],
        )
        parser.add_argument(
            "file",
            metavar="<file>",
            help="path to a MicroPython script or `-` for stdin",
            type=argparse.FileType(encoding="utf-8"),
        )
        parser.add_argument(
            "-n",
            "--name",
            metavar="<name>",
            required=False,
            help="Bluetooth device name or Bluetooth address for BLE connection; "
            "serial port name for USB connection",
        )

        parser.add_argument(
            "--start",
            help="Start the program immediately after downloading it.",
            action=argparse.BooleanOptionalAction,
            default=True,
        )

        parser.add_argument(
            "--wait",
            help="Wait for the program to complete before disconnecting. Only applies when starting program right away.",
            action=argparse.BooleanOptionalAction,
            default=True,
        )

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

        # Pick the right connection
        if 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 None:
                print("Pybricks Hub not found.", file=sys.stderr)
                exit(1)

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

pybricks / pybricksdev / 16817720475

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