diff --git a/README.md b/README.md index 413ae43..41336f5 100644 --- a/README.md +++ b/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]` | -| `--src`, `--lib` | Upload `src/` or a tree to `/lib` | +| `-u`, `--upload` | Recursive upload: `-u SRC [DEST]` (skips unchanged files via `file_hashes.json` on device) | +| `--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) | diff --git a/cli.py b/cli.py index 484e16f..e7691ec 100755 --- a/cli.py +++ b/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) - print(f"Upload complete: {files_copied} files, {dirs_created} directories created.", file=sys.stderr) + files_copied, dirs_created, files_skipped = conn.upload_directory( + 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: diff --git a/deploy_manifest.py b/deploy_manifest.py new file mode 100644 index 0000000..0fc653d --- /dev/null +++ b/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") diff --git a/device.py b/device.py index 56fe6cf..98ca506 100644 --- a/device.py +++ b/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 @@ -108,14 +116,42 @@ class DeviceConnection: self._fs_writefile_with_wdt(remote_path, data) finally: self.disconnect() + + 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): + 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() diff --git a/tests/test_deploy_manifest.py b/tests/test_deploy_manifest.py new file mode 100644 index 0000000..0ed03a5 --- /dev/null +++ b/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()