diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index b772e0e..dfe6ed5 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -5,20 +5,18 @@ 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 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}" @@ -50,44 +48,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 +58,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 +73,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) @@ -134,8 +98,8 @@ def add_parser(self, subparsers: argparse._SubParsersAction): parser.add_argument( "file", metavar="", - help="path to a MicroPython script or `-` for stdin", - type=argparse.FileType(), + help="path to a MicroPython script, `-` for stdin, or `repl` for interactive prompt", + type=str, ) parser.add_argument( "-n", @@ -213,11 +177,21 @@ 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: + # Handle builtin programs. + if args.file == "repl": + await hub.run(UserProgramId.REPL, args.wait) + else: + # 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(script_path, args.wait) + await hub.run(args.file, args.wait) else: - await hub.download(script_path) + await hub.download(args.file) finally: await hub.disconnect() 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() 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