Arduino Core 3.3.6
This commit is contained in:
Jason2866
+864
-34
@@ -12,15 +12,24 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import importlib.util
|
||||
import locale
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import shutil
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
from os.path import isfile, join
|
||||
from pathlib import Path
|
||||
from littlefs import LittleFS
|
||||
from fatfs import Partition, RamDisk, create_extended_partition
|
||||
from fatfs import create_esp32_wl_image
|
||||
from fatfs import calculate_esp32_wl_overhead
|
||||
from fatfs import is_esp32_wl_image, extract_fat_from_esp32_wl
|
||||
from fatfs.partition_extended import PartitionExtended
|
||||
from fatfs.wrapper import pyf_mkfs, PY_FR_OK as FR_OK
|
||||
import importlib.util
|
||||
|
||||
from SCons.Script import (
|
||||
ARGUMENTS,
|
||||
@@ -48,6 +57,15 @@ build_dir = Path(projectconfig.get("platformio", "build_dir"))
|
||||
# Configure Python environment through centralized platform management
|
||||
PYTHON_EXE, esptool_binary_path = platform.setup_python_env(env)
|
||||
|
||||
# Load SPIFFS generator from local module
|
||||
spiffsgen_path = platform_dir / "builder" / "spiffsgen.py"
|
||||
spec = importlib.util.spec_from_file_location("spiffsgen", str(spiffsgen_path))
|
||||
spiffsgen = importlib.util.module_from_spec(spec)
|
||||
sys.modules["spiffsgen"] = spiffsgen
|
||||
spec.loader.exec_module(spiffsgen)
|
||||
SpiffsFS = spiffsgen.SpiffsFS
|
||||
SpiffsBuildConfig = spiffsgen.SpiffsBuildConfig
|
||||
|
||||
# Load board configuration and determine MCU architecture
|
||||
board = env.BoardConfig()
|
||||
board_id = env.subst("$BOARD")
|
||||
@@ -391,12 +409,6 @@ def fetch_fs_size(env):
|
||||
env["FS_PAGE"] = int("0x100", 16)
|
||||
env["FS_BLOCK"] = int("0x1000", 16)
|
||||
|
||||
# FFat specific offsets, see:
|
||||
# https://github.com/lorol/arduino-esp32fatfs-plugin#notes-for-fatfs
|
||||
if filesystem == "fatfs":
|
||||
env["FS_START"] += 4096
|
||||
env["FS_SIZE"] -= 4096
|
||||
|
||||
|
||||
def __fetch_fs_size(target, source, env):
|
||||
"""
|
||||
@@ -414,6 +426,342 @@ def __fetch_fs_size(target, source, env):
|
||||
return (target, source)
|
||||
|
||||
|
||||
def build_fs_image(target, source, env):
|
||||
"""
|
||||
Build filesystem image using littlefs-python.
|
||||
|
||||
Args:
|
||||
target: SCons target (output .bin file)
|
||||
source: SCons source (directory with files)
|
||||
env: SCons environment object
|
||||
|
||||
Returns:
|
||||
int: 0 on success, 1 on failure
|
||||
"""
|
||||
|
||||
# Get parameters
|
||||
source_dir = str(source[0])
|
||||
target_file = str(target[0])
|
||||
fs_size = env["FS_SIZE"]
|
||||
block_size = env.get("FS_BLOCK", 4096)
|
||||
|
||||
# Calculate block count
|
||||
block_count = fs_size // block_size
|
||||
|
||||
# Get disk version from board config or project options
|
||||
# Default to LittleFS version 2.1 (0x00020001)
|
||||
disk_version_str = "2.1"
|
||||
|
||||
# Try to read from project config (env-specific or common section)
|
||||
for section in ["env:" + env["PIOENV"], "common"]:
|
||||
if projectconfig.has_option(section, "board_build.littlefs_version"):
|
||||
disk_version_str = projectconfig.get(section, "board_build.littlefs_version")
|
||||
break
|
||||
|
||||
# Parse version string and create proper version integer
|
||||
# LittleFS version format: (major << 16) | (minor << 0)
|
||||
try:
|
||||
version_parts = str(disk_version_str).split(".")
|
||||
major = int(version_parts[0])
|
||||
minor = int(version_parts[1]) if len(version_parts) > 1 else 0
|
||||
# Format: major in upper 16 bits, minor in lower 16 bits
|
||||
disk_version = (major << 16) | minor
|
||||
except (ValueError, IndexError):
|
||||
print(f"Warning: Invalid littlefs version '{disk_version_str}', using default 2.1")
|
||||
disk_version = (2 << 16) | 1
|
||||
|
||||
try:
|
||||
# Create LittleFS instance with Arduino / IDF compatible parameters
|
||||
fs = LittleFS(
|
||||
block_size=block_size,
|
||||
block_count=block_count,
|
||||
read_size=1, # Minimum read size
|
||||
prog_size=1, # Minimum program size
|
||||
cache_size=block_size, # Cache size = block size
|
||||
lookahead_size=32, # Default lookahead buffer
|
||||
block_cycles=500, # Wear leveling cycles
|
||||
name_max=64, # ESP-IDF default filename length
|
||||
disk_version=disk_version,
|
||||
mount=True
|
||||
)
|
||||
|
||||
# Add all files from source directory
|
||||
source_path = Path(source_dir)
|
||||
if source_path.exists():
|
||||
for item in source_path.rglob("*"):
|
||||
rel_path = item.relative_to(source_path)
|
||||
fs_path = rel_path.as_posix()
|
||||
|
||||
if item.is_dir():
|
||||
fs.makedirs(fs_path, exist_ok=True)
|
||||
# Set directory mtime attribute
|
||||
try:
|
||||
mtime = int(item.stat().st_mtime)
|
||||
fs.setattr(fs_path, 't', mtime.to_bytes(4, 'little'))
|
||||
except Exception:
|
||||
pass # Ignore timestamp errors
|
||||
else:
|
||||
# Ensure parent directories exist
|
||||
if rel_path.parent != Path("."):
|
||||
fs.makedirs(rel_path.parent.as_posix(), exist_ok=True)
|
||||
# Copy file
|
||||
with fs.open(fs_path, "wb") as dest:
|
||||
dest.write(item.read_bytes())
|
||||
# Set file mtime attribute (ESP-IDF compatible)
|
||||
try:
|
||||
mtime = int(item.stat().st_mtime)
|
||||
fs.setattr(fs_path, 't', mtime.to_bytes(4, 'little'))
|
||||
except Exception:
|
||||
pass # Ignore timestamp errors
|
||||
|
||||
# Write filesystem image
|
||||
with open(target_file, "wb") as f:
|
||||
f.write(fs.context.buffer)
|
||||
|
||||
return 0
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error building filesystem image: {e}")
|
||||
return 1
|
||||
|
||||
|
||||
def build_spiffs_image(target, source, env):
|
||||
"""
|
||||
Build SPIFFS filesystem image using spiffsgen.py.
|
||||
|
||||
Args:
|
||||
target: SCons target (output .bin file)
|
||||
source: SCons source (directory with files)
|
||||
env: SCons environment object
|
||||
|
||||
Returns:
|
||||
int: 0 on success, 1 on failure
|
||||
"""
|
||||
|
||||
# Get parameters
|
||||
source_dir = str(source[0])
|
||||
target_file = str(target[0])
|
||||
fs_size = env["FS_SIZE"]
|
||||
page_size = env.get("FS_PAGE", 256)
|
||||
block_size = env.get("FS_BLOCK", 4096)
|
||||
|
||||
# Get SPIFFS configuration from project config or use defaults
|
||||
obj_name_len = 32
|
||||
meta_len = 4
|
||||
use_magic = True
|
||||
use_magic_len = True
|
||||
aligned_obj_ix_tables = False
|
||||
|
||||
# Check common section first, then env-specific (so env-specific takes precedence)
|
||||
for section in ["common", "env:" + env["PIOENV"]]:
|
||||
if projectconfig.has_option(section, "board_build.spiffs.obj_name_len"):
|
||||
obj_name_len = int(projectconfig.get(section, "board_build.spiffs.obj_name_len"))
|
||||
if projectconfig.has_option(section, "board_build.spiffs.meta_len"):
|
||||
meta_len = int(projectconfig.get(section, "board_build.spiffs.meta_len"))
|
||||
if projectconfig.has_option(section, "board_build.spiffs.use_magic"):
|
||||
use_magic = projectconfig.getboolean(section, "board_build.spiffs.use_magic")
|
||||
if projectconfig.has_option(section, "board_build.spiffs.use_magic_len"):
|
||||
use_magic_len = projectconfig.getboolean(section, "board_build.spiffs.use_magic_len")
|
||||
if projectconfig.has_option(section, "board_build.spiffs.aligned_obj_ix_tables"):
|
||||
aligned_obj_ix_tables = projectconfig.getboolean(section, "board_build.spiffs.aligned_obj_ix_tables")
|
||||
|
||||
try:
|
||||
# Create SPIFFS build configuration
|
||||
spiffs_build_config = SpiffsBuildConfig(
|
||||
page_size=page_size,
|
||||
page_ix_len=2, # SPIFFS_PAGE_IX_LEN
|
||||
block_size=block_size,
|
||||
block_ix_len=2, # SPIFFS_BLOCK_IX_LEN
|
||||
meta_len=meta_len,
|
||||
obj_name_len=obj_name_len,
|
||||
obj_id_len=2, # SPIFFS_OBJ_ID_LEN
|
||||
span_ix_len=2, # SPIFFS_SPAN_IX_LEN
|
||||
packed=True,
|
||||
aligned=True,
|
||||
endianness='little',
|
||||
use_magic=use_magic,
|
||||
use_magic_len=use_magic_len,
|
||||
aligned_obj_ix_tables=aligned_obj_ix_tables
|
||||
)
|
||||
|
||||
# Create SPIFFS filesystem
|
||||
spiffs = SpiffsFS(fs_size, spiffs_build_config)
|
||||
|
||||
# Add all files from source directory
|
||||
source_path = Path(source_dir)
|
||||
if source_path.exists():
|
||||
for item in source_path.rglob("*"):
|
||||
if item.is_file():
|
||||
rel_path = item.relative_to(source_path)
|
||||
img_path = "/" + rel_path.as_posix()
|
||||
spiffs.create_file(img_path, str(item))
|
||||
|
||||
# Generate binary image
|
||||
image = spiffs.to_binary()
|
||||
|
||||
# Write to file
|
||||
with open(target_file, "wb") as f:
|
||||
f.write(image)
|
||||
|
||||
print(f"\nSuccessfully created SPIFFS image: {target_file}")
|
||||
return 0
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error building SPIFFS image: {e}")
|
||||
return 1
|
||||
|
||||
|
||||
def build_fatfs_image(target, source, env):
|
||||
"""
|
||||
Build FatFS filesystem image with ESP32 Wear Leveling support.
|
||||
|
||||
Uses fatfs-ng module to create ESP-IDF compatible WL-wrapped FAT images.
|
||||
|
||||
Args:
|
||||
target: SCons target (output .bin file)
|
||||
source: SCons source (directory with files)
|
||||
env: SCons environment object
|
||||
|
||||
Returns:
|
||||
int: 0 on success, 1 on failure
|
||||
"""
|
||||
|
||||
# Get parameters
|
||||
source_dir = str(source[0])
|
||||
target_file = str(target[0])
|
||||
fs_size = env["FS_SIZE"]
|
||||
sector_size = env.get("FS_SECTOR", 4096)
|
||||
|
||||
# ESP-IDF WL layout (following wl_fatfsgen.py):
|
||||
# [dummy sector] [FAT data] [state1] [state2] [config]
|
||||
# Total WL sectors: 1 dummy + 2 states + 1 config = 4 sectors
|
||||
wl_info = calculate_esp32_wl_overhead(fs_size, sector_size)
|
||||
|
||||
wl_reserved_sectors = wl_info['wl_overhead_sectors']
|
||||
fat_fs_size = wl_info['fat_size']
|
||||
sector_count = wl_info['fat_sectors']
|
||||
|
||||
try:
|
||||
# Create RAM disk with the FAT filesystem size (without WL overhead)
|
||||
storage = bytearray(fat_fs_size)
|
||||
disk = RamDisk(storage, sector_size=sector_size, sector_count=sector_count)
|
||||
|
||||
# Create partition, format, and mount
|
||||
base_partition = Partition(disk)
|
||||
|
||||
# Format the filesystem with proper workarea size for LFN support
|
||||
# Workarea needs to be at least sector_size, use 2x for safety with LFN
|
||||
workarea_size = sector_size * 2
|
||||
|
||||
# Create filesystem with parameters matching ESP-IDF expectations:
|
||||
# - n_fat=2: Two FAT copies for redundancy
|
||||
# - align=0: Auto-align (let FATFS decide)
|
||||
# - n_root=512: Number of root directory entries (FAT12/16 only, 0 for FAT32)
|
||||
# - au_size=0: Auto allocation unit size
|
||||
ret = pyf_mkfs(
|
||||
base_partition.pname,
|
||||
n_fat=2,
|
||||
align=0,
|
||||
n_root=512, # Standard root entries for FAT16
|
||||
au_size=0, # Auto
|
||||
workarea_size=workarea_size
|
||||
)
|
||||
if ret != FR_OK:
|
||||
raise Exception(f"Failed to format filesystem: error code {ret}")
|
||||
|
||||
# Mount the filesystem
|
||||
base_partition.mount()
|
||||
|
||||
# Wrap with extended partition for directory support
|
||||
partition = PartitionExtended(base_partition)
|
||||
|
||||
# Track skipped files
|
||||
skipped_files = []
|
||||
|
||||
# Add all files from source directory
|
||||
source_path = Path(source_dir)
|
||||
if source_path.exists():
|
||||
for item in source_path.rglob("*"):
|
||||
rel_path = item.relative_to(source_path)
|
||||
fs_path = "/" + rel_path.as_posix()
|
||||
|
||||
if item.is_dir():
|
||||
try:
|
||||
partition.mkdir(fs_path)
|
||||
except Exception:
|
||||
# Directory might already exist or be root
|
||||
pass
|
||||
else:
|
||||
# Ensure parent directories exist
|
||||
if rel_path.parent != Path("."):
|
||||
parent_path = "/" + rel_path.parent.as_posix()
|
||||
try:
|
||||
partition.mkdir(parent_path)
|
||||
except Exception:
|
||||
pass # Directory might already exist
|
||||
|
||||
# Copy file
|
||||
try:
|
||||
with partition.open(fs_path, "w") as dest:
|
||||
dest.write(item.read_bytes())
|
||||
except Exception as e:
|
||||
print(f"Warning: Failed to write file {rel_path}: {e}")
|
||||
skipped_files.append(str(rel_path))
|
||||
|
||||
# Unmount filesystem
|
||||
base_partition.unmount()
|
||||
|
||||
# Read boot sector parameters for validation
|
||||
bytes_per_sector = struct.unpack('<H', storage[11:13])[0]
|
||||
reserved_sectors = struct.unpack('<H', storage[14:16])[0]
|
||||
num_fats = storage[16]
|
||||
sectors_per_fat = struct.unpack('<H', storage[22:24])[0]
|
||||
total_sectors = struct.unpack('<H', storage[19:21])[0]
|
||||
|
||||
# Validate boot sector matches our expectations
|
||||
if bytes_per_sector != sector_size:
|
||||
raise Exception(f"Boot sector bytes_per_sector ({bytes_per_sector}) != sector_size ({sector_size})")
|
||||
|
||||
print("\nBoot sector validation:")
|
||||
print(f" Bytes per sector: {bytes_per_sector}")
|
||||
print(f" Reserved sectors: {reserved_sectors}")
|
||||
print(f" Number of FATs: {num_fats}")
|
||||
print(f" Sectors per FAT: {sectors_per_fat}")
|
||||
print(f" Total sectors: {total_sectors}")
|
||||
|
||||
# Wrap FAT image with ESP-IDF wear leveling layer
|
||||
# This uses the fatfs-ng module's ESP32WearLeveling implementation
|
||||
print("\nWrapping FAT image with ESP-IDF wear leveling...")
|
||||
print(f" Layout: {wl_info['layout']}")
|
||||
print(f" Partition size: {fs_size} bytes")
|
||||
print(f" FAT filesystem size: {fat_fs_size} bytes ({sector_count} sectors)")
|
||||
print(f" WL overhead: {wl_reserved_sectors} sectors ({wl_info['wl_overhead_size']} bytes)")
|
||||
|
||||
wl_image = create_esp32_wl_image(bytes(storage), fs_size, sector_size)
|
||||
|
||||
print(f" WL-wrapped image created ({len(wl_image)} bytes)")
|
||||
|
||||
# Write WL-wrapped image to file
|
||||
with open(target_file, "wb") as f:
|
||||
f.write(wl_image)
|
||||
|
||||
# Print summary
|
||||
if skipped_files:
|
||||
print(f"\nWarning: {len(skipped_files)} file(s) skipped:")
|
||||
for skipped in skipped_files[:10]: # Show first 10
|
||||
print(f" - {skipped}")
|
||||
if len(skipped_files) > 10:
|
||||
print(f" ... and {len(skipped_files) - 10} more")
|
||||
|
||||
print(f"\nSuccessfully created ESP-IDF WL-wrapped FAT image: {target_file}")
|
||||
|
||||
return 0
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error building FatFS image: {e}")
|
||||
return 1
|
||||
|
||||
|
||||
def check_lib_archive_exists():
|
||||
"""
|
||||
Check if lib_archive is set in platformio.ini configuration.
|
||||
@@ -427,14 +775,29 @@ def check_lib_archive_exists():
|
||||
return False
|
||||
|
||||
|
||||
def build_fs_router(target, source, env):
|
||||
"""Route to appropriate filesystem builder based on filesystem type."""
|
||||
fs_type = board.get("build.filesystem", "littlefs")
|
||||
if fs_type == "littlefs":
|
||||
return build_fs_image(target, source, env)
|
||||
elif fs_type == "fatfs":
|
||||
return build_fatfs_image(target, source, env)
|
||||
elif fs_type == "spiffs":
|
||||
return build_spiffs_image(target, source, env)
|
||||
else:
|
||||
print(f"Error: Unknown filesystem type '{fs_type}'. Supported types: littlefs, fatfs, spiffs")
|
||||
return 1
|
||||
|
||||
|
||||
def switch_off_ldf():
|
||||
"""
|
||||
Disables LDF (Library Dependency Finder) for uploadfs, uploadfsota, and buildfs targets.
|
||||
Disables LDF (Library Dependency Finder) for uploadfs, uploadfsota, buildfs,
|
||||
download_fs, and erase targets.
|
||||
|
||||
This optimization prevents unnecessary library dependency scanning and compilation
|
||||
when only filesystem operations are performed.
|
||||
"""
|
||||
fs_targets = {"uploadfs", "uploadfsota", "buildfs", "erase"}
|
||||
fs_targets = {"uploadfs", "uploadfsota", "buildfs", "erase", "download_fs"}
|
||||
if fs_targets & set(COMMAND_LINE_TARGETS):
|
||||
# Disable LDF by modifying project configuration directly
|
||||
env_section = "env:" + env["PIOENV"]
|
||||
@@ -494,22 +857,6 @@ env.Replace(
|
||||
ERASEFLAGS=["--chip", mcu, "--port", '"$UPLOAD_PORT"'],
|
||||
ERASETOOL=uploader_path,
|
||||
ERASECMD='$ERASETOOL $ERASEFLAGS erase-flash',
|
||||
MKFSTOOL="mk%s" % filesystem
|
||||
+ (
|
||||
(
|
||||
"_${PIOPLATFORM}_"
|
||||
+ (
|
||||
"espidf"
|
||||
if "espidf" in env.subst("$PIOFRAMEWORK")
|
||||
else "${PIOFRAMEWORK}"
|
||||
)
|
||||
)
|
||||
if filesystem == "spiffs"
|
||||
else ""
|
||||
),
|
||||
# Legacy `ESP32_SPIFFS_IMAGE_NAME` is used as the second fallback value
|
||||
# for backward compatibility
|
||||
|
||||
ESP32_FS_IMAGE_NAME=env.get(
|
||||
"ESP32_FS_IMAGE_NAME",
|
||||
env.get("ESP32_SPIFFS_IMAGE_NAME", filesystem),
|
||||
@@ -559,15 +906,7 @@ env.Append(
|
||||
),
|
||||
DataToBin=Builder(
|
||||
action=env.VerboseAction(
|
||||
" ".join(
|
||||
['"$MKFSTOOL"', "-c", "$SOURCES", "-s", "$FS_SIZE"]
|
||||
+ (
|
||||
["-p", "$FS_PAGE", "-b", "$FS_BLOCK"]
|
||||
if filesystem in ("littlefs", "spiffs")
|
||||
else []
|
||||
)
|
||||
+ ["$TARGET"]
|
||||
),
|
||||
build_fs_router,
|
||||
"Building FS image from '$SOURCES' directory to $TARGET",
|
||||
),
|
||||
emitter=__fetch_fs_size,
|
||||
@@ -776,6 +1115,486 @@ def coredump_analysis(target, source, env):
|
||||
print(f"Error: Failed to run coredump analysis: {e}")
|
||||
print(f'Make sure esp-coredump is installed: uv pip install --python "{PYTHON_EXE}" esp-coredump')
|
||||
|
||||
|
||||
def _get_unpack_dir(env):
|
||||
"""
|
||||
Get the unpack directory from project configuration.
|
||||
|
||||
Args:
|
||||
env: SCons environment object
|
||||
|
||||
Returns:
|
||||
str: Unpack directory path
|
||||
"""
|
||||
unpack_dir = "unpacked_fs"
|
||||
|
||||
# Read from project config (env-specific or common section)
|
||||
for section in ["env:" + env["PIOENV"], "common"]:
|
||||
if projectconfig.has_option(section, "board_build.unpack_dir"):
|
||||
unpack_dir = projectconfig.get(section, "board_build.unpack_dir")
|
||||
break
|
||||
|
||||
return unpack_dir
|
||||
|
||||
|
||||
def _prepare_unpack_dir(unpack_dir):
|
||||
"""
|
||||
Prepare the unpack directory by removing old content and creating fresh directory.
|
||||
|
||||
Args:
|
||||
unpack_dir: Directory path to prepare
|
||||
|
||||
Returns:
|
||||
Path: Path object for the unpack directory
|
||||
"""
|
||||
unpack_path = Path(get_project_dir()) / unpack_dir
|
||||
if unpack_path.exists():
|
||||
shutil.rmtree(unpack_path)
|
||||
unpack_path.mkdir(parents=True, exist_ok=True)
|
||||
return unpack_path
|
||||
|
||||
|
||||
def _download_partition_image(env, fs_type_filter=None):
|
||||
"""
|
||||
Common function to download partition table and filesystem image from device.
|
||||
|
||||
Args:
|
||||
env: SCons environment object
|
||||
fs_type_filter: List of partition subtypes to look for (e.g., [0x82, 0x83] for LittleFS/SPIFFS)
|
||||
or [0x81] for FAT. If None, accepts any data partition.
|
||||
|
||||
Returns:
|
||||
tuple: (fs_file_path, fs_start, fs_size, fs_subtype) or (None, None, None, None) on error
|
||||
"""
|
||||
# Ensure upload port is set
|
||||
if not env.subst("$UPLOAD_PORT"):
|
||||
env.AutodetectUploadPort()
|
||||
|
||||
upload_port = env.subst("$UPLOAD_PORT")
|
||||
download_speed = board.get("download.speed", "115200")
|
||||
|
||||
# Download partition table from device
|
||||
print(f"\nDownloading partition table from {upload_port}...\n")
|
||||
|
||||
build_dir = Path(env.subst("$BUILD_DIR"))
|
||||
build_dir.mkdir(parents=True, exist_ok=True)
|
||||
partition_file = build_dir / "partition_table_from_flash.bin"
|
||||
|
||||
esptool_cmd = [
|
||||
uploader_path.strip('"'),
|
||||
"--port", upload_port,
|
||||
"--baud", str(download_speed),
|
||||
"--before", "default-reset",
|
||||
"--after", "hard-reset",
|
||||
"read-flash",
|
||||
"0x8000", # Partition table offset
|
||||
"0x1000", # Partition table size (4KB)
|
||||
str(partition_file)
|
||||
]
|
||||
|
||||
try:
|
||||
result = subprocess.run(esptool_cmd, check=False)
|
||||
if result.returncode != 0:
|
||||
print("Error: Failed to download partition table")
|
||||
return None, None, None, None
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
return None, None, None, None
|
||||
|
||||
with open(partition_file, 'rb') as f:
|
||||
partition_data = f.read()
|
||||
|
||||
# Parse partition entries (format: 0xAA 0x50 followed by entry data)
|
||||
entries = [e for e in partition_data.split(b'\xaaP') if len(e) > 0]
|
||||
|
||||
fs_start = None
|
||||
fs_size = None
|
||||
fs_subtype = None
|
||||
|
||||
for entry in entries:
|
||||
if len(entry) < 32:
|
||||
continue
|
||||
|
||||
# Byte 0: Type (0x01 for data partitions)
|
||||
# Byte 1: SubType (0x81=FAT, 0x82=SPIFFS, 0x83=LittleFS)
|
||||
# Bytes 2-5: Offset (4 bytes, little-endian)
|
||||
# Bytes 6-9: Size (4 bytes, little-endian)
|
||||
|
||||
part_subtype = entry[1]
|
||||
|
||||
# Check if this partition matches our filter
|
||||
if fs_type_filter is None or part_subtype in fs_type_filter:
|
||||
fs_start = int.from_bytes(entry[2:6], byteorder='little', signed=False)
|
||||
fs_size = int.from_bytes(entry[6:10], byteorder='little', signed=False)
|
||||
fs_subtype = part_subtype
|
||||
break
|
||||
|
||||
if fs_start is None or fs_size is None:
|
||||
print("Error: No matching filesystem partition found in partition table")
|
||||
return None, None, None, None
|
||||
|
||||
print(f"\nFound filesystem partition (subtype {hex(fs_subtype)}):")
|
||||
print(f" Start: {hex(fs_start)}")
|
||||
print(f" Size: {hex(fs_size)} ({fs_size} bytes)")
|
||||
|
||||
# Download filesystem image
|
||||
fs_file = build_dir / f"downloaded_fs_{hex(fs_start)}_{hex(fs_size)}.bin"
|
||||
|
||||
print("\nDownloading filesystem from device...\n")
|
||||
|
||||
esptool_cmd = [
|
||||
uploader_path.strip('"'),
|
||||
"--port", upload_port,
|
||||
"--baud", str(download_speed),
|
||||
"--before", "default-reset",
|
||||
"--after", "hard-reset",
|
||||
"read-flash",
|
||||
hex(fs_start),
|
||||
hex(fs_size),
|
||||
str(fs_file)
|
||||
]
|
||||
|
||||
try:
|
||||
result = subprocess.run(esptool_cmd, check=False)
|
||||
if result.returncode != 0:
|
||||
print(f"Error: Download failed with code {result.returncode}")
|
||||
return None, None, None, None
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
return None, None, None, None
|
||||
|
||||
print(f"\nDownloaded to {fs_file}")
|
||||
|
||||
return fs_file, fs_start, fs_size, fs_subtype
|
||||
|
||||
|
||||
def _extract_littlefs(fs_file, fs_size, unpack_path, unpack_dir):
|
||||
"""Extract LittleFS filesystem."""
|
||||
# Read the downloaded filesystem image
|
||||
with open(fs_file, 'rb') as f:
|
||||
fs_data = f.read()
|
||||
|
||||
# Use ESP-IDF defaults
|
||||
block_size = 0x1000 # 4KB
|
||||
block_count = fs_size // block_size
|
||||
|
||||
# Create LittleFS instance and mount the image
|
||||
fs = LittleFS(
|
||||
block_size=block_size,
|
||||
block_count=block_count,
|
||||
mount=False
|
||||
)
|
||||
fs.context.buffer = bytearray(fs_data)
|
||||
fs.mount()
|
||||
|
||||
# Extract all files
|
||||
file_count = 0
|
||||
print("\nExtracted files:")
|
||||
for root, dirs, files in fs.walk("/"):
|
||||
if not root.endswith("/"):
|
||||
root += "/"
|
||||
|
||||
# Create directories
|
||||
for dir_name in dirs:
|
||||
src_path = root + dir_name
|
||||
dst_path = unpack_path / src_path[1:] # Remove leading '/'
|
||||
dst_path.mkdir(parents=True, exist_ok=True)
|
||||
print(f" [DIR] {src_path}")
|
||||
|
||||
# Extract files
|
||||
for file_name in files:
|
||||
src_path = root + file_name
|
||||
dst_path = unpack_path / src_path[1:] # Remove leading '/'
|
||||
dst_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with fs.open(src_path, "rb") as src:
|
||||
file_data = src.read()
|
||||
dst_path.write_bytes(file_data)
|
||||
|
||||
print(f" [FILE] {src_path} ({len(file_data)} bytes)")
|
||||
file_count += 1
|
||||
|
||||
fs.unmount()
|
||||
print(f"\nSuccessfully extracted {file_count} file(s) to {unpack_dir}")
|
||||
return 0
|
||||
|
||||
|
||||
def _parse_spiffs_config(fs_data, fs_size):
|
||||
"""
|
||||
Auto-detect SPIFFS configuration from the image.
|
||||
Tries common configurations and validates against the image.
|
||||
|
||||
Returns:
|
||||
dict: SPIFFS configuration parameters or None
|
||||
"""
|
||||
# Common ESP32/ESP8266 SPIFFS configurations
|
||||
common_configs = [
|
||||
# ESP32/ESP8266 defaults
|
||||
{'page_size': 256, 'block_size': 4096, 'obj_name_len': 32},
|
||||
# Alternative configurations
|
||||
{'page_size': 256, 'block_size': 8192, 'obj_name_len': 32},
|
||||
{'page_size': 512, 'block_size': 4096, 'obj_name_len': 32},
|
||||
{'page_size': 256, 'block_size': 4096, 'obj_name_len': 64},
|
||||
]
|
||||
|
||||
print("\nAuto-detecting SPIFFS configuration...")
|
||||
|
||||
for config in common_configs:
|
||||
try:
|
||||
# Try to parse with this configuration
|
||||
spiffs_build_config = SpiffsBuildConfig(
|
||||
page_size=config['page_size'],
|
||||
page_ix_len=2,
|
||||
block_size=config['block_size'],
|
||||
block_ix_len=2,
|
||||
meta_len=4,
|
||||
obj_name_len=config['obj_name_len'],
|
||||
obj_id_len=2,
|
||||
span_ix_len=2,
|
||||
packed=True,
|
||||
aligned=True,
|
||||
endianness='little',
|
||||
use_magic=True,
|
||||
use_magic_len=True,
|
||||
aligned_obj_ix_tables=False
|
||||
)
|
||||
|
||||
# Try to create and parse the filesystem
|
||||
spiffs = SpiffsFS(fs_size, spiffs_build_config)
|
||||
spiffs.from_binary(fs_data)
|
||||
|
||||
# If we got here without exception, this config works
|
||||
print(" Detected SPIFFS configuration:")
|
||||
print(f" Page size: {config['page_size']} bytes")
|
||||
print(f" Block size: {config['block_size']} bytes")
|
||||
print(f" Max filename length: {config['obj_name_len']}")
|
||||
|
||||
return {
|
||||
'page_size': config['page_size'],
|
||||
'block_size': config['block_size'],
|
||||
'obj_name_len': config['obj_name_len'],
|
||||
'meta_len': 4,
|
||||
'use_magic': True,
|
||||
'use_magic_len': True,
|
||||
'aligned_obj_ix_tables': False
|
||||
}
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
# If no config worked, return defaults
|
||||
print(" Could not auto-detect configuration, using ESP32/ESP8266 defaults")
|
||||
return {
|
||||
'page_size': 256,
|
||||
'block_size': 4096,
|
||||
'obj_name_len': 32,
|
||||
'meta_len': 4,
|
||||
'use_magic': True,
|
||||
'use_magic_len': True,
|
||||
'aligned_obj_ix_tables': False
|
||||
}
|
||||
|
||||
|
||||
def _extract_spiffs(fs_file, fs_size, unpack_path, unpack_dir):
|
||||
"""Extract SPIFFS filesystem with auto-detected configuration."""
|
||||
# Read the downloaded filesystem image
|
||||
with open(fs_file, 'rb') as f:
|
||||
fs_data = f.read()
|
||||
|
||||
# Auto-detect SPIFFS configuration
|
||||
config = _parse_spiffs_config(fs_data, fs_size)
|
||||
|
||||
# Create SPIFFS build configuration
|
||||
spiffs_build_config = SpiffsBuildConfig(
|
||||
page_size=config['page_size'],
|
||||
page_ix_len=2,
|
||||
block_size=config['block_size'],
|
||||
block_ix_len=2,
|
||||
meta_len=config['meta_len'],
|
||||
obj_name_len=config['obj_name_len'],
|
||||
obj_id_len=2,
|
||||
span_ix_len=2,
|
||||
packed=True,
|
||||
aligned=True,
|
||||
endianness='little',
|
||||
use_magic=config['use_magic'],
|
||||
use_magic_len=config['use_magic_len'],
|
||||
aligned_obj_ix_tables=config['aligned_obj_ix_tables']
|
||||
)
|
||||
|
||||
# Create SPIFFS filesystem and parse the image
|
||||
spiffs = SpiffsFS(fs_size, spiffs_build_config)
|
||||
spiffs.from_binary(fs_data)
|
||||
|
||||
# Extract files
|
||||
file_count = spiffs.extract_files(str(unpack_path))
|
||||
|
||||
if file_count == 0:
|
||||
print("\nNo files were extracted.")
|
||||
print("The filesystem may be empty, freshly formatted, or contain only deleted entries.")
|
||||
else:
|
||||
print(f"\nSuccessfully extracted {file_count} file(s) to {unpack_dir}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def _extract_fatfs(fs_file, unpack_path, unpack_dir):
|
||||
"""Extract FatFS filesystem."""
|
||||
# Read the downloaded filesystem image
|
||||
with open(fs_file, 'rb') as f:
|
||||
fs_data = bytearray(f.read())
|
||||
|
||||
# Check if the image looks like a valid FAT filesystem
|
||||
if len(fs_data) < 512:
|
||||
print("Error: Downloaded image is too small to be a valid FAT filesystem")
|
||||
return 1
|
||||
|
||||
# Try to detect and extract wear leveling layer
|
||||
sector_size = 4096 # Default ESP32 sector size
|
||||
|
||||
# Check if this is a wear-leveling wrapped image
|
||||
if is_esp32_wl_image(fs_data, sector_size):
|
||||
print("Detected Wear Leveling layer, extracting FAT data...")
|
||||
fat_data = extract_fat_from_esp32_wl(fs_data, sector_size)
|
||||
if fat_data is None:
|
||||
print("Error: Failed to extract FAT data from wear-leveling image")
|
||||
return 1
|
||||
fs_data = bytearray(fat_data)
|
||||
print(f" Extracted FAT data: {len(fs_data)} bytes")
|
||||
else:
|
||||
print("No Wear Leveling layer detected, treating as raw FAT image...")
|
||||
|
||||
# Read sector size from FAT boot sector (offset 0x0B, 2 bytes, little-endian)
|
||||
sector_size = int.from_bytes(fs_data[0x0B:0x0D], byteorder='little')
|
||||
|
||||
# Validate sector size
|
||||
if sector_size not in [512, 1024, 2048, 4096]:
|
||||
print(f"Error: Invalid sector size {sector_size}. Must be 512, 1024, 2048, or 4096")
|
||||
return 1
|
||||
|
||||
# Mount with fatfs-python
|
||||
fs_size_adjusted = len(fs_data)
|
||||
sector_count = fs_size_adjusted // sector_size
|
||||
disk = RamDisk(fs_data, sector_size=sector_size, sector_count=sector_count)
|
||||
partition = create_extended_partition(disk)
|
||||
partition.mount()
|
||||
|
||||
# Extract all files using PartitionExtended.walk() and read_file()
|
||||
print("Extracting files:\n")
|
||||
extracted_count = 0
|
||||
for root, dirs, files in partition.walk("/"):
|
||||
# Determine target directory
|
||||
if root == "/":
|
||||
abs_root = unpack_path
|
||||
else:
|
||||
rel_root = root[1:] if root.startswith("/") else root
|
||||
abs_root = unpack_path / rel_root
|
||||
abs_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Extract files in current directory
|
||||
for filename in files:
|
||||
# Construct source path
|
||||
if root == "/":
|
||||
src_file = "/" + filename
|
||||
else:
|
||||
src_file = root.rstrip("/") + "/" + filename
|
||||
|
||||
dst_file = abs_root / filename
|
||||
try:
|
||||
data = partition.read_file(src_file)
|
||||
dst_file.write_bytes(data)
|
||||
print(f" FILE: {src_file} ({len(data)} bytes)")
|
||||
extracted_count += 1
|
||||
except Exception as e:
|
||||
print(f" Warning: Failed to extract {src_file}: {e}")
|
||||
partition.unmount()
|
||||
|
||||
# Summary
|
||||
if extracted_count == 0:
|
||||
print("\nNo files were extracted.")
|
||||
print("The filesystem may be empty, freshly formatted, or contain only deleted entries.")
|
||||
else:
|
||||
print(f"\nSuccessfully extracted {extracted_count} file(s) to {unpack_dir}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def download_fs_action(target, source, env):
|
||||
"""Download and extract filesystem from device."""
|
||||
# Get unpack directory (use global env, not the parameter)
|
||||
unpack_dir = _get_unpack_dir(env)
|
||||
|
||||
# Download partition image
|
||||
fs_file, _fs_start, fs_size, fs_subtype = _download_partition_image(env, None)
|
||||
|
||||
if fs_file is None:
|
||||
return 1
|
||||
|
||||
# Read header for detailed filesystem detection
|
||||
with open(fs_file, 'rb') as f:
|
||||
header = f.read(16384) # Read more to check for offset FAT
|
||||
|
||||
# Detect filesystem type with improved logic
|
||||
fs_type = None
|
||||
|
||||
# 1. Check for LittleFS magic at offset 8 of the superblock
|
||||
if len(header) >= 16 and header[8:16] == b'littlefs':
|
||||
fs_type = "littlefs"
|
||||
|
||||
# 2. Check for FAT filesystem (with or without Wear Leveling)
|
||||
if fs_type is None:
|
||||
# Check multiple possible offsets for FAT boot sector
|
||||
# ESP32 with WL often has FAT at offset 0x1000 (4096)
|
||||
fat_offsets = [0, 4096, 8192]
|
||||
|
||||
for offset in fat_offsets:
|
||||
if len(header) >= offset + 512:
|
||||
boot_sector = header[offset:offset+512]
|
||||
|
||||
# Check for FAT boot signature at offset 510-511
|
||||
if boot_sector[510:512] == b'\x55\xAA':
|
||||
# Additional validation: check for FAT filesystem markers
|
||||
# Check for "FAT" string or "MSDOS" in boot sector
|
||||
if (b'FAT' in boot_sector[0:90] or
|
||||
b'MSDOS' in boot_sector[0:90] or
|
||||
b'MSWIN' in boot_sector[0:90]):
|
||||
# Verify bytes per sector
|
||||
bytes_per_sector = int.from_bytes(boot_sector[11:13], byteorder='little')
|
||||
if bytes_per_sector in [512, 1024, 2048, 4096]:
|
||||
fs_type = "fatfs"
|
||||
print(f" FAT boot sector found at offset 0x{offset:x}")
|
||||
break
|
||||
|
||||
# 3. Fall back to partition table subtype if no clear signature found
|
||||
if fs_type is None:
|
||||
if fs_subtype == 0x81:
|
||||
fs_type = "fatfs"
|
||||
elif fs_subtype == 0x82:
|
||||
# Subtype 0x82 can be either SPIFFS or LittleFS, default to SPIFFS
|
||||
fs_type = "spiffs"
|
||||
elif fs_subtype == 0x83:
|
||||
fs_type = "littlefs"
|
||||
else:
|
||||
print(f"Warning: Unknown partition subtype 0x{fs_subtype:02X}, defaulting to SPIFFS")
|
||||
fs_type = "spiffs"
|
||||
|
||||
print(f"\nDetected filesystem: {fs_type.upper()} (partition subtype: 0x{fs_subtype:02X})")
|
||||
|
||||
# Prepare unpack directory
|
||||
unpack_path = _prepare_unpack_dir(unpack_dir)
|
||||
|
||||
# Extract filesystem
|
||||
try:
|
||||
if fs_type == "littlefs":
|
||||
return _extract_littlefs(fs_file, fs_size, unpack_path, unpack_dir)
|
||||
elif fs_type == "spiffs":
|
||||
return _extract_spiffs(fs_file, fs_size, unpack_path, unpack_dir)
|
||||
elif fs_type == "fatfs":
|
||||
return _extract_fatfs(fs_file, unpack_path, unpack_dir)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
return 1
|
||||
|
||||
|
||||
#
|
||||
# Target: Build executable and linkable firmware or FS image
|
||||
#
|
||||
@@ -1013,6 +1832,17 @@ env.AddPlatformTarget(
|
||||
"Upload Filesystem Image OTA",
|
||||
)
|
||||
|
||||
# Target: Download Filesystem (auto-detect type)
|
||||
env.AddPlatformTarget(
|
||||
"download_fs",
|
||||
None,
|
||||
[
|
||||
env.VerboseAction(BeforeUpload, "Looking for upload port..."),
|
||||
env.VerboseAction(download_fs_action, "Downloading and extracting filesystem")
|
||||
],
|
||||
"Download and extract filesystem from device",
|
||||
)
|
||||
|
||||
# Target: Erase Flash and Upload
|
||||
env.AddPlatformTarget(
|
||||
"erase_upload",
|
||||
|
||||
@@ -45,6 +45,8 @@ PLATFORMIO_URL_VERSION_RE = re.compile(
|
||||
# Python dependencies required for ESP32 platform builds
|
||||
python_deps = {
|
||||
"platformio": "https://github.com/pioarduino/platformio-core/archive/refs/tags/v6.1.18.zip",
|
||||
"littlefs-python": ">=0.16.0",
|
||||
"fatfs-ng": ">=0.1.14",
|
||||
"pyyaml": ">=6.0.2",
|
||||
"rich-click": ">=1.8.6",
|
||||
"zopfli": ">=0.2.2",
|
||||
|
||||
@@ -0,0 +1,762 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# spiffsgen is a tool used to generate a spiffs image from a directory
|
||||
#
|
||||
# SPDX-FileCopyrightText: 2019-2024 Espressif Systems (Shanghai) CO LTD
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
import argparse
|
||||
import io
|
||||
import math
|
||||
import os
|
||||
import struct
|
||||
|
||||
try:
|
||||
import typing
|
||||
|
||||
TSP = typing.TypeVar('TSP', bound='SpiffsObjPageWithIdx')
|
||||
ObjIdsItem = typing.Tuple[int, typing.Type[TSP]]
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
SPIFFS_PH_FLAG_USED_FINAL_INDEX = 0xF8
|
||||
SPIFFS_PH_FLAG_USED_FINAL = 0xFC
|
||||
|
||||
SPIFFS_PH_FLAG_LEN = 1
|
||||
SPIFFS_PH_IX_SIZE_LEN = 4
|
||||
SPIFFS_PH_IX_OBJ_TYPE_LEN = 1
|
||||
SPIFFS_TYPE_FILE = 1
|
||||
|
||||
# Based on typedefs under spiffs_config.h
|
||||
SPIFFS_OBJ_ID_LEN = 2 # spiffs_obj_id
|
||||
SPIFFS_SPAN_IX_LEN = 2 # spiffs_span_ix
|
||||
SPIFFS_PAGE_IX_LEN = 2 # spiffs_page_ix
|
||||
SPIFFS_BLOCK_IX_LEN = 2 # spiffs_block_ix
|
||||
|
||||
|
||||
class SpiffsBuildConfig(object):
|
||||
def __init__(self,
|
||||
page_size, # type: int
|
||||
page_ix_len, # type: int
|
||||
block_size, # type: int
|
||||
block_ix_len, # type: int
|
||||
meta_len, # type: int
|
||||
obj_name_len, # type: int
|
||||
obj_id_len, # type: int
|
||||
span_ix_len, # type: int
|
||||
packed, # type: bool
|
||||
aligned, # type: bool
|
||||
endianness, # type: str
|
||||
use_magic, # type: bool
|
||||
use_magic_len, # type: bool
|
||||
aligned_obj_ix_tables # type: bool
|
||||
):
|
||||
if block_size % page_size != 0:
|
||||
raise RuntimeError('block size should be a multiple of page size')
|
||||
|
||||
self.page_size = page_size
|
||||
self.block_size = block_size
|
||||
self.obj_id_len = obj_id_len
|
||||
self.span_ix_len = span_ix_len
|
||||
self.packed = packed
|
||||
self.aligned = aligned
|
||||
self.obj_name_len = obj_name_len
|
||||
self.meta_len = meta_len
|
||||
self.page_ix_len = page_ix_len
|
||||
self.block_ix_len = block_ix_len
|
||||
self.endianness = endianness
|
||||
self.use_magic = use_magic
|
||||
self.use_magic_len = use_magic_len
|
||||
self.aligned_obj_ix_tables = aligned_obj_ix_tables
|
||||
|
||||
self.PAGES_PER_BLOCK = self.block_size // self.page_size
|
||||
self.OBJ_LU_PAGES_PER_BLOCK = int(math.ceil(self.block_size / self.page_size * self.obj_id_len / self.page_size))
|
||||
self.OBJ_USABLE_PAGES_PER_BLOCK = self.PAGES_PER_BLOCK - self.OBJ_LU_PAGES_PER_BLOCK
|
||||
|
||||
self.OBJ_LU_PAGES_OBJ_IDS_LIM = self.page_size // self.obj_id_len
|
||||
|
||||
self.OBJ_DATA_PAGE_HEADER_LEN = self.obj_id_len + self.span_ix_len + SPIFFS_PH_FLAG_LEN
|
||||
|
||||
pad = 4 - (4 if self.OBJ_DATA_PAGE_HEADER_LEN % 4 == 0 else self.OBJ_DATA_PAGE_HEADER_LEN % 4)
|
||||
|
||||
self.OBJ_DATA_PAGE_HEADER_LEN_ALIGNED = self.OBJ_DATA_PAGE_HEADER_LEN + pad
|
||||
self.OBJ_DATA_PAGE_HEADER_LEN_ALIGNED_PAD = pad
|
||||
self.OBJ_DATA_PAGE_CONTENT_LEN = self.page_size - self.OBJ_DATA_PAGE_HEADER_LEN
|
||||
|
||||
self.OBJ_INDEX_PAGES_HEADER_LEN = (self.OBJ_DATA_PAGE_HEADER_LEN_ALIGNED + SPIFFS_PH_IX_SIZE_LEN +
|
||||
SPIFFS_PH_IX_OBJ_TYPE_LEN + self.obj_name_len + self.meta_len)
|
||||
if aligned_obj_ix_tables:
|
||||
self.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED = (self.OBJ_INDEX_PAGES_HEADER_LEN + SPIFFS_PAGE_IX_LEN - 1) & ~(SPIFFS_PAGE_IX_LEN - 1)
|
||||
self.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED_PAD = self.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED - self.OBJ_INDEX_PAGES_HEADER_LEN
|
||||
else:
|
||||
self.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED = self.OBJ_INDEX_PAGES_HEADER_LEN
|
||||
self.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED_PAD = 0
|
||||
|
||||
self.OBJ_INDEX_PAGES_OBJ_IDS_HEAD_LIM = (self.page_size - self.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED) // self.block_ix_len
|
||||
self.OBJ_INDEX_PAGES_OBJ_IDS_LIM = (self.page_size - self.OBJ_DATA_PAGE_HEADER_LEN_ALIGNED) // self.block_ix_len
|
||||
|
||||
|
||||
class SpiffsFullError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class SpiffsPage(object):
|
||||
_endianness_dict = {
|
||||
'little': '<',
|
||||
'big': '>'
|
||||
}
|
||||
|
||||
_len_dict = {
|
||||
1: 'B',
|
||||
2: 'H',
|
||||
4: 'I',
|
||||
8: 'Q'
|
||||
}
|
||||
|
||||
def __init__(self, bix, build_config): # type: (int, SpiffsBuildConfig) -> None
|
||||
self.build_config = build_config
|
||||
self.bix = bix
|
||||
|
||||
def to_binary(self): # type: () -> bytes
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
class SpiffsObjPageWithIdx(SpiffsPage):
|
||||
def __init__(self, obj_id, build_config): # type: (int, SpiffsBuildConfig) -> None
|
||||
super(SpiffsObjPageWithIdx, self).__init__(0, build_config)
|
||||
self.obj_id = obj_id
|
||||
|
||||
def to_binary(self): # type: () -> bytes
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
class SpiffsObjLuPage(SpiffsPage):
|
||||
def __init__(self, bix, build_config): # type: (int, SpiffsBuildConfig) -> None
|
||||
SpiffsPage.__init__(self, bix, build_config)
|
||||
|
||||
self.obj_ids_limit = self.build_config.OBJ_LU_PAGES_OBJ_IDS_LIM
|
||||
self.obj_ids = list() # type: typing.List[ObjIdsItem]
|
||||
|
||||
def _calc_magic(self, blocks_lim): # type: (int) -> int
|
||||
# Calculate the magic value mirroring computation done by the macro SPIFFS_MAGIC defined in
|
||||
# spiffs_nucleus.h
|
||||
magic = 0x20140529 ^ self.build_config.page_size
|
||||
if self.build_config.use_magic_len:
|
||||
magic = magic ^ (blocks_lim - self.bix)
|
||||
# narrow the result to build_config.obj_id_len bytes
|
||||
mask = (2 << (8 * self.build_config.obj_id_len)) - 1
|
||||
return magic & mask
|
||||
|
||||
def register_page(self, page): # type: (TSP) -> None
|
||||
if not self.obj_ids_limit > 0:
|
||||
raise SpiffsFullError()
|
||||
|
||||
obj_id = (page.obj_id, page.__class__)
|
||||
self.obj_ids.append(obj_id)
|
||||
self.obj_ids_limit -= 1
|
||||
|
||||
def to_binary(self): # type: () -> bytes
|
||||
img = b''
|
||||
|
||||
for (obj_id, page_type) in self.obj_ids:
|
||||
if page_type == SpiffsObjIndexPage:
|
||||
obj_id ^= (1 << ((self.build_config.obj_id_len * 8) - 1))
|
||||
img += struct.pack(SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[self.build_config.obj_id_len], obj_id)
|
||||
|
||||
assert len(img) <= self.build_config.page_size
|
||||
|
||||
img += b'\xFF' * (self.build_config.page_size - len(img))
|
||||
|
||||
return img
|
||||
|
||||
def magicfy(self, blocks_lim): # type: (int) -> None
|
||||
# Only use magic value if no valid obj id has been written to the spot, which is the
|
||||
# spot taken up by the last obj id on last lookup page. The parent is responsible
|
||||
# for determining which is the last lookup page and calling this function.
|
||||
remaining = self.obj_ids_limit
|
||||
empty_obj_id_dict = {
|
||||
1: 0xFF,
|
||||
2: 0xFFFF,
|
||||
4: 0xFFFFFFFF,
|
||||
8: 0xFFFFFFFFFFFFFFFF
|
||||
}
|
||||
if remaining >= 2:
|
||||
for i in range(remaining):
|
||||
if i == remaining - 2:
|
||||
self.obj_ids.append((self._calc_magic(blocks_lim), SpiffsObjDataPage))
|
||||
break
|
||||
else:
|
||||
self.obj_ids.append((empty_obj_id_dict[self.build_config.obj_id_len], SpiffsObjDataPage))
|
||||
self.obj_ids_limit -= 1
|
||||
|
||||
|
||||
class SpiffsObjIndexPage(SpiffsObjPageWithIdx):
|
||||
def __init__(self, obj_id, span_ix, size, name, build_config
|
||||
): # type: (int, int, int, str, SpiffsBuildConfig) -> None
|
||||
super(SpiffsObjIndexPage, self).__init__(obj_id, build_config)
|
||||
self.span_ix = span_ix
|
||||
self.name = name
|
||||
self.size = size
|
||||
|
||||
if self.span_ix == 0:
|
||||
self.pages_lim = self.build_config.OBJ_INDEX_PAGES_OBJ_IDS_HEAD_LIM
|
||||
else:
|
||||
self.pages_lim = self.build_config.OBJ_INDEX_PAGES_OBJ_IDS_LIM
|
||||
|
||||
self.pages = list() # type: typing.List[int]
|
||||
|
||||
def register_page(self, page): # type: (SpiffsObjDataPage) -> None
|
||||
if not self.pages_lim > 0:
|
||||
raise SpiffsFullError()
|
||||
|
||||
self.pages.append(page.offset)
|
||||
self.pages_lim -= 1
|
||||
|
||||
def to_binary(self): # type: () -> bytes
|
||||
obj_id = self.obj_id ^ (1 << ((self.build_config.obj_id_len * 8) - 1))
|
||||
img = struct.pack(SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[self.build_config.obj_id_len] +
|
||||
SpiffsPage._len_dict[self.build_config.span_ix_len] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_FLAG_LEN],
|
||||
obj_id,
|
||||
self.span_ix,
|
||||
SPIFFS_PH_FLAG_USED_FINAL_INDEX)
|
||||
|
||||
# Add padding before the object index page specific information
|
||||
img += b'\xFF' * self.build_config.OBJ_DATA_PAGE_HEADER_LEN_ALIGNED_PAD
|
||||
|
||||
# If this is the first object index page for the object, add filename, type
|
||||
# and size information
|
||||
if self.span_ix == 0:
|
||||
img += struct.pack(SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_IX_SIZE_LEN] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_FLAG_LEN],
|
||||
self.size,
|
||||
SPIFFS_TYPE_FILE)
|
||||
|
||||
img += self.name.encode() + (b'\x00' * (
|
||||
(self.build_config.obj_name_len - len(self.name))
|
||||
+ self.build_config.meta_len
|
||||
+ self.build_config.OBJ_INDEX_PAGES_HEADER_LEN_ALIGNED_PAD))
|
||||
|
||||
# Finally, add the page index of data pages
|
||||
for page in self.pages:
|
||||
page = page >> int(math.log(self.build_config.page_size, 2))
|
||||
img += struct.pack(SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[self.build_config.page_ix_len], page)
|
||||
|
||||
assert len(img) <= self.build_config.page_size
|
||||
|
||||
img += b'\xFF' * (self.build_config.page_size - len(img))
|
||||
|
||||
return img
|
||||
|
||||
|
||||
class SpiffsObjDataPage(SpiffsObjPageWithIdx):
|
||||
def __init__(self, offset, obj_id, span_ix, contents, build_config
|
||||
): # type: (int, int, int, bytes, SpiffsBuildConfig) -> None
|
||||
super(SpiffsObjDataPage, self).__init__(obj_id, build_config)
|
||||
self.span_ix = span_ix
|
||||
self.contents = contents
|
||||
self.offset = offset
|
||||
|
||||
def to_binary(self): # type: () -> bytes
|
||||
img = struct.pack(SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[self.build_config.obj_id_len] +
|
||||
SpiffsPage._len_dict[self.build_config.span_ix_len] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_FLAG_LEN],
|
||||
self.obj_id,
|
||||
self.span_ix,
|
||||
SPIFFS_PH_FLAG_USED_FINAL)
|
||||
|
||||
img += self.contents
|
||||
|
||||
assert len(img) <= self.build_config.page_size
|
||||
|
||||
img += b'\xFF' * (self.build_config.page_size - len(img))
|
||||
|
||||
return img
|
||||
|
||||
|
||||
class SpiffsBlock(object):
|
||||
def _reset(self): # type: () -> None
|
||||
self.cur_obj_index_span_ix = 0
|
||||
self.cur_obj_data_span_ix = 0
|
||||
self.cur_obj_id = 0
|
||||
self.cur_obj_idx_page = None # type: typing.Optional[SpiffsObjIndexPage]
|
||||
|
||||
def __init__(self, bix, build_config): # type: (int, SpiffsBuildConfig) -> None
|
||||
self.build_config = build_config
|
||||
self.offset = bix * self.build_config.block_size
|
||||
self.remaining_pages = self.build_config.OBJ_USABLE_PAGES_PER_BLOCK
|
||||
self.pages = list() # type: typing.List[SpiffsPage]
|
||||
self.bix = bix
|
||||
|
||||
lu_pages = list()
|
||||
for i in range(self.build_config.OBJ_LU_PAGES_PER_BLOCK):
|
||||
page = SpiffsObjLuPage(self.bix, self.build_config)
|
||||
lu_pages.append(page)
|
||||
|
||||
self.pages.extend(lu_pages)
|
||||
|
||||
self.lu_page_iter = iter(lu_pages)
|
||||
self.lu_page = next(self.lu_page_iter)
|
||||
|
||||
self._reset()
|
||||
|
||||
def _register_page(self, page): # type: (TSP) -> None
|
||||
if isinstance(page, SpiffsObjDataPage):
|
||||
assert self.cur_obj_idx_page is not None
|
||||
self.cur_obj_idx_page.register_page(page) # can raise SpiffsFullError
|
||||
|
||||
try:
|
||||
self.lu_page.register_page(page)
|
||||
except SpiffsFullError:
|
||||
self.lu_page = next(self.lu_page_iter)
|
||||
try:
|
||||
self.lu_page.register_page(page)
|
||||
except AttributeError: # no next lookup page
|
||||
# Since the amount of lookup pages is pre-computed at every block instance,
|
||||
# this should never occur
|
||||
raise RuntimeError('invalid attempt to add page to a block when there is no more space in lookup')
|
||||
|
||||
self.pages.append(page)
|
||||
|
||||
def begin_obj(self, obj_id, size, name, obj_index_span_ix=0, obj_data_span_ix=0
|
||||
): # type: (int, int, str, int, int) -> None
|
||||
if not self.remaining_pages > 0:
|
||||
raise SpiffsFullError()
|
||||
self._reset()
|
||||
|
||||
self.cur_obj_id = obj_id
|
||||
self.cur_obj_index_span_ix = obj_index_span_ix
|
||||
self.cur_obj_data_span_ix = obj_data_span_ix
|
||||
|
||||
page = SpiffsObjIndexPage(obj_id, self.cur_obj_index_span_ix, size, name, self.build_config)
|
||||
self._register_page(page)
|
||||
|
||||
self.cur_obj_idx_page = page
|
||||
|
||||
self.remaining_pages -= 1
|
||||
self.cur_obj_index_span_ix += 1
|
||||
|
||||
def update_obj(self, contents): # type: (bytes) -> None
|
||||
if not self.remaining_pages > 0:
|
||||
raise SpiffsFullError()
|
||||
page = SpiffsObjDataPage(self.offset + (len(self.pages) * self.build_config.page_size),
|
||||
self.cur_obj_id, self.cur_obj_data_span_ix, contents, self.build_config)
|
||||
|
||||
self._register_page(page)
|
||||
|
||||
self.cur_obj_data_span_ix += 1
|
||||
self.remaining_pages -= 1
|
||||
|
||||
def end_obj(self): # type: () -> None
|
||||
self._reset()
|
||||
|
||||
def is_full(self): # type: () -> bool
|
||||
return self.remaining_pages <= 0
|
||||
|
||||
def to_binary(self, blocks_lim): # type: (int) -> bytes
|
||||
img = b''
|
||||
|
||||
if self.build_config.use_magic:
|
||||
for (idx, page) in enumerate(self.pages):
|
||||
if idx == self.build_config.OBJ_LU_PAGES_PER_BLOCK - 1:
|
||||
assert isinstance(page, SpiffsObjLuPage)
|
||||
page.magicfy(blocks_lim)
|
||||
img += page.to_binary()
|
||||
else:
|
||||
for page in self.pages:
|
||||
img += page.to_binary()
|
||||
|
||||
assert len(img) <= self.build_config.block_size
|
||||
|
||||
img += b'\xFF' * (self.build_config.block_size - len(img))
|
||||
return img
|
||||
|
||||
def _parse_from_binary(self, block_data): # type: (bytes) -> None
|
||||
"""Parse block data from binary image.
|
||||
|
||||
Args:
|
||||
block_data: Raw block bytes
|
||||
"""
|
||||
self._raw_data = block_data
|
||||
|
||||
|
||||
class SpiffsFS(object):
|
||||
def __init__(self, img_size, build_config): # type: (int, SpiffsBuildConfig) -> None
|
||||
if img_size % build_config.block_size != 0:
|
||||
raise RuntimeError('image size should be a multiple of block size')
|
||||
|
||||
self.img_size = img_size
|
||||
self.build_config = build_config
|
||||
|
||||
self.blocks = list() # type: typing.List[SpiffsBlock]
|
||||
self.blocks_lim = self.img_size // self.build_config.block_size
|
||||
self.remaining_blocks = self.blocks_lim
|
||||
self.cur_obj_id = 1 # starting object id
|
||||
|
||||
def _create_block(self): # type: () -> SpiffsBlock
|
||||
if self.is_full():
|
||||
raise SpiffsFullError('the image size has been exceeded')
|
||||
|
||||
block = SpiffsBlock(len(self.blocks), self.build_config)
|
||||
self.blocks.append(block)
|
||||
self.remaining_blocks -= 1
|
||||
return block
|
||||
|
||||
def is_full(self): # type: () -> bool
|
||||
return self.remaining_blocks <= 0
|
||||
|
||||
def create_file(self, img_path, file_path): # type: (str, str) -> None
|
||||
if len(img_path) > self.build_config.obj_name_len:
|
||||
raise RuntimeError("object name '%s' too long" % img_path)
|
||||
|
||||
name = img_path
|
||||
|
||||
with open(file_path, 'rb') as obj:
|
||||
contents = obj.read()
|
||||
|
||||
stream = io.BytesIO(contents)
|
||||
|
||||
try:
|
||||
block = self.blocks[-1]
|
||||
block.begin_obj(self.cur_obj_id, len(contents), name)
|
||||
except (IndexError, SpiffsFullError):
|
||||
block = self._create_block()
|
||||
block.begin_obj(self.cur_obj_id, len(contents), name)
|
||||
|
||||
contents_chunk = stream.read(self.build_config.OBJ_DATA_PAGE_CONTENT_LEN)
|
||||
|
||||
while contents_chunk:
|
||||
try:
|
||||
block = self.blocks[-1]
|
||||
try:
|
||||
# This can fail because either (1) all the pages in block have been
|
||||
# used or (2) object index has been exhausted.
|
||||
block.update_obj(contents_chunk)
|
||||
except SpiffsFullError:
|
||||
# If its (1), use the outer exception handler
|
||||
if block.is_full():
|
||||
raise SpiffsFullError
|
||||
# If its (2), write another object index page
|
||||
block.begin_obj(self.cur_obj_id, len(contents), name,
|
||||
obj_index_span_ix=block.cur_obj_index_span_ix,
|
||||
obj_data_span_ix=block.cur_obj_data_span_ix)
|
||||
continue
|
||||
except (IndexError, SpiffsFullError):
|
||||
# All pages in the block have been exhausted. Create a new block, copying
|
||||
# the previous state of the block to a new one for the continuation of the
|
||||
# current object
|
||||
prev_block = block
|
||||
block = self._create_block()
|
||||
block.cur_obj_id = prev_block.cur_obj_id
|
||||
block.cur_obj_idx_page = prev_block.cur_obj_idx_page
|
||||
block.cur_obj_data_span_ix = prev_block.cur_obj_data_span_ix
|
||||
block.cur_obj_index_span_ix = prev_block.cur_obj_index_span_ix
|
||||
continue
|
||||
|
||||
contents_chunk = stream.read(self.build_config.OBJ_DATA_PAGE_CONTENT_LEN)
|
||||
|
||||
block.end_obj()
|
||||
|
||||
self.cur_obj_id += 1
|
||||
|
||||
def to_binary(self): # type: () -> bytes
|
||||
img = b''
|
||||
all_blocks = []
|
||||
for block in self.blocks:
|
||||
all_blocks.append(block.to_binary(self.blocks_lim))
|
||||
bix = len(self.blocks)
|
||||
if self.build_config.use_magic:
|
||||
# Create empty blocks with magic numbers
|
||||
while self.remaining_blocks > 0:
|
||||
block = SpiffsBlock(bix, self.build_config)
|
||||
all_blocks.append(block.to_binary(self.blocks_lim))
|
||||
self.remaining_blocks -= 1
|
||||
bix += 1
|
||||
else:
|
||||
# Just fill remaining spaces FF's
|
||||
all_blocks.append(b'\xFF' * (self.img_size - len(all_blocks) * self.build_config.block_size))
|
||||
img += b''.join([blk for blk in all_blocks])
|
||||
return img
|
||||
|
||||
def from_binary(self, image_data): # type: (bytes) -> None
|
||||
"""Parse a SPIFFS binary image and populate the filesystem structure.
|
||||
|
||||
Args:
|
||||
image_data: Raw SPIFFS image bytes
|
||||
"""
|
||||
if len(image_data) != self.img_size:
|
||||
raise RuntimeError(f'image size mismatch: expected {self.img_size}, got {len(image_data)}')
|
||||
|
||||
# Parse blocks from the image
|
||||
blocks_count = self.img_size // self.build_config.block_size
|
||||
for bix in range(blocks_count):
|
||||
block_offset = bix * self.build_config.block_size
|
||||
block_data = image_data[block_offset:block_offset + self.build_config.block_size]
|
||||
block = SpiffsBlock(bix, self.build_config)
|
||||
block._parse_from_binary(block_data)
|
||||
self.blocks.append(block)
|
||||
|
||||
def extract_files(self, output_dir): # type: (str) -> int
|
||||
"""Extract all files from the SPIFFS filesystem to a directory.
|
||||
|
||||
Args:
|
||||
output_dir: Directory path where files will be extracted
|
||||
|
||||
Returns:
|
||||
int: Number of files extracted
|
||||
"""
|
||||
|
||||
# Build a map of object_id -> file info
|
||||
files_map = {} # obj_id -> {'name': str, 'size': int, 'data_pages': [(span_ix, page_data)]}
|
||||
|
||||
for block in self.blocks:
|
||||
# Parse lookup pages to find valid objects
|
||||
for page_idx in range(self.build_config.OBJ_LU_PAGES_PER_BLOCK):
|
||||
lu_page_offset = page_idx * self.build_config.page_size
|
||||
lu_page_data = block._raw_data[lu_page_offset:lu_page_offset + self.build_config.page_size]
|
||||
|
||||
# Parse object IDs from lookup page
|
||||
for i in range(0, len(lu_page_data), self.build_config.obj_id_len):
|
||||
if i + self.build_config.obj_id_len > len(lu_page_data):
|
||||
break
|
||||
|
||||
obj_id_bytes = lu_page_data[i:i + self.build_config.obj_id_len]
|
||||
obj_id = struct.unpack(
|
||||
SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[self.build_config.obj_id_len],
|
||||
obj_id_bytes
|
||||
)[0]
|
||||
|
||||
# Check if it's a valid object (not erased/empty)
|
||||
empty_values = {1: 0xFF, 2: 0xFFFF, 4: 0xFFFFFFFF, 8: 0xFFFFFFFFFFFFFFFF}
|
||||
if obj_id == empty_values[self.build_config.obj_id_len]:
|
||||
continue
|
||||
|
||||
# Check if it's an index page (MSB set)
|
||||
is_index = obj_id & (1 << ((self.build_config.obj_id_len * 8) - 1))
|
||||
real_obj_id = obj_id & ~(1 << ((self.build_config.obj_id_len * 8) - 1))
|
||||
|
||||
if is_index and real_obj_id not in files_map:
|
||||
files_map[real_obj_id] = {'name': None, 'size': 0, 'data_pages': []}
|
||||
|
||||
# Parse actual pages to get file metadata and content
|
||||
for page_idx in range(self.build_config.OBJ_LU_PAGES_PER_BLOCK, self.build_config.PAGES_PER_BLOCK):
|
||||
page_offset = page_idx * self.build_config.page_size
|
||||
page_data = block._raw_data[page_offset:page_offset + self.build_config.page_size]
|
||||
|
||||
# Parse page header
|
||||
header_fmt = (
|
||||
SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[self.build_config.obj_id_len] +
|
||||
SpiffsPage._len_dict[self.build_config.span_ix_len] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_FLAG_LEN]
|
||||
)
|
||||
header_size = struct.calcsize(header_fmt)
|
||||
|
||||
if len(page_data) < header_size:
|
||||
continue
|
||||
|
||||
obj_id, span_ix, flags = struct.unpack(header_fmt, page_data[:header_size])
|
||||
|
||||
# Check for valid page
|
||||
empty_id = {1: 0xFF, 2: 0xFFFF, 4: 0xFFFFFFFF, 8: 0xFFFFFFFFFFFFFFFF}[self.build_config.obj_id_len]
|
||||
if obj_id == empty_id:
|
||||
continue
|
||||
|
||||
is_index = obj_id & (1 << ((self.build_config.obj_id_len * 8) - 1))
|
||||
real_obj_id = obj_id & ~(1 << ((self.build_config.obj_id_len * 8) - 1))
|
||||
|
||||
if is_index and flags == SPIFFS_PH_FLAG_USED_FINAL_INDEX:
|
||||
# Index page - contains file metadata
|
||||
if real_obj_id not in files_map:
|
||||
files_map[real_obj_id] = {'name': None, 'size': 0, 'data_pages': []}
|
||||
|
||||
# Only first index page (span_ix == 0) has filename and size
|
||||
if span_ix == 0:
|
||||
# Skip to size and type fields
|
||||
offset = header_size + self.build_config.OBJ_DATA_PAGE_HEADER_LEN_ALIGNED_PAD
|
||||
size_type_fmt = (
|
||||
SpiffsPage._endianness_dict[self.build_config.endianness] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_IX_SIZE_LEN] +
|
||||
SpiffsPage._len_dict[SPIFFS_PH_IX_OBJ_TYPE_LEN]
|
||||
)
|
||||
size_type_size = struct.calcsize(size_type_fmt)
|
||||
|
||||
if offset + size_type_size <= len(page_data):
|
||||
file_size, obj_type = struct.unpack(size_type_fmt, page_data[offset:offset + size_type_size])
|
||||
offset += size_type_size
|
||||
|
||||
# Read filename
|
||||
name_end = offset + self.build_config.obj_name_len
|
||||
if name_end <= len(page_data):
|
||||
name_bytes = page_data[offset:name_end]
|
||||
# Find null terminator
|
||||
null_pos = name_bytes.find(b'\x00')
|
||||
if null_pos != -1:
|
||||
name_bytes = name_bytes[:null_pos]
|
||||
filename = name_bytes.decode('utf-8', errors='ignore')
|
||||
files_map[real_obj_id]['name'] = filename
|
||||
files_map[real_obj_id]['size'] = file_size
|
||||
|
||||
elif not is_index and flags == SPIFFS_PH_FLAG_USED_FINAL:
|
||||
# Data page - contains file content
|
||||
if real_obj_id in files_map:
|
||||
# Extract content (skip header, no padding on data pages)
|
||||
content_start = header_size
|
||||
content = page_data[content_start:content_start + self.build_config.OBJ_DATA_PAGE_CONTENT_LEN]
|
||||
files_map[real_obj_id]['data_pages'].append((span_ix, content))
|
||||
|
||||
# Extract files to output directory
|
||||
file_count = 0
|
||||
for obj_id, file_info in files_map.items():
|
||||
if file_info['name'] is None:
|
||||
continue
|
||||
|
||||
# Remove leading slash if present
|
||||
rel_path = file_info['name'].lstrip('/')
|
||||
file_path = os.path.join(output_dir, rel_path)
|
||||
if not rel_path:
|
||||
print(f" Warning: Skipping file with empty path (obj_id={obj_id})")
|
||||
continue
|
||||
|
||||
# Create parent directories
|
||||
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
||||
|
||||
# Sort data pages by span index
|
||||
file_info['data_pages'].sort(key=lambda x: x[0])
|
||||
|
||||
# Write file content
|
||||
with open(file_path, 'wb') as f:
|
||||
total_written = 0
|
||||
for span_ix, content in file_info['data_pages']:
|
||||
# Write only up to the file size
|
||||
remaining = file_info['size'] - total_written
|
||||
if remaining <= 0:
|
||||
break
|
||||
to_write = min(len(content), remaining)
|
||||
f.write(content[:to_write])
|
||||
total_written += to_write
|
||||
|
||||
file_count += 1
|
||||
print(f" Extracted: {file_info['name']} ({file_info['size']} bytes)")
|
||||
|
||||
return file_count
|
||||
|
||||
|
||||
class CustomHelpFormatter(argparse.HelpFormatter):
|
||||
"""
|
||||
Similar to argparse.ArgumentDefaultsHelpFormatter, except it
|
||||
doesn't add the default value if "(default:" is already present.
|
||||
This helps in the case of options with action="store_false", like
|
||||
--no-magic or --no-magic-len.
|
||||
"""
|
||||
def _get_help_string(self, action): # type: (argparse.Action) -> str
|
||||
if action.help is None:
|
||||
return ''
|
||||
if '%(default)' not in action.help and '(default:' not in action.help:
|
||||
if action.default is not argparse.SUPPRESS:
|
||||
defaulting_nargs = [argparse.OPTIONAL, argparse.ZERO_OR_MORE]
|
||||
if action.option_strings or action.nargs in defaulting_nargs:
|
||||
return action.help + ' (default: %(default)s)'
|
||||
return action.help
|
||||
|
||||
|
||||
def main(): # type: () -> None
|
||||
parser = argparse.ArgumentParser(description='SPIFFS Image Generator',
|
||||
formatter_class=CustomHelpFormatter)
|
||||
|
||||
parser.add_argument('image_size',
|
||||
help='Size of the created image')
|
||||
|
||||
parser.add_argument('base_dir',
|
||||
help='Path to directory from which the image will be created')
|
||||
|
||||
parser.add_argument('output_file',
|
||||
help='Created image output file path')
|
||||
|
||||
parser.add_argument('--page-size',
|
||||
help='Logical page size. Set to value same as CONFIG_SPIFFS_PAGE_SIZE.',
|
||||
type=int,
|
||||
default=256)
|
||||
|
||||
parser.add_argument('--block-size',
|
||||
help="Logical block size. Set to the same value as the flash chip's sector size (g_rom_flashchip.sector_size).",
|
||||
type=int,
|
||||
default=4096)
|
||||
|
||||
parser.add_argument('--obj-name-len',
|
||||
help='File full path maximum length. Set to value same as CONFIG_SPIFFS_OBJ_NAME_LEN.',
|
||||
type=int,
|
||||
default=32)
|
||||
|
||||
parser.add_argument('--meta-len',
|
||||
help='File metadata length. Set to value same as CONFIG_SPIFFS_META_LENGTH.',
|
||||
type=int,
|
||||
default=4)
|
||||
|
||||
parser.add_argument('--use-magic',
|
||||
dest='use_magic',
|
||||
help='Use magic number to create an identifiable SPIFFS image. Specify if CONFIG_SPIFFS_USE_MAGIC.',
|
||||
action='store_true')
|
||||
|
||||
parser.add_argument('--no-magic',
|
||||
dest='use_magic',
|
||||
help='Inverse of --use-magic (default: --use-magic is enabled)',
|
||||
action='store_false')
|
||||
|
||||
parser.add_argument('--use-magic-len',
|
||||
dest='use_magic_len',
|
||||
help='Use position in memory to create different magic numbers for each block. Specify if CONFIG_SPIFFS_USE_MAGIC_LENGTH.',
|
||||
action='store_true')
|
||||
|
||||
parser.add_argument('--no-magic-len',
|
||||
dest='use_magic_len',
|
||||
help='Inverse of --use-magic-len (default: --use-magic-len is enabled)',
|
||||
action='store_false')
|
||||
|
||||
parser.add_argument('--follow-symlinks',
|
||||
help='Take into account symbolic links during partition image creation.',
|
||||
action='store_true')
|
||||
|
||||
parser.add_argument('--big-endian',
|
||||
help='Specify if the target architecture is big-endian. If not specified, little-endian is assumed.',
|
||||
action='store_true')
|
||||
|
||||
parser.add_argument('--aligned-obj-ix-tables',
|
||||
action='store_true',
|
||||
help='Use aligned object index tables. Specify if SPIFFS_ALIGNED_OBJECT_INDEX_TABLES is set.')
|
||||
|
||||
parser.set_defaults(use_magic=True, use_magic_len=True)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if not os.path.exists(args.base_dir):
|
||||
raise RuntimeError('given base directory %s does not exist' % args.base_dir)
|
||||
|
||||
with open(args.output_file, 'wb') as image_file:
|
||||
image_size = int(args.image_size, 0)
|
||||
spiffs_build_default = SpiffsBuildConfig(args.page_size, SPIFFS_PAGE_IX_LEN,
|
||||
args.block_size, SPIFFS_BLOCK_IX_LEN, args.meta_len,
|
||||
args.obj_name_len, SPIFFS_OBJ_ID_LEN, SPIFFS_SPAN_IX_LEN,
|
||||
True, True, 'big' if args.big_endian else 'little',
|
||||
args.use_magic, args.use_magic_len, args.aligned_obj_ix_tables)
|
||||
|
||||
spiffs = SpiffsFS(image_size, spiffs_build_default)
|
||||
|
||||
for root, dirs, files in os.walk(args.base_dir, followlinks=args.follow_symlinks):
|
||||
for f in files:
|
||||
full_path = os.path.join(root, f)
|
||||
spiffs.create_file('/' + os.path.relpath(full_path, args.base_dir).replace('\\', '/'), full_path)
|
||||
|
||||
image = spiffs.to_binary()
|
||||
|
||||
image_file.write(image)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user