MicroPython/weather-info
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first save

Browse Source
This commit is contained in:
tiijay
2025-10-19 18:29:10 +02:00
commit b5a30adb27
1303 changed files with 234711 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.venv/lib/python3.12/site-packages/mpremote/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
try:
from importlib.metadata import version, PackageNotFoundError
try:
__version__ = version("mpremote")
except PackageNotFoundError:
# Error loading package version (e.g. running from source).
__version__ = "0.0.0-local"
except ImportError:
# importlib.metadata not available (e.g. CPython <3.8 without
# importlib_metadata compatibility package installed).
__version__ = "0.0.0-unknown"

6
.venv/lib/python3.12/site-packages/mpremote/__main__.py Normal file
View File

@@ -0,0 +1,6 @@
#!/usr/bin/env python3
import sys
from mpremote import main
sys.exit(main.main())

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/__init__.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/__main__.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/commands.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/console.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/main.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/mip.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/mp_errno.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/repl.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/romfs.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/transport.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
.venv/lib/python3.12/site-packages/mpremote/__pycache__/transport_serial.cpython-312.pyc Normal file
View File

Binary file not shown.

746
.venv/lib/python3.12/site-packages/mpremote/commands.py Normal file
View File

@@ -0,0 +1,746 @@
import binascii
import errno
import hashlib
import os
import sys
import tempfile
import zlib
import serial.tools.list_ports
from .transport import TransportError, TransportExecError, stdout_write_bytes
from .transport_serial import SerialTransport
from .romfs import make_romfs, VfsRomWriter
class CommandError(Exception):
pass
def do_connect(state, args=None):
dev = args.device[0] if args else "auto"
do_disconnect(state)
try:
if dev == "list":
# List attached devices.
for p in sorted(serial.tools.list_ports.comports()):
print(
"{} {} {:04x}:{:04x} {} {}".format(
p.device,
p.serial_number,
p.vid if isinstance(p.vid, int) else 0,
p.pid if isinstance(p.pid, int) else 0,
p.manufacturer,
p.product,
)
)
# Don't do implicit REPL command.
state.did_action()
elif dev == "auto":
# Auto-detect and auto-connect to the first available USB serial port.
for p in sorted(serial.tools.list_ports.comports()):
if p.vid is not None and p.pid is not None:
try:
state.transport = SerialTransport(p.device, baudrate=115200)
return
except TransportError as er:
if not er.args[0].startswith("failed to access"):
raise er
raise TransportError("no device found")
elif dev.startswith("id:"):
# Search for a device with the given serial number.
serial_number = dev[len("id:") :]
dev = None
for p in serial.tools.list_ports.comports():
if p.serial_number == serial_number:
state.transport = SerialTransport(p.device, baudrate=115200)
return
raise TransportError("no device with serial number {}".format(serial_number))
else:
# Connect to the given device.
if dev.startswith("port:"):
dev = dev[len("port:") :]
state.transport = SerialTransport(dev, baudrate=115200)
return
except TransportError as er:
msg = er.args[0]
if msg.startswith("failed to access"):
msg += " (it may be in use by another program)"
raise CommandError(msg)
def do_disconnect(state, _args=None):
if not state.transport:
return
try:
if state.transport.mounted:
if not state.transport.in_raw_repl:
state.transport.enter_raw_repl(soft_reset=False)
state.transport.umount_local()
if state.transport.in_raw_repl:
state.transport.exit_raw_repl()
except OSError:
# Ignore any OSError exceptions when shutting down, eg:
# - filesystem_command will close the connection if it had an error
# - umounting will fail if serial port disappeared
pass
state.transport.close()
state.transport = None
state._auto_soft_reset = True
def show_progress_bar(size, total_size, op="copying"):
if not sys.stdout.isatty():
return
verbose_size = 2048
bar_length = 20
if total_size < verbose_size:
return
elif size >= total_size:
# Clear progress bar when copy completes
print("\r" + " " * (13 + len(op) + bar_length) + "\r", end="")
else:
bar = size * bar_length // total_size
progress = size * 100 // total_size
print(
"\r ... {} {:3d}% [{}{}]".format(op, progress, "#" * bar, "-" * (bar_length - bar)),
end="",
)
def _remote_path_join(a, *b):
if not a:
a = "./"
result = a.rstrip("/")
for x in b:
result += "/" + x.strip("/")
return result
def _remote_path_dirname(a):
a = a.rsplit("/", 1)
if len(a) == 1:
return ""
else:
return a[0]
def _remote_path_basename(a):
return a.rsplit("/", 1)[-1]
def do_filesystem_cp(state, src, dest, multiple, check_hash=False):
if dest.startswith(":"):
dest_no_slash = dest.rstrip("/" + os.path.sep + (os.path.altsep or ""))
dest_exists = state.transport.fs_exists(dest_no_slash[1:])
dest_isdir = dest_exists and state.transport.fs_isdir(dest_no_slash[1:])
# A trailing / on dest forces it to be a directory.
if dest != dest_no_slash:
if not dest_isdir:
raise CommandError("cp: destination is not a directory")
dest = dest_no_slash
else:
dest_exists = os.path.exists(dest)
dest_isdir = dest_exists and os.path.isdir(dest)
if multiple:
if not dest_exists:
raise CommandError("cp: destination does not exist")
if not dest_isdir:
raise CommandError("cp: destination is not a directory")
# Download the contents of source.
try:
if src.startswith(":"):
data = state.transport.fs_readfile(src[1:], progress_callback=show_progress_bar)
filename = _remote_path_basename(src[1:])
else:
with open(src, "rb") as f:
data = f.read()
filename = os.path.basename(src)
except IsADirectoryError:
raise CommandError("cp: -r not specified; omitting directory")
# Write back to dest.
if dest.startswith(":"):
# If the destination path is just the directory, then add the source filename.
if dest_isdir:
dest = ":" + _remote_path_join(dest[1:], filename)
# Skip copy if the destination file is identical.
if check_hash:
try:
remote_hash = state.transport.fs_hashfile(dest[1:], "sha256")
source_hash = hashlib.sha256(data).digest()
# remote_hash will be None if the device doesn't support
# hashlib.sha256 (and therefore won't match).
if remote_hash == source_hash:
print("Up to date:", dest[1:])
return
except OSError:
pass
# Write to remote.
state.transport.fs_writefile(dest[1:], data, progress_callback=show_progress_bar)
else:
# If the destination path is just the directory, then add the source filename.
if dest_isdir:
dest = os.path.join(dest, filename)
# Write to local file.
with open(dest, "wb") as f:
f.write(data)
def do_filesystem_recursive_cp(state, src, dest, multiple, check_hash):
# Ignore trailing / on both src and dest. (Unix cp ignores them too)
src = src.rstrip("/" + os.path.sep + (os.path.altsep if os.path.altsep else ""))
dest = dest.rstrip("/" + os.path.sep + (os.path.altsep if os.path.altsep else ""))
# If the destination directory exists, then we copy into it. Otherwise we
# use the destination as the target.
if dest.startswith(":"):
dest_exists = state.transport.fs_exists(dest[1:])
else:
dest_exists = os.path.exists(dest)
# Recursively find all files to copy from a directory.
# `dirs` will be a list of dest split paths.
# `files` will be a list of `(dest split path, src joined path)`.
dirs = []
files = []
# For example, if src=/tmp/foo, with /tmp/foo/x.py and /tmp/foo/a/b/c.py,
# and if the destination directory exists, then we will have:
# dirs = [['foo'], ['foo', 'a'], ['foo', 'a', 'b']]
# files = [(['foo', 'x.py'], '/tmp/foo/x.py'), (['foo', 'a', 'b', 'c.py'], '/tmp/foo/a/b/c.py')]
# If the destination doesn't exist, then we will have:
# dirs = [['a'], ['a', 'b']]
# files = [(['x.py'], '/tmp/foo/x.py'), (['a', 'b', 'c.py'], '/tmp/foo/a/b/c.py')]
def _list_recursive(base, src_path, dest_path, src_join_fun, src_isdir_fun, src_listdir_fun):
src_path_joined = src_join_fun(base, *src_path)
if src_isdir_fun(src_path_joined):
if dest_path:
dirs.append(dest_path)
for entry in src_listdir_fun(src_path_joined):
_list_recursive(
base,
src_path + [entry],
dest_path + [entry],
src_join_fun,
src_isdir_fun,
src_listdir_fun,
)
else:
files.append(
(
dest_path,
src_path_joined,
)
)
if src.startswith(":"):
src_dirname = [_remote_path_basename(src[1:])]
dest_dirname = src_dirname if dest_exists else []
_list_recursive(
_remote_path_dirname(src[1:]),
src_dirname,
dest_dirname,
src_join_fun=_remote_path_join,
src_isdir_fun=state.transport.fs_isdir,
src_listdir_fun=lambda p: [x.name for x in state.transport.fs_listdir(p)],
)
else:
src_dirname = [os.path.basename(src)]
dest_dirname = src_dirname if dest_exists else []
_list_recursive(
os.path.dirname(src),
src_dirname,
dest_dirname,
src_join_fun=os.path.join,
src_isdir_fun=os.path.isdir,
src_listdir_fun=os.listdir,
)
# If no directories were encountered then we must have just had a file.
if not dirs:
return do_filesystem_cp(state, src, dest, multiple, check_hash)
def _mkdir(a, *b):
try:
if a.startswith(":"):
state.transport.fs_mkdir(_remote_path_join(a[1:], *b))
else:
os.mkdir(os.path.join(a, *b))
except FileExistsError:
pass
# Create the destination if necessary.
if not dest_exists:
_mkdir(dest)
# Create all sub-directories relative to the destination.
for d in dirs:
_mkdir(dest, *d)
# Copy all files, in sorted order to help it be deterministic.
files.sort()
for dest_path_split, src_path_joined in files:
if src.startswith(":"):
src_path_joined = ":" + src_path_joined
if dest.startswith(":"):
dest_path_joined = ":" + _remote_path_join(dest[1:], *dest_path_split)
else:
dest_path_joined = os.path.join(dest, *dest_path_split)
do_filesystem_cp(state, src_path_joined, dest_path_joined, False, check_hash)
def do_filesystem_recursive_rm(state, path, args):
if state.transport.fs_isdir(path):
if state.transport.mounted:
r_cwd = state.transport.eval("os.getcwd()")
abs_path = os.path.normpath(
os.path.join(r_cwd, path) if not os.path.isabs(path) else path
)
if isinstance(state.transport, SerialTransport) and abs_path.startswith(
f"{SerialTransport.fs_hook_mount}/"
):
raise CommandError(
f"rm -r not permitted on {SerialTransport.fs_hook_mount} directory"
)
for entry in state.transport.fs_listdir(path):
do_filesystem_recursive_rm(state, _remote_path_join(path, entry.name), args)
if path:
try:
state.transport.fs_rmdir(path)
if args.verbose:
print(f"removed directory: '{path}'")
except OSError as e:
if e.errno != errno.EINVAL: # not vfs mountpoint
raise CommandError(
f"rm -r: cannot remove :{path} {os.strerror(e.errno) if e.errno else ''}"
) from e
if args.verbose:
print(f"skipped: '{path}' (vfs mountpoint)")
else:
state.transport.fs_rmfile(path)
if args.verbose:
print(f"removed: '{path}'")
def human_size(size, decimals=1):
for unit in ["B", "K", "M", "G", "T"]:
if size < 1024.0 or unit == "T":
break
size /= 1024.0
return f"{size:.{decimals}f}{unit}" if unit != "B" else f"{int(size)}"
def do_filesystem_tree(state, path, args):
"""Print a tree of the device's filesystem starting at path."""
connectors = ("├── ", "└── ")
def _tree_recursive(path, prefix=""):
entries = state.transport.fs_listdir(path)
entries.sort(key=lambda e: e.name)
for i, entry in enumerate(entries):
connector = connectors[1] if i == len(entries) - 1 else connectors[0]
is_dir = entry.st_mode & 0x4000 # Directory
size_str = ""
# most MicroPython filesystems don't support st_size on directories, reduce clutter
if entry.st_size > 0 or not is_dir:
if args.size:
size_str = f"[{entry.st_size:>9}] "
elif args.human:
size_str = f"[{human_size(entry.st_size):>6}] "
print(f"{prefix}{connector}{size_str}{entry.name}")
if is_dir:
_tree_recursive(
_remote_path_join(path, entry.name),
prefix + (" " if i == len(entries) - 1 else "│ "),
)
if not path or path == ".":
path = state.transport.exec("import os;print(os.getcwd())").strip().decode("utf-8")
if not (path == "." or state.transport.fs_isdir(path)):
raise CommandError(f"tree: '{path}' is not a directory")
if args.verbose:
print(f":{path} on {state.transport.device_name}")
else:
print(f":{path}")
_tree_recursive(path)
def do_filesystem(state, args):
state.ensure_raw_repl()
state.did_action()
command = args.command[0]
paths = args.path
if command == "cat":
# Don't do verbose output for `cat` unless explicitly requested.
verbose = args.verbose is True
else:
verbose = args.verbose is not False
if command == "cp":
# Note: cp requires the user to specify local/remote explicitly via
# leading ':'.
# The last argument must be the destination.
if len(paths) <= 1:
raise CommandError("cp: missing destination path")
cp_dest = paths[-1]
paths = paths[:-1]
else:
# All other commands implicitly use remote paths. Strip the
# leading ':' if the user included them.
paths = [path[1:] if path.startswith(":") else path for path in paths]
# ls and tree implicitly lists the cwd.
if command in ("ls", "tree") and not paths:
paths = [""]
try:
# Handle each path sequentially.
for path in paths:
if verbose:
if command == "cp":
print("{} {} {}".format(command, path, cp_dest))
else:
print("{} :{}".format(command, path))
if command == "cat":
state.transport.fs_printfile(path)
elif command == "ls":
for result in state.transport.fs_listdir(path):
print(
"{:12} {}{}".format(
result.st_size, result.name, "/" if result.st_mode & 0x4000 else ""
)
)
elif command == "mkdir":
state.transport.fs_mkdir(path)
elif command == "rm":
if args.recursive:
do_filesystem_recursive_rm(state, path, args)
else:
state.transport.fs_rmfile(path)
elif command == "rmdir":
state.transport.fs_rmdir(path)
elif command == "touch":
state.transport.fs_touchfile(path)
elif command.endswith("sum") and command[-4].isdigit():
digest = state.transport.fs_hashfile(path, command[:-3])
print(digest.hex())
elif command == "cp":
if args.recursive:
do_filesystem_recursive_cp(
state, path, cp_dest, len(paths) > 1, not args.force
)
else:
do_filesystem_cp(state, path, cp_dest, len(paths) > 1, not args.force)
elif command == "tree":
do_filesystem_tree(state, path, args)
except OSError as er:
raise CommandError("{}: {}: {}.".format(command, er.strerror, os.strerror(er.errno)))
except TransportError as er:
raise CommandError("Error with transport:\n{}".format(er.args[0]))
def do_edit(state, args):
state.ensure_raw_repl()
state.did_action()
if not os.getenv("EDITOR"):
raise CommandError("edit: $EDITOR not set")
for src in args.files:
src = src.lstrip(":")
dest_fd, dest = tempfile.mkstemp(suffix=os.path.basename(src))
try:
print("edit :%s" % (src,))
state.transport.fs_touchfile(src)
data = state.transport.fs_readfile(src, progress_callback=show_progress_bar)
with open(dest_fd, "wb") as f:
f.write(data)
if os.system('%s "%s"' % (os.getenv("EDITOR"), dest)) == 0:
with open(dest, "rb") as f:
state.transport.fs_writefile(
src, f.read(), progress_callback=show_progress_bar
)
finally:
os.unlink(dest)
def _do_execbuffer(state, buf, follow):
state.ensure_raw_repl()
state.did_action()
try:
state.transport.exec_raw_no_follow(buf)
if follow:
ret, ret_err = state.transport.follow(timeout=None, data_consumer=stdout_write_bytes)
if ret_err:
stdout_write_bytes(ret_err)
sys.exit(1)
except TransportError as er:
raise CommandError(er.args[0])
except KeyboardInterrupt:
sys.exit(1)
def do_exec(state, args):
_do_execbuffer(state, args.expr[0], args.follow)
def do_eval(state, args):
buf = "print(" + args.expr[0] + ")"
_do_execbuffer(state, buf, True)
def do_run(state, args):
filename = args.path[0]
try:
with open(filename, "rb") as f:
buf = f.read()
except OSError:
raise CommandError(f"could not read file '{filename}'")
_do_execbuffer(state, buf, args.follow)
def do_mount(state, args):
state.ensure_raw_repl()
path = args.path[0]
state.transport.mount_local(path, unsafe_links=args.unsafe_links)
print(f"Local directory {path} is mounted at /remote")
def do_umount(state, path):
state.ensure_raw_repl()
state.transport.umount_local()
def do_resume(state, _args=None):
state._auto_soft_reset = False
def do_soft_reset(state, _args=None):
state.ensure_raw_repl(soft_reset=True)
state.did_action()
def do_rtc(state, args):
state.ensure_raw_repl()
state.did_action()
state.transport.exec("import machine")
if args.set:
import datetime
now = datetime.datetime.now()
timetuple = "({}, {}, {}, {}, {}, {}, {}, {})".format(
now.year,
now.month,
now.day,
now.weekday(),
now.hour,
now.minute,
now.second,
now.microsecond,
)
state.transport.exec("machine.RTC().datetime({})".format(timetuple))
else:
print(state.transport.eval("machine.RTC().datetime()"))
def _do_romfs_query(state, args):
state.ensure_raw_repl()
state.did_action()
# Detect the romfs and get its associated device.
state.transport.exec("import vfs")
if not state.transport.eval("hasattr(vfs,'rom_ioctl')"):
print("ROMFS is not enabled on this device")
return
num_rom_partitions = state.transport.eval("vfs.rom_ioctl(1)")
if num_rom_partitions <= 0:
print("No ROMFS partitions available")
return
for rom_id in range(num_rom_partitions):
state.transport.exec(f"dev=vfs.rom_ioctl(2,{rom_id})")
has_object = state.transport.eval("hasattr(dev,'ioctl')")
if has_object:
rom_block_count = state.transport.eval("dev.ioctl(4,0)")
rom_block_size = state.transport.eval("dev.ioctl(5,0)")
rom_size = rom_block_count * rom_block_size
print(
f"ROMFS{rom_id} partition has size {rom_size} bytes ({rom_block_count} blocks of {rom_block_size} bytes each)"
)
else:
rom_size = state.transport.eval("len(dev)")
print(f"ROMFS{rom_id} partition has size {rom_size} bytes")
romfs = state.transport.eval("bytes(memoryview(dev)[:12])")
print(f" Raw contents: {romfs.hex(':')} ...")
if not romfs.startswith(b"\xd2\xcd\x31"):
print(" Not a valid ROMFS")
else:
size = 0
for value in romfs[3:]:
size = (size << 7) | (value & 0x7F)
if not value & 0x80:
break
print(f" ROMFS image size: {size}")
def _do_romfs_build(state, args):
state.did_action()
if args.path is None:
raise CommandError("romfs build: source path not given")
input_directory = args.path
if args.output is None:
output_file = input_directory + ".romfs"
else:
output_file = args.output
romfs = make_romfs(input_directory, mpy_cross=args.mpy)
print(f"Writing {len(romfs)} bytes to output file {output_file}")
with open(output_file, "wb") as f:
f.write(romfs)
def _do_romfs_deploy(state, args):
state.ensure_raw_repl()
state.did_action()
transport = state.transport
if args.path is None:
raise CommandError("romfs deploy: source path not given")
rom_id = args.partition
romfs_filename = args.path
# Read in or create the ROMFS filesystem image.
if os.path.isfile(romfs_filename) and romfs_filename.endswith((".img", ".romfs")):
with open(romfs_filename, "rb") as f:
romfs = f.read()
else:
romfs = make_romfs(romfs_filename, mpy_cross=args.mpy)
print(f"Image size is {len(romfs)} bytes")
# Detect the ROMFS partition and get its associated device.
state.transport.exec("import vfs")
if not state.transport.eval("hasattr(vfs,'rom_ioctl')"):
raise CommandError("ROMFS is not enabled on this device")
transport.exec(f"dev=vfs.rom_ioctl(2,{rom_id})")
if transport.eval("isinstance(dev,int) and dev<0"):
raise CommandError(f"ROMFS{rom_id} partition not found on device")
has_object = transport.eval("hasattr(dev,'ioctl')")
if has_object:
rom_block_count = transport.eval("dev.ioctl(4,0)")
rom_block_size = transport.eval("dev.ioctl(5,0)")
rom_size = rom_block_count * rom_block_size
print(
f"ROMFS{rom_id} partition has size {rom_size} bytes ({rom_block_count} blocks of {rom_block_size} bytes each)"
)
else:
rom_size = transport.eval("len(dev)")
print(f"ROMFS{rom_id} partition has size {rom_size} bytes")
# Check if ROMFS image is valid
if not romfs.startswith(VfsRomWriter.ROMFS_HEADER):
print("Invalid ROMFS image")
sys.exit(1)
# Check if ROMFS filesystem image will fit in the target partition.
if len(romfs) > rom_size:
print("ROMFS image is too big for the target partition")
sys.exit(1)
# Prepare ROMFS partition for writing.
print(f"Preparing ROMFS{rom_id} partition for writing")
transport.exec("import vfs\ntry:\n vfs.umount('/rom')\nexcept:\n pass")
chunk_size = 4096
if has_object:
for offset in range(0, len(romfs), rom_block_size):
transport.exec(f"dev.ioctl(6,{offset // rom_block_size})")
chunk_size = min(chunk_size, rom_block_size)
else:
rom_min_write = transport.eval(f"vfs.rom_ioctl(3,{rom_id},{len(romfs)})")
chunk_size = max(chunk_size, rom_min_write)
# Detect capabilities of the device to use the fastest method of transfer.
has_bytes_fromhex = transport.eval("hasattr(bytes,'fromhex')")
try:
transport.exec("from binascii import a2b_base64")
has_a2b_base64 = True
except TransportExecError:
has_a2b_base64 = False
try:
transport.exec("from io import BytesIO")
transport.exec("from deflate import DeflateIO,RAW")
has_deflate_io = True
except TransportExecError:
has_deflate_io = False
# Deploy the ROMFS filesystem image to the device.
for offset in range(0, len(romfs), chunk_size):
romfs_chunk = romfs[offset : offset + chunk_size]
romfs_chunk += bytes(chunk_size - len(romfs_chunk))
if has_deflate_io:
# Needs: binascii.a2b_base64, io.BytesIO, deflate.DeflateIO.
compressor = zlib.compressobj(wbits=-9)
romfs_chunk_compressed = compressor.compress(romfs_chunk)
romfs_chunk_compressed += compressor.flush()
buf = binascii.b2a_base64(romfs_chunk_compressed).strip()
transport.exec(f"buf=DeflateIO(BytesIO(a2b_base64({buf})),RAW,9).read()")
elif has_a2b_base64:
# Needs: binascii.a2b_base64.
buf = binascii.b2a_base64(romfs_chunk)
transport.exec(f"buf=a2b_base64({buf})")
elif has_bytes_fromhex:
# Needs: bytes.fromhex.
buf = romfs_chunk.hex()
transport.exec(f"buf=bytes.fromhex('{buf}')")
else:
# Needs nothing special.
transport.exec("buf=" + repr(romfs_chunk))
print(f"\rWriting at offset {offset}", end="")
if has_object:
transport.exec(
f"dev.writeblocks({offset // rom_block_size},buf,{offset % rom_block_size})"
)
else:
transport.exec(f"vfs.rom_ioctl(4,{rom_id},{offset},buf)")
# Complete writing.
if not has_object:
transport.eval(f"vfs.rom_ioctl(5,{rom_id})")
print()
print("ROMFS image deployed")
def do_romfs(state, args):
if args.command[0] == "query":
_do_romfs_query(state, args)
elif args.command[0] == "build":
_do_romfs_build(state, args)
elif args.command[0] == "deploy":
_do_romfs_deploy(state, args)
else:
raise CommandError(
f"romfs: '{args.command[0]}' is not a command; pass romfs --help for a list"
)

176
.venv/lib/python3.12/site-packages/mpremote/console.py Normal file
View File

@@ -0,0 +1,176 @@
import sys, time
try:
import select, termios
except ImportError:
termios = None
select = None
import msvcrt, signal
class ConsolePosix:
def __init__(self):
self.infd = sys.stdin.fileno()
self.infile = sys.stdin.buffer
self.outfile = sys.stdout.buffer
if hasattr(self.infile, "raw"):
self.infile = self.infile.raw
if hasattr(self.outfile, "raw"):
self.outfile = self.outfile.raw
self.orig_attr = termios.tcgetattr(self.infd)
def enter(self):
# attr is: [iflag, oflag, cflag, lflag, ispeed, ospeed, cc]
attr = termios.tcgetattr(self.infd)
attr[0] &= ~(
termios.BRKINT | termios.ICRNL | termios.INPCK | termios.ISTRIP | termios.IXON
)
attr[1] = 0
attr[2] = attr[2] & ~(termios.CSIZE | termios.PARENB) | termios.CS8
attr[3] = 0
attr[6][termios.VMIN] = 1
attr[6][termios.VTIME] = 0
termios.tcsetattr(self.infd, termios.TCSANOW, attr)
def exit(self):
termios.tcsetattr(self.infd, termios.TCSANOW, self.orig_attr)
def waitchar(self, pyb_serial):
# TODO pyb_serial might not have fd
select.select([self.infd, pyb_serial.fd], [], [])
def readchar(self):
res = select.select([self.infd], [], [], 0)
if res[0]:
return self.infile.read(1)
else:
return None
def write(self, buf):
self.outfile.write(buf)
class ConsoleWindows:
KEY_MAP = {
b"H": b"A", # UP
b"P": b"B", # DOWN
b"M": b"C", # RIGHT
b"K": b"D", # LEFT
b"G": b"H", # POS1
b"O": b"F", # END
b"Q": b"6~", # PGDN
b"I": b"5~", # PGUP
b"s": b"1;5D", # CTRL-LEFT,
b"t": b"1;5C", # CTRL-RIGHT,
b"\x8d": b"1;5A", # CTRL-UP,
b"\x91": b"1;5B", # CTRL-DOWN,
b"w": b"1;5H", # CTRL-POS1
b"u": b"1;5F", # CTRL-END
b"\x98": b"1;3A", # ALT-UP,
b"\xa0": b"1;3B", # ALT-DOWN,
b"\x9d": b"1;3C", # ALT-RIGHT,
b"\x9b": b"1;3D", # ALT-LEFT,
b"\x97": b"1;3H", # ALT-POS1,
b"\x9f": b"1;3F", # ALT-END,
b"S": b"3~", # DEL,
b"\x93": b"3;5~", # CTRL-DEL
b"R": b"2~", # INS
b"\x92": b"2;5~", # CTRL-INS
b"\x94": b"Z", # Ctrl-Tab = BACKTAB,
}
def __init__(self):
self.ctrl_c = 0
def _sigint_handler(self, signo, frame):
self.ctrl_c += 1
def enter(self):
signal.signal(signal.SIGINT, self._sigint_handler)
def exit(self):
signal.signal(signal.SIGINT, signal.SIG_DFL)
def inWaiting(self):
return 1 if self.ctrl_c or msvcrt.kbhit() else 0
def waitchar(self, pyb_serial):
while not (self.inWaiting() or pyb_serial.inWaiting()):
time.sleep(0.01)
def readchar(self):
if self.ctrl_c:
self.ctrl_c -= 1
return b"\x03"
if msvcrt.kbhit():
ch = msvcrt.getch()
while ch in b"\x00\xe0": # arrow or function key prefix?
if not msvcrt.kbhit():
return None
ch = msvcrt.getch() # second call returns the actual key code
try:
ch = b"\x1b[" + self.KEY_MAP[ch]
except KeyError:
return None
return ch
def write(self, buf):
buf = buf.decode() if isinstance(buf, bytes) else buf
sys.stdout.write(buf)
sys.stdout.flush()
# for b in buf:
# if isinstance(b, bytes):
# msvcrt.putch(b)
# else:
# msvcrt.putwch(b)
if termios:
Console = ConsolePosix
VT_ENABLED = True
else:
Console = ConsoleWindows
# Windows VT mode ( >= win10 only)
# https://bugs.python.org/msg291732
import ctypes, os
from ctypes import wintypes
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
ERROR_INVALID_PARAMETER = 0x0057
ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004
def _check_bool(result, func, args):
if not result:
raise ctypes.WinError(ctypes.get_last_error())
return args
LPDWORD = ctypes.POINTER(wintypes.DWORD)
kernel32.GetConsoleMode.errcheck = _check_bool
kernel32.GetConsoleMode.argtypes = (wintypes.HANDLE, LPDWORD)
kernel32.SetConsoleMode.errcheck = _check_bool
kernel32.SetConsoleMode.argtypes = (wintypes.HANDLE, wintypes.DWORD)
def set_conout_mode(new_mode, mask=0xFFFFFFFF):
# don't assume StandardOutput is a console.
# open CONOUT$ instead
fdout = os.open("CONOUT$", os.O_RDWR)
try:
hout = msvcrt.get_osfhandle(fdout)
old_mode = wintypes.DWORD()
kernel32.GetConsoleMode(hout, ctypes.byref(old_mode))
mode = (new_mode & mask) | (old_mode.value & ~mask)
kernel32.SetConsoleMode(hout, mode)
return old_mode.value
finally:
os.close(fdout)
# def enable_vt_mode():
mode = mask = ENABLE_VIRTUAL_TERMINAL_PROCESSING
try:
set_conout_mode(mode, mask)
VT_ENABLED = True
except WindowsError:
VT_ENABLED = False

636
.venv/lib/python3.12/site-packages/mpremote/main.py Normal file
View File

@@ -0,0 +1,636 @@
"""
MicroPython Remote - Interaction and automation tool for MicroPython
MIT license; Copyright (c) 2019-2022 Damien P. George
This program provides a set of utilities to interact with and automate a
MicroPython device over a serial connection. Commands supported are:
mpremote -- auto-detect, connect and enter REPL
mpremote <device-shortcut> -- connect to given device
mpremote connect <device> -- connect to given device
mpremote disconnect -- disconnect current device
mpremote mount <local-dir> -- mount local directory on device
mpremote eval <string> -- evaluate and print the string
mpremote exec <string> -- execute the string
mpremote run <script> -- run the given local script
mpremote fs <command> <args...> -- execute filesystem commands on the device
mpremote repl -- enter REPL
"""
import argparse
import os, sys, time
from collections.abc import Mapping
from textwrap import dedent
import platformdirs
from .commands import (
CommandError,
do_connect,
do_disconnect,
do_edit,
do_filesystem,
do_mount,
do_umount,
do_exec,
do_eval,
do_run,
do_resume,
do_rtc,
do_soft_reset,
do_romfs,
)
from .mip import do_mip
from .repl import do_repl
_PROG = "mpremote"
def do_sleep(state, args):
time.sleep(args.ms[0])
def do_help(state, _args=None):
def print_commands_help(cmds, help_key):
max_command_len = max(len(cmd) for cmd in cmds.keys())
for cmd in sorted(cmds.keys()):
help_message_lines = dedent(help_key(cmds[cmd])).split("\n")
help_message = help_message_lines[0]
for line in help_message_lines[1:]:
help_message = "{}\n{}{}".format(help_message, " " * (max_command_len + 4), line)
print(" ", cmd, " " * (max_command_len - len(cmd) + 2), help_message, sep="")
print(_PROG, "-- MicroPython remote control")
print("See https://docs.micropython.org/en/latest/reference/mpremote.html")
print("\nList of commands:")
print_commands_help(
_COMMANDS, lambda x: x[1]().description
) # extract description from argparse
print("\nList of shortcuts:")
print_commands_help(_command_expansions, lambda x: x[2]) # (args, sub, help_message)
sys.exit(0)
def do_version(state, _args=None):
from . import __version__
print(f"{_PROG} {__version__}")
sys.exit(0)
def _bool_flag(cmd_parser, name, short_name, default, description):
# In Python 3.9+ this can be replaced with argparse.BooleanOptionalAction.
group = cmd_parser.add_mutually_exclusive_group()
group.add_argument(
"--" + name,
"-" + short_name,
action="store_true",
default=default,
help=description,
)
group.add_argument(
"--no-" + name,
action="store_false",
dest=name,
)
def argparse_connect():
cmd_parser = argparse.ArgumentParser(description="connect to given device")
cmd_parser.add_argument(
"device", nargs=1, help="Either list, auto, id:x, port:x, or any valid device name/path"
)
return cmd_parser
def argparse_sleep():
cmd_parser = argparse.ArgumentParser(description="sleep before executing next command")
cmd_parser.add_argument("ms", nargs=1, type=float, help="milliseconds to sleep for")
return cmd_parser
def argparse_edit():
cmd_parser = argparse.ArgumentParser(description="edit files on the device")
cmd_parser.add_argument("files", nargs="+", help="list of remote paths")
return cmd_parser
def argparse_mount():
cmd_parser = argparse.ArgumentParser(description="mount local directory on device")
_bool_flag(
cmd_parser,
"unsafe-links",
"l",
False,
"follow symbolic links pointing outside of local directory",
)
cmd_parser.add_argument("path", nargs=1, help="local path to mount")
return cmd_parser
def argparse_repl():
cmd_parser = argparse.ArgumentParser(description="connect to given device")
_bool_flag(cmd_parser, "escape-non-printable", "e", False, "escape non-printable characters")
cmd_parser.add_argument(
"--capture",
type=str,
required=False,
help="saves a copy of the REPL session to the specified path",
)
cmd_parser.add_argument(
"--inject-code", type=str, required=False, help="code to be run when Ctrl-J is pressed"
)
cmd_parser.add_argument(
"--inject-file",
type=str,
required=False,
help="path to file to be run when Ctrl-K is pressed",
)
return cmd_parser
def argparse_eval():
cmd_parser = argparse.ArgumentParser(description="evaluate and print the string")
cmd_parser.add_argument("expr", nargs=1, help="expression to execute")
return cmd_parser
def argparse_exec():
cmd_parser = argparse.ArgumentParser(description="execute the string")
_bool_flag(
cmd_parser, "follow", "f", True, "follow output until the expression completes (default)"
)
cmd_parser.add_argument("expr", nargs=1, help="expression to execute")
return cmd_parser
def argparse_run():
cmd_parser = argparse.ArgumentParser(description="run the given local script")
_bool_flag(
cmd_parser, "follow", "f", True, "follow output until the script completes (default)"
)
cmd_parser.add_argument("path", nargs=1, help="path to script to execute")
return cmd_parser
def argparse_rtc():
cmd_parser = argparse.ArgumentParser(description="get (default) or set the device RTC")
_bool_flag(cmd_parser, "set", "s", False, "set the RTC to the current local time")
return cmd_parser
def argparse_filesystem():
cmd_parser = argparse.ArgumentParser(
description="execute filesystem commands on the device",
add_help=False,
)
cmd_parser.add_argument("--help", action="help", help="show this help message and exit")
_bool_flag(cmd_parser, "recursive", "r", False, "recursive (for cp and rm commands)")
_bool_flag(
cmd_parser,
"force",
"f",
False,
"force copy even if file is unchanged (for cp command only)",
)
_bool_flag(
cmd_parser,
"verbose",
"v",
None,
"enable verbose output (defaults to True for all commands except cat)",
)
size_group = cmd_parser.add_mutually_exclusive_group()
size_group.add_argument(
"--size",
"-s",
default=False,
action="store_true",
help="show file size in bytes(tree command only)",
)
size_group.add_argument(
"--human",
"-h",
default=False,
action="store_true",
help="show file size in a more human readable way (tree command only)",
)
cmd_parser.add_argument(
"command",
nargs=1,
help="filesystem command (e.g. cat, cp, sha256sum, ls, rm, rmdir, touch, tree)",
)
cmd_parser.add_argument("path", nargs="+", help="local and remote paths")
return cmd_parser
def argparse_mip():
cmd_parser = argparse.ArgumentParser(
description="install packages from micropython-lib or third-party sources"
)
_bool_flag(cmd_parser, "mpy", "m", True, "download as compiled .mpy files (default)")
cmd_parser.add_argument(
"--target", type=str, required=False, help="destination direction on the device"
)
cmd_parser.add_argument(
"--index",
type=str,
required=False,
help="package index to use (defaults to micropython-lib)",
)
cmd_parser.add_argument("command", nargs=1, help="mip command (e.g. install)")
cmd_parser.add_argument(
"packages",
nargs="+",
help="list package specifications, e.g. name, name@version, github:org/repo, github:org/repo@branch, gitlab:org/repo, gitlab:org/repo@branch",
)
return cmd_parser
def argparse_romfs():
cmd_parser = argparse.ArgumentParser(description="manage ROM partitions")
_bool_flag(
cmd_parser,
"mpy",
"m",
True,
"automatically compile .py files to .mpy when building the ROMFS image (default)",
)
cmd_parser.add_argument(
"--partition",
"-p",
type=int,
default=0,
help="ROMFS partition to use",
)
cmd_parser.add_argument(
"--output",
"-o",
help="output file",
)
cmd_parser.add_argument("command", nargs=1, help="romfs command, one of: query, build, deploy")
cmd_parser.add_argument("path", nargs="?", help="path to directory to deploy")
return cmd_parser
def argparse_none(description):
return lambda: argparse.ArgumentParser(description=description)
# Map of "command" to tuple of (handler_func, argparse_func).
_COMMANDS = {
"connect": (
do_connect,
argparse_connect,
),
"sleep": (
do_sleep,
argparse_sleep,
),
"disconnect": (
do_disconnect,
argparse_none("disconnect current device"),
),
"edit": (
do_edit,
argparse_edit,
),
"resume": (
do_resume,
argparse_none("resume a previous mpremote session (will not auto soft-reset)"),
),
"soft-reset": (
do_soft_reset,
argparse_none("perform a soft-reset of the device"),
),
"mount": (
do_mount,
argparse_mount,
),
"umount": (
do_umount,
argparse_none("unmount the local directory"),
),
"repl": (
do_repl,
argparse_repl,
),
"eval": (
do_eval,
argparse_eval,
),
"exec": (
do_exec,
argparse_exec,
),
"run": (
do_run,
argparse_run,
),
"rtc": (
do_rtc,
argparse_rtc,
),
"fs": (
do_filesystem,
argparse_filesystem,
),
"mip": (
do_mip,
argparse_mip,
),
"help": (
do_help,
argparse_none("print help and exit"),
),
"version": (
do_version,
argparse_none("print version and exit"),
),
"romfs": (
do_romfs,
argparse_romfs,
),
}
# Additional commands aliases.
# The value can either be:
# - A command string.
# - A list of command strings, each command will be executed sequentially.
# - A dict of command: { [], help: ""}
_BUILTIN_COMMAND_EXPANSIONS = {
# Device connection shortcuts.
"devs": {
"command": "connect list",
"help": "list available serial ports",
},
# Filesystem shortcuts (use `cp` instead of `fs cp`).
"cat": "fs cat",
"cp": "fs cp",
"ls": "fs ls",
"mkdir": "fs mkdir",
"rm": "fs rm",
"rmdir": "fs rmdir",
"sha256sum": "fs sha256sum",
"touch": "fs touch",
"tree": "fs tree",
# Disk used/free.
"df": [
"exec",
"""
import os,vfs
_f = "{:<10}{:>9}{:>9}{:>9}{:>5} {}"
print(_f.format("filesystem", "size", "used", "avail", "use%", "mounted on"))
try:
_ms = vfs.mount()
except:
_ms = []
for _m in [""] + os.listdir("/"):
_m = "/" + _m
_s = os.stat(_m)
if _s[0] & 1 << 14:
_ms.append(("<unknown>",_m))
for _v,_p in _ms:
_s = os.statvfs(_p)
_sz = _s[0]*_s[2]
if _sz:
_av = _s[0]*_s[3]
_us = 100*(_sz-_av)//_sz
print(_f.format(str(_v), _sz, _sz-_av, _av, _us, _p))
""",
],
# Other shortcuts.
"reset": {
"command": [
"exec",
"--no-follow",
"import time, machine; time.sleep_ms(100); machine.reset()",
],
"help": "hard reset the device",
},
"bootloader": {
"command": [
"exec",
"--no-follow",
"import time, machine; time.sleep_ms(100); machine.bootloader()",
],
"help": "make the device enter its bootloader",
},
# Simple aliases.
"--help": "help",
"--version": "version",
}
# Add "a0", "a1", ..., "u0", "u1", ..., "c0", "c1", ... as aliases
# for "connect /dev/ttyACMn" (and /dev/ttyUSBn, COMn) etc.
for port_num in range(4):
for prefix, port in [("a", "/dev/ttyACM"), ("u", "/dev/ttyUSB"), ("c", "COM")]:
_BUILTIN_COMMAND_EXPANSIONS["{}{}".format(prefix, port_num)] = {
"command": "connect {}{}".format(port, port_num),
"help": 'connect to serial port "{}{}"'.format(port, port_num),
}
def load_user_config():
# Create empty config object.
config = __build_class__(lambda: None, "Config")()
config.commands = {}
# Get config file name.
path = platformdirs.user_config_dir(appname=_PROG, appauthor=False)
config_file = os.path.join(path, "config.py")
# Check if config file exists.
if not os.path.exists(config_file):
return config
# Exec the config file in its directory.
with open(config_file) as f:
config_data = f.read()
prev_cwd = os.getcwd()
os.chdir(path)
# Pass in the config path so that the config file can use it.
config.__dict__["config_path"] = path
config.__dict__["__file__"] = config_file
exec(config_data, config.__dict__)
os.chdir(prev_cwd)
return config
def prepare_command_expansions(config):
global _command_expansions
_command_expansions = {}
for command_set in (_BUILTIN_COMMAND_EXPANSIONS, config.commands):
for cmd, sub in command_set.items():
cmd = cmd.split()
if len(cmd) == 1:
args = ()
else:
args = tuple(c.split("=") for c in cmd[1:])
help_message = ""
if isinstance(sub, Mapping):
help_message = sub.get("help", "")
sub = sub["command"]
if isinstance(sub, str):
sub = sub.split()
_command_expansions[cmd[0]] = (args, sub, help_message)
def do_command_expansion(args):
def usage_error(cmd, exp_args, msg):
print(f"Command {cmd} {msg}; signature is:")
print(" ", cmd, " ".join("=".join(a) for a in exp_args))
sys.exit(1)
last_arg_idx = len(args)
pre = []
while args and args[0] in _command_expansions:
cmd = args.pop(0)
exp_args, exp_sub, _ = _command_expansions[cmd]
for exp_arg in exp_args:
if args and args[0] == "+":
break
exp_arg_name = exp_arg[0]
if args and "=" not in args[0]:
# Argument given without a name.
value = args.pop(0)
elif args and args[0].startswith(exp_arg_name + "="):
# Argument given with correct name.
value = args.pop(0).split("=", 1)[1]
else:
# No argument given, or argument given with a different name.
if len(exp_arg) == 1:
# Required argument (it has no default).
usage_error(cmd, exp_args, f"missing argument {exp_arg_name}")
else:
# Optional argument with a default.
value = exp_arg[1]
pre.append(f"{exp_arg_name}={value}")
args[0:0] = exp_sub
last_arg_idx = len(exp_sub)
if last_arg_idx < len(args) and "=" in args[last_arg_idx]:
# Extra unknown arguments given.
arg = args[last_arg_idx].split("=", 1)[0]
usage_error(cmd, exp_args, f"given unexpected argument {arg}")
# Insert expansion with optional setting of arguments.
if pre:
args[0:0] = ["exec", ";".join(pre)]
class State:
def __init__(self):
self.transport = None
self._did_action = False
self._auto_soft_reset = True
def did_action(self):
self._did_action = True
def run_repl_on_completion(self):
return not self._did_action
def ensure_connected(self):
if self.transport is None:
do_connect(self)
def ensure_raw_repl(self, soft_reset=None):
self.ensure_connected()
soft_reset = self._auto_soft_reset if soft_reset is None else soft_reset
if soft_reset or not self.transport.in_raw_repl:
self.transport.enter_raw_repl(soft_reset=soft_reset)
self._auto_soft_reset = False
def ensure_friendly_repl(self):
self.ensure_connected()
if self.transport.in_raw_repl:
self.transport.exit_raw_repl()
def main():
config = load_user_config()
prepare_command_expansions(config)
remaining_args = sys.argv[1:]
state = State()
try:
while remaining_args:
# Skip the terminator.
if remaining_args[0] == "+":
remaining_args.pop(0)
continue
# Rewrite the front of the list with any matching expansion.
do_command_expansion(remaining_args)
# The (potentially rewritten) command must now be a base command.
cmd = remaining_args.pop(0)
try:
handler_func, parser_func = _COMMANDS[cmd]
except KeyError:
raise CommandError(f"'{cmd}' is not a command")
# If this command (or any down the chain) has a terminator, then
# limit the arguments passed for this command. They will be added
# back after processing this command.
try:
terminator = remaining_args.index("+")
command_args = remaining_args[:terminator]
extra_args = remaining_args[terminator:]
except ValueError:
command_args = remaining_args
extra_args = []
# Special case: "fs ls" and "fs tree" can have only options and no path specified.
if (
cmd == "fs"
and len(command_args) >= 1
and command_args[0] in ("ls", "tree")
and sum(1 for a in command_args if not a.startswith("-")) == 1
):
command_args.append("")
# Use the command-specific argument parser.
cmd_parser = parser_func()
cmd_parser.prog = cmd
# Catch all for unhandled positional arguments (this is the next command).
cmd_parser.add_argument(
"next_command", nargs=argparse.REMAINDER, help=f"Next {_PROG} command"
)
args = cmd_parser.parse_args(command_args)
# Execute command.
handler_func(state, args)
# Get any leftover unprocessed args.
remaining_args = args.next_command + extra_args
# If no commands were "actions" then implicitly finish with the REPL
# using default args.
if state.run_repl_on_completion():
disconnected = do_repl(state, argparse_repl().parse_args([]))
# Handle disconnection message
if disconnected:
print("\ndevice disconnected")
return 0
except CommandError as e:
# Make sure existing stdout appears before the error message on stderr.
sys.stdout.flush()
print(f"{_PROG}: {e}", file=sys.stderr)
sys.stderr.flush()
return 1
finally:
do_disconnect(state)

219
.venv/lib/python3.12/site-packages/mpremote/mip.py Normal file
View File

@@ -0,0 +1,219 @@
# Micropython package installer
# Ported from micropython-lib/micropython/mip/mip.py.
# MIT license; Copyright (c) 2022 Jim Mussared
import urllib.error
import urllib.request
import json
import tempfile
import os
import os.path
from .commands import CommandError, show_progress_bar
_PACKAGE_INDEX = "https://micropython.org/pi/v2"
allowed_mip_url_prefixes = ("http://", "https://", "github:", "gitlab:")
# This implements os.makedirs(os.dirname(path))
def _ensure_path_exists(transport, path):
split = path.split("/")
# Handle paths starting with "/".
if not split[0]:
split.pop(0)
split[0] = "/" + split[0]
prefix = ""
for i in range(len(split) - 1):
prefix += split[i]
if not transport.fs_exists(prefix):
transport.fs_mkdir(prefix)
prefix += "/"
# Check if the specified path exists and matches the hash.
def _check_exists(transport, path, short_hash):
try:
remote_hash = transport.fs_hashfile(path, "sha256")
except FileNotFoundError:
return False
return remote_hash.hex()[: len(short_hash)] == short_hash
def _rewrite_url(url, branch=None):
if not branch:
branch = "HEAD"
if url.startswith("github:"):
url = url[7:].split("/")
url = (
"https://raw.githubusercontent.com/"
+ url[0]
+ "/"
+ url[1]
+ "/"
+ branch
+ "/"
+ "/".join(url[2:])
)
elif url.startswith("gitlab:"):
url = url[7:].split("/")
url = (
"https://gitlab.com/"
+ url[0]
+ "/"
+ url[1]
+ "/-/raw/"
+ branch
+ "/"
+ "/".join(url[2:])
)
return url
def _download_file(transport, url, dest):
if url.startswith(allowed_mip_url_prefixes):
try:
with urllib.request.urlopen(url) as src:
data = src.read()
except urllib.error.HTTPError as e:
if e.status == 404:
raise CommandError(f"File not found: {url}")
else:
raise CommandError(f"Error {e.status} requesting {url}")
except urllib.error.URLError as e:
raise CommandError(f"{e.reason} requesting {url}")
else:
if "\\" in url:
raise CommandError(f'Use "/" instead of "\\" in file URLs: {url!r}\n')
try:
with open(url, "rb") as f:
data = f.read()
except OSError as e:
raise CommandError(f"{e.strerror} opening {url}")
print("Installing:", dest)
_ensure_path_exists(transport, dest)
transport.fs_writefile(dest, data, progress_callback=show_progress_bar)
def _install_json(transport, package_json_url, index, target, version, mpy):
base_url = ""
if package_json_url.startswith(allowed_mip_url_prefixes):
try:
with urllib.request.urlopen(_rewrite_url(package_json_url, version)) as response:
package_json = json.load(response)
except urllib.error.HTTPError as e:
if e.status == 404:
raise CommandError(f"Package not found: {package_json_url}")
else:
raise CommandError(f"Error {e.status} requesting {package_json_url}")
except urllib.error.URLError as e:
raise CommandError(f"{e.reason} requesting {package_json_url}")
base_url = package_json_url.rpartition("/")[0]
elif package_json_url.endswith(".json"):
try:
with open(package_json_url, "r") as f:
package_json = json.load(f)
except OSError:
raise CommandError(f"Error opening {package_json_url}")
base_url = os.path.dirname(package_json_url)
else:
raise CommandError(f"Invalid url for package: {package_json_url}")
for target_path, short_hash in package_json.get("hashes", ()):
fs_target_path = target + "/" + target_path
if _check_exists(transport, fs_target_path, short_hash):
print("Exists:", fs_target_path)
else:
file_url = f"{index}/file/{short_hash[:2]}/{short_hash}"
_download_file(transport, file_url, fs_target_path)
for target_path, url in package_json.get("urls", ()):
fs_target_path = target + "/" + target_path
if base_url and not url.startswith(allowed_mip_url_prefixes):
url = f"{base_url}/{url}" # Relative URLs
_download_file(transport, _rewrite_url(url, version), fs_target_path)
for dep, dep_version in package_json.get("deps", ()):
_install_package(transport, dep, index, target, dep_version, mpy)
def _install_package(transport, package, index, target, version, mpy):
if package.startswith(allowed_mip_url_prefixes):
if package.endswith(".py") or package.endswith(".mpy"):
print(f"Downloading {package} to {target}")
_download_file(
transport, _rewrite_url(package, version), target + "/" + package.rsplit("/")[-1]
)
return
else:
if not package.endswith(".json"):
if not package.endswith("/"):
package += "/"
package += "package.json"
print(f"Installing {package} to {target}")
elif package.endswith(".json"):
pass
else:
if not version:
version = "latest"
print(f"Installing {package} ({version}) from {index} to {target}")
mpy_version = "py"
if mpy:
transport.exec("import sys")
mpy_version = transport.eval("getattr(sys.implementation, '_mpy', 0) & 0xFF") or "py"
package = f"{index}/package/{mpy_version}/{package}/{version}.json"
_install_json(transport, package, index, target, version, mpy)
def do_mip(state, args):
state.did_action()
if args.command[0] == "install":
state.ensure_raw_repl()
for package in args.packages:
version = None
if "@" in package:
package, version = package.split("@")
print("Install", package)
if args.index is None:
args.index = _PACKAGE_INDEX
if args.target is None:
state.transport.exec("import sys")
lib_paths = [
p
for p in state.transport.eval("sys.path")
if not p.startswith("/rom") and p.endswith("/lib")
]
if lib_paths and lib_paths[0]:
args.target = lib_paths[0]
else:
raise CommandError(
"Unable to find lib dir in sys.path, use --target to override"
)
if args.mpy is None:
args.mpy = True
try:
_install_package(
state.transport,
package,
args.index.rstrip("/"),
args.target,
version,
args.mpy,
)
except CommandError:
print("Package may be partially installed")
raise
print("Done")
else:
raise CommandError(f"mip: '{args.command[0]}' is not a command")

55
.venv/lib/python3.12/site-packages/mpremote/mp_errno.py Normal file
View File

@@ -0,0 +1,55 @@
import errno
import platform
# This table maps numeric values defined by `py/mperrno.h` to host errno code.
MP_ERRNO_TABLE = {
1: errno.EPERM,
2: errno.ENOENT,
3: errno.ESRCH,
4: errno.EINTR,
5: errno.EIO,
6: errno.ENXIO,
7: errno.E2BIG,
8: errno.ENOEXEC,
9: errno.EBADF,
10: errno.ECHILD,
11: errno.EAGAIN,
12: errno.ENOMEM,
13: errno.EACCES,
14: errno.EFAULT,
16: errno.EBUSY,
17: errno.EEXIST,
18: errno.EXDEV,
19: errno.ENODEV,
20: errno.ENOTDIR,
21: errno.EISDIR,
22: errno.EINVAL,
23: errno.ENFILE,
24: errno.EMFILE,
25: errno.ENOTTY,
26: errno.ETXTBSY,
27: errno.EFBIG,
28: errno.ENOSPC,
29: errno.ESPIPE,
30: errno.EROFS,
31: errno.EMLINK,
32: errno.EPIPE,
33: errno.EDOM,
34: errno.ERANGE,
95: errno.EOPNOTSUPP,
97: errno.EAFNOSUPPORT,
98: errno.EADDRINUSE,
103: errno.ECONNABORTED,
104: errno.ECONNRESET,
105: errno.ENOBUFS,
106: errno.EISCONN,
107: errno.ENOTCONN,
110: errno.ETIMEDOUT,
111: errno.ECONNREFUSED,
113: errno.EHOSTUNREACH,
114: errno.EALREADY,
115: errno.EINPROGRESS,
125: errno.ECANCELED,
}
if platform.system() != "Windows":
MP_ERRNO_TABLE[15] = errno.ENOTBLK

121
.venv/lib/python3.12/site-packages/mpremote/repl.py Normal file
View File

@@ -0,0 +1,121 @@
from .console import Console, ConsolePosix
from .transport import TransportError
def do_repl_main_loop(
state, console_in, console_out_write, *, escape_non_printable, code_to_inject, file_to_inject
):
while True:
try:
console_in.waitchar(state.transport.serial)
c = console_in.readchar()
if c:
if c in (b"\x1d", b"\x18"): # ctrl-] or ctrl-x, quit
break
elif c == b"\x04": # ctrl-D
# special handling needed for ctrl-D if filesystem is mounted
state.transport.write_ctrl_d(console_out_write)
elif c == b"\x0a" and code_to_inject is not None: # ctrl-j, inject code
state.transport.serial.write(code_to_inject)
elif c == b"\x0b" and file_to_inject is not None: # ctrl-k, inject script
console_out_write(bytes("Injecting %s\r\n" % file_to_inject, "utf8"))
state.transport.enter_raw_repl(soft_reset=False)
with open(file_to_inject, "rb") as f:
pyfile = f.read()
try:
state.transport.exec_raw_no_follow(pyfile)
except TransportError as er:
console_out_write(b"Error:\r\n")
console_out_write(er)
state.transport.exit_raw_repl()
else:
state.transport.serial.write(c)
n = state.transport.serial.inWaiting()
if n > 0:
dev_data_in = state.transport.serial.read(n)
if dev_data_in is not None:
if escape_non_printable:
# Pass data through to the console, with escaping of non-printables.
console_data_out = bytearray()
for c in dev_data_in:
if c in (8, 9, 10, 13, 27) or 32 <= c <= 126:
console_data_out.append(c)
else:
console_data_out.extend(b"[%02x]" % c)
else:
console_data_out = dev_data_in
console_out_write(console_data_out)
except OSError as er:
if _is_disconnect_exception(er):
return True
else:
raise
return False
def do_repl(state, args):
state.ensure_friendly_repl()
state.did_action()
escape_non_printable = args.escape_non_printable
capture_file = args.capture
code_to_inject = args.inject_code
file_to_inject = args.inject_file
print("Connected to MicroPython at %s" % state.transport.device_name)
print("Use Ctrl-] or Ctrl-x to exit this shell")
if escape_non_printable:
print("Escaping non-printable bytes/characters by printing their hex code")
if capture_file is not None:
print('Capturing session to file "%s"' % capture_file)
capture_file = open(capture_file, "wb")
if code_to_inject is not None:
code_to_inject = bytes(code_to_inject.replace("\\n", "\r\n"), "utf8")
print("Use Ctrl-J to inject", code_to_inject)
if file_to_inject is not None:
print('Use Ctrl-K to inject file "%s"' % file_to_inject)
console = Console()
console.enter()
def console_out_write(b):
console.write(b)
if capture_file is not None:
capture_file.write(b)
capture_file.flush()
try:
return do_repl_main_loop(
state,
console,
console_out_write,
escape_non_printable=escape_non_printable,
code_to_inject=code_to_inject,
file_to_inject=file_to_inject,
)
finally:
console.exit()
if capture_file is not None:
capture_file.close()
def _is_disconnect_exception(exception):
"""
Check if an exception indicates device disconnect.
Returns True if the exception indicates the device has disconnected,
False otherwise.
"""
if isinstance(exception, OSError):
if hasattr(exception, "args") and len(exception.args) > 0:
# IO error, device disappeared
if exception.args[0] == 5:
return True
# Check for common disconnect messages in the exception string
exception_str = str(exception)
disconnect_indicators = ["Write timeout", "Device disconnected", "ClearCommError failed"]
return any(indicator in exception_str for indicator in disconnect_indicators)
return False

148
.venv/lib/python3.12/site-packages/mpremote/romfs.py Normal file
View File

@@ -0,0 +1,148 @@
# MIT license; Copyright (c) 2022 Damien P. George
import struct, sys, os
try:
from mpy_cross import run as mpy_cross_run
except ImportError:
mpy_cross_run = None
class VfsRomWriter:
ROMFS_HEADER = b"\xd2\xcd\x31"
ROMFS_RECORD_KIND_UNUSED = 0
ROMFS_RECORD_KIND_PADDING = 1
ROMFS_RECORD_KIND_DATA_VERBATIM = 2
ROMFS_RECORD_KIND_DATA_POINTER = 3
ROMFS_RECORD_KIND_DIRECTORY = 4
ROMFS_RECORD_KIND_FILE = 5
def __init__(self):
self._dir_stack = [(None, bytearray())]
def _encode_uint(self, value):
encoded = [value & 0x7F]
value >>= 7
while value != 0:
encoded.insert(0, 0x80 | (value & 0x7F))
value >>= 7
return bytes(encoded)
def _pack(self, kind, payload):
return self._encode_uint(kind) + self._encode_uint(len(payload)) + payload
def _extend(self, data):
buf = self._dir_stack[-1][1]
buf.extend(data)
return len(buf)
def finalise(self):
_, data = self._dir_stack.pop()
encoded_kind = VfsRomWriter.ROMFS_HEADER
encoded_len = self._encode_uint(len(data))
if (len(encoded_kind) + len(encoded_len) + len(data)) % 2 == 1:
encoded_len = b"\x80" + encoded_len
data = encoded_kind + encoded_len + data
return data
def opendir(self, dirname):
self._dir_stack.append((dirname, bytearray()))
def closedir(self):
dirname, dirdata = self._dir_stack.pop()
dirdata = self._encode_uint(len(dirname)) + bytes(dirname, "ascii") + dirdata
self._extend(self._pack(VfsRomWriter.ROMFS_RECORD_KIND_DIRECTORY, dirdata))
def mkdata(self, data):
assert len(self._dir_stack) == 1
return self._extend(self._pack(VfsRomWriter.ROMFS_RECORD_KIND_DATA_VERBATIM, data)) - len(
data
)
def mkfile(self, filename, filedata):
filename = bytes(filename, "ascii")
payload = self._encode_uint(len(filename))
payload += filename
if isinstance(filedata, tuple):
sub_payload = self._encode_uint(filedata[0])
sub_payload += self._encode_uint(filedata[1])
payload += self._pack(VfsRomWriter.ROMFS_RECORD_KIND_DATA_POINTER, sub_payload)
else:
payload += self._pack(VfsRomWriter.ROMFS_RECORD_KIND_DATA_VERBATIM, filedata)
self._dir_stack[-1][1].extend(self._pack(VfsRomWriter.ROMFS_RECORD_KIND_FILE, payload))
def copy_recursively(vfs, src_dir, print_prefix, mpy_cross):
assert src_dir.endswith("/")
DIR = 1 << 14
mpy_cross_missed = 0
dir_contents = sorted(os.listdir(src_dir))
for name in dir_contents:
src_name = src_dir + name
st = os.stat(src_name)
if name == dir_contents[-1]:
# Last entry in the directory listing.
print_entry = "\\--"
print_recurse = " "
else:
# Not the last entry in the directory listing.
print_entry = "|--"
print_recurse = "| "
if st[0] & DIR:
# A directory, enter it and copy its contents recursively.
print(print_prefix + print_entry, name + "/")
vfs.opendir(name)
mpy_cross_missed += copy_recursively(
vfs, src_name + "/", print_prefix + print_recurse, mpy_cross
)
vfs.closedir()
else:
# A file.
did_mpy = False
name_extra = ""
if mpy_cross and name.endswith(".py"):
name_mpy = name[:-3] + ".mpy"
src_name_mpy = src_dir + name_mpy
if not os.path.isfile(src_name_mpy):
if mpy_cross_run is not None:
did_mpy = True
proc = mpy_cross_run(src_name)
proc.wait()
else:
mpy_cross_missed += 1
if did_mpy:
name_extra = " -> .mpy"
print(print_prefix + print_entry, name + name_extra)
if did_mpy:
name = name_mpy
src_name = src_name_mpy
with open(src_name, "rb") as src:
vfs.mkfile(name, src.read())
if did_mpy:
os.remove(src_name_mpy)
return mpy_cross_missed
def make_romfs(src_dir, *, mpy_cross):
if not src_dir.endswith("/"):
src_dir += "/"
vfs = VfsRomWriter()
# Build the filesystem recursively.
print("Building romfs filesystem, source directory: {}".format(src_dir))
print("/")
try:
mpy_cross_missed = copy_recursively(vfs, src_dir, "", mpy_cross)
except OSError as er:
print("Error: OSError {}".format(er), file=sys.stderr)
sys.exit(1)
if mpy_cross_missed:
print("Warning: `mpy_cross` module not found, .py files were not precompiled")
mpy_cross = False
return vfs.finalise()

211
.venv/lib/python3.12/site-packages/mpremote/transport.py Normal file
View File

@@ -0,0 +1,211 @@
#!/usr/bin/env python3
#
# This file is part of the MicroPython project, http://micropython.org/
#
# The MIT License (MIT)
#
# Copyright (c) 2023 Jim Mussared
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import ast, errno, hashlib, os, re, sys
from collections import namedtuple
from .mp_errno import MP_ERRNO_TABLE
def stdout_write_bytes(b):
b = b.replace(b"\x04", b"")
if hasattr(sys.stdout, "buffer"):
sys.stdout.buffer.write(b)
sys.stdout.buffer.flush()
else:
text = b.decode(sys.stdout.encoding, "strict")
sys.stdout.write(text)
class TransportError(Exception):
pass
class TransportExecError(TransportError):
def __init__(self, status_code, error_output):
self.status_code = status_code
self.error_output = error_output
super().__init__(error_output)
listdir_result = namedtuple("dir_result", ["name", "st_mode", "st_ino", "st_size"])
# Takes a Transport error (containing the text of an OSError traceback) and
# raises it as the corresponding OSError-derived exception.
def _convert_filesystem_error(e, info):
if "OSError" in e.error_output:
for code, estr in [
*errno.errorcode.items(),
(errno.EOPNOTSUPP, "EOPNOTSUPP"),
]:
if estr in e.error_output:
return OSError(code, info)
# Some targets don't render OSError with the name of the errno, so in these
# cases support an explicit mapping of errnos to known numeric codes.
error_lines = e.error_output.splitlines()
match = re.match(r"OSError: (\d+)$", error_lines[-1])
if match:
value = int(match.group(1), 10)
if value in MP_ERRNO_TABLE:
return OSError(MP_ERRNO_TABLE[value], info)
return e
class Transport:
def fs_listdir(self, src=""):
buf = bytearray()
def repr_consumer(b):
buf.extend(b.replace(b"\x04", b""))
cmd = "import os\nfor f in os.ilistdir(%s):\n print(repr(f), end=',')" % (
("'%s'" % src) if src else ""
)
try:
buf.extend(b"[")
self.exec(cmd, data_consumer=repr_consumer)
buf.extend(b"]")
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
return [
listdir_result(*f) if len(f) == 4 else listdir_result(*(f + (0,)))
for f in ast.literal_eval(buf.decode())
]
def fs_stat(self, src):
try:
self.exec("import os")
return os.stat_result(self.eval("os.stat(%s)" % ("'%s'" % src)))
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
def fs_exists(self, src):
try:
self.fs_stat(src)
return True
except OSError:
return False
def fs_isdir(self, src):
try:
mode = self.fs_stat(src).st_mode
return (mode & 0x4000) != 0
except OSError:
# Match CPython, a non-existent path is not a directory.
return False
def fs_printfile(self, src, chunk_size=256):
cmd = (
"with open('%s') as f:\n while 1:\n"
" b=f.read(%u)\n if not b:break\n print(b,end='')" % (src, chunk_size)
)
try:
self.exec(cmd, data_consumer=stdout_write_bytes)
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
def fs_readfile(self, src, chunk_size=256, progress_callback=None):
if progress_callback:
src_size = self.fs_stat(src).st_size
contents = bytearray()
try:
self.exec("f=open('%s','rb')\nr=f.read" % src)
while True:
chunk = self.eval("r({})".format(chunk_size))
if not chunk:
break
contents.extend(chunk)
if progress_callback:
progress_callback(len(contents), src_size)
self.exec("f.close()")
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
return contents
def fs_writefile(self, dest, data, chunk_size=256, progress_callback=None):
if progress_callback:
src_size = len(data)
written = 0
try:
self.exec("f=open('%s','wb')\nw=f.write" % dest)
while data:
chunk = data[:chunk_size]
self.exec("w(" + repr(chunk) + ")")
data = data[len(chunk) :]
if progress_callback:
written += len(chunk)
progress_callback(written, src_size)
self.exec("f.close()")
except TransportExecError as e:
raise _convert_filesystem_error(e, dest) from None
def fs_mkdir(self, path):
try:
self.exec("import os\nos.mkdir('%s')" % path)
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_rmdir(self, path):
try:
self.exec("import os\nos.rmdir('%s')" % path)
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_rmfile(self, path):
try:
self.exec("import os\nos.remove('%s')" % path)
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_touchfile(self, path):
try:
self.exec("f=open('%s','a')\nf.close()" % path)
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_hashfile(self, path, algo, chunk_size=256):
try:
self.exec("import hashlib\nh = hashlib.{algo}()".format(algo=algo))
except TransportExecError:
# hashlib (or hashlib.{algo}) not available on device. Do the hash locally.
data = self.fs_readfile(path, chunk_size=chunk_size)
return getattr(hashlib, algo)(data).digest()
try:
self.exec(
"buf = memoryview(bytearray({chunk_size}))\nwith open('{path}', 'rb') as f:\n while True:\n n = f.readinto(buf)\n if n == 0:\n break\n h.update(buf if n == {chunk_size} else buf[:n])\n".format(
chunk_size=chunk_size, path=path
)
)
return self.eval("h.digest()")
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None

1058
.venv/lib/python3.12/site-packages/mpremote/transport_serial.py Normal file
View File

File diff suppressed because it is too large Load Diff