8 Commits

Author SHA1 Message Date
1edcb8b1f7 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>
2026-05-17 19:14:51 +12:00
ccc215acbd fix(cli): lazy settings fetch and full device erase
Download settings only for --show or when applying edits; skip upload
when unchanged; erase entire device root including settings.json.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-16 21:14:54 +12:00
580fd11aca fix(cli): skip redundant settings uploads when unchanged
Only upload settings when edited values differ from current device settings to avoid unnecessary resets.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-04 22:48:54 +12:00
pi
d6331a105c feat(cli): add --reset-device-name and WDT feed during uploads
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-03 21:27:29 +12:00
2f3db9272b feat(cli): extend upload flags for src, patterns, lib, and --all
Support --patterns/--paterns, --all, --erase, and src upload excluding patterns/.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-03 14:54:13 +12:00
713cd6e9a1 feat(cli): add ls flag and default lib upload
Made-with: Cursor
2026-04-15 00:46:19 +12:00
9e72c62481 fix(led-tool): build cli without spec file dependency 2026-04-14 23:13:21 +12:00
pi
eee9327e15 docs: fix readme layout and cli flags
Made-with: Cursor
2026-04-12 00:13:49 +12:00
7 changed files with 528 additions and 78 deletions

View File

@@ -16,5 +16,5 @@ python_version = "3"
[scripts]
web = "python web.py"
cli = "python cli.py"
build = "pyinstaller --clean led-cli.spec"
build = "pyinstaller --clean --onefile --name led-cli --paths lib cli.py"
install = "pipenv install"

View File

@@ -1,8 +1,41 @@
# led-tool
- `-s, --show`: Display current settings from device
- `--no-download`: Don't download settings first (use empty settings)
**Default behavior:** Downloads settings and prints them. If any edit flags are provided, settings are modified and uploaded automatically (unless `--no-upload` is specified).
## Device Connection
CLI helpers for MicroPython LED devices: **`settings.json`** download/upload via **mpremote**, resets, follow/tail, recursive uploads, optional firmware flash, and **led-driver**-oriented flags (presets, transport, Wi-Fi fields).
The tools use an integrated mpremote transport to communicate with MicroPython devices. Make sure your device is connected and accessible via the specified serial port.## LicenseSee LICENSE file for details.
Connection is always via **`-p` / `--port`** (default `/dev/ttyACM0`). There is **no** separate “server IP” flag; the Pi or PC address is configured on the device in **`settings.json`** when using Wi-Fi mode.
## Common flags
| Flag | Purpose |
|------|--------|
| `-p`, `--port` | Serial device (default `/dev/ttyACM0`) |
| `-s`, `--show` | Print current settings from the device |
| `-n`, `--name` | Device **name** |
| `--reset-device-name` | Set **name** to firmware default (`led-` + STA MAC hex, same as `Settings.set_defaults` on led-driver) |
| `--id` | Numeric device id (ESP-NOW, 0255) |
| `--pin` | LED GPIO (`led_pin`) |
| `-b`, `--brightness` | Brightness 0255 |
| `-l`, `--leds` | LED count (`num_leds`) |
| `-d`, `--debug` | Debug 0 or 1 |
| `-o`, `--order` | LED colour order (`rgb`, `grb`, …) |
| `--preset` / `--pattern` | Create or replace a named preset in **led-driver** `presets.json` |
| `--default` | Startup preset name |
| `--transport` | `espnow` or `wifi` (`transport_type` on device) |
| `--ssid`, `--wifi-password`, `--wifi-channel` | Wi-Fi / channel fields for the driver |
| `-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]` (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) |
**Settings I/O:** With no arguments, **`led-cli`** prints help (no device access). **`--show`** downloads `settings.json` and prints it (read-only). Setting fields (`--name`, `-b`, …) download once, merge, print, and **upload only when a value actually changed**. **`--preset`** alone only touches **`presets.json`** (no `settings.json` download). Ordered actions (`--reset`, `--upload`, `-e`, …) run without touching settings unless you also pass **`--show`** or setting / preset edits as above.
Run **`python cli.py -h`** for the full epilog and argument list.
## License
See **LICENSE** in this directory.

272
cli.py
View File

@@ -2,8 +2,10 @@
"""
LED Bar Configuration CLI Tool
Command-line interface for downloading, editing, and uploading settings.json
to/from MicroPython devices via mpremote.
Command-line interface for editing settings.json on MicroPython devices via mpremote.
Settings are downloaded from the device only when displaying (--show) or when applying
setting changes; uploads occur only when something changed. Use --erase to clear the
device filesystem (including settings).
"""
import json
@@ -11,11 +13,12 @@ import argparse
import subprocess
import sys
import time
import shutil
from typing import Dict, Any, List, Optional
import tempfile
import os
from device import copy_file, DeviceConnection
from device import copy_file, DeviceConnection, firmware_default_device_name
def resolve_flash_binary(path: str) -> Optional[str]:
@@ -147,14 +150,14 @@ _FLAGS_WITH_VALUE = frozenset({
'-p', '--port', '-n', '--name', '--pin', '-b', '--brightness',
'-l', '--leds', '-d', '-debug', '--debug', '-o', '--order',
'--preset', '--pattern', '--default', '--transport', '--ssid',
'--wifi-password', '--wifi-channel',
'--wifi-password', '--wifi-channel', '--src', '--lib', '--patterns', '--paterns',
})
def _get_ordered_actions(argv: List[str]) -> List[tuple]:
"""
Scan argv and return list of (action_name, value) in order of appearance.
Actions: flash, pause, reset, upload, erase_all, rm, follow.
Actions: flash, pause, reset, upload, ls, erase_all, rm, follow.
"""
actions = []
i = 1
@@ -201,22 +204,51 @@ def _get_ordered_actions(argv: List[str]) -> List[tuple]:
i += 2
else:
i += 1
# Use empty string as remote_dir to map to root on device
actions.append(('upload', [local_dir, ""]))
# Upload source tree excluding patterns/ (handled by --patterns).
actions.append(('upload_src_no_patterns', local_dir))
continue
if arg == '--lib':
# Upload local DIR to /lib on device
if i + 1 < len(argv):
if arg in ('--patterns', '--paterns'):
# Upload local patterns DIR (default: ./src/patterns) to /patterns.
local_dir = os.path.join("src", "patterns")
if i + 1 < len(argv) and not argv[i + 1].startswith('-'):
local_dir = argv[i + 1]
actions.append(('upload', [local_dir, "lib"]))
i += 2
else:
raise ValueError("--lib requires a directory argument")
i += 1
actions.append(('upload', [local_dir, "patterns"]))
continue
if arg == '--all':
actions.append(('upload_src_no_patterns', "src"))
actions.append(('upload', [os.path.join("src", "patterns"), "patterns"]))
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"
if i + 1 < len(argv) and not argv[i + 1].startswith('-'):
local_dir = argv[i + 1]
i += 2
else:
i += 1
actions.append(('upload', [local_dir, "lib"]))
continue
if arg == '--ls':
actions.append(('ls', None))
i += 1
continue
if arg == '-e':
actions.append(('erase_all', None))
i += 1
continue
if arg == '--erase':
actions.append(('erase_all', None))
i += 1
continue
if arg == '--rm':
if i + 1 < len(argv):
actions.append(('rm', argv[i + 1]))
@@ -253,10 +285,13 @@ def main() -> None:
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Download and print current settings
# Show help (no device I/O)
%(prog)s
# Download, edit device name and brightness, then upload
# Show current settings.json from device (read-only)
%(prog)s --show
# Edit device name and brightness, then upload only if values changed
%(prog)s -n "LED-Strip-1" -b 128
# Set multiple parameters
@@ -283,6 +318,9 @@ Examples:
# Set name, num_leds, default pattern, and upload
%(prog)s --name "MyStrip" -l 60 --default rainbow
# Reset logical device name to firmware default (STA MAC based)
%(prog)s --reset-device-name
"""
)
@@ -297,6 +335,12 @@ Examples:
help="Device name"
)
parser.add_argument(
"--reset-device-name",
action="store_true",
help="Set name to firmware default (led-<STA MAC hex>, same as fresh settings.json on led-driver)",
)
parser.add_argument(
"--id",
dest="device_id",
@@ -429,20 +473,49 @@ Examples:
nargs="?",
const="src",
metavar="DIR",
help="Upload DIR recursively to device root (:/, no leading directory). If DIR is omitted, uses local ./src."
help="Upload DIR recursively to device root (:/, no leading directory), excluding patterns/. If DIR is omitted, uses local ./src."
)
parser.add_argument(
"--lib",
nargs="?",
const="lib",
metavar="DIR",
help="Upload DIR recursively to /lib on device"
help="Upload DIR recursively to /lib on device. If DIR is omitted, uses local ./lib."
)
parser.add_argument(
"-e",
"--all",
action="store_true",
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",
nargs="?",
const=os.path.join("src", "patterns"),
metavar="DIR",
help="Upload DIR recursively to /patterns on device. If DIR is omitted, uses local ./src/patterns."
)
parser.add_argument(
"--ls",
action="store_true",
help="List files on the device root (:/)"
)
parser.add_argument(
"-e", "--erase",
dest="erase_all",
action="store_true",
help="Erase all code on the device (delete all files except settings.json)"
help="Erase the device filesystem: delete all files and directories at device root (including settings.json and presets.json)",
)
parser.add_argument(
@@ -458,6 +531,9 @@ Examples:
)
try:
if len(sys.argv) <= 1:
parser.print_help()
return
args = parser.parse_args()
ordered_actions = _get_ordered_actions(sys.argv)
except ValueError as e:
@@ -521,22 +597,73 @@ 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)
elif action_name == 'upload_src_no_patterns':
src_dir = value
if not os.path.exists(src_dir):
print(f"Error: Directory does not exist: {src_dir}", file=sys.stderr)
sys.exit(1)
if not os.path.isdir(src_dir):
print(f"Error: Not a directory: {src_dir}", file=sys.stderr)
sys.exit(1)
try:
with tempfile.TemporaryDirectory() as temp_src:
for entry in sorted(os.listdir(src_dir)):
if entry == "patterns":
continue
src_entry = os.path.join(src_dir, entry)
dst_entry = os.path.join(temp_src, entry)
if os.path.isdir(src_entry):
shutil.copytree(src_entry, dst_entry)
else:
shutil.copy2(src_entry, dst_entry)
print(
f"Uploading {src_dir} (excluding patterns/) to device root on {port}...",
file=sys.stderr,
)
conn = DeviceConnection(port)
files_copied, dirs_created, files_skipped = conn.upload_directory(
temp_src, "", 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 src (excluding patterns): {e}", file=sys.stderr)
sys.exit(1)
elif action_name == 'ls':
try:
print(f"Listing files on device {port}...", file=sys.stderr)
conn = DeviceConnection(port)
items = conn.list_files('')
for name, is_dir, size in items:
marker = "d" if is_dir else "-"
print(f"{marker} {size:>8} {name}")
except Exception as e:
print(f"Error listing files: {e}", file=sys.stderr)
sys.exit(1)
elif action_name == 'erase_all':
try:
print(f"Erasing all code on device {port}...", file=sys.stderr)
print(f"Erasing device filesystem on {port}...", file=sys.stderr)
conn = DeviceConnection(port)
items = conn.list_files('')
for name, is_dir, size in items:
if name == "settings.json":
continue
conn.delete(name)
print("Erase complete.", file=sys.stderr)
print("Erase complete (device root is empty).", file=sys.stderr)
except Exception as e:
print(f"Error erasing device: {e}", file=sys.stderr)
sys.exit(1)
@@ -569,11 +696,35 @@ Examples:
sys.exit(1)
return # follow blocks; when interrupted we're done
default_name_from_device: Optional[str] = None
if args.reset_device_name:
if args.name is not None:
print(
"Error: use either --name or --reset-device-name, not both.",
file=sys.stderr,
)
sys.exit(1)
try:
print(
f"Reading firmware default device name (STA MAC) on {port}...",
file=sys.stderr,
)
default_name_from_device = firmware_default_device_name(port)
print(
f"Default name will be {default_name_from_device!r}.",
file=sys.stderr,
)
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
# Collect all edit parameters
edits: Dict[str, Any] = {}
if args.name is not None:
edits["name"] = args.name
elif default_name_from_device is not None:
edits["name"] = default_name_from_device
if args.pin is not None:
edits["led_pin"] = args.pin
@@ -609,32 +760,42 @@ Examples:
# Clamp into single-byte range; store as int in settings.json
edits["id"] = max(0, min(255, args.device_id))
# 1. Download: get current settings from device
settings_work = args.show or bool(edits)
if settings_work:
try:
print(f"Downloading settings from {args.port}...", file=sys.stderr)
settings = download_settings(args.port)
print("Settings downloaded successfully.", file=sys.stderr)
except Exception as e:
if "timeout" in str(e).lower() or "connection" in str(e).lower():
print(f"Error: Connection timeout. Check device connection on {args.port}", file=sys.stderr)
print(
f"Error: Connection timeout. Check device connection on {args.port}",
file=sys.stderr,
)
else:
print(f"Error downloading settings: {e}", file=sys.stderr)
sys.exit(1)
# If --show only, print and exit (unless presets or settings are being written)
if args.show:
print_settings(settings)
if not edits and args.preset is None:
return
# 2. Edit: apply edits to downloaded settings
if edits:
print(f"Applying {len(edits)} edit(s)...", file=sys.stderr)
settings.update(edits)
changed_edits: Dict[str, Any] = {}
for key, value in edits.items():
if settings.get(key) != value:
changed_edits[key] = value
if edits and not changed_edits:
print("No settings changes detected; skipping settings upload.", file=sys.stderr)
if changed_edits:
print(f"Applying {len(changed_edits)} setting change(s)...", file=sys.stderr)
settings.update(changed_edits)
print_settings(settings)
# 3a. Presets file (led-driver presets.json)
if args.preset is not None:
pattern = args.pattern if args.pattern is not None else "on"
try:
@@ -655,23 +816,62 @@ Examples:
print("Device will reset.", file=sys.stderr)
except Exception as e:
if "timeout" in str(e).lower() or "connection" in str(e).lower():
print(f"Error: Connection timeout. Check device connection on {args.port}", file=sys.stderr)
print(
f"Error: Connection timeout. Check device connection on {args.port}",
file=sys.stderr,
)
else:
print(f"Error uploading presets: {e}", file=sys.stderr)
sys.exit(1)
# 3b. Settings upload (resets device)
if edits:
if changed_edits:
try:
print(f"\nUploading settings to {args.port}...", file=sys.stderr)
upload_settings(args.port, settings)
print("Settings uploaded successfully. Device will reset.", file=sys.stderr)
except Exception as e:
if "timeout" in str(e).lower() or "connection" in str(e).lower():
print(f"Error: Connection timeout. Check device connection on {args.port}", file=sys.stderr)
print(
f"Error: Connection timeout. Check device connection on {args.port}",
file=sys.stderr,
)
else:
print(f"Error uploading settings: {e}", file=sys.stderr)
sys.exit(1)
return
if args.preset is not None:
pattern = args.pattern if args.pattern is not None else "on"
try:
print(f"Downloading presets from {args.port}...", file=sys.stderr)
presets_data = download_presets(args.port)
entry = presets_data.get(args.preset)
if not isinstance(entry, dict):
entry = {}
entry["p"] = pattern
presets_data[args.preset] = entry
print(
f"Writing preset {args.preset!r} (pattern={pattern}) to {PRESETS_REMOTE}...",
file=sys.stderr,
)
upload_presets(args.port, presets_data, reset=True)
print("Presets uploaded successfully.", file=sys.stderr)
print("Device will reset.", file=sys.stderr)
except Exception as e:
if "timeout" in str(e).lower() or "connection" in str(e).lower():
print(
f"Error: Connection timeout. Check device connection on {args.port}",
file=sys.stderr,
)
else:
print(f"Error uploading presets: {e}", file=sys.stderr)
sys.exit(1)
return
if ordered_actions:
return
parser.print_help()
if __name__ == "__main__":

45
deploy_manifest.py Normal file
View File

@@ -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")

136
device.py
View File

@@ -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
@@ -23,12 +31,15 @@ if lib_path not in sys.path and os.path.exists(lib_path):
sys.path.insert(0, lib_path)
from mpremote.transport_serial import SerialTransport
from mpremote.transport import TransportError
from mpremote.transport import TransportError, TransportExecError
class DeviceConnection:
"""Wrapper for device communication."""
#: Feed interval during uploads (led-driver uses WDT(timeout=10000) ms).
WDT_FEED_INTERVAL_SEC = 5.0
def __init__(self, device):
"""Connect to a device."""
self.device = device
@@ -54,6 +65,38 @@ class DeviceConnection:
pass
self.transport = None
def _feed_wdt(self) -> None:
"""Best-effort feed of the ESP task WDT between chunked FS writes."""
if self.transport is None:
return
try:
self.transport.exec(
"try:\n import machine\n machine.WDT(timeout=10000).feed()\nexcept Exception:\n pass\n"
)
except Exception:
pass
def _make_wdt_upload_progress_callback(self):
"""Progress hook for Transport.fs_writefile: feed WDT every WDT_FEED_INTERVAL_SEC."""
last_feed = [time.monotonic()]
def progress(written: int, total: int) -> None:
now = time.monotonic()
if now - last_feed[0] >= self.WDT_FEED_INTERVAL_SEC:
self._feed_wdt()
last_feed[0] = now
return progress
def _fs_writefile_with_wdt(self, remote_path: str, data: bytes) -> None:
"""Write file to device with periodic WDT feeds during long transfers."""
self._feed_wdt()
self.transport.fs_writefile(
remote_path,
data,
progress_callback=self._make_wdt_upload_progress_callback(),
)
def copy_from_device(self, remote_path, local_path):
"""Copy a file from device to local filesystem."""
self.connect()
@@ -70,17 +113,45 @@ class DeviceConnection:
try:
with open(local_path, 'rb') as f:
data = f.read()
self.transport.fs_writefile(remote_path, data)
self._fs_writefile_with_wdt(remote_path, data)
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
@@ -94,15 +165,19 @@ 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):
# Calculate relative path from local_dir
# Never upload Python bytecode trees (MicroPython does not use them).
dirs[:] = [d for d in dirs if d != "__pycache__"]
rel_path = os.path.relpath(root, local_dir)
# Build remote path
@@ -121,8 +196,12 @@ class DeviceConnection:
self.transport.fs_mkdir(remote_base)
dirs_created += 1
# Copy files
# Copy files (skip bytecode; __pycache__ dirs are pruned above)
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 == '/':
@@ -130,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.transport.fs_writefile(remote_file, data)
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()
@@ -235,6 +324,37 @@ class DeviceConnection:
raise TransportError(f"Failed to follow output: {e}") from e
def firmware_default_device_name(device: str) -> str:
"""
Default logical device name on led-driver firmware: led-<sta_mac_hex>,
matching Settings.set_defaults() in led-driver/src/settings.py.
"""
conn = DeviceConnection(device)
conn.connect()
try:
code = (
"import network, ubinascii\n"
"sta = network.WLAN(network.STA_IF)\n"
"sta.active(True)\n"
"mac = ubinascii.hexlify(sta.config('mac')).decode().lower()\n"
"print('led-' + mac)\n"
)
out = conn.transport.exec(code)
except TransportExecError as e:
raise RuntimeError(
"Could not read STA MAC from device (is this MicroPython with network?): "
+ (e.error_output or str(e)).strip()
) from e
finally:
conn.disconnect()
text = out.decode("utf-8", errors="replace").strip()
for line in reversed(text.splitlines()):
line = line.strip()
if line.startswith("led-"):
return line
raise RuntimeError("Device did not return a led-<mac> default name")
def copy_file(from_device, device, remote_path, local_path):
"""
Copy a file to/from device.

View File

@@ -10,7 +10,7 @@ pipenv install "$@"
if [ ! -f "dist/led-cli" ] || [ "cli.py" -nt "dist/led-cli" ] || [ "device.py" -nt "dist/led-cli" ]; then
echo ""
echo "Building binary..."
pipenv run pyinstaller --clean led-cli.spec
pipenv run pyinstaller --clean --onefile --name led-cli --paths lib cli.py
fi
# Ensure ~/.local/bin exists

View File

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