Merge branch 'dev' into feat/mini-hp-core

This commit is contained in:
maksyuki committed 2026-10-14 18:23:57 +08:00
commit 16f2a8118a
178 files changed
+19547 -1716

No files matched your search

+17
View File
@@ -16,6 +16,8 @@ CDC_PRIMITIVES = {
"async_fifo",
"async_reqack",
"axi4_async_bridge",
"tiny_axi4_async_bridge",
"tiny_apb_async_bridge",
"cdc_2phase",
"cdc_fifo",
"cdc_sync",
@@ -129,6 +131,21 @@ def validate(document_path: Path, root: Path) -> None:
raw_crossings = document.get("crossings")
if not isinstance(raw_crossings, list):
raise ValueError("crossings must be a list")
generated = document.get("generated_apb_crossings")
if generated is not None:
if (document.get("clock_contract") != "TINY-R2-P7" or
generated.get("generator") != "scripts/generate_tiny.py" or
generated.get("primitive") != "tiny_apb_async_bridge" or
generated.get("source") != "system"):
raise ValueError("unrecognized generated APB crossing contract")
expected = {"aon": ["clint", "archinfo"], "memory": ["xpi"],
"peripheral": ["gpio", "uart0", "timer0", "timer1", "i2c0", "dma",
"pwm", "rtc", "wdg", "rng", "crc", "ws2812", "crypto", "i2s"]}
if (generated.get("destinations") != expected or
generated.get("same_clock_isolation") != ["sdio"] or
generated.get("native_system") != ["sysctrl", "sram"] or
not set(expected).issubset(names)):
raise ValueError("P7 generated APB domain assignment mismatch")
crossing_names: set[str] = set()
for index, value in enumerate(raw_crossings):
field = f"crossings[{index}]"
+15
View File
@@ -39,6 +39,19 @@ def main() -> int:
if default_failure:
rejected.append(default_failure.group(0))
flow_result_path = args.result.with_name("result-sim.json")
flow_status = None
if flow_result_path.is_file():
try:
flow_result = json.loads(flow_result_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
flow_result = {}
flow_status = flow_result.get("status", "invalid")
# PIDs are namespace-local and can be reused. A durable terminal
# verdict, not a liveness guess from this process, establishes success.
if flow_status != "passed" or flow_result.get("exit_code") != 0:
missing.append("successful terminal run_flow result")
passed = not missing and not rejected
report = {
"schema_version": 1,
@@ -48,6 +61,8 @@ def main() -> int:
"missing_markers": missing,
"rejected_matches": sorted(set(rejected)),
}
if flow_status is not None:
report["flow_status"] = flow_status
atomic_write(args.result, json.dumps(report, indent=2, sort_keys=True) + "\n")
print(f"simulation check: {report['status']} ({args.result.resolve()})")
return int(not passed)
+115 -31
View File
@@ -16,13 +16,36 @@ from scripts.rtl.generate_pin_map import generate as generate_pin_map # noqa: E
from scripts.setup_helpers import atomic_write # noqa: E402
TINY = ROOT / "rtl/tiny"
RCU_TARGETS = {"gpio": 1, "uart0": 2, "i2c0": 3, "timer0": 4, "timer1": 5,
"pwm": 6, "i2s": 7, "sdio": 8, "dma": 9, "ws2812": 10,
"rng": 11, "crc": 12, "crypto": 13}
def read_topology(path: Path, memory_map: Path) -> dict:
document = json.loads(path.read_text())
regions = {r["symbol"] for r in json.loads(memory_map.read_text())["regions"]}
if document.get("schema_version") != 1 or document.get("dma_channels") != 4:
raise ValueError("Tiny requires topology schema 1 and four DMA channels")
if (document.get("schema_version") != 1 or document.get("dma_channels") != 8
or document.get("dma_request_mask") != "0x31ff"):
raise ValueError("Tiny P6 requires eight DMA channels and request mask 0x31ff")
if document.get("cpu") != {"module": "hazard3_cpu_2port", "local_ports": 2,
"data_outstanding": 1}:
raise ValueError("Tiny P4 requires independent I/D and one ordered data operation")
memory = document.get("main_sram", {})
if any(memory.get(key) != value for key, value in {
"groups": 4, "group_bytes": 32768, "macros_per_group": 8,
"macro_bytes": 4096, "base": "0x30000000",
"macro_instance": "u_soc.u_sram.gen_group[{group}].u_group.gen_bank[{bank}].u_ram.u_mem",
"external_attachment": "per-target-P5",
}.items()):
raise ValueError("Tiny SRAM geometry/P5 attachment differs from its frozen contract")
if document.get("fabric") != {
"owners": ["cpu_slow", "central_dma", "sdio"], "active_owners": [0, 1, 2],
"cpu_combined_outstanding": 1, "dma_direction_outstanding": 1,
"sdio_direction_outstanding": 1, "target_combined_outstanding": 1,
"sram_targets": [0, 1, 2, 3], "xpi_target": 4, "apb_target_base": 5,
"decode_error_target": 25, "protocol_error_target": 26, "target_count": 27,
}:
raise ValueError("Tiny P6 requires three active owners and 27 targets")
if document.get("irq_vector_width") != 32 or document.get("external_irq_count") != 30:
raise ValueError("Tiny requires a 32-bit IRQ vector and 30 external inputs")
names: set[str] = set()
@@ -42,14 +65,24 @@ def read_topology(path: Path, memory_map: Path) -> dict:
assigned.add(region)
if assigned != {r for r in regions if r.startswith("APB4_")}:
raise ValueError("Tiny topology must route every APB region")
if len(document["apb_targets"]) != 16:
raise ValueError("Tiny's fixed APB bridge requires sixteen targets")
if len(document["apb_targets"]) != 20:
raise ValueError("Tiny P6 requires twenty independent APB targets")
target_names = [target["name"] for target in document["apb_targets"]]
if len(set(target_names)) != 16 or any(re.fullmatch(r"[a-z][a-z0-9_]*", name) is None
if len(set(target_names)) != 20 or any(re.fullmatch(r"[a-z][a-z0-9_]*", name) is None
for name in target_names):
raise ValueError("Tiny APB target names must be unique identifiers")
if sorted(item["pin"] for item in document["gpio_alt_functions"]) != list(range(32)):
raise ValueError("Tiny GPIO map must contain all 32 pins exactly once")
inputs = [signal for pin in document["gpio_alt_functions"] for mode in ("alt0", "alt1")
for signal in pin[mode]["inputs"]]
if len(inputs) != len(set(inputs)):
raise ValueError("Tiny peripheral inputs require one exclusive route")
if document.get("private_crypto") != {
"banks": 6, "bank_bytes": 4096,
"macro_instance": "u_soc.u_crypto.u_storage.gen_bank[{bank}].gen_macro.u_sram.u_mem",
"clock_domain": "peripheral", "logical_domain": "PCLK",
}:
raise ValueError("Tiny requires six private Crypto banks on PCLK")
return document
@@ -65,7 +98,23 @@ def generate_bindings(document: dict, output: Path, pdk: str = "IHP130",
have_pll: bool = False) -> None:
platform = platform_values(pdk, have_pll)
comment = "// Generated by scripts/generate_tiny.py; do not edit.\n"
irq = [comment, "always_comb begin", " s_irq = '0;"]
constants = [comment, "`ifndef RETROSOC_TINY__TOPOLOGY_SVH",
"`define RETROSOC_TINY__TOPOLOGY_SVH",
f"`define RETROSOC_TINY__TARGET_COUNT {document['fabric']['target_count']}",
f"`define RETROSOC_TINY__LEAF_COUNT {len(document['apb_targets']) + 1}",
f"`define RETROSOC_TINY__DECODE_ERROR {document['fabric']['decode_error_target']}",
f"`define RETROSOC_TINY__PROTOCOL_ERROR {document['fabric']['protocol_error_target']}"]
constants += [f"`define RETROSOC_TINY__TARGET_{target['name'].upper()} {index + 5}"
for index, target in enumerate(document["apb_targets"])]
constants.append("`endif\n")
atomic_write(output / "rtl/tiny_topology.svh", "\n".join(constants))
# Keep board pulls independently resolved. A packed tri1 with partial
# external drivers loses its weak pulls in the locked Verilator 5.046.
board = [comment, *[f"tri1 s_gpio_{pin};" for pin in range(document["gpio_pins"])]]
board.append("wire [31:0] s_gpio = {" +
", ".join(f"s_gpio_{pin}" for pin in reversed(range(32))) + "};\n")
atomic_write(output / "rtl/tiny_tb_gpio.svh", "\n".join(board))
irq = [comment, "always_comb begin", " s_irq_async = '0;"]
header = ["#ifndef RETROSOC_GENERATED_IRQ_METADATA_H",
"#define RETROSOC_GENERATED_IRQ_METADATA_H", "#include <stdint.h>",
"#define RS_SOC_IRQ_VECTOR_WIDTH UINT32_C(32)",
@@ -73,7 +122,7 @@ def generate_bindings(document: dict, output: Path, pdk: str = "IHP130",
f"#define RS_SOC_ALLOCATED_IRQ_COUNT UINT32_C({len(document['interrupts']) - 2})"]
for item in document["interrupts"]:
bit, name = item["core_bit"], item["name"].upper()
irq.append(f" s_irq[{bit}] = {item['signal']};")
irq.append(f" s_irq_async[{bit}] = {item['signal']};")
header.append(f"#define RS_SOC_IRQ_{name} UINT32_C({bit})")
if bit >= 2:
header.append(f"#define RS_SOC_EXT_IRQ_{name} UINT32_C({bit - 2})")
@@ -90,7 +139,15 @@ def generate_bindings(document: dict, output: Path, pdk: str = "IHP130",
"#ifndef RETROSOC_GENERATED_SOC_FEATURES_H\n"
"#define RETROSOC_GENERATED_SOC_FEATURES_H\n"
"#define RS_SOC_TINY 1\n#define RS_SOC_HAS_PSRAM 0U\n"
"#define RS_SOC_DMA_CHANNELS 4U\n#define RS_SOC_ID 0x54494E59U\n"
"#define RS_SOC_DMA_CHANNELS 8U\n#define RS_SOC_ID 0x54494E59U\n"
"#define RS_SOC_DMA_REQUEST_MASK 0x31FFU\n"
"#define RS_SOC_HAS_I2C1 0U\n#define RS_SOC_HAS_SDIO1 0U\n"
"#define RS_SOC_HAS_UART0 1U\n#define RS_SOC_HAS_UART1 0U\n"
"#define RS_SOC_HAS_I2C0 1U\n"
"#define RS_SOC_HAS_RNG 1U\n#define RS_SOC_HAS_CRC 1U\n"
"#define RS_SOC_HAS_WS2812 1U\n#define RS_SOC_HAS_CRYPTO 1U\n"
"#define RS_SOC_HAS_I2S 1U\n#define RS_SOC_HAS_SDIO0 1U\n"
"#define RS_SOC_HAS_DVP 0U\n#define RS_SOC_RNG_QUALIFIED 0U\n"
f"#define RS_SOC_TECHNOLOGY_ID 0x{platform['technology']:08X}U\n"
f"#define RS_SOC_PLL_PRESENT {platform['pll_present']}U\n"
f"#define RS_SOC_GPIO_HAS_PULLS {platform['gpio_pulls']}U\n#endif\n")
@@ -106,33 +163,60 @@ def generate_bindings(document: dict, output: Path, pdk: str = "IHP130",
gpio.extend((f"assign u_gpio_if.{mode}_do_i[{pin}] = {value['do']};",
f"assign u_gpio_if.{mode}_oe_i[{pin}] = {value['oe']};"))
atomic_write(output / "rtl/tiny_gpio.svh", "\n".join(gpio) + "\n")
ports, declarations, routes, decode = [], [comment], [comment], [comment]
declarations, bridges, decode = [comment], [comment], [comment]
reset_assignments = []
for index, target in enumerate(document["apb_targets"]):
name = target["name"]
ports.append(f"apb4_if.master {name}")
clock = ("s_ref_clk" if name in ("clint", "archinfo") else
"s_backend_mem_clk" if name == "xpi" else
"s_exec_clk" if name in ("sysctrl", "sram", "sdio") else "s_backend_pclk")
target_bit = RCU_TARGETS.get(name)
ip_clock = f"s_{name}_clk" if target_bit is not None else clock
if target_bit is not None:
declarations.append(f"logic {ip_clock};")
declarations.append(f"logic s_{name}_rst_n;")
declarations.append(f"apb4_if u_{name}_apb4_if (.pclk(clk_i), .presetn(s_{name}_rst_n));")
reset_assignments.append(f"assign s_{name}_rst_n = s_leaf_rst_n[{index + 1}];")
routes.append(f".{name}(u_{name}_apb4_if)")
decode.append(f"if ({' || '.join(f'`SOC_ADDR_IS_{r}(s_addr_q)' for r in target['regions'])}) "
f"s_select = {len(document['apb_targets'])}'b{1 << index:016b};")
atomic_write(output / "rtl/tiny_apb_ports.svh", ",\n".join(ports) + "\n")
declarations.append(f"apb4_if u_{name}_apb4_if (.pclk({ip_clock}), .presetn(s_{name}_rst_n));")
if target_bit is None:
reset_assignments.append(f"assign s_{name}_rst_n = s_leaf_rst_n[{index + 1}];")
else:
reset_assignments.append(f"(* keep_hierarchy = 1 *) tiny_peripheral_leaf u_{name}_lifecycle ("
f".clk_i({clock}), .rst_n_i(s_leaf_rst_n[{index + 1}]), "
f".gate_i(s_gate_request[{target_bit}]), "
f".reset_i(s_local_reset[{target_bit}]), "
f".clk_o({ip_clock}), .rst_n_o(s_{name}_rst_n), "
f".gated_o(s_actual_gate[{target_bit}]), "
f".held_o(s_reset_held[{target_bit}]));")
extra_ready = " && s_i2s_reset_ready" if name == "i2s" else ""
if name == "dma":
# CDC-ready outputs are observed in SYS, matching APB readiness.
extra_ready = " && s_dma_ready_sys && s_dma_idle_sys"
reset_assignments.append(f"assign s_reset_ready[{target_bit}] = s_{name}_rst_n && "
f"s_apb_cdc_ready[{index}]{extra_ready};")
reset_assignments.append(f"assign s_leaf_clk[{index + 1}] = {clock};")
bridges.append(f"assign s_target_rst_n[{index + 5}] = s_fabric_rst_n;")
crossed = clock != "s_exec_clk" or target_bit is not None
endpoint = f"u_{name}_sys_apb4_if" if crossed else f"u_{name}_apb4_if"
bridges.append(f"tiny_axi42apb4 u_{name}_bridge ("
f".clk_i(s_exec_clk), .rst_n_i(s_fabric_rst_n), "
f".axi4(u_targets_axi4_if[{index + 5}]), .apb4({endpoint}));")
if crossed:
disabled = f"s_disabled_sys[{target_bit}]" if target_bit is not None else "1'b0"
declarations.append(f"apb4_if {endpoint} (.pclk(s_exec_clk), .presetn(s_fabric_rst_n));")
bridges.append(f"tiny_apb_async_bridge u_{name}_cfg_cdc ("
f".src_clk_i(s_exec_clk), .src_rst_n_i(s_fabric_rst_n), "
f".dst_clk_i({ip_clock}), .dst_rst_n_i(s_{name}_rst_n), "
f".disabled_i({disabled}), "
f".idle_o(s_apb_cdc_idle[{index}]), "
f".ready_o(s_apb_cdc_ready[{index}]), "
f".src_apb4({endpoint}), .dst_apb4(u_{name}_apb4_if));")
else:
bridges.append(f"assign s_apb_cdc_idle[{index}] = 1'b1;")
bridges.append(f"assign s_apb_cdc_ready[{index}] = 1'b1;")
decode.append(f"if ({' || '.join(f'`SOC_ADDR_IS_{r}(addr_i)' for r in target['regions'])}) "
f"return 5'd{index + 5};")
atomic_write(output / "rtl/tiny_apb_interfaces.svh", "\n".join(declarations + reset_assignments) + "\n")
atomic_write(output / "rtl/tiny_apb_connections.svh", ",\n".join(routes[1:]) + "\n")
atomic_write(output / "rtl/tiny_apb_decode.svh", "\n".join(decode) + "\n")
wiring = [comment]
for index, target in enumerate(document["apb_targets"]):
name = target["name"]
for field, value in {"paddr": "s_addr_q", "pwrite": "s_write_q",
"pwdata": "s_wdata_q", "pstrb": "s_wstrb_q", "pprot": "3'd0",
"psel": f"s_select[{index}] && s_access",
"penable": "s_enable"}.items():
wiring.append(f"assign {name}.{field} = {value};")
wiring.extend((f"assign s_ready[{index}] = {name}.pready;",
f"assign s_error[{index}] = {name}.pslverr;",
f"assign s_rdata[{index}] = {name}.prdata;"))
atomic_write(output / "rtl/tiny_apb_wiring.svh", "\n".join(wiring) + "\n")
atomic_write(output / "rtl/tiny_apb_bridges.svh", "\n".join(bridges) + "\n")
atomic_write(output / "rtl/tiny_fabric_decode.svh", "\n".join(decode) + "\n")
def generate_filelists(output: Path, generated: Path, defines: list[str],
@@ -141,7 +225,7 @@ def generate_filelists(output: Path, generated: Path, defines: list[str],
prefixes = {"top": TINY / "top", "dv": TINY / "dv", "ip": ROOT / "rtl/ip",
"tech": ROOT / "rtl/tech", "clusterip": ROOT / "rtl/managed/clusterip",
"hazard3": ROOT / "rtl/managed/hazard3/hdl", "physical": ROOT / "physical",
"cache": ROOT / ".cache"}
"cache": ROOT / ".cache", "tests": ROOT / "tests"}
suffix = "_yosys" if synthesis == "YOSYS" else "_verilator" if simulator == "VERILATOR" else ""
pdk_name = "pdk_ihp130.fl" if pdk == "IHP130" else f"pdk_ics55{suffix}.fl"
templates = [*sorted((TINY / "filelist").glob("*.fl")),
+2
View File
@@ -43,6 +43,7 @@ TEST_SOURCES = (
"crt/src/hal/fabric_monitor.c",
"crt/src/hal/sysctrl.c",
"crt/src/hal/clock.c",
"crt/src/hal/rcu.c",
"crt/src/hal/resource.c",
"crt/src/hal/extension.c",
"crt/src/hal/user_ip.c",
@@ -122,6 +123,7 @@ def main() -> int:
"-DRS_NPU_TEST_MMIO",
"-DRS_FABRIC_MONITOR_TEST_MMIO",
"-DRS_RESOURCE_TEST_MMIO",
"-DRS_TINY_RCU_TEST_MMIO",
"-I",
str(memory_map_root / "include"),
"-I",
+35 -16
View File
@@ -124,24 +124,22 @@ def main() -> int:
args.result.parent.mkdir(parents=True, exist_ok=True)
started = datetime.now(timezone.utc)
start_clock = time.monotonic()
atomic_write(
args.result,
json.dumps(
{
"schema_version": 1,
"tool": args.tool,
"command": command,
"started_at": started.isoformat(),
"status": "running",
"log": str(args.log.resolve()),
},
indent=2,
sort_keys=True,
)
+ "\n",
)
running_result = {
"schema_version": 1,
"tool": args.tool,
"command": command,
"started_at": started.isoformat(),
"status": "running",
"parent_pid": os.getpid(),
"log": str(args.log.resolve()),
}
atomic_write(args.result, json.dumps(running_result, indent=2, sort_keys=True) + "\n")
def interrupt_flow(_signum: int, _frame: object) -> None:
# A launcher can deliver SIGINT followed by SIGTERM. Do not let a
# second signal interrupt process-group cleanup or result persistence.
signal.signal(signal.SIGINT, signal.SIG_IGN)
signal.signal(signal.SIGTERM, signal.SIG_IGN)
raise KeyboardInterrupt
signal.signal(signal.SIGINT, interrupt_flow)
@@ -169,6 +167,8 @@ def main() -> int:
pass_fds=jobserver_fds(),
start_new_session=True,
)
running_result["child_pid"] = process.pid
atomic_write(args.result, json.dumps(running_result, indent=2, sort_keys=True) + "\n")
assert process.stdout is not None
if args.stream_bytes:
while chunk := os.read(process.stdout.fileno(), 4096):
@@ -193,11 +193,25 @@ def main() -> int:
returncode = process.wait()
except KeyboardInterrupt:
error_message = "interrupted"
# Persist cancellation before the bounded cleanup wait. If the outer
# executor subsequently kills this wrapper, the file must not imply
# that the job is still running or that its child finished normally.
cancelled = dict(running_result, status="interrupted", exit_code=130,
interrupted_at=datetime.now(timezone.utc).isoformat(),
child_cleanup_pending=process is not None)
atomic_write(args.result, json.dumps(cancelled, indent=2, sort_keys=True) + "\n")
if process is not None:
terminate_process_group(process)
returncode = 130
except (OSError, subprocess.SubprocessError) as error:
error_message = str(error)
if process is not None:
terminate_process_group(process)
except BaseException as error:
error_message = f"unexpected {type(error).__name__}: {error}"
if process is not None:
terminate_process_group(process)
returncode = 1
finished = datetime.now(timezone.utc)
result = {
@@ -211,8 +225,13 @@ def main() -> int:
"peak_rss_kib": resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss,
"exit_code": returncode,
"status": "interrupted" if returncode == 130 else ("passed" if returncode == 0 else "failed"),
"parent_pid": os.getpid(),
"log": str(args.log.resolve()),
}
if process is not None:
result["child_pid"] = process.pid
result["child_exit_code"] = process.returncode
result["child_cleanup_pending"] = process.returncode is None
if error_message:
result["error"] = error_message
if args.success_marker:
+1 -1
View File
@@ -15,7 +15,7 @@ from scripts.setup_helpers import ensure_git_repo # noqa: E402
TINY_SOURCES = (
"hazard3", "cluster_common", "cluster_archinfo", "cluster_pwm",
"cluster_rtc", "cluster_wdg", "pdk_ihp130",
"cluster_rtc", "cluster_wdg", "cluster_rng", "cluster_crc", "pdk_ihp130",
)
+31 -4
View File
@@ -114,10 +114,12 @@ def verify_locked_views(root: Path) -> dict:
return verified
def inspect_netlist(document: dict) -> dict:
def expand_netlist(document: dict) -> tuple:
"""Resolve hierarchy aliases without assigning a product clock contract."""
modules = document["modules"]
top = "retrosoc_tiny_asic"
leaves = []
boundaries = {}
aliases = {}
def canonical(bit):
@@ -159,6 +161,7 @@ def inspect_netlist(document: dict) -> dict:
path = f"{scope}/{name}"
attrs = modules.get(kind, {}).get("attributes", {})
if kind in modules and not int(str(attrs.get("blackbox", "0")), 2):
boundaries[path] = {"path": path, "type": kind, "ports": ports}
visit(kind, path, ports)
else:
directions = cell.get("port_directions", {
@@ -168,21 +171,44 @@ def inspect_netlist(document: dict) -> dict:
"directions": directions})
visit(top, top, {})
for c in leaves:
for c in [*leaves, *boundaries.values()]:
for port, bits in c["ports"].items():
c["ports"][port] = [canonical(bit) for bit in bits]
return modules, top, leaves, canonical, boundaries
def inspect_netlist(document: dict) -> dict:
# Keep the original P1 parked-PLL/direct-SYS contract fail-closed.
modules, top, leaves, canonical, _ = expand_netlist(document)
clock = modules[top]["cells"]["u_clock_buffer"]["connections"]["clk_o"]
if len(clock) != 1 or not isinstance(clock[0], int):
raise ValueError("missing live SYS clock")
system = canonical(f"{top}:{clock[0]}")
if system in ("0", "1", "x", "z"):
raise ValueError("SYS is not a live clock net")
ram = [c for c in leaves if c["type"] == RAM]
all_ram = [c for c in leaves if c["type"] == RAM]
crypto_names = {f"u_soc.u_crypto.u_storage.gen_bank[{n}].gen_macro.u_sram.u_mem"
for n in range(6)}
crypto = [c for c in all_ram if any(c["path"].replace("/", ".").endswith("." + name)
for name in crypto_names)]
observed_crypto = {name for name in crypto_names for c in crypto
if c["path"].replace("/", ".").endswith("." + name)}
if len(all_ram) != 38 or len(crypto) != 6 or observed_crypto != crypto_names:
raise ValueError("Tiny P6 requires exactly 32 main SRAM and six private Crypto macros")
if any(c["ports"].get("CLK") != [system] for c in crypto):
raise ValueError("private Crypto SRAM does not use SAFE24 PCLK")
ram = [c for c in all_ram if c not in crypto]
pll = [c for c in leaves if c["type"] == "PLL_TOP"]
if any(c["type"] == "ics55_ecos_sram_4096x32_m8" for c in leaves):
raise ValueError("Tiny instantiated the unsupported 16 KiB SRAM interface stub")
if len(ram) != 32 or len(pll) != 1:
raise ValueError("expected 32 main SRAM macros and one PLL")
expected_ram = {f"u_soc.u_sram.gen_group[{g}].u_group.gen_bank[{n}].u_ram.u_mem"
for g in range(4) for n in range(8)}
observed_ram = {next((name for name in expected_ram
if c["path"].replace("/", ".").endswith("." + name)), "") for c in ram}
if observed_ram != expected_ram:
raise ValueError("Tiny P4 SRAM group/macro hierarchy is missing or stale")
if any(c["ports"].get("CLK") != [system] for c in ram):
raise ValueError("main SRAM does not share direct SYS")
constants = {"0": 0, "1": 1}
@@ -202,7 +228,7 @@ def inspect_netlist(document: dict) -> dict:
enable = pll[0]["ports"].get("EN", [])
if len(enable) != 1 or constants.get(enable[0]) != 0:
raise ValueError("PLL enable is not proven constant zero")
cpu = [c for c in leaves if "u_hazard3_cpu_1port" in c["path"]
cpu = [c for c in leaves if "u_hazard3_cpu_2port" in c["path"]
and ("CK" in c["ports"] or "CKN" in c["ports"])]
if not cpu or any(c["ports"].get("CK", c["ports"].get("CKN")) != [system] for c in cpu):
raise ValueError("CPU sequential clocks do not share direct SYS")
@@ -213,6 +239,7 @@ def inspect_netlist(document: dict) -> dict:
"system_clock_source": {"port": "extclk_i_pad", "net": reference, "path": clock_path},
"cpu_clocked_cells": len(cpu),
"sram": [{"path": c["path"], "clock": c["ports"]["CLK"]} for c in ram],
"crypto_sram": [{"path": c["path"], "clock": c["ports"]["CLK"]} for c in crypto],
"pll": pll[0], "pll_enable": 0, "qualification": "SAFE24 digital connectivity only"}
+255
View File
@@ -0,0 +1,255 @@
#!/usr/bin/env python3
"""Replay immutable P3 images on P4 RTL without relabeling image provenance."""
from __future__ import annotations
import argparse
from pathlib import Path
import sys
import subprocess
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from scripts import tiny_r2_baseline as b # noqa: E402
from scripts import tiny_r2_p3 as p3 # noqa: E402
PHASE = "TINY-R2-P4"
HARDWARE_KEYS = ("SOC", "PDK", "HAVE_PLL", "HAVE_SRAM_IF", "HAVE_SRAM_MACRO",
"SRAM_SIZE_KIB", "PDK_BEHAV", "HAVE_HP", "EXT_CLK_HZ", "AUD_CLK_HZ",
"CLINT_TIMEBASE_HZ", "MGMT_CPU_CLK_HZ", "JTAG_IDCODE")
LOCAL_FIELDS = {"window", "port", "accepted", "completed", "latency_max", "admission_wait"}
BANK_FIELDS = {"window", "group", "issues_i", "issues_d", "issues_external", "conflicts",
"wait_i", "wait_d", "wait_external"}
def inputs(root: Path, variant: Path) -> dict:
result = p3.inputs(root, variant)
# Publication bindings are tested source consumers, not generated PDFs.
names = subprocess.check_output(
["git", "ls-files", "--cached", "--others", "--exclude-standard", "-z", "--", "publications"],
cwd=root, text=True).split("\0")
for name in names:
path = root / name
if name and path.is_file() and path.suffix in (".py", ".json", ".typ"):
result[name] = p3.sha256(path)
return result
def compatible_hardware(image: dict, model: dict) -> None:
a, z = (c["configuration"] for c in (image, model))
if any(key not in a or key not in z or a[key] != z[key] for key in HARDWARE_KEYS):
raise ValueError("retained image and candidate hardware configuration differ")
def parse_log(log: str, flow: dict) -> dict:
samples = p3.parse_log(log, flow)
rows = [b.parse_fields(line, LOCAL_FIELDS) for line in log.splitlines()
if line.startswith("R2_P4_LOCAL ")]
rows.sort(key=lambda row: (row["window"], row["port"]))
if [(r["window"], r["port"]) for r in rows] != [(w, p) for w in range(1, 9) for p in range(2)]:
raise ValueError("missing or duplicate P4 local observations")
for row in rows:
if row["completed"] > row["accepted"] or row["accepted"] - row["completed"] > 1:
raise ValueError("P4 local request/response accounting mismatch")
samples["local"] = rows
banks = [b.parse_fields(line, BANK_FIELDS) for line in log.splitlines()
if line.startswith("R2_P4_BANK ")]
banks.sort(key=lambda row: (row["window"], row["group"]))
if [(r["window"], r["group"]) for r in banks] != [(w, g) for w in range(1, 9) for g in range(4)]:
raise ValueError("missing or duplicate P4 bank observations")
samples["banks"] = banks
return samples
def reference_cases(path: Path, names: list[str]) -> list[dict]:
report = b.read_json(path)
if report.get("phase") != p3.PHASE or report.get("status") != "measurement_complete":
raise ValueError("reference is not a completed P3 measurement report")
matrix = b.read_json(b.check_artifact(report["matrix"]))
if matrix.get("status") != "passed":
raise ValueError("reference matrix did not pass")
if not names or len(set(names)) != len(names):
raise ValueError("select distinct retained cases")
result = []
for name in names:
matching = [c for c in report["cases"] if c["name"] == name]
original = [c for c in matrix["cases"] if c["name"] == name]
if len(matching) != 1 or len(original) != 1 or matching[0]["image"] != original[0]["image"]:
raise ValueError("unknown or mismatched retained case")
case = matching[0]
p3.validate_image(b.check_artifact(case["image"]))
# Recheck the original logs, not just a report's claimed numeric values.
samples = p3.validate_simulation(original[0]["simulations"]["verilator"],
case["image"], "verilator")
if samples != case["samples"]:
raise ValueError("reference report differs from its retained execution")
result.append(case)
return result
def replay(args) -> None:
root, variant = args.root, args.variant_root
parent = variant / "meta/tiny-r2-p4/replay" / args.simulator
directory = b.new_attempt(parent, "attempt-")
record = {"phase": PHASE, "status": "running", "repetitions": 1, "cases": []}
b.dump(parent / "latest.json", {"path": str(directory / "record.json")})
try:
reference = b.artifact(args.reference)
cases = reference_cases(args.reference, args.cases.split(","))
configuration = p3.configuration(variant)
for case in cases:
compatible_hardware(b.read_json(b.check_artifact(case["image"]))["configuration"],
configuration)
before = inputs(root, variant)
tools = b.tools_identity(root)
command = args.command[1:] if args.command[:1] == ["--"] else args.command
if not command:
raise ValueError("fresh model compilation command required")
p3.execute_make(root, directory, "compile-model", command)
if before != inputs(root, variant):
raise ValueError("source changed while compiling candidate model")
runtime = b.runtime_inputs(root, variant, args.simulator)
model = variant / "sim" / args.simulator / ("emu" if args.simulator == "verilator" else "simv")
record.update(reference=reference, configuration=configuration, tools=tools,
model=b.artifact(model), source_digest=b.digest_map(before),
source=b.snapshot(root, directory / "inputs", before | runtime),
compilation={s: b.artifact(directory / f"compile-model.{s}") for s in ("json", "log")})
launch = [str(model)]
if args.simulator == "iverilog":
record["runtime_tool"] = b.vvp_identity(args.vvp, tools["iverilog"])
launch.insert(0, record["runtime_tool"]["executable"]["path"])
for case in cases:
image = p3.validate_image(b.check_artifact(case["image"]))
begin, end = image["layout"]["descriptor_range"]
cmd = [*launch, "+firmware=" + image["artifacts"]["tiny_p3.hex"]["path"],
"+tiny_r2_p3", f"+baseline_tcd_base={begin:x}", f"+baseline_tcd_end={end:x}",
"+max_cycles=100000000"]
label = case["name"]
flow = b.execute(root, directory, label,
["timeout", "--foreground", f"{args.timeout}s", *cmd], quiet=True)
samples = parse_log((directory / f"{label}.log").read_text(), flow)
p3.validate_image(b.check_artifact(case["image"]))
record["cases"].append({"name": label, "image": case["image"], "samples": samples,
"baseline_samples": case["samples"],
"command": cmd, "timeout_seconds": args.timeout,
"flow": b.artifact(directory / f"{label}.json"),
"log": b.artifact(directory / f"{label}.log")})
b.dump(directory / "record.json", record)
if before != inputs(root, variant) or runtime != b.runtime_inputs(root, variant, args.simulator):
raise ValueError("candidate inputs changed during execution")
b.check_artifact(record["model"])
b.check_artifact(reference)
for case in cases:
p3.validate_image(b.check_artifact(case["image"]))
for tool in tools.values():
b.check_artifact(tool["executable"])
record["status"] = "passed"
except (OSError, ValueError) as error:
record.update(status="failed", error=str(error))
raise
finally:
b.dump(directory / "record.json", record)
print(f"P4 {args.simulator}: one execution per image; {directory}")
def validate_replay(path: Path, simulator: str) -> dict:
record = b.read_json(path)
if record.get("phase") != PHASE or record.get("status") != "passed" or record.get("repetitions") != 1:
raise ValueError("not a completed single-execution P4 replay")
for field in ("model", "reference"):
b.check_artifact(record[field])
for value in record["compilation"].values():
b.check_artifact(value)
for field in ("inputs", "snapshot", "patch"):
b.check_artifact(record["source"][field])
captured = b.read_json(Path(record["source"]["inputs"]["path"]))
if b.digest_map(captured) != record["source"]["input_digest"]:
raise ValueError("candidate source digest mismatch")
for tool in record["tools"].values():
b.check_artifact(tool["executable"])
b.check_artifact(tool["archive"])
prefix = [record["model"]["path"]]
if simulator == "iverilog":
runtime = b.check_artifact(record["runtime_tool"]["executable"])
compiler = Path(record["tools"]["iverilog"]["executable"]["path"])
if runtime.resolve().parent != compiler.resolve().parent:
raise ValueError("unmatched Icarus runtime")
prefix.insert(0, str(runtime))
reference = reference_cases(Path(record["reference"]["path"]), [c["name"] for c in record["cases"]])
for case, original in zip(record["cases"], reference, strict=True):
if case["image"] != original["image"] or case["baseline_samples"] != original["samples"]:
raise ValueError("comparison changed the retained image or reference measurements")
image = p3.validate_image(b.check_artifact(case["image"]))
compatible_hardware(image["configuration"], record["configuration"])
first, end = image["layout"]["descriptor_range"]
expected = [*prefix, "+firmware=" + image["artifacts"]["tiny_p3.hex"]["path"],
"+tiny_r2_p3", f"+baseline_tcd_base={first:x}", f"+baseline_tcd_end={end:x}",
"+max_cycles=100000000"]
flow = b.read_json(b.check_artifact(case["flow"]))
if case["command"] != expected or case["timeout_seconds"] <= 0 or flow["command"] != [
"timeout", "--foreground", f"{case['timeout_seconds']}s", *expected]:
raise ValueError("model/image execution provenance mismatch")
if parse_log(b.check_artifact(case["log"]).read_text(), flow) != case["samples"]:
raise ValueError("reported candidate samples differ from the log")
return record
def report(args) -> None:
base = args.variant_root / "meta/tiny-r2-p4"
runs = {}
for simulator in ("verilator", "iverilog"):
pointer = b.read_json(base / "replay" / simulator / "latest.json")
path = Path(pointer["path"])
runs[simulator] = (validate_replay(path, simulator), b.artifact(path))
before = b.digest_map(inputs(args.root, args.variant_root))
if any(run[0]["source_digest"] != before for run in runs.values()):
raise ValueError("candidate source changed after measurement")
cases = {c["name"]: c for c in runs["verilator"][0]["cases"]}
if set(cases) != {"perf-o2-plain-flat", "perf-o2-plain-banked"}:
raise ValueError("P4 comparison requires both approved placement images")
cross = runs["iverilog"][0]["cases"]
if len(cross) != 1 or cross[0]["name"] != "perf-o2-plain-banked":
raise ValueError("P4 requires one banked Icarus execution")
if any(cross[0][key] != cases["perf-o2-plain-banked"][key] for key in ("image", "samples")):
raise ValueError("cross-simulator image/results differ")
comparisons = []
for name, case in cases.items():
rows = []
for old, new in zip(case["baseline_samples"]["cases"], case["samples"]["cases"], strict=True):
rows.append({"name": new["name"], "baseline_cycles": old["cycles"],
"candidate_cycles": new["cycles"], "baseline_instructions": old["instructions"],
"candidate_instructions": new["instructions"],
"cycle_ratio": old["cycles"] / new["cycles"]})
comparisons.append({"name": name, "image": case["image"], "workloads": rows,
"local": case["samples"]["local"], "banks": case["samples"]["banks"]})
b.dump(base / "report.json", {"phase": PHASE, "status": "measurement_complete",
"source_digest": before, "comparisons": comparisons,
"executions": {key: value[1] for key, value in runs.items()},
"qualification": "matched-binary SAFE24 simulation; not phase/physical acceptance"})
print(f"P4 matched-image report: {base / 'report.json'}")
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, required=True)
parser.add_argument("--variant-root", type=Path, required=True)
parser.add_argument("--report", action="store_true")
parser.add_argument("--reference", type=Path)
parser.add_argument("--cases")
parser.add_argument("--simulator", choices=("verilator", "iverilog"))
parser.add_argument("--vvp", default="vvp")
parser.add_argument("--timeout", type=int, default=21600)
parser.add_argument("command", nargs=argparse.REMAINDER)
args = parser.parse_args()
if args.timeout <= 0:
parser.error("timeout must be positive")
if args.report:
report(args)
else:
if args.reference is None or not args.cases or not args.simulator:
parser.error("replay requires reference, cases and simulator")
replay(args)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+260
View File
@@ -0,0 +1,260 @@
#!/usr/bin/env python3
"""Replay immutable P3 images on P5 RTL without relabeling image provenance."""
from __future__ import annotations
import argparse
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from scripts import tiny_r2_baseline as b # noqa: E402
from scripts import tiny_r2_p3 as p3 # noqa: E402
from scripts import tiny_r2_p4 as previous # noqa: E402
PHASE = "TINY-R2-P5"
HARDWARE_KEYS = ("SOC", "PDK", "HAVE_PLL", "HAVE_SRAM_IF", "HAVE_SRAM_MACRO",
"SRAM_SIZE_KIB", "PDK_BEHAV", "HAVE_HP", "EXT_CLK_HZ", "AUD_CLK_HZ",
"CLINT_TIMEBASE_HZ", "MGMT_CPU_CLK_HZ", "JTAG_IDCODE")
FABRIC_FIELDS = {"window", "active_targets_max", "dma_rw_overlap_cycles", "dma_rw_beat_overlap_cycles"}
def inputs(root: Path, variant: Path) -> dict:
result = previous.inputs(root, variant)
for name in ("AGENTS.md", "docs/engineering.md", "docs/rtl-coding-style.md",
"docs/rtl-coding-style-compliance.md", "rtl/rtl_style_manifest.json",
"rtl/rtl_style_audit.json", "rtl/rtl_readiness.json", ".verible-format",
".github/workflows/_regression.yml"):
result[name] = b.sha256(root / name)
for directory in (root / "quality", root / "rtl/tiny/mk"):
for path in directory.rglob("*"):
if path.is_file() and path.suffix in (".json", ".mk"):
result[str(path.relative_to(root))] = b.sha256(path)
return result
def compatible_hardware(image: dict, model: dict) -> None:
a, z = (c["configuration"] for c in (image, model))
if any(key not in a or key not in z or a[key] != z[key] for key in HARDWARE_KEYS):
raise ValueError("retained image and candidate hardware configuration differ")
def parse_log(log: str, flow: dict) -> dict:
samples = previous.parse_log(log, flow)
rows = [b.parse_fields(line, FABRIC_FIELDS) for line in log.splitlines()
if line.startswith("R2_P5_FABRIC ")]
rows.sort(key=lambda row: row["window"])
if [row["window"] for row in rows] != list(range(1, 9)):
raise ValueError("missing or duplicate P5 fabric observations")
for row in rows:
if any(value < 0 for value in row.values()) or row["active_targets_max"] > 23:
raise ValueError("invalid P5 fabric observation")
if row["dma_rw_beat_overlap_cycles"] > row["dma_rw_overlap_cycles"]:
raise ValueError("P5 beat overlap exceeds transaction overlap")
if row["dma_rw_overlap_cycles"] and row["active_targets_max"] < 2:
raise ValueError("P5 overlap without independent targets")
samples["fabric"] = rows
return samples
def reference_cases(path: Path, names: list[str]) -> list[dict]:
report = b.read_json(path)
if report.get("phase") != previous.PHASE or report.get("status") != "measurement_complete":
raise ValueError("reference is not a completed P4 measurement report")
if not names or len(set(names)) != len(names):
raise ValueError("select distinct retained cases")
original = previous.validate_replay(b.check_artifact(report["executions"]["verilator"]), "verilator")
if original["source_digest"] != report["source_digest"]:
raise ValueError("P4 report source differs from retained execution")
result = []
for name in names:
matches = [c for c in original["cases"] if c["name"] == name]
comparisons = [c for c in report["comparisons"] if c["name"] == name]
if len(matches) != 1 or len(comparisons) != 1:
raise ValueError("unknown retained P4 case")
case, comparison = matches[0], comparisons[0]
if comparison["image"] != case["image"]:
raise ValueError("P4 comparison changed the retained image")
rows = comparison["workloads"]
if len(rows) != len(case["samples"]["cases"]):
raise ValueError("P4 comparison workload count differs")
for row, sample in zip(rows, case["samples"]["cases"], strict=True):
if (row["name"], row["candidate_cycles"], row["candidate_instructions"]) != (
sample["name"], sample["cycles"], sample["instructions"]):
raise ValueError("P4 comparison differs from retained samples")
if comparison["local"] != case["samples"]["local"] or comparison["banks"] != case["samples"]["banks"]:
raise ValueError("P4 comparison altered local observations")
result.append({"name": name, "image": case["image"], "samples": case["samples"]})
return result
def replay(args) -> None:
root, variant = args.root, args.variant_root
parent = variant / "meta/tiny-r2-p5/replay" / args.simulator
directory = b.new_attempt(parent, "attempt-")
record = {"phase": PHASE, "status": "running", "repetitions": 1, "cases": []}
b.dump(directory / "record.json", record)
b.dump(parent / "latest.json", {"path": str(directory / "record.json")})
try:
reference = b.artifact(args.reference)
cases = reference_cases(args.reference, args.cases.split(","))
configuration = p3.configuration(variant)
for case in cases:
compatible_hardware(b.read_json(b.check_artifact(case["image"]))["configuration"],
configuration)
before = inputs(root, variant)
tools = b.tools_identity(root)
command = args.command[1:] if args.command[:1] == ["--"] else args.command
if not command:
raise ValueError("fresh model compilation command required")
p3.execute_make(root, directory, "compile-model", command)
if before != inputs(root, variant):
raise ValueError("source changed while compiling candidate model")
runtime = b.runtime_inputs(root, variant, args.simulator)
model = variant / "sim" / args.simulator / ("emu" if args.simulator == "verilator" else "simv")
record.update(reference=reference, configuration=configuration, tools=tools,
model=b.artifact(model), source_digest=b.digest_map(before),
source=b.snapshot(root, directory / "inputs", before | runtime),
compilation={s: b.artifact(directory / f"compile-model.{s}") for s in ("json", "log")})
launch = [str(model)]
if args.simulator == "iverilog":
record["runtime_tool"] = b.vvp_identity(args.vvp, tools["iverilog"])
launch.insert(0, record["runtime_tool"]["executable"]["path"])
for case in cases:
image = p3.validate_image(b.check_artifact(case["image"]))
begin, end = image["layout"]["descriptor_range"]
cmd = [*launch, "+firmware=" + image["artifacts"]["tiny_p3.hex"]["path"],
"+tiny_r2_p3", "+tiny_r2_p5", f"+baseline_tcd_base={begin:x}", f"+baseline_tcd_end={end:x}",
"+max_cycles=100000000"]
label = case["name"]
flow = b.execute(root, directory, label,
["timeout", "--foreground", f"{args.timeout}s", *cmd], quiet=True)
samples = parse_log((directory / f"{label}.log").read_text(), flow)
p3.validate_image(b.check_artifact(case["image"]))
record["cases"].append({"name": label, "image": case["image"], "samples": samples,
"baseline_samples": case["samples"],
"command": cmd, "timeout_seconds": args.timeout,
"flow": b.artifact(directory / f"{label}.json"),
"log": b.artifact(directory / f"{label}.log")})
b.dump(directory / "record.json", record)
if before != inputs(root, variant) or runtime != b.runtime_inputs(root, variant, args.simulator):
raise ValueError("candidate inputs changed during execution")
b.check_artifact(record["model"])
b.check_artifact(reference)
for case in cases:
p3.validate_image(b.check_artifact(case["image"]))
for tool in tools.values():
b.check_artifact(tool["executable"])
record["status"] = "passed"
except (OSError, ValueError) as error:
record.update(status="failed", error=str(error))
raise
finally:
b.dump(directory / "record.json", record)
print(f"P5 {args.simulator}: one execution per image; {directory}")
def validate_replay(path: Path, simulator: str) -> dict:
record = b.read_json(path)
if record.get("phase") != PHASE or record.get("status") != "passed" or record.get("repetitions") != 1:
raise ValueError("not a completed single-execution P5 replay")
for field in ("model", "reference"):
b.check_artifact(record[field])
for value in record["compilation"].values():
b.check_artifact(value)
for field in ("inputs", "snapshot", "patch"):
b.check_artifact(record["source"][field])
captured = b.read_json(Path(record["source"]["inputs"]["path"]))
if b.digest_map(captured) != record["source"]["input_digest"]:
raise ValueError("candidate source digest mismatch")
for tool in record["tools"].values():
b.check_artifact(tool["executable"])
b.check_artifact(tool["archive"])
prefix = [record["model"]["path"]]
if simulator == "iverilog":
runtime = b.check_artifact(record["runtime_tool"]["executable"])
compiler = Path(record["tools"]["iverilog"]["executable"]["path"])
if runtime.resolve().parent != compiler.resolve().parent:
raise ValueError("unmatched Icarus runtime")
prefix.insert(0, str(runtime))
reference = reference_cases(Path(record["reference"]["path"]), [c["name"] for c in record["cases"]])
for case, original in zip(record["cases"], reference, strict=True):
if case["image"] != original["image"] or case["baseline_samples"] != original["samples"]:
raise ValueError("comparison changed the retained image or reference measurements")
image = p3.validate_image(b.check_artifact(case["image"]))
compatible_hardware(image["configuration"], record["configuration"])
first, end = image["layout"]["descriptor_range"]
expected = [*prefix, "+firmware=" + image["artifacts"]["tiny_p3.hex"]["path"],
"+tiny_r2_p3", "+tiny_r2_p5", f"+baseline_tcd_base={first:x}", f"+baseline_tcd_end={end:x}",
"+max_cycles=100000000"]
flow = b.read_json(b.check_artifact(case["flow"]))
if case["command"] != expected or case["timeout_seconds"] <= 0 or flow["command"] != [
"timeout", "--foreground", f"{case['timeout_seconds']}s", *expected]:
raise ValueError("model/image execution provenance mismatch")
if parse_log(b.check_artifact(case["log"]).read_text(), flow) != case["samples"]:
raise ValueError("reported candidate samples differ from the log")
return record
def report(args) -> None:
base = args.variant_root / "meta/tiny-r2-p5"
runs = {}
for simulator in ("verilator", "iverilog"):
pointer = b.read_json(base / "replay" / simulator / "latest.json")
path = Path(pointer["path"])
runs[simulator] = (validate_replay(path, simulator), b.artifact(path))
before = b.digest_map(inputs(args.root, args.variant_root))
if any(run[0]["source_digest"] != before for run in runs.values()):
raise ValueError("candidate source changed after measurement")
cases = {c["name"]: c for c in runs["verilator"][0]["cases"]}
if set(cases) != {"perf-o2-plain-flat", "perf-o2-plain-banked"}:
raise ValueError("P5 comparison requires both approved placement images")
cross = runs["iverilog"][0]["cases"]
if len(cross) != 1 or cross[0]["name"] != "perf-o2-plain-banked":
raise ValueError("P5 requires one banked Icarus execution")
if any(cross[0][key] != cases["perf-o2-plain-banked"][key] for key in ("image", "samples")):
raise ValueError("cross-simulator image/results differ")
comparisons = []
for name, case in cases.items():
rows = []
for old, new in zip(case["baseline_samples"]["cases"], case["samples"]["cases"], strict=True):
rows.append({"name": new["name"], "baseline_cycles": old["cycles"],
"candidate_cycles": new["cycles"], "baseline_instructions": old["instructions"],
"candidate_instructions": new["instructions"],
"cycle_ratio": old["cycles"] / new["cycles"]})
comparisons.append({"name": name, "image": case["image"], "workloads": rows,
"local": case["samples"]["local"], "banks": case["samples"]["banks"],
"fabric": case["samples"]["fabric"]})
b.dump(base / "report.json", {"phase": PHASE, "status": "measurement_complete",
"reference_phase": previous.PHASE,
"source_digest": before, "comparisons": comparisons,
"executions": {key: value[1] for key, value in runs.items()},
"qualification": "matched-binary SAFE24 simulation; not phase/physical acceptance"})
print(f"P5 matched-image report: {base / 'report.json'}")
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, required=True)
parser.add_argument("--variant-root", type=Path, required=True)
parser.add_argument("--report", action="store_true")
parser.add_argument("--reference", type=Path)
parser.add_argument("--cases")
parser.add_argument("--simulator", choices=("verilator", "iverilog"))
parser.add_argument("--vvp", default="vvp")
parser.add_argument("--timeout", type=int, default=21600)
parser.add_argument("command", nargs=argparse.REMAINDER)
args = parser.parse_args()
if args.timeout <= 0:
parser.error("timeout must be positive")
if args.report:
report(args)
else:
if args.reference is None or not args.cases or not args.simulator:
parser.error("replay requires reference, cases and simulator")
replay(args)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+386
View File
@@ -0,0 +1,386 @@
#!/usr/bin/env python3
"""Capture and check source-bound Tiny P6 acceptance without promoting a phase."""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
from pathlib import Path
import re
import sys
import tarfile
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from scripts import tiny_r2_baseline as b # noqa: E402
from scripts import tiny_r2_p5 as previous # noqa: E402
from scripts.check_simulation import DEFAULT_FAILURE # noqa: E402
PHASE = "TINY-R2-P6"
MODEL = "tests/rtl/sdio_native_model.sv"
VERIFICATION_UPDATE_PATHS = {MODEL, "scripts/tiny_r2_p6.py", "tests/test_tiny_r2_p6.py"}
CONSUMER_UPDATE_PATHS = {
"publications/programming_reference.py", "tests/test_publication_programming.py",
"physical/commercial/config/products/tiny.mk", "physical/commercial/tcl/common/products/tiny.tcl",
"tests/test_commercial_flow.py", "scripts/tiny_r2_p6.py", "tests/test_tiny_r2_p6.py",
}
COLLECTION_UPDATE_PATHS = {"scripts/tiny_r2_p6.py", "tests/test_tiny_r2_p6.py"}
RECORDS = {
"UART": {"pads": 1, "echo_bytes": 4, "flow_rejected": 1},
"DMA": {"channels": 8, "rejected_requests": 5},
"RNG": {"unqualified": 1, "diagnostic": 1, "irq": 1},
"CRC": {"dma": 1, "tails": 4, "empty": 1},
"CRYPTO": {"init": 1, "aes_sha": 1, "dma45": 1, "rsa_public": 1, "zeroize": 1},
"RECOVERY": {"audio_stops": 2, "ws_underflow": 1, "ws_abort": 1},
"SDIO": {"dma_bytes": 2048, "pio_bytes": 512, "width1": 1, "width4": 1, "irq": 1, "errors": 2},
}
def parse_log(content: str, flow: dict) -> dict:
if (flow.get("status") != "passed" or flow.get("exit_code") != 0
or not re.search(r"SIM_TEST_PASS Tiny cycles=[0-9]+(?:\r?\n|$)", content)
or DEFAULT_FAILURE.search(content)):
raise ValueError("P6 requires successful command, terminal marker and no forbidden errors")
result = {}
for name, expected in RECORDS.items():
rows = [b.parse_fields(line, set(expected)) for line in content.splitlines()
if line.startswith("R2_P6_" + name + " ")]
if rows != [expected]:
raise ValueError(f"missing, duplicate or invalid P6 {name} record")
result[name.lower()] = rows[0]
fields = {"bits", "frames", "ws_frames", "background_jobs", "tcd", "elapsed_us", "cycles", "instructions"}
audio = [b.parse_fields(line, fields) for line in content.splitlines()
if line.startswith("R2_P6_AUDIO ")]
if [row["bits"] for row in audio] != [16, 24]:
raise ValueError("P6 needs exactly both audio formats")
for row in audio:
if (row["frames"] != 256 or row["ws_frames"] != 2 or
not 1 <= row["background_jobs"] <= 64 or
row["tcd"] != int(row["bits"] == 24) or not 0 < row["elapsed_us"] < 50000 or
not 0 < row["instructions"] <= row["cycles"]):
raise ValueError("invalid P6 audio/contention observation")
result["audio"] = audio
wires = [b.parse_fields(line, {"bits", "samples"}) for line in content.splitlines()
if line.startswith("R2_P6_I2S_WIRE ")]
if wires != [{"bits": 16, "samples": 512}, {"bits": 24, "samples": 512}]:
raise ValueError("missing or incorrect independent audio wire observations")
result["audio_wires"] = wires
fields = {"owner", "read_bytes", "write_bytes", "descriptor_bytes", "ar_stall_max",
"aw_stall_max", "w_stall_max", "r_stall_max", "b_stall_max"}
bus = sorted([b.parse_fields(line, fields) for line in content.splitlines()
if line.startswith("R2_P6_AXI ")], key=lambda row: row["owner"])
if [row["owner"] for row in bus] != [0, 1, 2]:
raise ValueError("P6 needs each actual AXI owner observation")
for row in bus:
if (any(v < 0 for v in row.values()) or row["read_bytes"] == 0 or row["write_bytes"] == 0
or row["descriptor_bytes"] > row["read_bytes"]):
raise ValueError("invalid P6 AXI payload/backpressure observation")
if bus[1]["descriptor_bytes"] == 0:
raise ValueError("no actual background descriptor fetch observed")
result["bus"] = bus
contention = [b.parse_fields(line, {"sdio_dma_cycles", "cpu_fetches"})
for line in content.splitlines() if line.startswith("R2_P6_CONTENTION ")]
if len(contention) != 1 or any(value <= 0 for value in contention[0].values()):
raise ValueError("missing actual SDIO/DMA overlap and CPU progress")
result["contention"] = contention[0]
fields = {"ws_frames", "ws_wait_max", "fifo_max", "underflows", "aborts", "clock_stops"}
devices = [b.parse_fields(line, fields) for line in content.splitlines()
if line.startswith("R2_P6_DEVICES ")]
if len(devices) != 1:
raise ValueError("missing or duplicate independent device observations")
d = devices[0]
if (d["ws_frames"] != 4 or not 0 <= d["ws_wait_max"] <= 4800 or
not 1 <= d["fifo_max"] <= 16 or d["underflows"] != 1 or
d["aborts"] != 1 or d["clock_stops"] != 2):
raise ValueError("P6 device coverage/service budget failed")
result["devices"] = d
return result
def inputs(variants: list[Path]) -> dict:
result = {}
for variant in variants:
config = b.configuration(variant)["configuration"]
if config.get("APP") != "ci_smoke":
raise ValueError("P6 qualification requires the ci_smoke composition")
result.update(previous.inputs(ROOT, variant))
for name in (".bake.toml", ".clang-format", "docs/misra-c-2012.md",
"quality/misra/deviations.md", "docs/ip/dma.md", "docs/ip/i2s.md",
"docs/ip/sdio.md", "docs/ip/crypto.md", "docs/ip/ws2812.md",
"physical/commercial/config/products/tiny.mk",
"physical/commercial/tcl/common/products/tiny.tcl"):
result[name] = b.sha256(ROOT / name)
return result
def current_tools() -> dict:
tools = b.tools_identity(ROOT)
tools["vvp"] = b.vvp_identity("vvp", tools["iverilog"])
return tools
def checked_flow(path: Path, captured_at: str) -> dict:
flow = b.read_json(path)
if flow.get("status") != "passed" or flow.get("exit_code") != 0:
raise ValueError(f"unsuccessful gate: {path}")
if (not flow.get("started_at") or not flow.get("finished_at")
or flow["started_at"] < captured_at or flow["finished_at"] < flow["started_at"]):
raise ValueError(f"gate lacks successful post-capture chronology: {path}")
return flow
def check_model_declaration_move(before: str, after: str) -> None:
"""Permit only relocation of one unchanged continuous net declaration.
Verilator/sv2v accept its forward reference; native Icarus needs the integer
first. Every other source byte, including all procedural behavior, must match.
"""
pattern = re.compile(r"^ wire \[3:0\] s_token_lanes = [^;]+;\n", re.M)
old, new = pattern.findall(before), pattern.findall(after)
if (len(old) != 1 or old != new or pattern.sub("", before) != pattern.sub("", after)
or after.index(" integer s_data_bytes;") > after.index(new[0])):
raise ValueError("not the permitted SDIO model declaration-only relocation")
def snapshot_model(source: dict) -> str:
with tarfile.open(b.check_artifact(source["snapshot"])) as archive:
return archive.extractfile(MODEL).read().decode()
def check_update(directory: Path, saved: dict, update: dict, current: dict) -> dict:
if (update.get("phase") != PHASE or update.get("kind") != "native_model_declaration_order"
or update["original_source"] != b.artifact(directory / "source.json")):
raise ValueError("verification update does not bind the original campaign")
old = b.read_json(b.check_artifact(saved["source"]["inputs"]))
new = b.read_json(b.check_artifact(update["source"]["inputs"]))
changed = {name for name in old.keys() | new.keys() if old.get(name) != new.get(name)}
if (MODEL not in changed or not changed <= VERIFICATION_UPDATE_PATHS
or sorted(changed) != update["changed_inputs"] or new != current):
raise ValueError("verification update changed DUT/firmware/constraints or unapproved inputs")
check_model_declaration_move(snapshot_model(saved["source"]), snapshot_model(update["source"]))
for source in (saved["source"], update["source"]):
for name in ("inputs", "snapshot", "patch"):
b.check_artifact(source[name])
if update["tools"] != saved["tools"] or update["captured_at"] < saved["captured_at"]:
raise ValueError("verification update changed tools or chronology")
return old
def capture_verification_update(directory: Path, variants: list[Path]) -> None:
path = directory / "verification-update.json"
if path.exists():
raise ValueError("never overwrite a verification update")
saved = b.read_json(directory / "source.json")
current = inputs(variants)
old = b.read_json(b.check_artifact(saved["source"]["inputs"]))
changed = sorted(name for name in old.keys() | current.keys() if old.get(name) != current.get(name))
if MODEL not in changed or not set(changed) <= VERIFICATION_UPDATE_PATHS:
raise ValueError("only the native model and its collector tests may change")
check_model_declaration_move(snapshot_model(saved["source"]), (ROOT / MODEL).read_text())
update = {"phase": PHASE, "kind": "native_model_declaration_order",
"original_source": b.artifact(directory / "source.json"), "changed_inputs": changed,
"source": b.snapshot(ROOT, directory / "inputs-native-model", current),
"tools": current_tools(), "captured_at": datetime.now(timezone.utc).isoformat()}
check_update(directory, saved, update, current)
b.dump(path, update)
def check_consumer_update(directory: Path, previous_update: dict, update: dict, current: dict) -> None:
"""Permit only consumers not used by the running RTL/smoke flows."""
if (update.get("phase") != PHASE or update.get("kind") != "publication_and_commercial_consumers"
or update["previous_update"] != b.artifact(directory / "verification-update.json")):
raise ValueError("consumer update does not bind the verified model snapshot")
old = b.read_json(b.check_artifact(previous_update["source"]["inputs"]))
new = b.read_json(b.check_artifact(update["source"]["inputs"]))
changed = {name for name in old.keys() | new.keys() if old.get(name) != new.get(name)}
if (not changed or not changed <= CONSUMER_UPDATE_PATHS or
sorted(changed) != update["changed_inputs"] or new != current):
raise ValueError("consumer update changed simulation, firmware, smoke constraints or unrelated inputs")
for field in ("inputs", "snapshot", "patch"):
b.check_artifact(update["source"][field])
if (update["tools"] != previous_update["tools"] or
update["captured_at"] < previous_update["captured_at"]):
raise ValueError("consumer update changed tools or chronology")
def capture_consumer_update(directory: Path, variants: list[Path]) -> None:
path = directory / "consumer-update.json"
if path.exists():
raise ValueError("never overwrite a consumer update")
previous_update = b.read_json(directory / "verification-update.json")
current = inputs(variants)
old = b.read_json(b.check_artifact(previous_update["source"]["inputs"]))
changed = sorted(name for name in old.keys() | current.keys() if old.get(name) != current.get(name))
if not changed or not set(changed) <= CONSUMER_UPDATE_PATHS:
raise ValueError("unexpected non-consumer source changes")
update = {"phase": PHASE, "kind": "publication_and_commercial_consumers",
"previous_update": b.artifact(directory / "verification-update.json"),
"changed_inputs": changed,
"source": b.snapshot(ROOT, directory / "inputs-consumers", current),
"tools": current_tools(), "captured_at": datetime.now(timezone.utc).isoformat()}
check_consumer_update(directory, previous_update, update, current)
b.dump(path, update)
def check_collection_update(directory: Path, previous_update: dict, update: dict, current: dict) -> None:
if (update.get("phase") != PHASE or update.get("kind") != "terminal_log_collection"
or update["previous_update"] != b.artifact(directory / "consumer-update.json")):
raise ValueError("collection update does not bind the consumer snapshot")
old = b.read_json(b.check_artifact(previous_update["source"]["inputs"]))
new = b.read_json(b.check_artifact(update["source"]["inputs"]))
changed = {name for name in old.keys() | new.keys() if old.get(name) != new.get(name)}
if (not changed or not changed <= COLLECTION_UPDATE_PATHS or
sorted(changed) != update["changed_inputs"] or new != current):
raise ValueError("collection update changed inputs outside the collector/tests")
for field in ("inputs", "snapshot", "patch"):
b.check_artifact(update["source"][field])
if (update["tools"] != previous_update["tools"] or
update["captured_at"] < previous_update["captured_at"]):
raise ValueError("collection update changed tools or chronology")
def capture_collection_update(directory: Path, variants: list[Path]) -> None:
path = directory / "collection-update.json"
if path.exists():
raise ValueError("never overwrite a collection update")
previous_update = b.read_json(directory / "consumer-update.json")
current = inputs(variants)
old = b.read_json(b.check_artifact(previous_update["source"]["inputs"]))
changed = sorted(name for name in old.keys() | current.keys() if old.get(name) != current.get(name))
if not changed or not set(changed) <= COLLECTION_UPDATE_PATHS:
raise ValueError("unexpected non-collector source changes")
update = {"phase": PHASE, "kind": "terminal_log_collection",
"previous_update": b.artifact(directory / "consumer-update.json"),
"changed_inputs": changed,
"source": b.snapshot(ROOT, directory / "inputs-collection", current),
"tools": current_tools(), "captured_at": datetime.now(timezone.utc).isoformat()}
check_collection_update(directory, previous_update, update, current)
b.dump(path, update)
def capture(directory: Path, variants: list[Path]) -> None:
if (directory / "source.json").exists():
raise ValueError("capture needs a fresh evidence directory; never overwrite provenance")
if {b.configuration(v)["configuration"]["PDK"] for v in variants} != {"ICS55", "IHP130"}:
raise ValueError("capture both Tiny PDK input sets before validation")
source = b.snapshot(ROOT, directory / "inputs", inputs(variants))
tools = current_tools()
b.dump(directory / "source.json", {"phase": PHASE, "source": source, "tools": tools,
"captured_at": datetime.now(timezone.utc).isoformat(),
"variants": [str(p.resolve()) for p in variants],
"configurations": [b.configuration(p) for p in variants]})
def report(directory: Path, gates: list[Path], variants: list[Path]) -> None:
saved = b.read_json(directory / "source.json")
if saved.get("phase") != PHASE:
raise ValueError("wrong evidence phase")
if saved["tools"] != current_tools():
raise ValueError("locked tool identity changed after capture")
selected = [Path(p) for p in saved["variants"]]
current = inputs(selected)
original = b.read_json(b.check_artifact(saved["source"]["inputs"]))
update_path = directory / "verification-update.json"
update = b.read_json(update_path) if update_path.exists() else None
consumer_path = directory / "consumer-update.json"
consumer = b.read_json(consumer_path) if consumer_path.exists() else None
collection_path = directory / "collection-update.json"
collection = b.read_json(collection_path) if collection_path.exists() else None
consumer_inputs = current
if collection:
if not consumer:
raise ValueError("collection update requires the consumer snapshot")
check_collection_update(directory, consumer, collection, current)
consumer_inputs = b.read_json(b.check_artifact(consumer["source"]["inputs"]))
verified_inputs = current
if consumer:
if not update:
raise ValueError("consumer update requires the model snapshot")
check_consumer_update(directory, update, consumer, consumer_inputs)
verified_inputs = b.read_json(b.check_artifact(update["source"]["inputs"]))
if update:
original = check_update(directory, saved, update, verified_inputs)
elif current != original:
raise ValueError("P6 technical inputs changed after capture")
for name in ("snapshot", "patch"):
b.check_artifact(saved["source"][name])
checked = {}
for path in gates:
gate = checked_flow(path, saved["captured_at"])
checked[str(path)] = {"result": b.artifact(path), "log": b.artifact(Path(gate["log"]))}
simulations = {}
selections = [(variant, simulator) for variant in variants for simulator in ("iverilog", "verilator")
if (variant / "sim" / simulator / "result-sim.json").is_file()]
for variant, simulator in selections:
config = b.configuration(variant)
if config["configuration"].get("APP") != "ci_smoke":
raise ValueError("simulation must use the complete ci_smoke composition")
pdk = config["configuration"]["PDK"]
key = pdk + "/" + simulator
if key in simulations or simulator not in ("iverilog", "verilator"):
raise ValueError("duplicate or unsupported simulation variant")
folder = variant / "sim" / simulator
model_source = update if update and simulator == "iverilog" else saved
flow = checked_flow(folder / "result-sim.json", model_source["captured_at"])
compilation = checked_flow(folder / "result-compile.json", model_source["captured_at"])
if flow["started_at"] < compilation["finished_at"]:
raise ValueError("simulation predates completed model compilation")
check = b.read_json(folder / "result-sim-check.json")
if check.get("status") != "passed":
raise ValueError("missing strict simulation verdict")
samples = parse_log((folder / "sim.log").read_text(), flow)
paths = [folder / name for name in ("result-sim.json", "result-sim-check.json", "sim.log",
"result-compile.json", "compile.log")]
paths += [folder / ("emu" if simulator == "verilator" else "simv"),
variant / "sw/firmware", variant / "sw/retrosoc_fw.hex", variant / "sw/retrosoc_fw.map",
variant / "sw/retrosoc_fw.bin", variant / "sw/ld2_all_sram.lds",
variant / "sw/memory_regions.ld",
variant / "meta/manifest.json"]
paths += sorted(path for path in (variant / "sw/include").rglob("*") if path.is_file())
runtime = b.runtime_inputs(ROOT, variant, simulator)
if update and simulator == "verilator":
if runtime.get(MODEL) != current[MODEL]:
raise ValueError("missing native card model runtime input")
runtime[MODEL] = original[MODEL]
simulations[key] = {"configuration": config, "simulator": simulator, "samples": samples,
"source": model_source["source"], "runtime_inputs": runtime,
"artifacts": [b.artifact(path) for path in paths]}
if set(simulations) != {p + "/" + s for p in ("ICS55", "IHP130") for s in ("iverilog", "verilator")}:
raise ValueError("P6 needs both simulators on both Tiny PDKs")
# Physical/quality reports remain independent gates; this status never
# promotes functional collection into phase or physical qualification.
b.dump(directory / "report.json", {"phase": PHASE, "status": "functional_evidence_collected",
"tools": saved["tools"],
"verification_update": b.artifact(update_path) if update else None,
"consumer_update": b.artifact(consumer_path) if consumer else None,
"collection_update": b.artifact(collection_path) if collection else None,
"source": saved["source"], "source_inputs_rechecked": len(current), "gates": checked,
"simulations": simulations, "limitations": ["human review pending", "timing observational",
"no QFN64/package, routed/PVT/CDC/RDC or publication qualification"]})
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("action", choices=("capture", "capture-verification-update", "capture-consumer-update",
"capture-collection-update", "report"))
parser.add_argument("--evidence-root", type=Path, required=True)
parser.add_argument("--variant-root", action="append", type=Path, required=True)
parser.add_argument("--gate", action="append", type=Path, default=[])
args = parser.parse_args()
directory = args.evidence_root.resolve()
if not directory.is_relative_to(ROOT / "build"):
parser.error("evidence must stay under the repository build root")
if args.action == "capture":
capture(directory, args.variant_root)
elif args.action == "capture-verification-update":
capture_verification_update(directory, args.variant_root)
elif args.action == "capture-consumer-update":
capture_consumer_update(directory, args.variant_root)
elif args.action == "capture-collection-update":
capture_collection_update(directory, args.variant_root)
else:
report(directory, args.gate, args.variant_root)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+187
View File
@@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""P7 structural clock/macro evidence; separate from the frozen P1 audit."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from scripts import tiny_r2_baseline as b # noqa: E402
from scripts.config_key import variant_id # noqa: E402
from scripts.tiny_ics55_platform import expand_netlist, inspect_clock_source # noqa: E402
PHASE = "TINY-R2-P7"
RAM_TYPES = {"ICS55": ("ics55_ecos_sram_1024x32_m8", "CLK"),
"IHP130": ("RM_IHPSG13_1P_1024x32_c2_bm_bist", "A_CLK")}
def inspect_netlist(document: dict, pdk: str) -> dict:
modules, top, leaves, canonical, boundaries = expand_netlist(document)
def boundary(suffix: str, kind: str) -> dict:
found = [cell for name, cell in boundaries.items()
if name.replace("/", ".").endswith("." + suffix)
and cell["type"].split("$")[0] == kind]
if len(found) != 1:
raise ValueError(f"missing or ambiguous P7 clock boundary: {suffix}")
return found[0]
def pin(cell: dict, name: str) -> str:
bits = cell["ports"].get(name, [])
if len(bits) != 1 or bits[0] in ("0", "1", "x", "z"):
raise ValueError(f"missing/constant P7 clock: {cell['path']}/{name}")
return bits[0]
root = boundary("u_soc.u_clock_backend.u_sys_mux", "safe_clock_mux")
memory = boundary("u_soc.u_clock_backend.u_mem_mux", "safe_clock_mux")
peripheral = boundary("u_soc.u_clock_backend.u_pclk_mux", "safe_clock_mux")
cpu_gate = boundary("u_soc.u_cpu.u_cpu_gate", "tiny_clock_gate")
crypto_gate = boundary("u_soc.u_crypto_lifecycle", "tiny_peripheral_leaf")
sys_clock = pin(root, "clk_o")
pclk = pin(peripheral, "clk_o")
if pin(cpu_gate, "clk_i") != sys_clock or pin(memory, "clk0_i") != sys_clock:
raise ValueError("CPU gate/MEM source is not common SYS")
if pin(crypto_gate, "clk_i") != pclk:
raise ValueError("private Crypto storage gate is not sourced from PCLK")
cpu_clock = pin(cpu_gate, "clk_o")
crypto_clock = pin(crypto_gate, "clk_o")
ram_type, ram_pin = RAM_TYPES[pdk]
macros = [cell for cell in leaves if cell["type"] == ram_type]
main_names = {f"u_soc.u_sram.gen_group[{g}].u_group.gen_bank[{n}].u_ram.u_mem"
for g in range(4) for n in range(8)}
crypto_names = {f"u_soc.u_crypto.u_storage.gen_bank[{n}].gen_macro.u_sram.u_mem"
for n in range(6)}
observed = set()
for cell in macros:
name = next((name for name in main_names | crypto_names
if cell["path"].replace("/", ".").endswith("." + name)), None)
if name is None or name in observed:
raise ValueError("extra, duplicated or renamed physical SRAM macro")
observed.add(name)
expected_clock = sys_clock if name in main_names else crypto_clock
if pin(cell, ram_pin) != expected_clock:
raise ValueError(f"physical SRAM clock mismatch: {name}")
if len(macros) != 38 or observed != main_names | crypto_names:
raise ValueError("P7 requires 32 main SRAM plus six private Crypto macros")
cpu = [cell for cell in leaves if "u_hazard3_cpu_2port" in cell["path"]
and ("CK" in cell["ports"] or "CKN" in cell["ports"] or "CLK" in cell["ports"])]
if not cpu:
raise ValueError("missing CPU clock endpoints")
for cell in cpu:
for name in ("CK", "CKN", "CLK"):
if name in cell["ports"] and pin(cell, name) not in (sys_clock, cpu_clock):
raise ValueError("CPU has an unapproved clock or a hidden divider")
pll = [cell for cell in leaves if cell["type"] == "PLL_TOP"]
if pdk == "ICS55":
if len(pll) != 1 or pin(pll[0], "CKOUT1") != pin(root, "clk1_i"):
raise ValueError("P7 requires one PLL CKOUT1 connected to the SYS mux")
if pin(pll[0], "REFCLK") != pin(root, "clk0_i"):
raise ValueError("PLL reference and safe SYS source differ")
constants = {"0": 0, "1": 1}
for cell in leaves:
if cell["type"] in ("TIELOH7R", "TIEHIH7R"):
for bit in cell["ports"]["Z"]:
constants[bit] = int(cell["type"] == "TIEHIH7R")
if pll[0]["ports"].get("EN", ["x"])[0] in constants:
raise ValueError("P7 PLL enable must be controlled, not tied on/off")
external = modules[top]["ports"]["extclk_i_pad"]["bits"]
if len(external) != 1:
raise ValueError("missing external reference input")
inspect_clock_source(leaves, pin(root, "clk0_i"),
canonical(f"{top}:{external[0]}"), constants)
elif pll:
raise ValueError("no-PLL compatibility instantiated a PLL")
return {"system_clock": sys_clock, "memory_clock": pin(memory, "clk_o"),
"peripheral_clock": pclk, "cpu_clock": cpu_clock,
"main_macros": 32, "private_crypto_macros": 6, "pll_macros": len(pll),
"cpu_clocked_cells": len(cpu), "macro_bindings": macros,
"clock_boundaries": [root, memory, peripheral, cpu_gate, crypto_gate],
"qualification": "structural domain binding only; no physical operating-point qualification"}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("action", choices=("audit-netlist", "verify-netlist", "capture-source", "verify-source"))
parser.add_argument("--variant-root", type=Path, required=True)
parser.add_argument("--netlist", type=Path)
parser.add_argument("--config", type=Path)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
try:
if args.action in ("capture-source", "verify-source"):
inputs = b.source_inputs(ROOT, "ICS55") | b.source_inputs(ROOT, "IHP130")
for folder in ("tests", "configs/ci"):
for path in (ROOT / folder).rglob("*"):
if path.is_file() and path.suffix in (".py", ".sv", ".svh", ".c", ".S", ".ld", ".mk"):
inputs[str(path.relative_to(ROOT))] = b.sha256(path)
contract = b.artifact(ROOT / "docs/ip/tiny-soc.md")
if args.action == "capture-source":
if args.output.exists():
raise ValueError("source evidence already exists; use a new attempt instead of overwriting")
b.dump(args.output, {"phase": PHASE, "repository": b.git_info(ROOT),
"specification": contract, "source_inputs": inputs,
"source_digest": b.digest_map(inputs), "tools": b.tools_identity(ROOT),
"snapshot": b.snapshot(ROOT, args.output.parent / "source", inputs)})
else:
recorded = b.read_json(args.output)
if (recorded.get("phase") != PHASE or recorded.get("source_inputs") != inputs or
recorded.get("specification") != contract):
raise ValueError("P7 sources changed after the retained source snapshot")
print(f"{PHASE} {args.action}: PASS")
return 0
if args.netlist is None or args.config is None:
raise ValueError("netlist actions require --netlist and --config")
manifest = b.configuration(args.variant_root)
pdk = manifest["configuration"]["PDK"]
config = dict(line.split("=", 1) for line in args.config.read_text().splitlines() if "=" in line)
for key, value in {"SOC": "TINY", "PDK": pdk,
"HAVE_PLL": "YES" if pdk == "ICS55" else "NO",
"HAVE_SRAM_MACRO": "YES", "SRAM_SIZE_KIB": "128",
"TOP_DESIGN": "retrosoc_tiny_asic", "PERIOD_PS": "41667"}.items():
if config.get(key) != value:
raise ValueError(f"P7 consumed configuration mismatch: {key}")
values = [f"{k}={v}" for k, v in manifest["configuration"].items()
if k not in ("SIMU", "SYNTH", "SYNTH_RECIPE", "STA")]
expected = variant_id(manifest["profile"], values,
ROOT / "dependencies/dependencies.lock.json",
"2000-01-01-00-00").rsplit("-", 1)[1]
if config.get("CONFIG_DIGEST") != expected or args.variant_root.name.rsplit("-", 1)[1] != expected:
raise ValueError("P7 consumed configuration digest mismatch")
paths = (args.netlist, Path(str(args.netlist) + ".json"), args.config)
if args.action == "audit-netlist" and not paths[1].is_file():
# A mapped Verilog export is also a valid audit input. Serialize it
# without optimization or remapping; retain the conversion result.
# This supports the pre-P7 IHP130 recipe which did not emit JSON.
flow = b.read_json(args.variant_root / "syn/yosys/result-synth.json")
if flow.get("status") != "passed" or flow.get("exit_code") != 0:
raise ValueError("cannot derive audit JSON from an unsuccessful synthesis")
b.execute(ROOT, args.output.parent, "p7-netlist-json",
["timeout", "--foreground", "120s", "yosys", "-Q", "-T", "-p",
f'read_verilog "{paths[0]}"; write_json "{paths[1]}"'], quiet=True)
inputs = [b.artifact(path) for path in paths]
auditors = [b.artifact(ROOT / "scripts" / name) for name in
("tiny_r2_p7.py", "tiny_ics55_platform.py", "tiny_r2_baseline.py", "config_key.py")]
if args.action == "audit-netlist":
result = {"phase": PHASE, "status": "passed", "pdk": pdk,
"audit_scripts": auditors, "inputs": inputs,
"configuration": manifest["configuration"], "config_digest": expected,
"binding": inspect_netlist(json.loads(paths[1].read_text()), pdk)}
b.dump(args.output, result)
else:
result = b.read_json(args.output)
if (result.get("status") != "passed" or result.get("phase") != PHASE or
result.get("audit_scripts") != auditors or result.get("config_digest") != expected or
result.get("inputs") != inputs):
raise ValueError("P7 structural evidence is missing, stale, or not passed")
except (OSError, ValueError, KeyError) as error:
print(f"{PHASE}: {error}", file=sys.stderr)
return 1
print(f"{PHASE} {args.action}: PASS (structural evidence only)")
return 0
if __name__ == "__main__":
raise SystemExit(main())