From a3d454e5afdab6ffb995e5ce9c939289d09787b5 Mon Sep 17 00:00:00 2001 From: Laurens Valk Date: Tue, 5 Aug 2025 11:58:37 +0200 Subject: [PATCH 1/3] cli: Simplify path handling. Instead of converting a path to an open file, getting the file name, closing it again, and then opening it again, just use the given name. Sometimes simplicity is better. And it will let us use run repl in a future commit. --- pybricksdev/cli/__init__.py | 69 ++++++++++--------------------------- tests/test_cli.py | 2 +- 2 files changed, 19 insertions(+), 52 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index b772e0e..0b5c815 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -5,14 +5,11 @@ import argparse import asyncio -import contextlib import logging -import os import sys from abc import ABC, abstractmethod -from os import PathLike, path +from os import path from tempfile import NamedTemporaryFile -from typing import ContextManager, TextIO import argcomplete from argcomplete.completers import FilesCompleter @@ -50,44 +47,6 @@ async def run(self, args: argparse.Namespace): 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( @@ -98,7 +57,7 @@ def add_parser(self, subparsers: argparse._SubParsersAction): "file", metavar="", help="path to a MicroPython script or `-` for stdin", - type=argparse.FileType(), + type=str, ) parser.add_argument( "--abi", @@ -113,8 +72,12 @@ def add_parser(self, subparsers: argparse._SubParsersAction): 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) + if args.file == "-": + with NamedTemporaryFile(suffix=".py", delete=False) as temp: + temp.write(sys.stdin.buffer.read()) + args.file = temp.name + + mpy = await compile_multi_file(args.file, args.abi) print_mpy(mpy) @@ -135,7 +98,7 @@ def add_parser(self, subparsers: argparse._SubParsersAction): "file", metavar="", help="path to a MicroPython script or `-` for stdin", - type=argparse.FileType(), + type=str, ) parser.add_argument( "-n", @@ -213,11 +176,15 @@ def is_pybricks_usb(dev): # 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) + if args.file == "-": + with NamedTemporaryFile(suffix=".py", delete=False) as temp: + temp.write(sys.stdin.buffer.read()) + args.file = temp.name + + if args.start: + await hub.run(args.file, args.wait) + else: + await hub.download(args.file) finally: await hub.disconnect() diff --git a/tests/test_cli.py b/tests/test_cli.py index 679f886..303f973 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -45,7 +45,7 @@ def test_add_parser(self): with patch("builtins.open", mock_file): args = parser.parse_args(["ble", "test.py"]) assert args.conntype == "ble" - assert args.file.name == "test.py" + assert args.file == "test.py" assert args.name is None # Test with optional name argument From 15392006eca3bf9486183e0e88047640b1faca68 Mon Sep 17 00:00:00 2001 From: Laurens Valk Date: Tue, 5 Aug 2025 12:19:36 +0200 Subject: [PATCH 2/3] connections: Allow starting by program identifier. This allows starting slots and other builtins like the REPL. --- pybricksdev/connections/pybricks.py | 27 +++++++++++++-------------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/pybricksdev/connections/pybricks.py b/pybricksdev/connections/pybricks.py index fcdfee9..b20dfcc 100644 --- a/pybricksdev/connections/pybricks.py +++ b/pybricksdev/connections/pybricks.py @@ -566,7 +566,7 @@ async def download(self, script_path: str) -> None: async def run( self, - py_path: Optional[str] = None, + program: str | int, wait: bool = True, print_output: bool = True, line_handler: bool = True, @@ -575,8 +575,8 @@ async def run( Compiles and runs a user program. Args: - py_path: The path to the .py file to compile. If None, runs a - previously downloaded program. + program: The path to the .py file to compile. If an integer is + given, runs a slot or builtin program with that identifier. wait: If true, wait for the user program to stop before returning. print_output: If true, echo stdout of the hub to ``sys.stdout``. line_handler: If true enable hub stdout line handler features. @@ -592,24 +592,23 @@ async def run( self.print_output = print_output self._enable_line_handler = line_handler self.script_dir = os.getcwd() - if py_path is not None: - self.script_dir, _ = os.path.split(py_path) + if isinstance(program, str): + self.script_dir, _ = os.path.split(program) # maintain compatibility with older firmware (Pybricks profile < 1.2.0). if self._mpy_abi_version: - if py_path is None: + if not isinstance(program, str): raise RuntimeError( - "Hub does not support running stored program. Provide a py_path to run" + "Hub does not support running stored program. Provide a path to run" ) - await self._legacy_run(py_path, wait) + await self._legacy_run(program, wait) return - # Download the program if a path is provided - if py_path is not None: - await self.download(py_path) - - # Start the program - await self.start_user_program() + if isinstance(program, str): + await self.download(program) + await self.start_user_program() + else: + await self.start_user_program(program) if wait: await self._wait_for_user_program_stop() From 4c5501cdc5a88882bd6310070bcdd2f9b081e934 Mon Sep 17 00:00:00 2001 From: Laurens Valk Date: Tue, 5 Aug 2025 12:20:12 +0200 Subject: [PATCH 3/3] cli: Enable starting REPL. --- pybricksdev/cli/__init__.py | 25 ++++++++++++++++--------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 0b5c815..dfe6ed5 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -16,6 +16,7 @@ from pybricksdev import __name__ as MODULE_NAME from pybricksdev import __version__ as MODULE_VERSION +from pybricksdev.ble.pybricks import UserProgramId PROG_NAME = ( f"{path.basename(sys.executable)} -m {MODULE_NAME}" @@ -97,7 +98,7 @@ def add_parser(self, subparsers: argparse._SubParsersAction): parser.add_argument( "file", metavar="", - help="path to a MicroPython script or `-` for stdin", + help="path to a MicroPython script, `-` for stdin, or `repl` for interactive prompt", type=str, ) parser.add_argument( @@ -176,15 +177,21 @@ def is_pybricks_usb(dev): # Connect to the address and run the script await hub.connect() try: - if args.file == "-": - with NamedTemporaryFile(suffix=".py", delete=False) as temp: - temp.write(sys.stdin.buffer.read()) - args.file = temp.name - - if args.start: - await hub.run(args.file, args.wait) + # Handle builtin programs. + if args.file == "repl": + await hub.run(UserProgramId.REPL, args.wait) else: - await hub.download(args.file) + # If using stdin, save to temporary file first. + if args.file == "-": + with NamedTemporaryFile(suffix=".py", delete=False) as temp: + temp.write(sys.stdin.buffer.read()) + args.file = temp.name + + # Download program and optionally start it. + if args.start: + await hub.run(args.file, args.wait) + else: + await hub.download(args.file) finally: await hub.disconnect()