• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pybricks / pybricksdev / 16748339611

05 Aug 2025 11:08AM UTC coverage: 53.601% (-0.05%) from 53.65%
16748339611

push

github

laurensvalk
cli: Enable starting REPL.

64 of 243 branches covered (26.34%)

Branch coverage included in aggregate %.

1953 of 3520 relevant lines covered (55.48%)

0.55 hits per line

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

44.67
pybricksdev/cli/__init__.py
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 logging
1✔
9
import sys
1✔
10
from abc import ABC, abstractmethod
1✔
11
from os import path
1✔
12
from tempfile import NamedTemporaryFile
1✔
13

14
import argcomplete
1✔
15
from argcomplete.completers import FilesCompleter
1✔
16

17
from pybricksdev import __name__ as MODULE_NAME
1✔
18
from pybricksdev import __version__ as MODULE_VERSION
1✔
19

20
PROG_NAME = (
1✔
21
    f"{path.basename(sys.executable)} -m {MODULE_NAME}"
22
    if sys.argv[0].endswith("__main__.py")
23
    else path.basename(sys.argv[0])
24
)
25

26

27
class Tool(ABC):
1✔
28
    """Common base class for tool implementations."""
29

30
    @abstractmethod
1✔
31
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
32
        """
33
        Overriding methods must at least do the following::
34

35
            parser = subparsers.add_parser('tool', ...)
36
            parser.tool = self
37

38
        Then additional arguments can be added using the ``parser`` object.
39
        """
40
        pass
×
41

42
    @abstractmethod
1✔
43
    async def run(self, args: argparse.Namespace):
1✔
44
        """
45
        Overriding methods should provide an implementation to run the tool.
46
        """
47
        pass
×
48

49

50
class Compile(Tool):
1✔
51
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
52
        parser = subparsers.add_parser(
×
53
            "compile",
54
            help="compile a Pybricks program without running it",
55
        )
56
        parser.add_argument(
×
57
            "file",
58
            metavar="<file>",
59
            help="path to a MicroPython script or `-` for stdin",
60
            type=str,
61
        )
62
        parser.add_argument(
×
63
            "--abi",
64
            metavar="<n>",
65
            help="the MPY ABI version, one of %(choices)s (default: %(default)s)",
66
            default=6,
67
            choices=[5, 6],
68
            type=int,
69
        )
70
        parser.tool = self
×
71

72
    async def run(self, args: argparse.Namespace):
1✔
73
        from pybricksdev.compile import compile_multi_file, print_mpy
×
74

75
        if args.file == "-":
×
76
            with NamedTemporaryFile(suffix=".py", delete=False) as temp:
×
77
                temp.write(sys.stdin.buffer.read())
×
78
            args.file = temp.name
×
79

80
        mpy = await compile_multi_file(args.file, args.abi)
×
81
        print_mpy(mpy)
×
82

83

84
class Run(Tool):
1✔
85
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
86
        parser = subparsers.add_parser(
1✔
87
            "run",
88
            help="run a Pybricks program",
89
        )
90
        parser.tool = self
1✔
91
        parser.add_argument(
1✔
92
            "conntype",
93
            metavar="<connection type>",
94
            help="connection type: %(choices)s",
95
            choices=["ble", "usb"],
96
        )
97
        parser.add_argument(
1✔
98
            "file",
99
            metavar="<file>",
100
            help="path to a MicroPython script, `-` for stdin, or `repl` for interactive prompt",
101
            type=str,
102
        )
103
        parser.add_argument(
1✔
104
            "-n",
105
            "--name",
106
            metavar="<name>",
107
            required=False,
108
            help="Bluetooth device name or Bluetooth address for BLE connection; "
109
            "serial port name for USB connection",
110
        )
111

112
        parser.add_argument(
1✔
113
            "--start",
114
            help="Start the program immediately after downloading it.",
115
            action=argparse.BooleanOptionalAction,
116
            default=True,
117
        )
118

119
        parser.add_argument(
1✔
120
            "--wait",
121
            help="Wait for the program to complete before disconnecting. Only applies when starting program right away.",
122
            action=argparse.BooleanOptionalAction,
123
            default=True,
124
        )
125

126
    async def run(self, args: argparse.Namespace):
1✔
127

128
        # Pick the right connection
129
        if args.conntype == "ble":
1✔
130
            from pybricksdev.ble import find_device as find_ble
1✔
131
            from pybricksdev.connections.pybricks import PybricksHubBLE
1✔
132

133
            # It is a Pybricks Hub with BLE. Device name or address is given.
134
            print(f"Searching for {args.name or 'any hub with Pybricks service'}...")
1✔
135
            device_or_address = await find_ble(args.name)
1✔
136
            hub = PybricksHubBLE(device_or_address)
1✔
137
        elif args.conntype == "usb":
1✔
138
            from usb.core import find as find_usb
1✔
139

140
            from pybricksdev.connections.pybricks import PybricksHubUSB
1✔
141
            from pybricksdev.usb import (
1✔
142
                EV3_USB_PID,
143
                LEGO_USB_VID,
144
                MINDSTORMS_INVENTOR_USB_PID,
145
                NXT_USB_PID,
146
                SPIKE_ESSENTIAL_USB_PID,
147
                SPIKE_PRIME_USB_PID,
148
            )
149

150
            def is_pybricks_usb(dev):
1✔
151
                return (
×
152
                    (dev.idVendor == LEGO_USB_VID)
153
                    and (
154
                        dev.idProduct
155
                        in [
156
                            NXT_USB_PID,
157
                            EV3_USB_PID,
158
                            SPIKE_PRIME_USB_PID,
159
                            SPIKE_ESSENTIAL_USB_PID,
160
                            MINDSTORMS_INVENTOR_USB_PID,
161
                        ]
162
                    )
163
                    and dev.product.endswith("Pybricks")
164
                )
165

166
            device_or_address = find_usb(custom_match=is_pybricks_usb)
1✔
167

168
            if device_or_address is None:
1✔
169
                print("Pybricks Hub not found.", file=sys.stderr)
×
170
                exit(1)
×
171

172
            hub = PybricksHubUSB(device_or_address)
1✔
173
        else:
174
            raise ValueError(f"Unknown connection type: {args.conntype}")
×
175

176
        # Connect to the address and run the script
177
        await hub.connect()
1✔
178
        try:
1✔
179
            # Handle builtin programs.
180
            if args.file == "repl":
1✔
181
                await hub.run(128)
×
182
            else:
183
                # If using stdin, save to temporary file first.
184
                if args.file == "-":
1✔
185
                    with NamedTemporaryFile(suffix=".py", delete=False) as temp:
×
186
                        temp.write(sys.stdin.buffer.read())
×
187
                    args.file = temp.name
×
188

189
                # Download program and optionally start it.
190
                if args.start:
1✔
191
                    await hub.run(args.file, args.wait)
×
192
                else:
193
                    await hub.download(args.file)
1✔
194
        finally:
195
            await hub.disconnect()
1✔
196

197

198
class Flash(Tool):
1✔
199
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
200
        parser = subparsers.add_parser(
×
201
            "flash", help="flash firmware on a LEGO Powered Up device"
202
        )
203
        parser.tool = self
×
204

205
        parser.add_argument(
×
206
            "firmware",
207
            metavar="<firmware-file>",
208
            type=argparse.FileType(mode="rb"),
209
            help="the firmware .zip file",
210
        ).completer = FilesCompleter(allowednames=(".zip",))
211

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

216
    def run(self, args: argparse.Namespace):
1✔
217
        from pybricksdev.cli.flash import flash_firmware
×
218

219
        return flash_firmware(args.firmware, args.name)
×
220

221

222
class DFUBackup(Tool):
1✔
223
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
224
        parser = subparsers.add_parser("backup", help="backup firmware using DFU")
×
225
        parser.tool = self
×
226
        parser.add_argument(
×
227
            "firmware",
228
            metavar="<firmware-file>",
229
            type=argparse.FileType(mode="wb"),
230
            help="the firmware .bin file",
231
        ).completer = FilesCompleter(allowednames=(".bin",))
232

233
    async def run(self, args: argparse.Namespace):
1✔
234
        from pybricksdev.dfu import backup_dfu
×
235

236
        backup_dfu(args.firmware)
×
237

238

239
class DFURestore(Tool):
1✔
240
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
241
        parser = subparsers.add_parser(
×
242
            "restore",
243
            help="restore firmware using DFU",
244
        )
245
        parser.tool = self
×
246
        parser.add_argument(
×
247
            "firmware",
248
            metavar="<firmware-file>",
249
            type=argparse.FileType(mode="rb"),
250
            help="the firmware .bin file",
251
        ).completer = FilesCompleter(allowednames=(".bin",))
252

253
    async def run(self, args: argparse.Namespace):
1✔
254
        from pybricksdev.dfu import restore_dfu
×
255

256
        restore_dfu(args.firmware)
×
257

258

259
class DFU(Tool):
1✔
260
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
261
        self.parser = subparsers.add_parser(
×
262
            "dfu",
263
            help="use DFU to backup or restore firmware",
264
        )
265
        self.parser.tool = self
×
266
        self.subparsers = self.parser.add_subparsers(
×
267
            metavar="<action>", dest="action", help="the action to perform"
268
        )
269

270
        for tool in DFUBackup(), DFURestore():
×
271
            tool.add_parser(self.subparsers)
×
272

273
    def run(self, args: argparse.Namespace):
1✔
274
        if args.action not in self.subparsers.choices:
×
275
            self.parser.error(
×
276
                f'Missing name of action: {"|".join(self.subparsers.choices.keys())}'
277
            )
278

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

281

282
class OADFlash(Tool):
1✔
283
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
284
        parser = subparsers.add_parser(
×
285
            "flash",
286
            help="update firmware on a LEGO Powered Up device using TI OAD",
287
        )
288
        parser.tool = self
×
289
        parser.add_argument(
×
290
            "firmware",
291
            metavar="<firmware-file>",
292
            type=argparse.FileType(mode="rb"),
293
            help="the firmware .oda file",
294
        ).completer = FilesCompleter(allowednames=(".oda",))
295

296
    async def run(self, args: argparse.Namespace):
1✔
297
        from pybricksdev.cli.oad import flash_oad_image
×
298

299
        await flash_oad_image(args.firmware)
×
300

301

302
class OADInfo(Tool):
1✔
303
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
304
        parser = subparsers.add_parser(
×
305
            "info",
306
            help="get information about firmware on a LEGO Powered Up device using TI OAD",
307
        )
308
        parser.tool = self
×
309

310
    async def run(self, args: argparse.Namespace):
1✔
311
        from pybricksdev.cli.oad import dump_oad_info
×
312

313
        await dump_oad_info()
×
314

315

316
class OAD(Tool):
1✔
317
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
318
        self.parser = subparsers.add_parser(
×
319
            "oad",
320
            help="update firmware on a LEGO Powered Up device using TI OAD",
321
        )
322
        self.parser.tool = self
×
323
        self.subparsers = self.parser.add_subparsers(
×
324
            metavar="<action>", dest="action", help="the action to perform"
325
        )
326

327
        for tool in OADFlash(), OADInfo():
×
328
            tool.add_parser(self.subparsers)
×
329

330
    def run(self, args: argparse.Namespace):
1✔
331
        if args.action not in self.subparsers.choices:
×
332
            self.parser.error(
×
333
                f'Missing name of action: {"|".join(self.subparsers.choices.keys())}'
334
            )
335

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

338

339
class LWP3Repl(Tool):
1✔
340
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
341
        parser = subparsers.add_parser(
×
342
            "repl",
343
            help="interactive REPL for sending and receiving LWP3 messages",
344
        )
345
        parser.tool = self
×
346

347
    def run(self, args: argparse.Namespace):
1✔
348
        from pybricksdev.cli.lwp3.repl import repl, setup_repl_logging
×
349

350
        setup_repl_logging()
×
351
        return repl()
×
352

353

354
class LWP3(Tool):
1✔
355
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
356
        self.parser = subparsers.add_parser(
×
357
            "lwp3", help="interact with devices using LWP3"
358
        )
359
        self.parser.tool = self
×
360
        self.subparsers = self.parser.add_subparsers(
×
361
            metavar="<lwp3-tool>", dest="lwp3_tool", help="the tool to run"
362
        )
363

364
        for tool in (LWP3Repl(),):
×
365
            tool.add_parser(self.subparsers)
×
366

367
    def run(self, args: argparse.Namespace):
1✔
368
        if args.lwp3_tool not in self.subparsers.choices:
×
369
            self.parser.error(
×
370
                f'Missing name of tool: {"|".join(self.subparsers.choices.keys())}'
371
            )
372

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

375

376
class Udev(Tool):
1✔
377
    def add_parser(self, subparsers: argparse._SubParsersAction):
1✔
378
        parser = subparsers.add_parser("udev", help="print udev rules to stdout")
×
379
        parser.tool = self
×
380

381
    async def run(self, args: argparse.Namespace):
1✔
382
        from importlib.resources import read_text
×
383

384
        from pybricksdev import resources
×
385

386
        print(read_text(resources, resources.UDEV_RULES))
×
387

388

389
def main():
1✔
390
    """Runs ``pybricksdev`` command line interface."""
391

392
    if sys.platform == "win32":
×
393
        # Hack around bad side-effects of pythoncom on Windows
394
        try:
×
395
            from bleak_winrt._winrt import MTA, init_apartment
×
396
        except ImportError:
×
397
            from winrt._winrt import MTA, init_apartment
×
398

399
        init_apartment(MTA)
×
400

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

408
    parser.add_argument(
×
409
        "-v", "--version", action="version", version=f"{MODULE_NAME} v{MODULE_VERSION}"
410
    )
411
    parser.add_argument(
×
412
        "-d", "--debug", action="store_true", help="enable debug logging"
413
    )
414

415
    subparsers = parser.add_subparsers(
×
416
        metavar="<tool>",
417
        dest="tool",
418
        help="the tool to use",
419
    )
420

421
    for tool in Compile(), Run(), Flash(), DFU(), OAD(), LWP3(), Udev():
×
422
        tool.add_parser(subparsers)
×
423

424
    argcomplete.autocomplete(parser)
×
425
    args = parser.parse_args()
×
426

427
    logging.basicConfig(
×
428
        format="%(asctime)s: %(levelname)s: %(name)s: %(message)s",
429
        level=logging.DEBUG if args.debug else logging.WARNING,
430
    )
431

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

435
    asyncio.run(subparsers.choices[args.tool].tool.run(args))
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc