Public Access
twislaandClaude Opus 5.5
5b199c2436
scripts/sd_put.sh <file> [card path] sends a file (by default into /updates, for Update from SD) without taking the card out. The serial driver drops bytes once its receive buffer is full, so the transfer is stop-and-wait: 1 KB chunks, each acknowledged once the Storage Service has written it, into a 2 KB receive buffer. The device checks the SHA-256 before renaming <path>.part into place, and gives up after 5 s of silence or a card job that never returns. FileReceiver holds the logic, with 12 host tests. About 55 KB/s: 1.6 MB in under 30 s. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EhqxQ49eCju4CzKYNjZzwT
80 lines
2.6 KiB
Python
Executable File
80 lines
2.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Copies a file to the Cardputer's SD card over the USB serial console (the `sd put` command).
|
|
|
|
Usage: scripts/sd_put.sh <file> [card path]
|
|
The card path defaults to /updates/<file name>, where Settings > Firmware finds Update Files.
|
|
The device acknowledges each chunk once it's on the card, checks the SHA-256 of the whole file,
|
|
and only then renames <card path>.part to <card path>.
|
|
"""
|
|
import glob
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
import serial
|
|
|
|
DEBUG = bool(os.environ.get("SD_PUT_DEBUG")) # also show the console lines in between
|
|
|
|
|
|
def find_port():
|
|
ports = sorted(glob.glob("/dev/serial/by-id/*Espressif*"))
|
|
return os.path.realpath(ports[0]) if ports else None
|
|
|
|
|
|
def reply(port, timeout):
|
|
"""The next `sd put:` line from the device; everything else on the console is skipped."""
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
line = port.readline().decode(errors="replace").strip()
|
|
if line.startswith("sd put:"):
|
|
return line[len("sd put:"):].strip()
|
|
if line and DEBUG:
|
|
print(f"\n console: {line}", file=sys.stderr)
|
|
return "error no answer from the device"
|
|
|
|
|
|
def fail(answer):
|
|
print()
|
|
sys.exit(f"device: {answer}")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
sys.exit(__doc__)
|
|
path = sys.argv[1]
|
|
dest = sys.argv[2] if len(sys.argv) > 2 else "/updates/" + os.path.basename(path)
|
|
port_name = find_port()
|
|
if not port_name:
|
|
sys.exit("No Cardputer found on USB")
|
|
data = open(path, "rb").read()
|
|
sha = hashlib.sha256(data).hexdigest()
|
|
|
|
with serial.Serial(port_name, 115200, timeout=0.5) as port:
|
|
port.reset_input_buffer()
|
|
# The leading newline ends any half-typed command.
|
|
port.write(f"\nsd put {dest} {len(data)} {sha}\n".encode())
|
|
answer = reply(port, 10)
|
|
if not answer.startswith("ready "):
|
|
fail(answer)
|
|
chunk = int(answer.split()[1])
|
|
|
|
start, sent = time.time(), 0
|
|
while sent < len(data):
|
|
port.write(data[sent:sent + chunk])
|
|
sent = min(sent + chunk, len(data))
|
|
print(f"\rsending {sent * 100 // len(data):3d}%", end="", flush=True)
|
|
if sent < len(data):
|
|
answer = reply(port, 10)
|
|
if answer != f"ok {sent}":
|
|
fail(answer)
|
|
answer = reply(port, 15)
|
|
if not answer.startswith("done "):
|
|
fail(answer)
|
|
seconds = time.time() - start
|
|
print(f"\rdevice: {answer}, {seconds:.1f} s ({len(data) / 1024 / seconds:.0f} KB/s)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|