Merge pull request #143 from radrogow/add/repl

Add repl module and entrypoint

Author: BrianPugh

README.rst

@@ -103,7 +103,7 @@ littlefs-python comes bundled with a command-line tool, ``littlefs-python``, tha
 .. code:: console
 
    $ littlefs-python --help
-   usage: littlefs-python [-h] [--version] {create,extract,list} ...
+   usage: littlefs-python [-h] [--version] {create,extract,list,repl} ...
 
    Create, extract and inspect LittleFS filesystem images. Use one of the
    commands listed below, the '-h' / '--help' option can be used on each command
@@ -114,10 +114,11 @@ littlefs-python comes bundled with a command-line tool, ``littlefs-python``, tha
      --version             show program's version number and exit
 
    Available Commands:
-     {create,extract,list}
+     {create,extract,list,repl}
        create              Create LittleFS image from file/directory contents.
        extract             Extract LittleFS image contents to a directory.
        list                List LittleFS image contents.
+       repl                Inspect an existing LittleFS image through an interactive shell.
 
 To create a littlefs binary image:
 
@@ -135,6 +136,17 @@ To extract the contents of a littlefs binary image:
 
    $ littlefs-python extract lfs.bin output/ --block-size=4096
 
+To inspect or debug an existing image without extracting it first you can start a
+simple REPL. It provides shell-like commands such as ``ls``, ``tree``, ``put``, ``get``
+and ``rm`` that operate directly on the image data:
+
+.. code:: console
+
+   $ littlefs-python repl lfs.bin --block-size=4096
+   Mounted remote littlefs.
+   littlefs> ls
+   README.rst
+
 Development Setup
 =================
 

src/littlefs/__init__.py

@@ -34,6 +34,7 @@
     "LittleFS",
     "LittleFSError",
     "UserContext",
+    "UserContextFile",
     "UserContextWinDisk",
     "__LFS_DISK_VERSION__",
     "__LFS_VERSION__",
@@ -48,7 +49,7 @@
     # Package not installed
     pass
 
-from .context import UserContext, UserContextWinDisk
+from .context import UserContext, UserContextFile, UserContextWinDisk
 
 if TYPE_CHECKING:
     from .lfs import LFSStat

src/littlefs/__main__.py

@@ -6,6 +6,8 @@
 
 from littlefs import LittleFS, __version__
 from littlefs.errors import LittleFSError
+from littlefs.repl import LittleFSRepl
+from littlefs.context import UserContextFile
 
 # Dictionary mapping suffixes to their size in bytes
 _suffix_map = {
@@ -110,7 +112,7 @@ def create(parser: argparse.ArgumentParser, args: argparse.Namespace) -> int:
         compact_fs.fs_grow(args.block_count)
         data = compact_fs.context.buffer
         if not args.no_pad:
-            data = data.ljust(args.fs_size, b"\xFF")
+            data = data.ljust(args.fs_size, b"\xff")
     else:
         data = fs.context.buffer
 
@@ -188,6 +190,47 @@ def extract(parser: argparse.ArgumentParser, args: argparse.Namespace) -> int:
     return 0
 
 
+def repl(parser: argparse.ArgumentParser, args: argparse.Namespace) -> int:
+    """Inspect an existing LittleFS image through an interactive shell."""
+
+    source: Path = args.source
+    if not source.is_file():
+        parser.error(f"Source image '{source}' does not exist.")
+
+    image_size = source.stat().st_size
+    if not image_size or image_size % args.block_size:
+        parser.error(
+            f"Image size ({image_size} bytes) is not a multiple of the supplied block size ({args.block_size})."
+        )
+
+    block_count = image_size // args.block_size
+    if block_count == 0:
+        parser.error("Image is smaller than a single block; cannot mount.")
+
+    context = UserContextFile(str(source))
+    fs = LittleFS(
+        context=context,
+        block_size=args.block_size,
+        block_count=block_count,
+        name_max=args.name_max,
+        mount=False,
+    )
+
+    shell = LittleFSRepl(fs)
+    try:
+        try:
+            shell.do_mount()
+        except LittleFSError as exc:
+            parser.error(f"Failed to mount '{source}': {exc}")
+        shell.cmdloop()
+    finally:
+        if shell._mounted:
+            with suppress(LittleFSError):
+                fs.unmount()
+
+    return 0
+
+
 def get_parser():
     if sys.argv[0].endswith("__main__.py"):
         prog = f"python -m littlefs"
@@ -299,6 +342,19 @@ def add_command(handler, name="", help=""):
         help="LittleFS block size.",
     )
 
+    parser_repl = add_command(repl)
+    parser_repl.add_argument(
+        "source",
+        type=Path,
+        help="Source LittleFS filesystem binary.",
+    )
+    parser_repl.add_argument(
+        "--block-size",
+        type=size_parser,
+        required=True,
+        help="LittleFS block size.",
+    )
+
     return parser
 
 

src/littlefs/context.py

@@ -1,6 +1,7 @@
 import logging
 import typing
 import ctypes
+import os
 
 if typing.TYPE_CHECKING:
     from .lfs import LFSConfig
@@ -78,6 +79,60 @@ def sync(self, cfg: "LFSConfig") -> int:
         return 0
 
 
+class UserContextFile(UserContext):
+    """File-backed context using the standard library"""
+
+    def __init__(self, file_path: str, *, create: bool = False) -> None:
+        mode = "r+b"
+        if not os.path.exists(file_path):
+            if not create:
+                raise FileNotFoundError(f"Context file '{file_path}' does not exist")
+            mode = "w+b"
+
+        self._path = file_path
+        self._fh = open(file_path, mode)
+
+    def read(self, cfg: "LFSConfig", block: int, off: int, size: int) -> bytearray:
+        logging.getLogger(__name__).debug("LFS Read : Block: %d, Offset: %d, Size=%d" % (block, off, size))
+        start = block * cfg.block_size + off
+        self._fh.seek(start)
+        data = self._fh.read(size)
+
+        if len(data) < size:
+            data += b"\xff" * (size - len(data))
+
+        return bytearray(data)
+
+    def prog(self, cfg: "LFSConfig", block: int, off: int, data: bytes) -> int:
+        logging.getLogger(__name__).debug("LFS Prog : Block: %d, Offset: %d, Data=%r" % (block, off, data))
+        start = block * cfg.block_size + off
+        self._fh.seek(start)
+        self._fh.write(data)
+        return 0
+
+    def erase(self, cfg: "LFSConfig", block: int) -> int:
+        logging.getLogger(__name__).debug("LFS Erase: Block: %d" % block)
+        start = block * cfg.block_size
+        self._fh.seek(start)
+        self._fh.write(b"\xff" * cfg.block_size)
+        return 0
+
+    def sync(self, cfg: "LFSConfig") -> int:
+        self._fh.flush()
+        os.fsync(self._fh.fileno())
+        return 0
+
+    def close(self) -> None:
+        if not self._fh.closed:
+            self._fh.close()
+
+    def __del__(self):
+        try:
+            self.close()
+        except Exception:
+            pass
+
+
 try:
     import win32file
 except ImportError: