Allow starting REPL

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="<file>",
             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="<file>",
-            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()
 

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()

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