#!/usr/bin/env python3 """Multi-version portability orchestration (roadmap R6, H4.1). Given one or more firmware ELF targets (a demo or re-hosted build, a raw ``.fwimage``, a future version), run the same deterministic scenario against each on the QEMU backend: 1. bounded probe: relocate into virt RAM and run/step a fixed budget, capturing the first exception via ``parse_qemu_faults``; 2. symbol DB build: ELF symbols plus optional Ghidra exports / an existing green-eye JSON DB, saved per target; 3. scripted demo run with handler hooks and a control-block watchpoint, recorded as a normal green-eye session; 4. cross-target comparison with ``greeneye.analyze.compare_sessions`` plus explicit hit/watch/write profile deltas and the docs/06 M1-M6 skeleton (metrics without a reference capture are marked "not measured"). Outputs ``portability-report.json`` and ``portability-report.md`` under ``--out DIR`` with stable ordering. Unsupported targets are skipped with a reason instead of aborting the matrix. Run from the tool root: python3 harness/portability_matrix.py --target firmware/uproc_demo/build/uproc_demo.elf --out dumps/matrix """ from __future__ import annotations import argparse import hashlib import os import struct from collections import Counter from dataclasses import dataclass from pathlib import Path try: from .common import ( LOAD_ADDR, MAX_LOG_BYTES, SymbolDB, counter_delta, ensure_dir, load_target, open_flat_session, qemu_available, read_log_window, sha256_file, write_json, write_text, ) except ImportError: # direct script execution (python3 harness/portability_matrix.py) from common import ( # type: ignore[no-redef] LOAD_ADDR, MAX_LOG_BYTES, SymbolDB, counter_delta, ensure_dir, load_target, open_flat_session, qemu_available, read_log_window, sha256_file, write_json, write_text, ) from greeneye.analyze import compare_sessions, parse_qemu_faults # noqa: E402 from greeneye.record import load_events # noqa: E402 FORMAT = "greeneye-portability-matrix/1" RPC_HANDLERS = ("rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault") CTRL_SIZE = 400 CTRL_RPC_COUNT = 16 CTRL_STATE = 8 CTRL_HANDLER_COUNT = 328 METRIC_DEFS = ( ("M1", "Inventory coverage"), ("M2", "Reply agreement"), ("M3", "Ordering agreement"), ("M4", "State agreement"), ("M5", "Error-path agreement"), ("M6", "Memory side-effects"), ) NO_REFERENCE = "agreement is not measured: needs a reference capture (docs/06 5)" @dataclass class TargetSpec: path: str name: str @dataclass class MatrixOptions: out: str probe_instructions: int = 2000 probe_duration: float = 0.5 scenario_duration: float = 0.4 max_stops: int = 5000 max_rows: int = 40 symbols_json: str | None = None ghidra_names: str | None = None ghidra_index: str | None = None ghidra_exports: str | None = None ghidra_callgraph: str | None = None ghidra_bias: int = 0 scenario_script: str | None = None log_bytes: int = MAX_LOG_BYTES def _skipped_probe(reason: str | None) -> dict: return { "status": "skipped", "reason": reason, "session": None, "instructions": None, "stopped_pc": None, "stopped_symbol": None, "fault_count": 0, "faults_capped": False, "first_fault": None, } def _skipped_scenario(reason: str | None) -> dict: return { "status": "skipped", "reason": reason, "session": None, "hooks": [], "uart_banner": None, "ctrl": None, "snapshot": None, "event_counts": {}, "bp": {}, "watch": {}, "writes": {}, "fault_count": 0, "faults_capped": False, "first_fault": None, } # --------------------------------------------------------------------------- # per-target steps # --------------------------------------------------------------------------- def build_symbols(spec: TargetSpec, bias: int, out_dir: str, options: MatrixOptions): """ELF symbols + optional Ghidra exports / JSON DB, saved per target.""" db = SymbolDB() db.load_elf(spec.path, bias=bias, source="elf") sources = ["elf"] if options.symbols_json: if not os.path.isfile(options.symbols_json): raise FileNotFoundError(f"symbol DB not found: {options.symbols_json}") db.load_json(options.symbols_json) sources.append(os.path.basename(options.symbols_json)) ghidra_inputs = any( ( options.ghidra_names, options.ghidra_index, options.ghidra_exports, options.ghidra_callgraph, ) ) if ghidra_inputs: from greeneye.ghidra import build_symbol_db gdb = build_symbol_db( names_tsv=options.ghidra_names, index_tsv=options.ghidra_index, exports_tsv=options.ghidra_exports, callgraph_tsv=options.ghidra_callgraph, bias=options.ghidra_bias, ) for symbol in gdb: db.add(symbol) sources.append("ghidra") rel = os.path.join("symbols", f"{spec.name}.json") out_path = os.path.join(out_dir, rel) ensure_dir(os.path.dirname(out_path)) db.save_json( out_path, meta={"bias": bias, "sources": sources, "firmware": os.path.basename(spec.path)}, ) return db, sources, rel def run_probe(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict: if not qemu_ok: return _skipped_probe("qemu-system-riscv64 not found") result = _skipped_probe(None) result["instructions"] = options.probe_instructions session_rel = os.path.join("sessions", f"{spec.name}-probe") try: with open_flat_session( image, spec.path, record=os.path.join(out_dir, session_rel), name=f"{spec.name}-probe", symbols=db, log_faults=True, ) as fs: session = fs.session if options.probe_instructions > 0: session.step(options.probe_instructions) else: session.run(duration=options.probe_duration) result["stopped_pc"] = session.pc symbol = session.symbols.resolve(session.pc) result["stopped_symbol"] = symbol.name if symbol else None text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes) faults = parse_qemu_faults(text) result["fault_count"] = len(faults) result["faults_capped"] = capped result["first_fault"] = faults[0] if faults else None result["session"] = session_rel if text: fs.session.recorder.meta["qemu_log_blob"] = fs.session.recorder.blob("qemu.log", text.encode("utf-8")) fs.session.recorder.meta["faults_capped"] = capped result["status"] = "ok" except Exception as exc: # a failed step must not abort the matrix result["status"] = "error" result["reason"] = f"{type(exc).__name__}: {exc}" return result def install_scenario(session) -> list[str]: hooks: list[str] = [] for name in RPC_HANDLERS: symbol = session.symbols.get(name) if symbol is None: continue session.hook(symbol.addr, on_enter=lambda hit: None, listener=f"matrix:{name}") hooks.append(name) return hooks def read_ctrl(session, symbol) -> dict | None: if symbol is None: return None raw = session.memory.read(symbol.addr, CTRL_SIZE) u32 = lambda offset: struct.unpack_from(" dict: counts: Counter = Counter() bp: Counter = Counter() watch: Counter = Counter() writes: Counter = Counter() for event in events: kind = event.get("type", "?") counts[kind] += 1 if kind == "bp_hit": bp[event.get("symbol") or f"{event.get('pc', 0):#x}"] += 1 elif kind == "watch_hit": watch[f"{event.get('addr', 0):#x}:{event.get('kind') or 'w'}"] += 1 elif kind == "mem_write": writes[f"{(event.get('addr', 0) >> 12) << 12:#x}"] += 1 return { "event_counts": dict(sorted(counts.items())), "bp": {key: bp[key] for key in sorted(bp)}, "watch": {key: watch[key] for key in sorted(watch)}, "writes": {key: writes[key] for key in sorted(writes)}, } def run_scenario(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict: if not qemu_ok: return _skipped_scenario("qemu-system-riscv64 not found") result = _skipped_scenario(None) session_rel = os.path.join("sessions", f"{spec.name}-scenario") try: with open_flat_session( image, spec.path, record=os.path.join(out_dir, session_rel), name=f"{spec.name}-scenario", symbols=db, log_faults=True, ) as fs: session = fs.session result["hooks"] = install_scenario(session) ctrl_symbol = session.symbols.get("gsp_ctrl") ctrl_addr = ctrl_symbol.addr if ctrl_symbol else None if ctrl_addr is not None: session.watch(ctrl_addr + CTRL_RPC_COUNT, size=8, kind="w", listener="matrix:rpc_count") if options.scenario_script: import greeneye as ge script_path = Path(options.scenario_script) namespace = { "ge": ge, "greeneye": ge, "session": session, "s": session, "memory": session.memory, "symbols": session.symbols, "recorder": session.recorder, } exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace) else: session.run(duration=options.scenario_duration, max_stops=options.max_stops) if ctrl_addr is not None: session.memory.write_u64(ctrl_addr + 32, 1) # PING session.memory.write_u64(ctrl_addr + 40, 0x1234) session.run(duration=options.scenario_duration, max_stops=options.max_stops) result["ctrl"] = read_ctrl(session, ctrl_symbol) if ctrl_addr is not None: data = session.snapshot(ctrl_addr, CTRL_SIZE, name="ctrl") result["snapshot"] = {"label": "ctrl", "size": len(data), "sha256": hashlib.sha256(data).hexdigest()} uart = fs.target.read_uart() result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes) faults = parse_qemu_faults(text) result["fault_count"] = len(faults) result["faults_capped"] = capped result["first_fault"] = faults[0] if faults else None result.update(profile_from_events(load_events(os.path.join(out_dir, session_rel)))) result["session"] = session_rel result["status"] = "ok" except Exception as exc: # a failed step must not abort the matrix result["status"] = "error" result["reason"] = f"{type(exc).__name__}: {exc}" return result # --------------------------------------------------------------------------- # metrics + comparisons # --------------------------------------------------------------------------- def metric_rows(probe: dict, scenario: dict) -> list[dict]: """docs/06 M1-M6 skeleton; observed data is local, agreement stays unmeasured.""" bp = scenario.get("bp") or {} ctrl = scenario.get("ctrl") or {} snapshot = scenario.get("snapshot") or {} probe_fault = (probe.get("first_fault") or {}).get("desc") scenario_fault = (scenario.get("first_fault") or {}).get("desc") rows = [ { "id": "M1", "name": "Inventory coverage", "status": "not measured", "reason": NO_REFERENCE, "observed": { "rpc_handlers_hit": sorted(k for k in bp if k in RPC_HANDLERS), "other_hit_sites": sorted(k for k in bp if k not in RPC_HANDLERS), "declared_handlers": ctrl.get("handler_count"), }, }, { "id": "M2", "name": "Reply agreement", "status": "not measured", "reason": NO_REFERENCE, "observed": {"note": "reply fields are not decoded by the builtin scenario", "result": ctrl.get("result")}, }, { "id": "M3", "name": "Ordering agreement", "status": "not measured", "reason": NO_REFERENCE, "observed": {"events": sum((scenario.get("event_counts") or {}).values())}, }, { "id": "M4", "name": "State agreement", "status": "not measured", "reason": NO_REFERENCE, "observed": {"final_state": ctrl.get("state"), "rpc_state_hits": bp.get("rpc_state", 0)}, }, { "id": "M5", "name": "Error-path agreement", "status": "not measured", "reason": NO_REFERENCE, "observed": { "probe_first_fault": probe_fault, "scenario_first_fault": scenario_fault, "scenario_fault_count": scenario.get("fault_count", 0), }, }, { "id": "M6", "name": "Memory side-effects", "status": "not measured", "reason": NO_REFERENCE, "observed": { "written_pages": len(scenario.get("writes") or {}), "ctrl_snapshot_sha256": snapshot.get("sha256"), }, }, ] return rows def compare_targets(a: dict, b: dict, out_dir: str, options: MatrixOptions) -> dict: entry = { "a": a["name"], "b": b["name"], "status": "ok", "reason": None, "compare_report": None, "deltas": {"bp": [], "watch": [], "writes": []}, "all_zero": True, "metrics": [], } a_scenario = a.get("scenario") or {} b_scenario = b.get("scenario") or {} if a_scenario.get("status") != "ok" or b_scenario.get("status") != "ok": entry["status"] = "skipped" entry["reason"] = "a scenario session is not available on both targets" return entry a_dir = os.path.join(out_dir, a_scenario.get("session") or "") b_dir = os.path.join(out_dir, b_scenario.get("session") or "") try: report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows) rel = os.path.join("compare", f"{a['name']}--{b['name']}.md") write_text(os.path.join(out_dir, rel), report) entry["compare_report"] = rel except Exception as exc: entry["reason"] = f"{type(exc).__name__}: {exc}" deltas = {"bp": [], "watch": [], "writes": []} for name in ("bp", "watch", "writes"): deltas[name] = counter_delta(a_scenario.get(name) or {}, b_scenario.get(name) or {}) entry["deltas"] = deltas entry["all_zero"] = all(row["delta"] == 0 for rows in deltas.values() for row in rows) a_metrics = {row["id"]: row for row in a.get("metrics") or []} b_metrics = {row["id"]: row for row in b.get("metrics") or []} entry["metrics"] = [ { "id": metric_id, "name": name, "status": "not measured", "reason": NO_REFERENCE, "a_observed": (a_metrics.get(metric_id) or {}).get("observed"), "b_observed": (b_metrics.get(metric_id) or {}).get("observed"), } for metric_id, name in METRIC_DEFS ] return entry # --------------------------------------------------------------------------- # report assembly + rendering # --------------------------------------------------------------------------- def analyze_target(spec: TargetSpec, options: MatrixOptions, qemu_ok: bool) -> dict: target = { "name": spec.name, "path": spec.path, "sha256": "", "kind": "elf", "entry": None, "status": "skipped", "skip_reason": None, "probe": _skipped_probe(None), "symbols": {"count": 0, "sources": [], "path": None}, "scenario": _skipped_scenario(None), "metrics": metric_rows({}, {}), } if not os.path.isfile(spec.path): target["skip_reason"] = "file not found" return target target["sha256"] = sha256_file(spec.path) try: image = load_target(spec.path) except Exception as exc: target["kind"] = "unknown" target["skip_reason"] = f"{type(exc).__name__}: {exc}" return target target["entry"] = f"{image.entry:#x}" bias = LOAD_ADDR - image.min_vaddr db = SymbolDB() try: db, sources, sym_rel = build_symbols(spec, bias, options.out, options) target["symbols"] = {"count": len(db), "sources": sources, "path": sym_rel} except Exception as exc: target["symbols"] = {"count": 0, "sources": [], "path": None, "reason": f"{type(exc).__name__}: {exc}"} db.load_elf(spec.path, bias=bias, source="elf") target["symbols"]["count"] = len(db) target["symbols"]["sources"] = ["elf"] target["probe"] = run_probe(spec, image, db, options.out, options, qemu_ok) target["scenario"] = run_scenario(spec, image, db, options.out, options, qemu_ok) target["metrics"] = metric_rows(target["probe"], target["scenario"]) statuses = {target["probe"]["status"], target["scenario"]["status"]} if statuses == {"ok"}: target["status"] = "ok" elif "ok" in statuses: target["status"] = "partial" elif statuses == {"skipped"}: target["status"] = "skipped" target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason") else: target["status"] = "error" target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason") return target def run_matrix(specs: list[TargetSpec], options: MatrixOptions) -> dict: ensure_dir(options.out) qemu_ok = qemu_available() targets = [analyze_target(spec, options, qemu_ok) for spec in specs] comparisons: list[dict] = [] for i in range(len(targets)): for j in range(i + 1, len(targets)): comparisons.append(compare_targets(targets[i], targets[j], options.out, options)) report = { "format": FORMAT, "tool": "harness/portability_matrix.py", "backend": "qemu", "qemu_available": qemu_ok, "options": { "probe_instructions": options.probe_instructions, "probe_duration": options.probe_duration, "scenario_duration": options.scenario_duration, "scenario_script": options.scenario_script, "max_rows": options.max_rows, "symbols_json": options.symbols_json, "ghidra_bias": options.ghidra_bias, }, "targets": targets, "comparisons": comparisons, "skipped": [ {"name": target["name"], "reason": target["skip_reason"]} for target in targets if target["status"] == "skipped" ], } return report def _table(headers: list[str], rows: list[list]) -> list[str]: lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] for row in rows: lines.append("| " + " | ".join(str(cell) for cell in row) + " |") return lines def _profile_tables(title: str, profile: dict, max_rows: int) -> list[str]: lines = [f"**{title}**", ""] if not profile: return lines + ["_none observed_", ""] rows = [[f"`{key}`", count] for key, count in sorted(profile.items(), key=lambda kv: (-int(kv[1]), str(kv[0])))[:max_rows]] lines += _table(["key", "hits"], rows) + [""] return lines def render_markdown(report: dict, options: MatrixOptions) -> str: lines = [ "# green-eye portability delta report", "", "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/portability_matrix.py`.", "", f"* format: `{report['format']}` (green-eye {_greeneye_version()})", f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})", f"* targets: {len(report['targets'])}", "", "## Targets", "", ] lines += _table( ["target", "status", "sha256", "probe", "symbols", "scenario events"], [ [ f"`{target['name']}`", target["status"], f"`{(target['sha256'] or '')[:16]}`", f"{target['probe']['status']} ({target['probe']['fault_count']} faults)", target["symbols"]["count"], sum((target["scenario"].get("event_counts") or {}).values()), ] for target in report["targets"] ], ) lines += [""] for target in report["targets"]: lines += [f"### {target['name']}", ""] lines += [f"* path: `{target['path']}`", f"* entry: `{target['entry']}`", f"* sha256: `{target['sha256']}`"] if target["skip_reason"]: lines += [f"* note: {target['skip_reason']}"] lines += [""] probe = target["probe"] lines += ["#### Probe", ""] fault = probe.get("first_fault") stopped_pc = probe.get("stopped_pc") probe_rows = [ ["status", probe["status"]], ["instructions", probe.get("instructions")], ["stopped pc", f"`{stopped_pc:#x}`" if stopped_pc is not None else None], ["stopped symbol", probe.get("stopped_symbol")], ["fault count", probe.get("fault_count")], ["faults capped", probe.get("faults_capped")], ] if fault: probe_rows.append( ["first fault", f"cause {fault.get('cause')} {fault.get('desc')} @ epc `{fault.get('epc', 0):#x}`"] ) if probe.get("reason"): probe_rows.append(["reason", probe["reason"]]) lines += _table(["field", "value"], probe_rows) + [""] lines += ["#### Symbols", ""] lines += _table( ["count", "sources", "path"], [[target["symbols"]["count"], ", ".join(target["symbols"]["sources"]), target["symbols"]["path"]]], ) + [""] scenario = target["scenario"] lines += ["#### Scenario", ""] lines += _table( ["status", "session", "hooks", "uart banner", "faults"], [[ scenario["status"], f"`{scenario.get('session')}`", ", ".join(scenario.get("hooks") or []), scenario.get("uart_banner"), scenario.get("fault_count"), ]], ) + [""] lines += _profile_tables("Breakpoint hits", scenario.get("bp") or {}, options.max_rows) lines += _profile_tables("Watchpoint hits", scenario.get("watch") or {}, options.max_rows) lines += _profile_tables("Written pages", scenario.get("writes") or {}, options.max_rows) lines += ["#### M1-M6", ""] lines += [f"> {NO_REFERENCE}", ""] lines += _table( ["#", "metric", "status", "observed"], [ [row["id"], row["name"], row["status"], _short_json(row["observed"])] for row in target["metrics"] ], ) + [""] lines += ["## Cross-target comparisons", ""] if not report["comparisons"]: lines += ["_single target: no comparison_", ""] for comp in report["comparisons"]: lines += [f"### {comp['a']} vs {comp['b']}", ""] lines += [f"* status: {comp['status']}", f"* zero deltas: {'yes' if comp['all_zero'] else 'no'}"] if comp.get("compare_report"): lines += [f"* compare report: `{comp['compare_report']}`"] if comp.get("reason"): lines += [f"* note: {comp['reason']}"] lines += [""] for name in ("bp", "watch", "writes"): rows = [row for row in comp["deltas"][name] if row["delta"] != 0] rows.sort(key=lambda row: (-abs(row["delta"]), row["key"])) lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""] if rows: lines += _table( ["key", comp["a"], comp["b"], "delta"], [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]], ) + [""] else: lines += ["_no deltas_", ""] lines += ["**M1-M6 (not measured: reference capture required)**", ""] lines += _table( ["#", "metric", "A observed", "B observed"], [ [row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])] for row in comp["metrics"] ], ) + [""] if report["skipped"]: lines += ["## Skipped targets", ""] lines += _table(["target", "reason"], [[f"`{item['name']}`", item["reason"]] for item in report["skipped"]]) lines += [""] return "\n".join(lines) def _short_json(value, limit: int = 72) -> str: import json if value is None: return "-" text = json.dumps(value, sort_keys=True) return f"`{text}`" if len(text) <= limit else f"`{text[:limit - 1]}...`" def _greeneye_version() -> str: try: import greeneye return greeneye.__version__ except Exception: # pragma: no cover return "unknown" # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="portability_matrix", description=__doc__.splitlines()[0]) parser.add_argument("targets", nargs="*", help="firmware ELF paths") parser.add_argument("--target", action="append", default=[], help="firmware ELF path (repeatable)") parser.add_argument("--out", required=True, help="output directory for the report, sessions and symbols") parser.add_argument("--symbols", dest="symbols_json", help="existing greeneye symbol DB JSON") parser.add_argument("--ghidra-names", help="Ghidra names.tsv export") parser.add_argument("--ghidra-index", help="Ghidra decompiled_index.tsv export") parser.add_argument("--ghidra-exports", help="Ghidra exports.tsv") parser.add_argument("--ghidra-callgraph", help="Ghidra callgraph.tsv") parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0) parser.add_argument("--probe-instructions", type=int, default=2000, help="probe step budget (0 = wall-clock run)") parser.add_argument("--probe-duration", type=float, default=0.5, help="probe wall budget when --probe-instructions=0") parser.add_argument("--scenario-duration", type=float, default=0.4, help="budget for each scenario phase") parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per scenario phase") parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports") parser.add_argument("--scenario-script", help="replace the builtin scenario with a green-eye script") return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) targets = list(args.targets) + list(args.target) if not targets: parser.error("at least one target is required (positional or --target)") options = MatrixOptions( out=os.path.abspath(args.out), probe_instructions=args.probe_instructions, probe_duration=args.probe_duration, scenario_duration=args.scenario_duration, max_stops=args.max_stops, max_rows=args.max_rows, symbols_json=args.symbols_json, ghidra_names=args.ghidra_names, ghidra_index=args.ghidra_index, ghidra_exports=args.ghidra_exports, ghidra_callgraph=args.ghidra_callgraph, ghidra_bias=args.ghidra_bias, scenario_script=args.scenario_script, ) specs = _name_specs(targets) report = run_matrix(specs, options) json_path = write_json(os.path.join(options.out, "portability-report.json"), report) md_path = write_text(os.path.join(options.out, "portability-report.md"), render_markdown(report, options)) print(f"[matrix] {len(specs)} target(s), out={options.out}") for target in report["targets"]: print(f"[matrix] {target['name']}: status={target['status']} probe={target['probe']['status']} " f"scenario={target['scenario']['status']}") print(f"[matrix] wrote {json_path}") print(f"[matrix] wrote {md_path}") return 0 def _name_specs(paths: list[str]) -> list[TargetSpec]: seen: dict[str, int] = {} specs: list[TargetSpec] = [] for path in paths: base = Path(path).stem seen[base] = seen.get(base, 0) + 1 name = base if seen[base] == 1 else f"{base}-{seen[base]}" specs.append(TargetSpec(path=os.path.abspath(path), name=name)) return specs if __name__ == "__main__": # pragma: no cover raise SystemExit(main())