public/kaptan
1
Fork 0
Code Issues 24 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move firmware workspace under firmware

Browse Source
This commit is contained in:
Onur Özkan committed 2026-08-10 16:58:55 +03:00
1 parent 153b0bf262
commit f03fc119a4
222 files changed
+56 -13
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/analyze-fw.py
-122
View File
@@ -1,122 +0,0 @@
#!/usr/bin/env python3
"""Show flash, static RAM and ELF-section use for embedded firmware.
Usage:
scripts/analyze-fw.py <firmware.elf> <memory.x>
"""
import re
from pathlib import Path
import subprocess
import sys
REGION = re.compile(
r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*:\s*"
r"ORIGIN\s*=\s*([^,]+),\s*LENGTH\s*=\s*([^\s/}]+)",
re.MULTILINE,
)
def number(text):
"""Parse linker numbers such as 0x2001_0000, 176K, and 1M."""
text = text.strip().replace("_", "")
scale = {"K": 1024, "M": 1024 * 1024}.get(text[-1:].upper(), 1)
return int(text[:-1] if scale != 1 else text, 0) * scale
def run(command):
"""Run a tool and return stdout."""
return subprocess.run(command, check=True, text=True, capture_output=True).stdout
def llvm_size():
"""Find the llvm-size installed by the active Rust toolchain."""
sysroot = run(["rustc", "--print", "sysroot"]).strip()
host = next(
line.split(": ", 1)[1]
for line in run(["rustc", "-vV"]).splitlines()
if line.startswith("host: ")
)
tool = Path(sysroot) / "lib" / "rustlib" / host / "bin" / "llvm-size"
if not tool.is_file():
raise RuntimeError(
"llvm-size is unavailable; install it with: rustup component add llvm-tools-preview"
)
return tool
def regions(memory_x):
"""Return every named linker region as (name, origin, length)."""
return [
(name, number(origin), number(length))
for name, origin, length in REGION.findall(memory_x.read_text())
]
def sections(output):
"""Return llvm-size sections as (name, bytes, address)."""
result = []
for line in output.splitlines():
fields = line.split()
if len(fields) == 3 and fields[0].startswith("."):
result.append((fields[0], int(fields[1], 0), int(fields[2], 0)))
return result
def region_for(address, layout):
"""Return the linker region containing an address, if any."""
for name, origin, length in layout:
if origin <= address < origin + length:
return name
return "outside"
def main():
if len(sys.argv) != 3:
raise ValueError(f"usage: {sys.argv[0]} <firmware.elf> <memory.x>")
elf = Path(sys.argv[1])
memory_x = Path(sys.argv[2])
if not elf.is_file():
raise ValueError(f"ELF does not exist: {elf}")
if not memory_x.is_file():
raise ValueError(f"linker layout does not exist: {memory_x}")
layout = regions(memory_x)
if not layout:
raise ValueError(f"no MEMORY regions found in: {memory_x}")
section_list = sections(run([str(llvm_size()), "-A", str(elf)]))
if not section_list:
raise ValueError(f"no sections found in: {elf}")
print(f"\nFirmware: {elf}")
print(f"ELF file size: {elf.stat().st_size:,} bytes")
print("\nLinker regions:")
print(" Region Origin Used / Capacity Remaining")
for name, origin, length in layout:
used = sum(
size
for _, size, address in section_list
if size and origin <= address < origin + length
)
remaining = length - used
print(
f" {name:<12} 0x{origin:08X} {used / 1024:>7.2f} KiB /"
f" {length / 1024:>7.2f} KiB {remaining / 1024:>7.2f} KiB"
f" ({100 * used / length:5.2f}%)"
)
print("\nELF sections:")
print(" Region Address Size Section")
for name, size, address in sorted(section_list, key=lambda section: section[2]):
if size:
print(f" {region_for(address, layout):<12} 0x{address:08X} {size:>10,} {name}")
if __name__ == "__main__":
try:
main()
except (IndexError, OSError, RuntimeError, ValueError, subprocess.CalledProcessError) as error:
print(f"error: {error}", file=sys.stderr)
raise SystemExit(1) from error
scripts/create-dfu-image.sh
-36
View File
@@ -1,36 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
if [[ $# -ne 1 ]]; then
printf 'Usage: %s <firmware.elf>\n' "$0" >&2
exit 2
fi
readonly input_elf="$1"
readonly firmware_name="$(basename "${input_elf%.elf}")"
readonly board_name="${firmware_name#kaptan-}"
readonly output_image="${input_elf%.elf}.bin"
readonly script_directory="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
readonly workspace_root="$(cd -- "$script_directory/.." && pwd)"
readonly memory_x="$workspace_root/boards/$board_name/memory.x"
readonly host="$(rustc -vV | sed -n 's/^host: //p')"
readonly llvm_objcopy="$(rustc --print sysroot)/lib/rustlib/$host/bin/llvm-objcopy"
if [[ ! -f "$input_elf" ]]; then
printf 'firmware ELF does not exist: %s\n' "$input_elf" >&2
exit 1
fi
if [[ ! -f "$memory_x" ]]; then
printf 'cannot find the linker layout for board %q: %s\n' "$board_name" "$memory_x" >&2
exit 1
fi
if [[ ! -x "$llvm_objcopy" ]]; then
printf 'llvm-objcopy is unavailable; install it with: rustup component add llvm-tools-preview\n' >&2
exit 1
fi
"$llvm_objcopy" -O binary "$input_elf" "$output_image"
printf 'DFU image: %s\n' "$output_image"
"$script_directory/analyze-fw.py" "$input_elf" "$memory_x"
scripts/mavlink-debug.py
-84
View File
@@ -1,84 +0,0 @@
#!/usr/bin/env python3
"""Small MAVLink command-line debugger for a Kaptan flight controller."""
import argparse
import os
import sys
import time
def parse_arguments():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("serial_device", help="serial device connected to the flight controller")
parser.add_argument("command", choices=("arm", "disarm", "read"))
parser.add_argument("--baud", type=int, default=115200, help="serial baud rate (default: 115200)")
parser.add_argument("--system-id", type=int, default=1, help="target MAVLink system ID (default: 1)")
parser.add_argument(
"--component-id", type=int, default=1, help="target MAVLink component ID (default: 1)"
)
return parser.parse_args()
def print_messages(connection, timeout):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
message = connection.recv_match(blocking=True, timeout=deadline - time.monotonic())
if message is None:
return None
print(message)
yield message
def send_arm_command(connection, mavutil, arguments, armed):
for _ in range(3):
connection.mav.command_long_send(
arguments.system_id,
arguments.component_id,
mavutil.mavlink.MAV_CMD_COMPONENT_ARM_DISARM,
0,
1.0 if armed else 0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
)
for message in print_messages(connection, timeout=1.0):
if (
message.get_type() == "COMMAND_ACK"
and message.command == mavutil.mavlink.MAV_CMD_COMPONENT_ARM_DISARM
):
return
action = "arm" if armed else "disarm"
raise SystemExit(f"timed out waiting for {action} acknowledgement")
def read_messages(connection):
try:
while True:
message = connection.recv_match(blocking=True, timeout=1.0)
if message is not None:
print(message)
except KeyboardInterrupt:
pass
def main():
arguments = parse_arguments()
# Kaptan currently accepts MAVLink 2 frames only.
os.environ["MAVLINK20"] = "1"
from pymavlink import mavutil
connection = mavutil.mavlink_connection(arguments.serial_device, baud=arguments.baud)
if arguments.command == "read":
read_messages(connection)
else:
send_arm_command(connection, mavutil, arguments, armed=arguments.command == "arm")
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
sys.exit(130)
scripts/report-test-coverage.sh
-141
View File
@@ -1,141 +0,0 @@
#!/usr/bin/env sh
# Per-workspace-crate line and branch coverage report.
#
# This is deliberately a report over every host-testable workspace package,
# not a misleading aggregate. STM32F7 platform/board images are target-only:
# their `panic-halt` handler cannot link into the host test harness, so they
# are reported as N/A and must be covered by target checks plus bench
# evidence instead.
#
# Prerequisites (kept out of the firmware dependency graph):
# cargo install cargo-llvm-cov --locked
# rustup toolchain install nightly --component llvm-tools-preview
# python3 (standard library only, for strict source-level accounting)
set -eu
if ! cargo llvm-cov --version >/dev/null 2>&1; then
echo "cargo-llvm-cov is required; install it with: cargo install cargo-llvm-cov --locked" >&2
exit 127
fi
if ! rustup toolchain list | grep -q '^nightly'; then
echo "nightly with llvm-tools-preview is required for branch coverage" >&2
exit 127
fi
if ! python3 --version >/dev/null 2>&1; then
echo "python3 is required for strict source-level coverage accounting" >&2
exit 127
fi
python3 scripts/test-summarize-test-coverage.py
# Keep instrumentation products outside a possibly shared target directory.
: "${CARGO_TARGET_DIR:=/tmp/kaptan-test-coverage}"
export CARGO_TARGET_DIR
mkdir -p "$CARGO_TARGET_DIR"
# `cargo llvm-cov` merges every instrumented object below `CARGO_TARGET_DIR`.
# A shared target directory therefore makes a later strict-core report include
# an unexecuted copy of a shared dependency built while testing an earlier
# package. Keep reports together but isolate each strict-core invocation. The
# non-core packages retain the shared directory: several intentionally
# target-only crates have no host-instrumented object to export by themselves.
coverage_root=$CARGO_TARGET_DIR
core_incomplete=0
report_package() {
package=$1
source_path=$2
shift 2
case "$package" in
kaptan-math|kaptan-navigation|kaptan-control|kaptan-safety)
package_target_dir="$coverage_root/$package"
lcov_report="$coverage_root/$package.lcov"
json_report="$coverage_root/$package.json"
report_format=strict-core
;;
*)
package_target_dir=$coverage_root
lcov_report="$coverage_root/$package.lcov"
report_format=lcov
;;
esac
case "$report_format" in
strict-core)
# Keep the strict core isolated, then export both representations
# from the same test run. LCOV `DA` defines source lines; JSON
# coordinates define branch identities across generic instances.
RUSTC_BOOTSTRAP=1 CARGO_TARGET_DIR="$package_target_dir" cargo llvm-cov \
--branch \
--no-report \
-p "$package" \
"$@"
RUSTC_BOOTSTRAP=1 CARGO_TARGET_DIR="$package_target_dir" cargo llvm-cov report \
--lcov \
--output-path "$lcov_report"
RUSTC_BOOTSTRAP=1 CARGO_TARGET_DIR="$package_target_dir" cargo llvm-cov report \
--json \
--output-path "$json_report"
totals=$(python3 scripts/summarize-test-coverage.py \
"$lcov_report" "$json_report" "$source_path")
;;
lcov)
RUSTC_BOOTSTRAP=1 CARGO_TARGET_DIR="$package_target_dir" cargo llvm-cov \
--branch \
--lcov \
--output-path "$lcov_report" \
-p "$package" \
"$@"
totals=$(awk -v source_path="$source_path" '
/^SF:/ { selected = index($0, source_path) != 0 }
selected && /^LF:/ { lines_found += substr($0, 4) }
selected && /^LH:/ { lines_hit += substr($0, 4) }
selected && /^BRF:/ { branches_found += substr($0, 5) }
selected && /^BRH:/ { branches_hit += substr($0, 5) }
END {
printf "%d %d %d %d", lines_hit, lines_found, branches_hit, branches_found
}
' "$lcov_report")
;;
esac
set -- $totals
printf '%-26s lines %s/%s, branches %s/%s\n' "$package" "$1" "$2" "$3" "$4"
case "$package" in
kaptan-math|kaptan-navigation|kaptan-control|kaptan-safety)
if [ "$1" -ne "$2" ] || [ "$3" -ne "$4" ]; then
core_incomplete=1
fi
;;
esac
}
echo "Host test coverage (reports: $CARGO_TARGET_DIR; strict core unions LCOV DA lines and JSON branches)"
report_package kaptan-actuator /libs/kaptan-actuator/
report_package kaptan-board /libs/kaptan-board/
report_package kaptan-control /libs/kaptan-control/
report_package kaptan-data-channel /libs/kaptan-data-channel/
report_package kaptan-dma /libs/kaptan-dma/
report_package kaptan-drivers /libs/kaptan-drivers/
report_package kaptan-i2c /libs/kaptan-i2c/
report_package kaptan-math /libs/kaptan-math/
report_package kaptan-mavlink /libs/kaptan-mavlink/ --features dialect-kaptan
report_package kaptan-navigation /libs/kaptan-navigation/ --features sitl
report_package kaptan-safety /libs/kaptan-safety/
report_package kaptan-sensor /libs/kaptan-sensor/
report_package kaptan-spi /libs/kaptan-spi/
report_package kaptan-time /libs/kaptan-time/
report_package kaptan-tx /libs/kaptan-tx/
report_package kaptan-sitl /tools/sitl/
echo "platform-stm32f7 target-only/N/A (run cargo check -p platform-stm32f7 --lib)"
echo "kaptan-geprc-f722-v2 target-only/N/A (run its thumbv7em-none-eabihf check)"
echo "kaptan-kakute-f7-mini-v3 target-only/N/A (run its thumbv7em-none-eabihf check)"
if [ "$core_incomplete" -ne 0 ]; then
echo "Strict core coverage is incomplete; reports above identify each crate." >&2
exit 1
fi
scripts/summarize-test-coverage.py
-197
View File
@@ -1,197 +0,0 @@
#!/usr/bin/env python3
"""Summarize strict-core LCOV lines and LLVM JSON branches by source location.
`llvm-cov` emits one coverage record for every generic monomorphization. Its
per-file totals add those records, which can report an uncovered duplicate even
when the corresponding source line or branch has been exercised by another
monomorphization. This helper instead applies the gate to source locations:
* a source line is identified by its LCOV `DA` file and line number;
* a branch outcome is identified by its full source span; and
* a location is covered when any matching monomorphization executed it.
JSON segments are deliberately not used as a line denominator: they mark LLVM
instrumentation-region boundaries, not every source line represented by an
LCOV `DA` record. Replacing `DA` with segments would silently lower the strict
line denominator.
Macro-expansion records are deliberately rejected. Their source span denotes
the invocation, not necessarily the expanded predicate, so folding them into
the source-level key would make the gate unsound. Extend this helper with an
explicit expansion identity before adding macro-expanded strict-core coverage.
"""
from __future__ import annotations
import argparse
import json
from collections.abc import Iterable
from pathlib import Path
import sys
class CoverageFormatError(ValueError):
"""Coverage records cannot be safely summarized by source location."""
def selected_files(report: dict[object, object], source_path: str) -> Iterable[dict[object, object]]:
"""Yield selected file records after rejecting ambiguous macro expansion."""
exports = report.get("data")
if not isinstance(exports, list):
raise CoverageFormatError("LLVM JSON has no data array")
for export in exports:
if not isinstance(export, dict):
raise CoverageFormatError("LLVM JSON data array contains a non-object")
files = export.get("files")
if not isinstance(files, list):
raise CoverageFormatError("LLVM JSON export has no files array")
for file_record in files:
if not isinstance(file_record, dict):
raise CoverageFormatError("LLVM JSON files array contains a non-object")
filename = file_record.get("filename")
if not isinstance(filename, str) or source_path not in filename:
continue
expansions = file_record.get("expansions", [])
if not isinstance(expansions, list):
raise CoverageFormatError(f"{filename}: expansions is not an array")
if expansions:
raise CoverageFormatError(
f"{filename}: macro-expansion coverage is present; "
"refusing to merge it by invocation source span"
)
yield file_record
def lcov_line_totals(report: str, source_path: str) -> tuple[int, int, set[str]]:
"""Return the union of selected LCOV `DA` lines and their source files."""
lines: dict[tuple[str, int], bool] = {}
files: set[str] = set()
filename: str | None = None
for record in report.splitlines():
if record.startswith("SF:"):
filename = record[3:]
if source_path in filename:
files.add(filename)
continue
if record == "end_of_record":
filename = None
continue
if filename is None or source_path not in filename or not record.startswith("DA:"):
continue
fields = record[3:].split(",", 2)
if len(fields) < 2:
raise CoverageFormatError(f"{filename}: malformed LCOV DA record")
try:
line = int(fields[0])
count = int(fields[1])
except ValueError as error:
raise CoverageFormatError(f"{filename}: non-integer LCOV DA record") from error
if line <= 0 or count < 0:
raise CoverageFormatError(f"{filename}: invalid LCOV DA location or count")
key = (filename, line)
lines[key] = lines.get(key, False) or count > 0
if not files:
raise CoverageFormatError(f"no LCOV source file contains {source_path!r}")
if not lines:
raise CoverageFormatError(f"no LCOV DA records found under {source_path!r}")
return sum(lines.values()), len(lines), files
def branch_totals(report: dict[object, object], source_path: str) -> tuple[int, int, set[str]]:
"""Return the union of selected JSON branch outcomes and source files."""
branches: dict[tuple[str, int, int, int, int], list[bool]] = {}
files: set[str] = set()
for file_record in selected_files(report, source_path):
filename = file_record["filename"]
assert isinstance(filename, str)
files.add(filename)
raw_branches = file_record.get("branches")
if not isinstance(raw_branches, list):
raise CoverageFormatError(f"{filename}: branches is not an array")
for branch in raw_branches:
if (
not isinstance(branch, list)
or len(branch) != 9
or not all(type(value) is int for value in branch)
):
raise CoverageFormatError(f"{filename}: invalid branch record")
(
start_line,
start_column,
end_line,
end_column,
true_count,
false_count,
file_id,
expanded_file_id,
_kind,
) = branch
if (
start_line <= 0
or start_column <= 0
or end_line < start_line
or end_column <= 0
or (end_line == start_line and end_column < start_column)
or true_count < 0
or false_count < 0
or file_id < 0
or expanded_file_id < 0
):
raise CoverageFormatError(f"{filename}: invalid branch location or count")
if file_id != expanded_file_id:
raise CoverageFormatError(
f"{filename}:{start_line}:{start_column}: macro-expanded branch is present; "
"refusing to merge it by invocation source span"
)
key = (filename, start_line, start_column, end_line, end_column)
outcomes = branches.setdefault(key, [False, False])
outcomes[0] = outcomes[0] or true_count > 0
outcomes[1] = outcomes[1] or false_count > 0
if not files:
raise CoverageFormatError(f"no LLVM JSON source file contains {source_path!r}")
branch_hit = sum(sum(outcomes) for outcomes in branches.values())
return branch_hit, 2 * len(branches), files
def coverage_totals(
lcov_report: str, json_report: dict[object, object], source_path: str
) -> tuple[int, int, int, int]:
"""Return strict source-level line and branch coverage for one core crate."""
lines_hit, lines_found, lcov_files = lcov_line_totals(lcov_report, source_path)
branch_hit, branches_found, json_files = branch_totals(json_report, source_path)
if lcov_files != json_files:
raise CoverageFormatError(
"LCOV/JSON source-file mismatch: "
f"LCOV={sorted(lcov_files)!r}, JSON={sorted(json_files)!r}"
)
return lines_hit, lines_found, branch_hit, branches_found
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("lcov_report", type=Path, help="LCOV report from cargo llvm-cov")
parser.add_argument("json_report", type=Path, help="LLVM JSON report from cargo llvm-cov")
parser.add_argument("source_path", help="workspace-relative path fragment to include")
arguments = parser.parse_args()
try:
lcov_report = arguments.lcov_report.read_text(encoding="utf-8")
with arguments.json_report.open(encoding="utf-8") as report_file:
json_report = json.load(report_file)
if not isinstance(json_report, dict):
raise CoverageFormatError("LLVM JSON root is not an object")
print(*coverage_totals(lcov_report, json_report, arguments.source_path))
except (CoverageFormatError, OSError, json.JSONDecodeError) as error:
print(f"coverage summary error: {error}", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/test-summarize-test-coverage.py
-77
View File
@@ -1,77 +0,0 @@
#!/usr/bin/env python3
"""Synthetic regression tests for strict source-level coverage accounting."""
from __future__ import annotations
import importlib.util
import unittest
def load_summary_module():
spec = importlib.util.spec_from_file_location(
"summarize_test_coverage", "scripts/summarize-test-coverage.py"
)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
SUMMARY = load_summary_module()
SOURCE = "/libs/kaptan-navigation/"
FILENAME = "/repo/libs/kaptan-navigation/src/eskf.rs"
def json_report(*, expansions=None, expanded_file_id=0):
return {
"data": [
{
"files": [
{
"filename": FILENAME,
"expansions": [] if expansions is None else expansions,
# Segments intentionally differ from DA lines: strict
# line coverage must not lower its denominator to them.
"segments": [[10, 1, 3, True, True, False]],
"branches": [
[10, 5, 10, 10, 0, 3, 0, expanded_file_id, 4],
[10, 5, 10, 10, 2, 0, 0, expanded_file_id, 4],
[11, 5, 11, 10, 0, 0, 0, expanded_file_id, 4],
],
}
]
}
]
}
LCOV = """SF:/repo/libs/kaptan-navigation/src/eskf.rs
DA:10,0
DA:11,0
end_of_record
SF:/repo/libs/kaptan-navigation/src/eskf.rs
DA:10,3
DA:12,1
end_of_record
"""
class CoverageSummaryTests(unittest.TestCase):
def test_unions_da_lines_and_full_span_branch_outcomes(self):
self.assertEqual(
SUMMARY.coverage_totals(LCOV, json_report(), SOURCE),
# DA lines: 10 is covered by the second generic instance; 11 is
# uncovered; 12 is covered. Branch outcomes: the two duplicate
# source spans cover both outcomes, while line 11 covers neither.
(2, 3, 2, 4),
)
def test_rejects_macro_expansion_records(self):
with self.assertRaises(SUMMARY.CoverageFormatError):
SUMMARY.coverage_totals(LCOV, json_report(expansions=[{}]), SOURCE)
with self.assertRaises(SUMMARY.CoverageFormatError):
SUMMARY.coverage_totals(LCOV, json_report(expanded_file_id=1), SOURCE)
if __name__ == "__main__":
unittest.main()