#!/usr/bin/env python3
"""Stage a completed Option 3 matrix as audited, directly downloadable releases."""
from __future__ import annotations
import argparse
import hashlib
import html
import json
from pathlib import Path
import re
import shutil
import sys
from urllib.parse import quote, urljoin
sys.path.insert(0, str(Path(__file__).resolve().parent))
from firmware_memory_manifest import validate_package
ROOT = Path(__file__).resolve().parents[1]
FIRMWARE_SUFFIXES = {".bin", ".uf2", ".hex", ".zip"}
def is_companion(target):
return "companion" in target.lower() or "comp_radio" in target.lower()
def validate_manifest(manifest):
target = manifest["target"]
if manifest.get("schema_version", 0) < 2 or not manifest.get("verified"):
raise ValueError(f"{target}: firmware capability qualification failed")
if not is_companion(target) and not manifest.get("ota_update_verified"):
raise ValueError(f"{target}: infrastructure has no verified wireless updater")
if "companion_radio_full" in target.lower():
required = {"companion.usb_mota_source", "companion.mota_sender",
"companion.temp_radio", "companion.ota_cli"}
required.add("companion.wifi_ota_seeder" if manifest["platform"] == "ESP32_PLATFORM"
else "companion.ble_mota_source")
verified = {item["capability"] for item in manifest["verification"] if item["present"]}
if not required <= verified:
raise ValueError(f"{target}: Full Companion lacks verified MOTA sending")
def collect_artifacts(directory, version):
records = []
accounted = set()
for path in sorted(directory.glob("*.capabilities.json")):
stem = path.name.removesuffix(".capabilities.json")
files = [candidate for candidate in directory.glob(stem + ".*")
if candidate.suffix in FIRMWARE_SUFFIXES]
merged = directory / (stem + "-merged.bin")
if merged.is_file():
files.append(merged)
if not files:
continue # A measured rejected attempt may leave only a manifest.
if not stem.endswith("-" + version):
raise ValueError(f"stale or mixed-version artifact: {stem}")
manifest = json.loads(path.read_text())
validate_manifest(manifest)
extensions = {item.suffix for item in files}
platform = manifest["platform"]
if platform == "ESP32_PLATFORM" and (not merged.is_file() or not (directory / (stem + ".bin")).is_file()):
raise ValueError(f"{stem}: ESP32 application/merged pair incomplete")
if platform == "NRF52_PLATFORM":
if ".uf2" not in extensions or (manifest.get("ota_update_verified") and ".zip" not in extensions):
raise ValueError(f"{stem}: nRF52 UF2/DFU artifacts incomplete")
if any(item.stat().st_size == 0 for item in files):
raise ValueError(f"{stem}: empty firmware artifact")
memory = validate_package(directory / stem)
manifest["runtime_ram"] = {key: memory[key] for key in (
"passed", "available_internal_bytes", "required_heap_bytes", "elf_sha256")}
accounted.update(files)
records.append({"manifest": manifest, "files": sorted(files) + [path, directory / (stem + ".memory.json")]})
unaccounted = {path for path in directory.iterdir() if path.suffix in FIRMWARE_SUFFIXES} - accounted
if unaccounted:
raise ValueError("firmware without qualification: " + ", ".join(sorted(p.name for p in unaccounted)))
if not records:
raise ValueError("no qualified firmware artifacts")
return records
def category(record):
manifest = record["manifest"]
if is_companion(manifest["target"]):
return "companion"
if manifest["build_profile"] == "full":
return "full-profiles"
if "lora_ota" in manifest["target"].lower():
return "lora-ota"
if "logging" in manifest["artifact_target"].lower():
return "logging"
if any(role in manifest["target"].lower() for role in ("sensor", "terminal")):
return "utility"
return "repeater-room"
def sha256(path):
with path.open("rb") as stream:
return hashlib.file_digest(stream, "sha256").hexdigest()
def completed_matrix_failures(status, allow_partial=False):
if status.get("state") == "completed" and status.get("exit_code") == "0":
return []
if not allow_partial:
raise ValueError("the Option 3 matrix has not completed successfully")
if status.get("state") != "failed" or status.get("exit_code") != "1":
raise ValueError("partial publication requires a finished matrix, not a running or interrupted build")
log = Path(status.get("log", ""))
if not log.is_file():
raise ValueError("partial publication requires the matrix failure summary")
text = log.read_text(errors="replace")
summaries = list(re.finditer(r"^Logging matrix completed with (\d+) failed build\(s\):\s*$", text, re.M))
if not summaries:
raise ValueError("partial publication requires the matrix failure summary")
summary = summaries[-1]
count = int(summary[1])
lines = text[summary.end():].lstrip("\r\n").splitlines()
failures = []
for line in lines[:count]:
match = re.fullmatch(r" ([\w.+-]+) \(([^()]+)\) -> (.+)", line)
if not match:
raise ValueError("matrix failure summary is incomplete")
failures.append({"target": match[1], "profile": match[2],
"log_file": Path(match[3]).name})
if not count or len(failures) != count:
raise ValueError("matrix failure summary is incomplete")
return failures
def portable_profile_exclusions(status):
if not status.get("log"):
return [] # Older successful status files did not require a log path.
text = Path(status["log"]).read_text(errors="replace")
summaries = list(re.finditer(
r"^(\d+) standard ESP32 target\(s\) exceeded the portable OTA slot and were deferred to the expanded FULL pass:\s*$",
text, re.M))
if not summaries:
if re.search(r"^DEFERRED: ", text, re.M):
raise ValueError("portable-profile exclusion summary is missing")
return []
summary = summaries[-1]
count = int(summary[1])
names = []
for line in text[summary.end():].lstrip("\r\n").splitlines()[:count]:
match = re.fullmatch(r" ([\w.+-]+)", line)
if not match:
raise ValueError("portable-profile exclusion summary is incomplete")
names.append(match[1])
if not count or len(names) != count or len(set(names)) != count:
raise ValueError("portable-profile exclusion summary is incomplete")
return [{"target": name, "profile": "standard", "reason": "portable_slot_overflow"}
for name in names]
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
parser.add_argument("--build-status", required=True, type=Path)
parser.add_argument("--version", default="1.17.1.5")
parser.add_argument("--commit", required=True)
parser.add_argument("--repo", default="mikecarper/MeshCore")
parser.add_argument("--allow-partial", action="store_true",
help="Publish qualified outputs from a finished matrix with failures, listing every failed attempt")
args = parser.parse_args()
status = dict(line.split("=", 1) for line in args.build_status.read_text().splitlines() if "=" in line)
failures = completed_matrix_failures(status, args.allow_partial)
portable_exclusions = portable_profile_exclusions(status)
output_directory = Path(status["working_directory"]) / status["output_directory"]
if output_directory.resolve() != args.input.resolve():
raise ValueError("build status belongs to another output directory")
firmware_label = f"v{args.version}-halo-keymind-cascade-dev"
if status.get("source_commit") != args.commit or status.get("firmware_version") != firmware_label:
raise ValueError("build status belongs to another source revision or version")
if status.get("firmware_profile") != "cascade":
raise ValueError("matrix did not use Cascade runtime defaults")
radio = {key: status["radio_" + key] for key in ("frequency", "bandwidth", "sf", "cr")}
if args.output.exists() and any(args.output.iterdir()):
raise ValueError("staging directory must be empty; existing releases are never overwritten")
version = f"{firmware_label}-{args.commit[:8]}"
records = collect_artifacts(args.input, version)
base_tag = version
groups = []
titles = {"companion": "Companion builds", "repeater-room": "Repeater and Room Server builds",
"utility": "Sensor and Terminal utilities", "logging": "USB packet logging builds",
"lora-ota": "LoRa OTA builds", "full-profiles": "Expanded FULL ESP32 profiles"}
for name in titles:
current = []
count = 0
chunks = []
for record in (r for r in records if category(r) == name):
if count + len(record["files"]) > 900:
chunks.append(current)
current, count = [], 0
current.append(record)
count += len(record["files"])
if current:
chunks.append(current)
for index, chunk in enumerate(chunks):
key = name + (f"-{index + 1}" if index else "")
tag = base_tag if key == "companion" else f"{key}-{base_tag}"
title = f"MeshCore {args.version} Dev - {titles[name]}"
if index:
title += f" (part {index + 1})"
groups.append({"key": key, "tag": tag, "title": title, "prerelease": True, "records": chunk})
links = "\n".join(f"- [{g['key']}](https://github.com/{args.repo}/releases/tag/{g['tag']})" for g in groups)
source_url = f"https://github.com/{args.repo}/blob/{args.commit}"
rows = []
for group in groups:
destination = args.output / group["key"]
destination.mkdir(parents=True)
summaries = []
for record in group["records"]:
manifest = record["manifest"]
file_links = []
for path in record["files"]:
shutil.copy2(path, destination / path.name)
if path.suffix in FIRMWARE_SUFFIXES:
url = f"https://github.com/{args.repo}/releases/download/{group['tag']}/{quote(path.name)}"
label = "merged.bin (USB)" if path.name.endswith("-merged.bin") else path.suffix[1:]
file_links.append(f'{html.escape(label)}')
methods = ", ".join(manifest.get("ota_update_methods", [])) or "USB"
summaries.append({**manifest, "files": [path.name for path in record["files"]]})
rows.append(f"
{html.escape(manifest['artifact_target'])}
{html.escape(manifest['build_profile'])}
{html.escape(methods)}
{' · '.join(file_links)}
")
(destination / "TARGET-MANIFEST.json").write_text(json.dumps(summaries, indent=2) + "\n")
columns = ("artifact_target", "target", "platform", "build_profile",
"ota_update_methods", "files")
(destination / "TARGET-MANIFEST.tsv").write_text(
"\t".join(columns) + "\n" + "".join(
"\t".join(",".join(row.get(key, [])) if isinstance(row.get(key), list)
else str(row.get(key, "")) for key in columns) + "\n"
for row in summaries))
guide = (ROOT / "docs/full_companion_features.md").read_text()
guide = re.sub(r"\]\(([^)]+)\)",
lambda match: "](" + urljoin(source_url + "/docs/", match[1]) + ")", guide)
(destination / "FULL-COMPANION-FEATURES.md").write_text(guide)
exclusions = args.input / "ota-excluded-targets.txt"
if exclusions.is_file():
shutil.copy2(exclusions, destination / exclusions.name)
partial_note = ""
portable_note = ""
if portable_exclusions:
report = {"source": args.commit, "version": args.version,
"portable_slot_bytes": 1310720, "exclusions": portable_exclusions}
(destination / "PORTABLE-PROFILE-EXCLUSIONS.json").write_text(json.dumps(report, indent=2) + "\n")
(destination / "PORTABLE-PROFILE-EXCLUSIONS.md").write_text(
"# Portable image limits\n\n"
"These standard profiles exceeded the 1.25 MiB (1,310,720-byte) application slot. "
"The matrix attempted their expanded Full alternatives instead. Check TARGET-MANIFEST.json "
"for qualified Full images and use their matching partition layout. "
"Smaller images that passed qualification remain available.\n\n"
+ "".join(f"- `{item['target']}`\n" for item in portable_exclusions))
report_url = f"https://github.com/{args.repo}/releases/download/{group['tag']}/PORTABLE-PROFILE-EXCLUSIONS.md"
portable_note = (f"**Portable image limits:** {len(portable_exclusions)} standard profile(s) exceeded "
f"the 1.25 MiB slot and were redirected to the Full pass. [Affected profiles]({report_url}).\n\n")
if failures:
report = {"source": args.commit, "version": args.version, "matrix_state": status["state"],
"matrix_exit_code": int(status["exit_code"]), "failures": failures}
(destination / "BUILD-FAILURES.json").write_text(json.dumps(report, indent=2) + "\n")
(destination / "BUILD-FAILURES.md").write_text(
"# Build attempts awaiting repair\n\n"
f"Source: `{args.commit}`. The matrix finished with exit code {status['exit_code']}.\n\n"
"Published files passed their individual capability, size and memory checks. "
"These failed attempts are omitted pending repair; another profile for the same board may be available.\n\n"
"| Target | Profile | Build log |\n| --- | --- | --- |\n"
+ "".join(f"| `{f['target']}` | `{f['profile']}` | `{f['log_file']}` |\n" for f in failures))
report_url = f"https://github.com/{args.repo}/releases/download/{group['tag']}/BUILD-FAILURES.md"
partial_note = (f"**Partial matrix:** {len(failures)} build attempt(s) failed and await repair. "
f"Only individually qualified firmware is published. [Missing builds]({report_url}).\n\n")
body = (f"> **Development prerelease.** Firmware identifier: **{base_tag}**.\n\n"
f"# MeshCore {args.version} Dev — USA Cascade\n\n"
f"Source: `{args.commit}`. USA/Canada: **{radio['frequency']} MHz, BW{radio['bandwidth']}, SF{radio['sf']}, CR{radio['cr']}**; Cascade defaults.\n\n"
f"This page contains **{len(summaries)} qualified firmware profiles**. Full Companions send MOTA and normally update over USB. Infrastructure has a verified wireless update path; see each capability manifest.\n\n"
f"[Feature on/off and update directions]({source_url}/docs/full_companion_features.md) · "
f"[Release details]({source_url}/docs/releases/{args.version}.md)\n\n"
"Match the exact board, radio, display, and storage variant. ESP32 WiFi updates use the application `.bin`; the merged image installs boot/partition data over USB. nRF52 `.zip` files are native application DFU packages. LoRa MOTA installation requires an exact destination package; nRF52 also requires its matching OTAFIX bootloader. Full Companions do not LoRa-install onto themselves.\n\n"
"Firmware and capability checks passed in the build matrix. Hardware update testing across every board was not performed.\n\n"
+ partial_note + portable_note + links + "\n")
(args.output / (group["key"] + "-notes.md")).write_text(body)
(destination / "BUILD-NOTES.txt").write_text(body)
picker = (""
f"MeshCore {html.escape(args.version)} USA Cascade"
""
f"
MeshCore {html.escape(args.version)} USA Cascade
"
"
Match your exact hardware. Full Companions can send MOTA; USB is their normal update method.