feat: add xmodem cp upload/download command

meshtastic/__main__.py

@@ -37,6 +37,7 @@
 except ImportError as e:
     have_test = False
 
+import meshtastic.file_transfer_cli as file_transfer_cli
 import meshtastic.ota
 import meshtastic.util
 import meshtastic.serial_interface
@@ -443,6 +444,116 @@ def onConnected(interface):
             # Must turn off encryption on primary channel
             interface.getNode(args.dest, **getNode_kwargs).turnOffEncryptionOnPrimaryChannel()
 
+        if args.ls is not None:
+            closeNow = True
+            remote_dir = args.ls
+            depth = int(getattr(args, "ls_depth", 0) or 0)
+            node = interface.localNode
+            rows = node.listDir(remote_dir, depth=depth)
+            if rows is None:
+                meshtastic.util.our_exit("listDir failed", 1)
+            for path, sz in rows:
+                print(f"{sz}\t{path}")
+
+        if args.upload is not None:
+            closeNow = True
+            node = interface.localNode
+            local_tokens = list(args.upload[:-1])
+            remote_base = args.upload[-1]
+            try:
+                upload_pairs = file_transfer_cli.plan_upload(local_tokens, remote_base)
+            except file_transfer_cli.FileTransferCliError as e:
+                meshtastic.util.our_exit(str(e), 1)
+
+            def _upload_progress(sent, total):
+                pct = 100 * sent // total if total else 0
+                bar = '#' * (pct // 5) + '.' * (20 - pct // 5)
+                print(f"\r  [{bar}] {pct}%", end="", flush=True)
+
+            for i, (lp, devp) in enumerate(upload_pairs, start=1):
+                print(f"Uploading ({i}/{len(upload_pairs)}) {lp} → {devp}")
+                ok = node.uploadFile(lp, devp, on_progress=_upload_progress)
+                print(f"\r  {'OK' if ok else 'FAILED'}: {lp} → {devp}        ")
+                if not ok:
+                    meshtastic.util.our_exit("Upload failed", 1)
+
+        if args.download is not None:
+            closeNow = True
+            node = interface.localNode
+            rpath, lpath = args.download
+            lpath_abs = os.path.abspath(os.path.expanduser(lpath))
+            if os.path.isdir(lpath_abs):
+                meshtastic.util.our_exit(
+                    "ERROR: --download LOCAL must be a file path, not a directory (use --download-tree or --download-glob).",
+                    1,
+                )
+            parent = os.path.dirname(lpath_abs)
+            if parent:
+                os.makedirs(parent, exist_ok=True)
+            print(f"Downloading {rpath} → {lpath}")
+            def _download_progress(received, _total):
+                print(f"\r  {received} bytes received...", end="", flush=True)
+            ok = node.downloadFile(rpath, lpath_abs, on_progress=_download_progress)
+            print(f"\r  {'OK' if ok else 'FAILED'}: {rpath} → {lpath_abs}        ")
+            if not ok:
+                meshtastic.util.our_exit("Download failed", 1)
+
+        if args.download_tree is not None:
+            closeNow = True
+            node = interface.localNode
+            rdir, ldir = args.download_tree
+            depth = 255
+            rows = node.listDir(rdir.rstrip("/") or "/", depth=depth)
+            if rows is None:
+                meshtastic.util.our_exit("listDir failed", 1)
+            try:
+                tree_pairs = file_transfer_cli.plan_download_tree(rdir, ldir, rows)
+            except file_transfer_cli.FileTransferCliError as e:
+                meshtastic.util.our_exit(str(e), 1)
+            if not tree_pairs:
+                meshtastic.util.our_exit("No files matched for --download-tree", 1)
+
+            def _download_progress(received, _total):
+                print(f"\r  {received} bytes received...", end="", flush=True)
+
+            for i, (devp, locp) in enumerate(tree_pairs, start=1):
+                os.makedirs(os.path.dirname(locp) or ".", exist_ok=True)
+                print(f"Downloading ({i}/{len(tree_pairs)}) {devp} → {locp}")
+                ok = node.downloadFile(devp, locp, on_progress=_download_progress)
+                print(f"\r  {'OK' if ok else 'FAILED'}: {devp} → {locp}        ")
+                if not ok:
+                    meshtastic.util.our_exit("Download failed", 1)
+
+        if args.download_glob is not None:
+            closeNow = True
+            node = interface.localNode
+            pattern, ldir = args.download_glob
+            try:
+                base, _rel = file_transfer_cli.split_remote_glob_pattern(pattern)
+            except file_transfer_cli.FileTransferCliError as e:
+                meshtastic.util.our_exit(str(e), 1)
+            depth = 255
+            rows = node.listDir(base, depth=depth)
+            if rows is None:
+                meshtastic.util.our_exit("listDir failed", 1)
+            try:
+                glob_pairs = file_transfer_cli.plan_download_glob(pattern, ldir, rows)
+            except file_transfer_cli.FileTransferCliError as e:
+                meshtastic.util.our_exit(str(e), 1)
+            if not glob_pairs:
+                meshtastic.util.our_exit("No files matched for --download-glob", 1)
+
+            def _download_progress(received, _total):
+                print(f"\r  {received} bytes received...", end="", flush=True)
+
+            for i, (devp, locp) in enumerate(glob_pairs, start=1):
+                os.makedirs(os.path.dirname(locp) or ".", exist_ok=True)
+                print(f"Downloading ({i}/{len(glob_pairs)}) {devp} → {locp}")
+                ok = node.downloadFile(devp, locp, on_progress=_download_progress)
+                print(f"\r  {'OK' if ok else 'FAILED'}: {devp} → {locp}        ")
+                if not ok:
+                    meshtastic.util.our_exit("Download failed", 1)
+
         if args.reboot:
             closeNow = True
             waitForAckNak = True
@@ -1876,6 +1987,74 @@ def addLocalActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPars
         default=None
     )
 
+    xfer = group.add_mutually_exclusive_group()
+    xfer.add_argument(
+        "--upload",
+        help=(
+            "Upload local files to the device via XModem (requires matching firmware).  "
+            "Usage: --upload LOCAL [LOCAL ...] REMOTE.  "
+            "Last argument is the device path: for a single plain file LOCAL, REMOTE is the exact destination file path; "
+            "otherwise REMOTE is a directory prefix and relative paths are preserved.  "
+            "LOCAL may be files, directories (recursive), or globs (quote patterns with **).  "
+            "Each device path must be <= 128 UTF-8 bytes."
+        ),
+        nargs="+",
+        metavar="SPEC",
+        default=None,
+    )
+    xfer.add_argument(
+        "--download",
+        help=(
+            "Download one file from the device.  "
+            "Usage: --download REMOTE_FILE LOCAL_FILE.  "
+            "For a directory tree use --download-tree; for remote globs use --download-glob."
+        ),
+        nargs=2,
+        metavar=("REMOTE", "LOCAL"),
+        default=None,
+    )
+    xfer.add_argument(
+        "--download-tree",
+        help=(
+            "Download a full remote directory tree via MFLIST + XModem.  "
+            "Usage: --download-tree REMOTE_DIR LOCAL_DIR.  "
+            "Only rows with size > 0 are treated as files."
+        ),
+        nargs=2,
+        metavar=("REMOTE_DIR", "LOCAL_DIR"),
+        default=None,
+    )
+    xfer.add_argument(
+        "--download-glob",
+        help=(
+            "Download remote files matching a glob (MFLIST at the literal base + filter).  "
+            "Usage: --download-glob 'REMOTE_PATTERN' LOCAL_DIR.  "
+            "Pattern must include * ? or [; ** matches across / (relative to the literal base)."
+        ),
+        nargs=2,
+        metavar=("REMOTE_PATTERN", "LOCAL_DIR"),
+        default=None,
+    )
+
+    group.add_argument(
+        "--ls",
+        help=(
+            "List files on the device under REMOTE_DIR via XMODEM MFLIST (requires matching firmware).  "
+            "Output: size_bytes<TAB>path (one per line)."
+        ),
+        nargs="?",
+        const="/",
+        default=None,
+        metavar="REMOTE_DIR",
+    )
+
+    group.add_argument(
+        "--ls-depth",
+        help="Max directory depth for --ls (0 = files in REMOTE_DIR only).",
+        type=int,
+        default=0,
+    )
+
     return parser
 
 def addRemoteActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: