feat(cli): skip unchanged files using device file_hashes.json

Read and update file_hashes.json on deploy; add --force-upload and tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

README.md

@@ -25,8 +25,9 @@ Connection is always via **`-p` / `--port`** (default `/dev/ttyACM0`). There is
 | `-r`, `--reset` | Reset the device |
 | `-f`, `--follow` | Follow serial output (optional timeout seconds) |
 | `--pause` | Sleep N seconds (for chained actions) |
-| `-u`, `--upload` | Recursive upload: `-u SRC [DEST]` |
+| `-u`, `--upload` | Recursive upload: `-u SRC [DEST]` (skips unchanged files via `file_hashes.json` on device) |
-| `--src`, `--lib` | Upload `src/` or a tree to `/lib` |
+| `--src`, `--lib`, `--all` | Deploy led-driver trees to flash root, `patterns/`, and `lib/` |
+| `--force-upload` | Upload every file; ignore `file_hashes.json` |
 | `-e`, `--erase` | Erase everything at device root (including `settings.json` and `presets.json`) |
 | `--rm` | Remove a path on the device |
 | `--flash` | Flash a firmware binary (uses **esptool** on the host) |

cli.py

@@ -223,6 +223,10 @@ def _get_ordered_actions(argv: List[str]) -> List[tuple]:
             actions.append(('upload', ["lib", "lib"]))
             i += 1
             continue
+        if arg == '--force-upload':
+            # Handled via argparse; skip during action scan
+            i += 1
+            continue
         if arg == '--lib':
             # Upload local DIR (default: ./lib) to /lib on device
             local_dir = "lib"
@@ -486,6 +490,12 @@ Examples:
         help="Upload ./src (excluding patterns), ./src/patterns, and ./lib."
     )
 
+    parser.add_argument(
+        "--force-upload",
+        action="store_true",
+        help="Upload every file; ignore file_hashes.json on the device",
+    )
+
     parser.add_argument(
         "--patterns", "--paterns",
         dest="patterns_dir",
@@ -587,8 +597,14 @@ Examples:
                 else:
                     print(f"Uploading {upload_dir} to device on {port}...", file=sys.stderr)
                 conn = DeviceConnection(port)
-                files_copied, dirs_created = conn.upload_directory(upload_dir, remote_dir)
+                files_copied, dirs_created, files_skipped = conn.upload_directory(
-                print(f"Upload complete: {files_copied} files, {dirs_created} directories created.", file=sys.stderr)
+                    upload_dir, remote_dir, force=args.force_upload
+                )
+                print(
+                    f"Upload complete: {files_copied} uploaded, {files_skipped} skipped, "
+                    f"{dirs_created} directories created.",
+                    file=sys.stderr,
+                )
             except Exception as e:
                 print(f"Error uploading directory: {e}", file=sys.stderr)
                 sys.exit(1)
@@ -616,9 +632,12 @@ Examples:
                         file=sys.stderr,
                     )
                     conn = DeviceConnection(port)
-                    files_copied, dirs_created = conn.upload_directory(temp_src, "")
+                    files_copied, dirs_created, files_skipped = conn.upload_directory(
+                        temp_src, "", force=args.force_upload
+                    )
                     print(
-                        f"Upload complete: {files_copied} files, {dirs_created} directories created.",
+                        f"Upload complete: {files_copied} uploaded, {files_skipped} skipped, "
+                        f"{dirs_created} directories created.",
                         file=sys.stderr,
                     )
             except Exception as e:

deploy_manifest.py

@@ -0,0 +1,45 @@
+"""
+Host-side helpers for file_hashes.json (same format as led-driver/src/file_hashes.py).
+"""
+
+import hashlib
+import json
+import os
+
+MANIFEST_VERSION = 1
+MANIFEST_FILENAME = "file_hashes.json"
+HASH_ALGO = "sha256"
+
+
+def normalize_remote_path(remote_file: str) -> str:
+    """Device-relative path with forward slashes (no leading slash)."""
+    return remote_file.replace("\\", "/").lstrip("/")
+
+
+def sha256_hex_file(path: str) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        for chunk in iter(lambda: f.read(65536), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def parse_manifest(data: bytes) -> dict:
+    """Return path -> hash map from manifest bytes."""
+    try:
+        doc = json.loads(data.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError):
+        return {}
+    if not isinstance(doc, dict):
+        return {}
+    files = doc.get("files")
+    return dict(files) if isinstance(files, dict) else {}
+
+
+def build_manifest_bytes(files: dict) -> bytes:
+    doc = {
+        "version": MANIFEST_VERSION,
+        "algorithm": HASH_ALGO,
+        "files": files,
+    }
+    return json.dumps(doc, separators=(",", ":")).encode("utf-8")

device.py

@@ -10,6 +10,14 @@ import os
 import time
 import serial
 
+from deploy_manifest import (
+    MANIFEST_FILENAME,
+    build_manifest_bytes,
+    normalize_remote_path,
+    parse_manifest,
+    sha256_hex_file,
+)
+
 # Add lib directory to path - handle both normal execution and PyInstaller bundle
 if getattr(sys, 'frozen', False):
     # Running as a PyInstaller bundle
@@ -109,13 +117,41 @@ class DeviceConnection:
         finally:
             self.disconnect()
 
-    def upload_directory(self, local_dir, remote_dir=None):
+    def _manifest_remote_path(self) -> str:
+        return "/" + MANIFEST_FILENAME
+
+    def read_hash_manifest(self) -> dict:
+        """Load path -> sha256 hex map from file_hashes.json on the device."""
+        remote = self._manifest_remote_path()
+        if not self.transport.fs_exists(remote):
+            return {}
+        try:
+            data = self.transport.fs_readfile(remote)
+            return parse_manifest(data)
+        except Exception:
+            return {}
+
+    def write_hash_manifest(self, files: dict) -> None:
+        """Write merged file_hashes.json to the device root."""
+        self._fs_writefile_with_wdt(
+            self._manifest_remote_path(),
+            build_manifest_bytes(files),
+        )
+    
+    def upload_directory(self, local_dir, remote_dir=None, *, force: bool = False):
         """
         Upload a directory recursively to the device.
         
+        Skips files whose sha256 matches file_hashes.json on the device unless
+        force is True. Updates the manifest after upload.
+        
         Args:
             local_dir: Local directory path to upload
             remote_dir: Remote directory path (default: root, uses basename of local_dir)
+            force: Upload every file even when the manifest hash matches
+        
+        Returns:
+            (files_copied, dirs_created, files_skipped)
         """
         import os
         
@@ -129,11 +165,14 @@ class DeviceConnection:
                 remote_dir = os.path.basename(os.path.abspath(local_dir))
             
             # Ensure remote directory exists
-            if not self.transport.fs_exists(remote_dir):
+            if remote_dir and not self.transport.fs_exists(remote_dir):
                 self.transport.fs_mkdir(remote_dir)
             
+            manifest = {} if force else self.read_hash_manifest()
+            
             # Walk through local directory and copy files
             files_copied = 0
+            files_skipped = 0
             dirs_created = 0
             
             for root, dirs, files in os.walk(local_dir):
@@ -161,6 +200,8 @@ class DeviceConnection:
                 for file in files:
                     if file.endswith((".pyc", ".pyo")):
                         continue
+                    if file == MANIFEST_FILENAME:
+                        continue
                     local_file = os.path.join(root, file)
                     # Handle root directory case properly
                     if remote_base == '/':
@@ -168,13 +209,23 @@ class DeviceConnection:
                     else:
                         remote_file = '/'.join([remote_base, file])
                     
+                    manifest_key = normalize_remote_path(remote_file)
+                    local_hash = sha256_hex_file(local_file)
+                    if not force and manifest.get(manifest_key) == local_hash:
+                        print(f"Skipping (unchanged): {remote_file}", file=sys.stderr)
+                        files_skipped += 1
+                        continue
+                    
                     print(f"Uploading: {remote_file}", file=sys.stderr)
                     with open(local_file, 'rb') as f:
                         data = f.read()
                     self._fs_writefile_with_wdt(remote_file, data)
+                    manifest[manifest_key] = local_hash
                     files_copied += 1
             
-            return files_copied, dirs_created
+            self.write_hash_manifest(manifest)
+            
+            return files_copied, dirs_created, files_skipped
         finally:
             self.disconnect()
     

tests/test_deploy_manifest.py

@@ -0,0 +1,52 @@
+"""Tests for deploy_manifest helpers (host Python)."""
+
+import json
+import os
+import tempfile
+import unittest
+
+from deploy_manifest import (
+    MANIFEST_FILENAME,
+    build_manifest_bytes,
+    normalize_remote_path,
+    parse_manifest,
+    sha256_hex_file,
+)
+
+
+class DeployManifestTests(unittest.TestCase):
+    def test_normalize_remote_path(self):
+        self.assertEqual(normalize_remote_path("/patterns/chase.py"), "patterns/chase.py")
+        self.assertEqual(normalize_remote_path("main.py"), "main.py")
+
+    def test_round_trip_manifest(self):
+        files = {"main.py": "abc", "patterns/x.py": "def"}
+        blob = build_manifest_bytes(files)
+        parsed = parse_manifest(blob)
+        self.assertEqual(parsed, files)
+        doc = json.loads(blob.decode())
+        self.assertEqual(doc["version"], 1)
+        self.assertEqual(doc["algorithm"], "sha256")
+
+    def test_parse_manifest_invalid(self):
+        self.assertEqual(parse_manifest(b"not json"), {})
+        self.assertEqual(parse_manifest(b'{"files": "nope"}'), {})
+
+    def test_sha256_hex_file(self):
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"hello")
+            path = f.name
+        try:
+            self.assertEqual(
+                sha256_hex_file(path),
+                "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824",
+            )
+        finally:
+            os.unlink(path)
+
+    def test_manifest_filename(self):
+        self.assertEqual(MANIFEST_FILENAME, "file_hashes.json")
+
+
+if __name__ == "__main__":
+    unittest.main()