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

pybricks / pybricksdev / 16748407221

05 Aug 2025 11:11AM UTC coverage: 53.613% (+0.01%) from 53.601%
16748407221

push

github

laurensvalk
cli: Enable starting REPL.

64 of 243 branches covered (26.34%)

Branch coverage included in aggregate %.

1954 of 3521 relevant lines covered (55.5%)

0.55 hits per line

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

44.95
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
from pybricksdev.ble.pybricks import UserProgramId
1✔
20

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

27

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

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

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

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

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

50

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

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

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

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

84

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

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

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

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

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

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

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

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

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

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

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

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

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

198

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

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

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

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

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

222

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

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

237
        backup_dfu(args.firmware)
×
238

239

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

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

257
        restore_dfu(args.firmware)
×
258

259

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

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

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

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

282

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

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

300
        await flash_oad_image(args.firmware)
×
301

302

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

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

314
        await dump_oad_info()
×
315

316

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

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

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

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

339

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

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

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

354

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

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

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

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

376

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

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

385
        from pybricksdev import resources
×
386

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

389

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

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

400
        init_apartment(MTA)
×
401

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

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

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

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

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

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

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

436
    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