16748212402

Committed 05 Aug 2025 11:02AM UTC coverage: 53.606% (-0.01%) from 53.618%

Build # 16748212402

Build Type

push

github

Committed by

laurensvalk

Commit Message

setup: Don't flake8 pystone.

This got included since a poetry update but it should not be. Also delete unused Jupyter notebook.

Run Details

63 of 241 branches covered (26.14%)

Branch coverage included in aggregate %.

1959 of 3531 relevant lines covered (55.48%)

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

pybricks / pybricksdev / 16748212402

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