mirror of
https://github.com/retroSoC/retroSoC.git
synced 2026-11-04 22:32:36 +00:00
feat: add comprehensive tests for APU-P7 and APU-P8 functionality
- Introduced differential tests for the APU-P7 frontend profile-1 BAM to ensure consistency with ROM and generator outputs. - Implemented concurrent decode and continuous KWS smoke tests for APU-P7 to validate system stability under load. - Added accuracy qualification tests for APU-P7 to verify through-hardware performance against expected results. - Created contention tests for APU-P8 to assess USB2/SDIO0/APU Gateway A interactions. - Developed quiesced loader admission tests for APU-P8 to ensure proper handling of DMA operations. - Added post-fault visibility tests for APU-P8 HP interconnect to confirm fault handling capabilities. - Enhanced register parity tests to include ABI digest verification for consistency across implementations. - Updated publication register tests to reflect changes in capability constants for P5 and P7 profiles.
This commit is contained in:
50 files changed
+9154
-45
No files matched your search
@@ -0,0 +1,534 @@
|
||||
"""APU V1 ABI digest tool: canonical table inventory, serialization, and CRC-32.
|
||||
|
||||
Per docs/ip/apu.md (ABI_DIGEST at offset 0x054 and the Phase8 completion
|
||||
contract) the digest is the nonzero CRC-32 over the complete frozen V1 tables:
|
||||
|
||||
- the APB V1 register/field table and the 128-byte job-descriptor layout
|
||||
(canonical source rtl/ip/multimedia/apu_define.svh, mirrored by
|
||||
crt/include/retrosoc/hal/apu_regs.h; both sides are re-verified equal here),
|
||||
- the hardware discovery values (IP_ID, IP_VERSION, CAPABILITY0/1, IRQ mask)
|
||||
reported by the current release (P7-qualified configuration),
|
||||
- the format/output-mode/PCM-format/operation identifier tables
|
||||
(crt/include/retrosoc/hal/apu.h enums; operation IDs 0/1 are the frozen V1
|
||||
contract values of docs/ip/apu.md "Descriptor ABI" and have no named code
|
||||
constant),
|
||||
- the supported APUMC V1/V2 container constants (scripts/apu_isa.py),
|
||||
- the APUM model container constants (scripts/apu_kws.py),
|
||||
- the microcode class/opcode/predicate/wait-source/primitive/trap tables, the
|
||||
instruction field positions, the entry word layout, and the P3/P4/P5
|
||||
primitive masks (scripts/apu_isa.py abi_manifest()),
|
||||
- the reserved MP3 identifiers: format ID 1, capability bit 1, the three-entry
|
||||
APUMC layout whose entry 1 is the unsupported-MP3 trap stub
|
||||
(rtl/ip/multimedia/apu_p5_codecs.apus), the trap instruction/detail, and the
|
||||
P5 rejection tuple (code 3/stage 2/detail 0x01000001).
|
||||
|
||||
The deferred P6 design archive is documentation-only and contributes nothing;
|
||||
no P6 constant exists in any parsed source.
|
||||
|
||||
Serialization (the release identity; keep stable): canonical ASCII JSON with
|
||||
recursively sorted keys, separators `,` and `:`, no insignificant whitespace,
|
||||
no terminal newline, and every integer encoded as a JSON decimal number. The
|
||||
digest is the reflected CRC-32/ISO-HDLC (binascii.crc32, the same variant as
|
||||
apu_isa.crc32_iso_hdlc) over that complete byte stream.
|
||||
|
||||
CLI: default prints the digest and section summary; --check compares the
|
||||
computed digest against the handwritten rtl/ip/multimedia/apu_reg.sv AbiDigest
|
||||
localparam and the crt/include/retrosoc/hal/apu_regs.h RS_APU_DIGEST_IMPLEMENTED
|
||||
define, requiring all three to match and be nonzero; --json PATH dumps the
|
||||
canonical inventory for review.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import binascii
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
import apu_kws # noqa: E402
|
||||
from apu_isa import ( # noqa: E402
|
||||
APUMC_ABI_V2,
|
||||
ControlOpcode,
|
||||
Entry,
|
||||
Instruction,
|
||||
InstructionClass,
|
||||
abi_manifest,
|
||||
)
|
||||
|
||||
RTL_DEFINE = ROOT / "rtl/ip/multimedia/apu_define.svh"
|
||||
RTL_REGISTER = ROOT / "rtl/ip/multimedia/apu_reg.sv"
|
||||
RTL_MICROCODE = ROOT / "rtl/ip/multimedia/apu_p5_codecs.apus"
|
||||
C_HEADER = ROOT / "crt/include/retrosoc/hal/apu_regs.h"
|
||||
C_HAL = ROOT / "crt/include/retrosoc/hal/apu.h"
|
||||
|
||||
SECTION_ORDER = (
|
||||
"discovery",
|
||||
"apb",
|
||||
"kws_model",
|
||||
"formats",
|
||||
"apumc",
|
||||
"apum",
|
||||
"isa",
|
||||
"mp3_stub",
|
||||
)
|
||||
|
||||
|
||||
def _rtl_defines() -> dict[str, int]:
|
||||
"""Parse every `define in the canonical hardware/software ABI header."""
|
||||
|
||||
values: dict[str, int] = {}
|
||||
pattern = re.compile(
|
||||
r"^`define\s+((?:APB4_APU__|RETROSOC_APU_KWS__)\w+)\s+"
|
||||
r"((?:\d+)'[hHdDbB][0-9a-fA-F_]+|\d+)\s*$"
|
||||
)
|
||||
for line in RTL_DEFINE.read_text(encoding="utf-8").splitlines():
|
||||
match = pattern.match(line)
|
||||
if match is None:
|
||||
continue
|
||||
name, literal = match.groups()
|
||||
if "'" in literal:
|
||||
_, value = literal.split("'", maxsplit=1)
|
||||
radix = {"h": 16, "d": 10, "b": 2}[value[0].lower()]
|
||||
values[name] = int(value[1:].replace("_", ""), radix)
|
||||
else:
|
||||
values[name] = int(literal, 10)
|
||||
return values
|
||||
|
||||
|
||||
def _c_abi_defines() -> dict[str, int]:
|
||||
"""Parse the handwritten RS_APU_ABI_* mirror of the SVH define file."""
|
||||
|
||||
values: dict[str, int] = {}
|
||||
pattern = re.compile(
|
||||
r"^#define\s+RS_APU_ABI_(\w+)\s+"
|
||||
r"(?:UINT32_C\((0[xX][0-9a-fA-F]+)\)|(\d+)[uU]?)\s*$"
|
||||
)
|
||||
for line in C_HEADER.read_text(encoding="utf-8").splitlines():
|
||||
match = pattern.match(line)
|
||||
if match is None:
|
||||
continue
|
||||
name, hexadecimal, decimal = match.groups()
|
||||
values[name] = int(hexadecimal, 16) if hexadecimal is not None else int(decimal, 10)
|
||||
return values
|
||||
|
||||
|
||||
def _c_discovery_defines() -> dict[str, int]:
|
||||
"""Parse the implemented hardware discovery values from the C header."""
|
||||
|
||||
values: dict[str, int] = {}
|
||||
pattern = re.compile(
|
||||
r"^#define\s+RS_APU_(IP_ID_VALUE|IP_VERSION_VALUE|IRQ_ALL"
|
||||
r"|CAPABILITY0_IMPLEMENTED|CAPABILITY0_P7_IMPLEMENTED"
|
||||
r"|CAPABILITY1_IMPLEMENTED|CAPABILITY1_P7_IMPLEMENTED)"
|
||||
r"\s+UINT32_C\((0[xX][0-9a-fA-F]+)\)\s*$"
|
||||
)
|
||||
for line in C_HEADER.read_text(encoding="utf-8").splitlines():
|
||||
match = pattern.match(line)
|
||||
if match is not None:
|
||||
values[match.group(1)] = int(match.group(2), 16)
|
||||
values["CAPABILITY0_IMPLEMENTED"] = values["CAPABILITY0_P7_IMPLEMENTED"]
|
||||
values["CAPABILITY1_IMPLEMENTED"] = values["CAPABILITY1_P7_IMPLEMENTED"]
|
||||
return values
|
||||
|
||||
|
||||
def _hex_literal(pattern: str, text: str) -> int:
|
||||
match = re.search(pattern, text)
|
||||
if match is None:
|
||||
raise ValueError(f"pattern not found in {RTL_REGISTER.name}: {pattern}")
|
||||
return int(match.group(1).replace("_", ""), 16)
|
||||
|
||||
|
||||
def _rtl_discovery_values() -> dict[str, int]:
|
||||
"""Extract the release-selected discovery localparams from apu_reg.sv."""
|
||||
|
||||
text = RTL_REGISTER.read_text(encoding="utf-8")
|
||||
branch = r"localparam logic \[31:0\] %s = (?:\([^)]*\)|\w+) \? 32'h([0-9a-fA-F_]+) :"
|
||||
return {
|
||||
"IP_ID_VALUE": _hex_literal(
|
||||
r"localparam logic \[31:0\] IpId = 32'h([0-9a-fA-F_]+);", text
|
||||
),
|
||||
"IP_VERSION_VALUE": _hex_literal(branch % "IpVersion", text),
|
||||
"CAPABILITY0_IMPLEMENTED": _hex_literal(branch % "Capability0", text),
|
||||
"CAPABILITY1_IMPLEMENTED": _hex_literal(branch % "Capability1", text),
|
||||
"IRQ_ALL": _hex_literal(
|
||||
r"localparam logic \[10:0\] IrqMask = 11'h([0-9a-fA-F_]+);", text
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _c_enum_values(enum_name: str) -> dict[str, int]:
|
||||
"""Parse a simple `typedef enum { NAME = n, ... } name;` block from apu.h."""
|
||||
|
||||
text = C_HAL.read_text(encoding="utf-8")
|
||||
match = re.search(r"typedef enum \{([^}]*)\}\s+" + enum_name + r"\s*;", text)
|
||||
if match is None:
|
||||
raise ValueError(f"enum {enum_name} not found in {C_HAL.name}")
|
||||
return {
|
||||
name: int(value)
|
||||
for name, value in re.findall(r"(\w+)\s*=\s*(\d+)\s*,?", match.group(1))
|
||||
}
|
||||
|
||||
|
||||
def _manifest_rtl_exact() -> dict[str, int]:
|
||||
"""Map every apu_isa manifest value to its handwritten SVH define name."""
|
||||
|
||||
manifest = abi_manifest()
|
||||
exact = {
|
||||
"APUMC_MAGIC": manifest["apumc"]["magic"],
|
||||
"APUMC_ABI": manifest["apumc"]["abi"],
|
||||
"APUMC_ABI_V1": manifest["apumc"]["abi_v1"],
|
||||
"APUMC_ABI_V2": manifest["apumc"]["abi_v2"],
|
||||
"APUMC_V1": manifest["apumc"]["abi_v1"],
|
||||
"APUMC_V2": manifest["apumc"]["abi_v2"],
|
||||
"APUMC_HEADER_BYTES": manifest["apumc"]["header_bytes"],
|
||||
"APUMC_ENTRY_BYTES": manifest["apumc"]["entry_bytes"],
|
||||
"APUMC_ENTRY_COUNT": manifest["apumc"]["entry_count"],
|
||||
"APUMC_MAX_INSTRUCTIONS": manifest["apumc"]["max_instructions"],
|
||||
"APUMC_MAX_INSTRUCTIONS_V1": manifest["apumc"]["max_instructions_v1"],
|
||||
"APUMC_MAX_INSTRUCTIONS_V2": manifest["apumc"]["max_instructions_v2"],
|
||||
"APUMC_P3_PRIMITIVE_MASK": manifest["p3_implemented_primitive_mask"],
|
||||
"APUMC_P4_PRIMITIVE_MASK": manifest["p4_implemented_primitive_mask"],
|
||||
"APUMC_P5_PRIMITIVE_MASK": manifest["p5_implemented_primitive_mask"],
|
||||
"LOCAL_DATA_BYTES": manifest["p4_local_data_bytes"],
|
||||
"LOCAL_TABLE_SCRATCH_BYTES": manifest["p4_local_table_scratch_bytes"],
|
||||
}
|
||||
for name, value in manifest["classes"].items():
|
||||
exact[f"MC_CLASS_{name.upper()}"] = value
|
||||
for name, value in manifest["control_opcodes"].items():
|
||||
exact[f"MC_CONTROL_{name.upper()}"] = value
|
||||
for name, value in manifest["scalar_opcodes"].items():
|
||||
exact[f"MC_SCALAR_{name.upper()}"] = value
|
||||
for class_name, opcodes in manifest["deferred_opcodes"].items():
|
||||
for name, value in opcodes.items():
|
||||
exact[f"MC_{class_name.upper()}_{name.upper()}"] = value
|
||||
predicate_names = {
|
||||
"always": "ALWAYS",
|
||||
"eq": "EQ",
|
||||
"ne": "NE",
|
||||
"slt": "SIGNED_LT",
|
||||
"sge": "SIGNED_GE",
|
||||
"ult": "UNSIGNED_LT",
|
||||
"uge": "UNSIGNED_GE",
|
||||
"input_exhausted": "INPUT_EXHAUSTED",
|
||||
"input_ready": "INPUT_READY",
|
||||
"output_ready": "OUTPUT_READY",
|
||||
"kernel_done": "KERNEL_DONE",
|
||||
"transport_done": "TRANSPORT_DONE",
|
||||
}
|
||||
for name, value in manifest["predicates"].items():
|
||||
exact[f"MC_PRED_{predicate_names[name]}"] = value
|
||||
for name, value in manifest["wait_sources"].items():
|
||||
exact[f"MC_WAIT_{name.upper()}"] = value
|
||||
for name, value in manifest["primitives"].items():
|
||||
exact[f"MC_PRIMITIVE_{name.upper()}"] = value
|
||||
for name, value in manifest["instruction_fields"].items():
|
||||
exact[f"MC_INSTRUCTION_{name.upper()}"] = value
|
||||
for name, value in manifest["entry_words"].items():
|
||||
exact[f"APUMC_ENTRY_{name.upper()}"] = value
|
||||
trap_names = {
|
||||
"illegal": "ILLEGAL",
|
||||
"pc_range": "PC_RANGE",
|
||||
"call_stack": "CALL_STACK",
|
||||
"loop": "LOOP",
|
||||
"local_range": "LOCAL_RANGE",
|
||||
"unavailable": "UNAVAILABLE",
|
||||
"watchdog": "WATCHDOG",
|
||||
"retired_budget": "RETIRED_BUDGET",
|
||||
"engine": "ENGINE",
|
||||
"explicit": "EXPLICIT",
|
||||
}
|
||||
for name, value in manifest["trap_reasons"].items():
|
||||
if name != "reserved":
|
||||
exact[f"MC_TRAP_{trap_names[name]}"] = value
|
||||
return exact
|
||||
|
||||
|
||||
def _require_equal(what: str, left: dict[str, int], right: dict[str, int]) -> None:
|
||||
if left != right:
|
||||
only_left = sorted(set(left) - set(right))
|
||||
only_right = sorted(set(right) - set(left))
|
||||
differ = sorted(key for key in set(left) & set(right) if left[key] != right[key])
|
||||
raise ValueError(f"{what} drift: only-left {only_left}, only-right {only_right}, "
|
||||
f"differing {[(key, left[key], right[key]) for key in differ]}")
|
||||
|
||||
|
||||
def _mp3_stub_section(apb: dict[str, int]) -> dict[str, int]:
|
||||
"""Reconstruct the reserved-MP3 APUMC stub contract from the released image."""
|
||||
|
||||
lines = RTL_MICROCODE.read_text(encoding="utf-8").splitlines()
|
||||
entry_lines = [line for line in lines if line.startswith(".entry ")]
|
||||
if len(entry_lines) != 3:
|
||||
raise ValueError(f"{RTL_MICROCODE.name} must declare exactly three .entry lines")
|
||||
mp3_tokens = entry_lines[1].split()
|
||||
if mp3_tokens[:5] != [".entry", "1", "mp3", "mp3", "mp3"]:
|
||||
raise ValueError(f"unexpected MP3 entry directive: {entry_lines[1]}")
|
||||
numbers = [int(token, 0) for token in mp3_tokens[5:]]
|
||||
if numbers != [1, 1, 0, 0, 0, 0, 0]:
|
||||
raise ValueError(f"unexpected MP3 stub entry fields: {entry_lines[1]}")
|
||||
try:
|
||||
label_index = lines.index("mp3:")
|
||||
except ValueError as error:
|
||||
raise ValueError("mp3 label missing from released microcode image") from error
|
||||
body = [line.strip() for line in lines[label_index + 1:] if line.strip()]
|
||||
trap_match = re.fullmatch(r"trap 0x([0-9a-fA-F]+)", body[0])
|
||||
if trap_match is None:
|
||||
raise ValueError(f"mp3 label must be followed by its trap stub, found: {body[0]}")
|
||||
trap_detail = int(trap_match.group(1), 16)
|
||||
if trap_detail != 0x01000001:
|
||||
raise ValueError(f"unexpected MP3 stub trap detail 0x{trap_detail:08x}")
|
||||
|
||||
stub = Entry(
|
||||
format_id=1,
|
||||
entry_pc=0,
|
||||
first_pc=0,
|
||||
last_pc=0,
|
||||
max_loop_count=numbers[0],
|
||||
max_retired=numbers[1],
|
||||
scratch_base=numbers[2],
|
||||
scratch_bytes=numbers[3],
|
||||
primitive_mask=numbers[4],
|
||||
table_offset=numbers[5],
|
||||
table_bytes=numbers[6],
|
||||
)
|
||||
words = stub.words(APUMC_ABI_V2)
|
||||
if words != (0x00000001, 0, 0, 0, 1, 1, 0, 0):
|
||||
raise ValueError(f"MP3 stub entry words drifted: {[hex(word) for word in words]}")
|
||||
trap_instruction = Instruction(
|
||||
instruction_class=int(InstructionClass.CONTROL),
|
||||
opcode=int(ControlOpcode.TRAP),
|
||||
immediate=trap_detail,
|
||||
).encode()
|
||||
if trap_instruction != 0x0200000001000001:
|
||||
raise ValueError(f"MP3 stub trap instruction drifted: 0x{trap_instruction:016x}")
|
||||
|
||||
section = {
|
||||
"format_id": 1,
|
||||
"capability_bit": apb["CAPABILITY0_MP3"],
|
||||
"entry_count": apb["APUMC_ENTRY_COUNT"],
|
||||
"entry_index": 1,
|
||||
"trap_instruction": trap_instruction,
|
||||
"trap_detail": trap_detail,
|
||||
"rejection_error_code": apb["ERROR_CODE_UNSUPPORTED"],
|
||||
"rejection_error_stage": apb["ERROR_STAGE_RING"],
|
||||
"rejection_detail": trap_detail,
|
||||
}
|
||||
for index, word in enumerate(words):
|
||||
section[f"entry_words[{index}]"] = word
|
||||
return section
|
||||
|
||||
|
||||
def canonical_inventory() -> dict[str, dict[str, int]]:
|
||||
"""Build the complete frozen V1 table inventory, failing on any drift."""
|
||||
|
||||
rtl = _rtl_defines()
|
||||
apb = {
|
||||
name.removeprefix("APB4_APU__"): value
|
||||
for name, value in rtl.items()
|
||||
if name.startswith("APB4_APU__")
|
||||
}
|
||||
kws_model = {
|
||||
name.removeprefix("RETROSOC_APU_KWS__"): value
|
||||
for name, value in rtl.items()
|
||||
if name.startswith("RETROSOC_APU_KWS__")
|
||||
}
|
||||
_require_equal("APB define/C mirror", apb, _c_abi_defines())
|
||||
_require_equal("ISA manifest/SVH", _manifest_rtl_exact(),
|
||||
{name: apb[name] for name in _manifest_rtl_exact()})
|
||||
_require_equal(
|
||||
"APUM container/SVH",
|
||||
{
|
||||
"APUM_MAGIC": apu_kws.APUM_MAGIC,
|
||||
"APUM_ABI": apu_kws.APUM_ABI,
|
||||
"APUM_IMAGE_BYTES": apu_kws.APUM_BYTES,
|
||||
"APUM_PAYLOAD_CRC": apu_kws.APUM_PAYLOAD_CRC,
|
||||
},
|
||||
{name: apb[name] for name in ("APUM_MAGIC", "APUM_ABI", "APUM_IMAGE_BYTES",
|
||||
"APUM_PAYLOAD_CRC")},
|
||||
)
|
||||
discovery_c = _c_discovery_defines()
|
||||
discovery_rtl = _rtl_discovery_values()
|
||||
_require_equal("discovery C/RTL", discovery_rtl,
|
||||
{name: discovery_c[name] for name in discovery_rtl})
|
||||
|
||||
formats = {name.lower().removeprefix("rs_apu_"): value
|
||||
for name, value in _c_enum_values("rs_apu_format_t").items()}
|
||||
output_modes = {name.lower().removeprefix("rs_apu_"): value
|
||||
for name, value in _c_enum_values("rs_apu_output_t").items()}
|
||||
pcm_formats = {name.lower().removeprefix("rs_apu_"): value
|
||||
for name, value in _c_enum_values("rs_apu_pcm_t").items()}
|
||||
_require_equal(
|
||||
"format ID/capability bit",
|
||||
{name: formats[name] for name in ("wav", "mp3", "flac")},
|
||||
{name: apb[f"CAPABILITY0_{name.upper()}"] for name in ("wav", "mp3", "flac")},
|
||||
)
|
||||
|
||||
manifest = abi_manifest()
|
||||
isa: dict[str, int] = {}
|
||||
for group in ("classes", "control_opcodes", "scalar_opcodes", "predicates",
|
||||
"wait_sources", "primitives", "trap_reasons", "instruction_fields",
|
||||
"entry_words"):
|
||||
for name, value in manifest[group].items():
|
||||
isa[f"{group}.{name}"] = value
|
||||
for class_name, opcodes in manifest["deferred_opcodes"].items():
|
||||
for name, value in opcodes.items():
|
||||
isa[f"deferred_opcodes.{class_name}.{name}"] = value
|
||||
for phase in ("p3", "p4", "p5"):
|
||||
isa[f"implemented_primitive_mask.{phase}"] = manifest[
|
||||
f"{phase}_implemented_primitive_mask"]
|
||||
isa["local_memory.p4_data_bytes"] = manifest["p4_local_data_bytes"]
|
||||
isa["local_memory.p4_table_scratch_bytes"] = manifest["p4_local_table_scratch_bytes"]
|
||||
|
||||
apumc = {name: value for name, value in manifest["apumc"].items()}
|
||||
apum = {
|
||||
"magic": apu_kws.APUM_MAGIC,
|
||||
"abi": apu_kws.APUM_ABI,
|
||||
"image_bytes": apu_kws.APUM_BYTES,
|
||||
"header_bytes": apu_kws.APUM_HEADER_BYTES,
|
||||
"operator_offset": apu_kws.APUM_OPERATOR_OFFSET,
|
||||
"operator_count": apu_kws.APUM_OPERATOR_COUNT,
|
||||
"operator_bytes": apu_kws.APUM_OPERATOR_BYTES,
|
||||
"tensor_offset": apu_kws.APUM_TENSOR_OFFSET,
|
||||
"tensor_count": apu_kws.APUM_TENSOR_COUNT,
|
||||
"tensor_bytes": apu_kws.APUM_TENSOR_BYTES,
|
||||
"parameter_offset": apu_kws.APUM_PARAMETER_OFFSET,
|
||||
"parameter_bytes": apu_kws.APUM_PARAMETER_BYTES,
|
||||
"parameter_end": apu_kws.APUM_PARAMETER_END,
|
||||
"payload_crc": apu_kws.APUM_PAYLOAD_CRC,
|
||||
}
|
||||
|
||||
discovery = {
|
||||
"ip_id": discovery_c["IP_ID_VALUE"],
|
||||
"ip_version": discovery_c["IP_VERSION_VALUE"],
|
||||
"capability0": discovery_c["CAPABILITY0_IMPLEMENTED"],
|
||||
"capability0_p7": discovery_c["CAPABILITY0_P7_IMPLEMENTED"],
|
||||
"capability1": discovery_c["CAPABILITY1_IMPLEMENTED"],
|
||||
"capability1_p7": discovery_c["CAPABILITY1_P7_IMPLEMENTED"],
|
||||
"irq_all": discovery_c["IRQ_ALL"],
|
||||
}
|
||||
format_section = {f"format.{name}": value for name, value in formats.items()}
|
||||
format_section.update({f"output_mode.{name}": value
|
||||
for name, value in output_modes.items()})
|
||||
format_section.update({f"pcm_format.{name}": value
|
||||
for name, value in pcm_formats.items()})
|
||||
format_section["operation.file_decode"] = 0
|
||||
format_section["operation.kws_memory_window"] = 1
|
||||
|
||||
inventory = {
|
||||
"discovery": discovery,
|
||||
"apb": apb,
|
||||
"kws_model": kws_model,
|
||||
"formats": format_section,
|
||||
"apumc": apumc,
|
||||
"apum": apum,
|
||||
"isa": isa,
|
||||
"mp3_stub": _mp3_stub_section(apb),
|
||||
}
|
||||
if tuple(inventory) != SECTION_ORDER:
|
||||
raise ValueError("section order drift")
|
||||
return inventory
|
||||
|
||||
|
||||
def serialize_inventory(inventory: dict[str, dict[str, int]]) -> bytes:
|
||||
"""Serialize the inventory to the documented release-identity byte stream."""
|
||||
|
||||
return json.dumps(
|
||||
inventory,
|
||||
ensure_ascii=True,
|
||||
separators=(",", ":"),
|
||||
sort_keys=True,
|
||||
).encode("ascii")
|
||||
|
||||
|
||||
def compute_abi_digest() -> int:
|
||||
"""Return the CRC-32/ISO-HDLC digest over the serialized V1 tables."""
|
||||
|
||||
return binascii.crc32(serialize_inventory(canonical_inventory())) & 0xFFFFFFFF
|
||||
|
||||
|
||||
def rtl_implemented_digest() -> int:
|
||||
"""Extract the handwritten AbiDigest localparam from apu_reg.sv."""
|
||||
|
||||
text = RTL_REGISTER.read_text(encoding="utf-8")
|
||||
match = re.search(
|
||||
r"localparam logic \[31:0\] AbiDigest = EnableP7 \? "
|
||||
r"32'h([0-9a-fA-F_]+) : 32'd0;",
|
||||
text,
|
||||
)
|
||||
if match is None:
|
||||
raise ValueError(f"AbiDigest localparam not found in {RTL_REGISTER.name}")
|
||||
return int(match.group(1).replace("_", ""), 16)
|
||||
|
||||
|
||||
def c_implemented_digest() -> int:
|
||||
"""Extract the handwritten P7 release digest from apu_regs.h."""
|
||||
|
||||
text = C_HEADER.read_text(encoding="utf-8")
|
||||
match = re.search(
|
||||
r"^#define\s+RS_APU_DIGEST_P7_IMPLEMENTED\s+UINT32_C\((0[xX][0-9a-fA-F]+)\)\s*$",
|
||||
text,
|
||||
re.MULTILINE,
|
||||
)
|
||||
if match is None:
|
||||
raise ValueError(f"RS_APU_DIGEST_P7_IMPLEMENTED not found in {C_HEADER.name}")
|
||||
return int(match.group(1), 16)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
parser.add_argument("--check", action="store_true",
|
||||
help="require RTL/C handwritten constants to match the computed digest")
|
||||
parser.add_argument("--json", metavar="PATH",
|
||||
help="dump the canonical inventory as JSON for review")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
try:
|
||||
inventory = canonical_inventory()
|
||||
digest = binascii.crc32(serialize_inventory(inventory)) & 0xFFFFFFFF
|
||||
except ValueError as error:
|
||||
print(f"apu_abi_digest: error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if args.json is not None:
|
||||
payload = {
|
||||
"digest": f"0x{digest:08x}",
|
||||
"serialization": "canonical ASCII JSON; recursively sorted keys; compact "
|
||||
"separators; decimal integers; no terminal newline; "
|
||||
"CRC-32/ISO-HDLC",
|
||||
"sections": {
|
||||
name: {key: f"0x{value:016x}" for key, value in sorted(entries.items())}
|
||||
for name, entries in inventory.items()
|
||||
},
|
||||
}
|
||||
Path(args.json).write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
||||
print(f"wrote canonical inventory to {args.json}")
|
||||
|
||||
print(f"0x{digest:08x}")
|
||||
for name in SECTION_ORDER:
|
||||
print(f" {name}: {len(inventory[name])} entries")
|
||||
print(f" serialized: {len(serialize_inventory(inventory))} bytes")
|
||||
|
||||
if args.check:
|
||||
try:
|
||||
rtl_digest = rtl_implemented_digest()
|
||||
c_digest = c_implemented_digest()
|
||||
except ValueError as error:
|
||||
print(f"apu_abi_digest: error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
ok = digest != 0 and rtl_digest == digest and c_digest == digest
|
||||
print(f" computed: 0x{digest:08x}")
|
||||
print(f" apu_reg.sv: 0x{rtl_digest:08x}")
|
||||
print(f" apu_regs.h: 0x{c_digest:08x}")
|
||||
if not ok:
|
||||
print("apu_abi_digest: FAIL: digest mismatch or zero digest", file=sys.stderr)
|
||||
return 1
|
||||
print("apu_abi_digest: OK: computed digest matches both handwritten constants")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,349 @@
|
||||
"""Bit-accurate model (BAM) of the frozen APU-P7 frontend numerical profile 1.
|
||||
|
||||
Implements docs/ip/apu.md "P7 frontend numerical profile 1" (lines 2322-2398)
|
||||
exactly, mirroring the RTL arithmetic in rtl/ip/multimedia/apu_kws_engine.sv
|
||||
operation by operation:
|
||||
|
||||
PCM 16000 x S16 -> positive peak P=max(1,max(x[0..15999]))
|
||||
-> 49 frames of 480 samples (hop 320; the last 160 samples only feed P)
|
||||
-> Hann Q2.30, real FFT input RNE(x*H/2^15), imaginary zero, zero-pad 512
|
||||
-> bit-reversed radix-2 DIT FFT (Q2.30 cos/-sin twiddles, one RNE/2^30 per
|
||||
complex multiply, RNE/2 per butterfly sum/diff; components = DFT*64,
|
||||
signed-32 with fault on overflow)
|
||||
-> bins 0..256, nearest-integer sqrt of the unsigned-64 component squares
|
||||
(ties even)
|
||||
-> 40 mel triangles Q2.30 (DC bin zeroed), M=RNE(sum/2^32) unsigned UQ28.4
|
||||
-> exact rational z=(1000000*M+16*P)/(16000000*P), exact 2^e*m
|
||||
normalization, log ROM interpolation with f=65536 allowed
|
||||
-> DCT Q2.30, C[j]=RNE(sum(log*D)/2^30) signed Q8.24
|
||||
-> q=TZ(C[j]/(2^24*S)+83) with the exact binary32 scale S=0x3f15af17
|
||||
(2^24*S == 9809687 == 0x95AF17), low-8-bit wrap to the INT8 feature byte.
|
||||
|
||||
All arithmetic is exact Python integer arithmetic. ROM constants are imported
|
||||
from scripts/generate_apu_kws_rtl_constants.py (the same generators that emit
|
||||
rtl/ip/multimedia/apu_kws_rom.svh), never duplicated. Where the RTL faults
|
||||
(FFT signed-32 component overflow) or wraps (log output truncation to
|
||||
signed-32), this model does the same; :class:`FrontendFault` marks the fault
|
||||
cases, which valid S16 audio never reaches.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import struct
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
import generate_apu_kws_rtl_constants as gen # noqa: E402
|
||||
|
||||
WINDOW_SAMPLES: Final = 16000
|
||||
WINDOW_BYTES: Final = 32000
|
||||
FRAME_COUNT: Final = 49
|
||||
FRAME_SAMPLES: Final = 480
|
||||
FRAME_HOP: Final = 320
|
||||
FFT_POINTS: Final = 512
|
||||
FFT_BINS: Final = 257
|
||||
MEL_BANDS: Final = 40
|
||||
MFCC_COEFFICIENTS: Final = 10
|
||||
FEATURE_BYTES: Final = FRAME_COUNT * MFCC_COEFFICIENTS
|
||||
QUANT_SCALE_BITS: Final = 0x3F15AF17
|
||||
QUANT_DIVISOR: Final = 0x0095AF17
|
||||
QUANT_BIAS: Final = 83
|
||||
_S32_MIN: Final = -(1 << 31)
|
||||
_S32_MAX: Final = (1 << 31) - 1
|
||||
|
||||
|
||||
class FrontendFault(RuntimeError):
|
||||
"""The input drives an RTL checked-arithmetic fault range."""
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def hann_q30() -> tuple[int, ...]:
|
||||
"""Hann window coefficients H[n]=RNE(2^30*(1-cos(2*pi*n/480))/2)."""
|
||||
return tuple(gen._hann())
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def twiddle_q30() -> tuple[tuple[int, ...], tuple[int, ...]]:
|
||||
"""FFT twiddles: RNE-quantized Q2.30 cos and negative sin of 2*pi*k/512."""
|
||||
real, imag = gen._twiddle()
|
||||
return tuple(real), tuple(imag)
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def mel_q30() -> tuple[int, ...]:
|
||||
"""Forty mel triangle weight rows, 257 bins each, Q2.30, DC zeroed (flat)."""
|
||||
return tuple(gen._mel_weights())
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def dct_q30() -> tuple[int, ...]:
|
||||
"""DCT coefficients D[j,b]=RNE(2^30*sqrt(2/40)*cos(pi*(b+1/2)*j/40)) (flat)."""
|
||||
return tuple(gen._dct())
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def log_q24() -> tuple[int, ...]:
|
||||
"""Log ROM L[i]=RNE(2^24*ln(1+i/1024)) for i=0..1024, signed Q8.24."""
|
||||
return tuple(gen._log_rom())
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def ln2_q24() -> int:
|
||||
"""LN2=RNE(2^24*ln(2)) signed Q8.24, the ROM localparam ApuKwsLn2Q24."""
|
||||
return gen._rne(gen.TWO.ln() * Decimal(1 << 24))
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _bit_reverse9() -> tuple[int, ...]:
|
||||
return tuple(int(f"{index:09b}"[::-1], 2) for index in range(FFT_POINTS))
|
||||
|
||||
|
||||
def _s32(value: int) -> int:
|
||||
"""Truncate to signed 32 bits, mirroring the RTL 32'(...) conversion."""
|
||||
value &= 0xFFFFFFFF
|
||||
return value - (1 << 32) if value > _S32_MAX else value
|
||||
|
||||
|
||||
def _trunc_div(numerator: int, denominator: int) -> int:
|
||||
"""TZ(numerator/denominator): truncation toward zero."""
|
||||
return numerator // denominator if numerator >= 0 else -((-numerator) // denominator)
|
||||
|
||||
|
||||
def rne_shift(value: int, shift: int) -> int:
|
||||
"""RNE(value / 2^shift) with ties to even, sign symmetric (RTL rne_shift)."""
|
||||
if shift == 0:
|
||||
return value
|
||||
magnitude = -value if value < 0 else value
|
||||
quotient = magnitude >> shift
|
||||
remainder = magnitude & ((1 << shift) - 1)
|
||||
half = 1 << (shift - 1)
|
||||
if remainder > half or (remainder == half and (quotient & 1) != 0):
|
||||
quotient += 1
|
||||
return -quotient if value < 0 else quotient
|
||||
|
||||
|
||||
def rne_uq32(value: int) -> int:
|
||||
"""Unsigned RNE(value / 2^32) with ties to even (RTL rne_uq32)."""
|
||||
if value < 0 or value >= 1 << 64:
|
||||
raise FrontendFault(f"mel reduction outside u64: {value}")
|
||||
quotient = value >> 32
|
||||
remainder = value & 0xFFFFFFFF
|
||||
if remainder > 0x80000000 or (remainder == 0x80000000 and (quotient & 1) != 0):
|
||||
quotient += 1
|
||||
return quotient
|
||||
|
||||
|
||||
def nearest_isqrt(value: int) -> int:
|
||||
"""Nearest-integer square root of an unsigned-64 value, ties even."""
|
||||
if value < 0 or value >= 1 << 64:
|
||||
raise FrontendFault(f"magnitude square outside u64: {value}")
|
||||
root = math.isqrt(value)
|
||||
lower = value - root * root
|
||||
upper = (root + 1) * (root + 1) - value
|
||||
if upper < lower or (upper == lower and (root & 1) != 0):
|
||||
root += 1
|
||||
return root
|
||||
|
||||
|
||||
def log_mel_q24(mel: int, peak: int) -> int:
|
||||
"""Exact rational log stage: Q8.24 ln(M/(16*P)+1e-6) (RTL log_mel_q24)."""
|
||||
if not 0 <= mel <= 0xFFFFFFFF:
|
||||
raise FrontendFault(f"mel value outside UQ28.4 word: {mel}")
|
||||
if not 1 <= peak <= 0xFFFF:
|
||||
raise FrontendFault(f"window peak outside u16: {peak}")
|
||||
numerator = 1_000_000 * mel + 16 * peak
|
||||
denominator = 16_000_000 * peak
|
||||
exponent = 0
|
||||
if numerator >= denominator:
|
||||
normalized = denominator
|
||||
while numerator >= normalized << 1:
|
||||
normalized <<= 1
|
||||
exponent += 1
|
||||
else:
|
||||
normalized = denominator
|
||||
while numerator < normalized:
|
||||
numerator <<= 1
|
||||
exponent -= 1
|
||||
scaled = (numerator - normalized) * 1024
|
||||
index = scaled // normalized
|
||||
remainder = scaled - index * normalized
|
||||
fraction_scaled = remainder * 65536
|
||||
fraction = fraction_scaled // normalized
|
||||
fraction_remainder = fraction_scaled - fraction * normalized
|
||||
if 2 * fraction_remainder > normalized or (
|
||||
2 * fraction_remainder == normalized and (fraction & 1) != 0
|
||||
):
|
||||
fraction += 1 # fraction may be 65536 (spec step 5)
|
||||
table = log_q24()
|
||||
interpolation = rne_shift((table[index + 1] - table[index]) * fraction, 16)
|
||||
return _s32(exponent * ln2_q24() + table[index] + interpolation)
|
||||
|
||||
|
||||
def quantize_coefficient(coefficient: int) -> int:
|
||||
"""Frozen step-7 quantizer: q=TZ(C/(2^24*S)+83) with S=0x3f15af17.
|
||||
|
||||
C/(2^24*S)+83 == (C + 83*9809687)/9809687 as an exact rational; truncating
|
||||
the integer division matches both the frozen formula and the RTL
|
||||
expression (s_dct_coefficient + 64'sh3087_c475) / 64'sh0095_af17.
|
||||
"""
|
||||
return _trunc_div(coefficient + QUANT_BIAS * QUANT_DIVISOR, QUANT_DIVISOR)
|
||||
|
||||
|
||||
def wrap_int8(value: int) -> int:
|
||||
"""Low-8-bit wrap of a quantized coefficient, interpreted signed INT8."""
|
||||
wrapped = value & 0xFF
|
||||
return wrapped - 256 if wrapped >= 128 else wrapped
|
||||
|
||||
|
||||
def samples_from_pcm(pcm: bytes) -> tuple[int, ...]:
|
||||
"""Decode one 32000-byte little-endian S16 window into 16000 samples."""
|
||||
if len(pcm) != WINDOW_BYTES:
|
||||
raise ValueError(f"profile-1 window must be {WINDOW_BYTES} bytes, got {len(pcm)}")
|
||||
return struct.unpack(f"<{WINDOW_SAMPLES}h", pcm)
|
||||
|
||||
|
||||
def _fft_frame(windowed: list[int], real: list[int], imag: list[int]) -> None:
|
||||
"""One 512-point profile-1 FFT in place; windowed holds 480 FFT inputs."""
|
||||
twiddle_real, twiddle_imag = twiddle_q30()
|
||||
bit_reverse = _bit_reverse9()
|
||||
for index in range(FFT_POINTS):
|
||||
real[index] = 0
|
||||
imag[index] = 0
|
||||
for index in range(FRAME_SAMPLES):
|
||||
real[bit_reverse[index]] = windowed[index]
|
||||
length = 2
|
||||
while length <= FFT_POINTS:
|
||||
half = length // 2
|
||||
step = FFT_POINTS // length
|
||||
for group in range(0, FFT_POINTS, length):
|
||||
for j in range(half):
|
||||
left = group + j
|
||||
right = left + half
|
||||
wr = twiddle_real[step * j]
|
||||
wi = twiddle_imag[step * j]
|
||||
right_r = real[right]
|
||||
right_i = imag[right]
|
||||
tw_r = _s32(rne_shift(right_r * wr - right_i * wi, 30))
|
||||
tw_i = _s32(rne_shift(right_r * wi + right_i * wr, 30))
|
||||
left_r = real[left]
|
||||
left_i = imag[left]
|
||||
sum_r = rne_shift(left_r + tw_r, 1)
|
||||
sum_i = rne_shift(left_i + tw_i, 1)
|
||||
dif_r = rne_shift(left_r - tw_r, 1)
|
||||
dif_i = rne_shift(left_i - tw_i, 1)
|
||||
for component in (sum_r, sum_i, dif_r, dif_i):
|
||||
if component < _S32_MIN or component > _S32_MAX:
|
||||
raise FrontendFault(f"FFT component overflow: {component}")
|
||||
real[left] = sum_r
|
||||
imag[left] = sum_i
|
||||
real[right] = dif_r
|
||||
imag[right] = dif_i
|
||||
length *= 2
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FrontendStages:
|
||||
"""Every observable profile-1 stage for one 16000-sample window."""
|
||||
|
||||
samples: tuple[int, ...]
|
||||
peak: int
|
||||
windowed: tuple[tuple[int, ...], ...] # 49 x 480, RNE(x*H/2^15)
|
||||
fft_real: tuple[tuple[int, ...], ...] # 49 x 512 signed-32, unscaled DFT*64
|
||||
fft_imag: tuple[tuple[int, ...], ...]
|
||||
magnitude: tuple[tuple[int, ...], ...] # 49 x 257
|
||||
mel: tuple[tuple[int, ...], ...] # 49 x 40 unsigned UQ28.4
|
||||
logs: tuple[tuple[int, ...], ...] # 49 x 40 signed Q8.24
|
||||
coeffs: tuple[tuple[int, ...], ...] # 49 x 10 signed Q8.24
|
||||
quantized: tuple[tuple[int, ...], ...] # 49 x 10 pre-wrap q
|
||||
features: bytes # 490 wrapped INT8 feature bytes
|
||||
|
||||
|
||||
def frontend_stages(samples: tuple[int, ...] | list[int]) -> FrontendStages:
|
||||
"""Run profile 1 on one 16000-sample S16 window and return every stage."""
|
||||
if len(samples) != WINDOW_SAMPLES:
|
||||
raise ValueError(f"profile-1 window must be {WINDOW_SAMPLES} samples")
|
||||
samples = tuple(int(value) for value in samples)
|
||||
if any(value < -32768 or value > 32767 for value in samples):
|
||||
raise ValueError("profile-1 input must be signed S16")
|
||||
hann = hann_q30()
|
||||
mel_weights = mel_q30()
|
||||
dct = dct_q30()
|
||||
peak = max(1, max(samples))
|
||||
|
||||
windowed_rows: list[tuple[int, ...]] = []
|
||||
fft_real_rows: list[tuple[int, ...]] = []
|
||||
fft_imag_rows: list[tuple[int, ...]] = []
|
||||
magnitude_rows: list[tuple[int, ...]] = []
|
||||
mel_rows: list[tuple[int, ...]] = []
|
||||
real = [0] * FFT_POINTS
|
||||
imag = [0] * FFT_POINTS
|
||||
for row in range(FRAME_COUNT):
|
||||
frame = samples[FRAME_HOP * row : FRAME_HOP * row + FRAME_SAMPLES]
|
||||
windowed = [rne_shift(value * coefficient, 15) for value, coefficient in zip(frame, hann)]
|
||||
_fft_frame(windowed, real, imag)
|
||||
windowed_rows.append(tuple(windowed))
|
||||
fft_real_rows.append(tuple(real))
|
||||
fft_imag_rows.append(tuple(imag))
|
||||
magnitude = tuple(
|
||||
nearest_isqrt(real[bin_index] * real[bin_index] + imag[bin_index] * imag[bin_index])
|
||||
for bin_index in range(FFT_BINS)
|
||||
)
|
||||
magnitude_rows.append(magnitude)
|
||||
mel_rows.append(
|
||||
tuple(
|
||||
rne_uq32(
|
||||
sum(
|
||||
magnitude[bin_index] * mel_weights[band * FFT_BINS + bin_index]
|
||||
for bin_index in range(FFT_BINS)
|
||||
)
|
||||
)
|
||||
for band in range(MEL_BANDS)
|
||||
)
|
||||
)
|
||||
|
||||
logs_rows = tuple(
|
||||
tuple(log_mel_q24(mel_rows[row][band], peak) for band in range(MEL_BANDS))
|
||||
for row in range(FRAME_COUNT)
|
||||
)
|
||||
coeffs_rows: list[tuple[int, ...]] = []
|
||||
for row in range(FRAME_COUNT):
|
||||
coeffs = []
|
||||
for coefficient in range(MFCC_COEFFICIENTS):
|
||||
total = sum(
|
||||
logs_rows[row][band] * dct[coefficient * MEL_BANDS + band]
|
||||
for band in range(MEL_BANDS)
|
||||
)
|
||||
if total < -(1 << 63) or total >= 1 << 63:
|
||||
raise FrontendFault(f"DCT reduction outside s64: {total}")
|
||||
coeffs.append(rne_shift(total, 30))
|
||||
coeffs_rows.append(tuple(coeffs))
|
||||
quantized_rows = tuple(
|
||||
tuple(quantize_coefficient(coeffs_rows[row][j]) for j in range(MFCC_COEFFICIENTS))
|
||||
for row in range(FRAME_COUNT)
|
||||
)
|
||||
features = bytes(value & 0xFF for row in quantized_rows for value in row)
|
||||
return FrontendStages(
|
||||
samples=samples,
|
||||
peak=peak,
|
||||
windowed=tuple(windowed_rows),
|
||||
fft_real=tuple(fft_real_rows),
|
||||
fft_imag=tuple(fft_imag_rows),
|
||||
magnitude=tuple(magnitude_rows),
|
||||
mel=tuple(mel_rows),
|
||||
logs=logs_rows,
|
||||
coeffs=tuple(coeffs_rows),
|
||||
quantized=quantized_rows,
|
||||
features=features,
|
||||
)
|
||||
|
||||
|
||||
def mfcc_profile1(pcm: bytes) -> bytes:
|
||||
"""Map one 32000-byte padded PCM window to its 490 profile-1 MFCC bytes."""
|
||||
return frontend_stages(samples_from_pcm(pcm)).features
|
||||
@@ -0,0 +1,219 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate the embedded APU release-evidence assets for the apu_release app.
|
||||
|
||||
Reads the deterministic P5 APUMC microcode bundle, the frozen P7 APUM KWS
|
||||
model, and one locked KWS corpus utterance; synthesizes a small deterministic
|
||||
mono 48 kHz S16 WAV fixture; computes the frozen expected PCM through the
|
||||
bit-accurate codec model; and emits a C header with every byte array and CRC
|
||||
the LP/HP evidence flow needs. The generator is deterministic: no network, no
|
||||
clock, and no randomness.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import struct
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
|
||||
import sys
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from apu_codecs import decode_wav, process_pcm # noqa: E402
|
||||
from apu_kws import APUM_ABI, APUM_BYTES, APUM_MAGIC, APUM_PAYLOAD_CRC, APUM_SHA256 # noqa: E402
|
||||
|
||||
APUMC_MAGIC = 0x41504D43
|
||||
APUMC_ABI_VERSIONS = (0x00010000, 0x00020000)
|
||||
APUMC_HEADER_BYTES = 64
|
||||
APUMC_HEADER_CRC_WORD = 11
|
||||
|
||||
WAV_RATE = 48000
|
||||
WAV_CHANNELS = 1
|
||||
WAV_BITS = 16
|
||||
WAV_FRAMES = 480
|
||||
|
||||
KWS_WINDOW_BYTES = 32000
|
||||
KWS_EXPECTED_CLASS = 7 # "Stop" in the frozen class order
|
||||
KWS_THRESHOLD = 1
|
||||
KWS_DEBOUNCE = 1
|
||||
# Locked corpus manifest row 000000 (docs/ip/apu-kws-corpus.tsv).
|
||||
KWS_CORPUS_RELPATH = "stop/563aa4e6_nohash_3.wav"
|
||||
KWS_WAV_SHA256 = "6ed51beca5b58fbbb06a16af68f120066218cf52c3fd4237f222a8ea9c972199"
|
||||
KWS_PCM_SHA256 = "d843b234a0f585d143c88aed0acc1cca5ed02150f3cf5b77cb0489ad5429d725"
|
||||
|
||||
|
||||
def _require(condition: bool, message: str) -> None:
|
||||
if not condition:
|
||||
raise ValueError(message)
|
||||
|
||||
|
||||
def _crc32(data: bytes) -> int:
|
||||
return zlib.crc32(data) & 0xFFFFFFFF
|
||||
|
||||
|
||||
def _sha256(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _load_apumc(path: Path) -> tuple[bytes, int, int]:
|
||||
image = path.read_bytes()
|
||||
_require(len(image) >= APUMC_HEADER_BYTES, f"APUMC bundle too small: {path}")
|
||||
magic, abi = struct.unpack_from("<II", image, 0)
|
||||
_require(magic == APUMC_MAGIC, f"APUMC magic mismatch in {path}")
|
||||
_require(abi in APUMC_ABI_VERSIONS, f"APUMC ABI 0x{abi:08x} is not supported")
|
||||
header_crc = struct.unpack_from("<I", image, APUMC_HEADER_CRC_WORD * 4)[0]
|
||||
payload_crc = _crc32(image[APUMC_HEADER_BYTES:])
|
||||
_require(header_crc == payload_crc, f"APUMC header CRC disagrees with payload CRC in {path}")
|
||||
return image, _crc32(image), payload_crc
|
||||
|
||||
|
||||
def _load_apum(path: Path) -> tuple[bytes, int]:
|
||||
image = path.read_bytes()
|
||||
_require(len(image) == APUM_BYTES, f"APUM model is not {APUM_BYTES} bytes: {path}")
|
||||
magic, abi = struct.unpack_from("<II", image, 0)
|
||||
_require(magic == APUM_MAGIC, f"APUM magic mismatch in {path}")
|
||||
_require(abi == APUM_ABI, f"APUM ABI 0x{abi:08x} mismatch in {path}")
|
||||
payload_crc = struct.unpack_from("<I", image, 0x24)[0]
|
||||
_require(payload_crc == APUM_PAYLOAD_CRC, f"APUM payload CRC mismatch in {path}")
|
||||
_require(_sha256(image) == APUM_SHA256, f"APUM image SHA-256 mismatch in {path}")
|
||||
return image, _crc32(image)
|
||||
|
||||
|
||||
def _synthesize_wav() -> bytes:
|
||||
samples = [((index * 48271 + 12345) % 65536) - 32768 for index in range(WAV_FRAMES)]
|
||||
payload = struct.pack(f"<{WAV_FRAMES}h", *samples)
|
||||
alignment = WAV_CHANNELS * (WAV_BITS // 8)
|
||||
fmt = struct.pack("<HHIIHH", 1, WAV_CHANNELS, WAV_RATE, WAV_RATE * alignment, alignment, WAV_BITS)
|
||||
chunks = b"fmt " + struct.pack("<I", len(fmt)) + fmt
|
||||
chunks += b"data" + struct.pack("<I", len(payload)) + payload
|
||||
return b"RIFF" + struct.pack("<I", len(chunks) + 4) + b"WAVE" + chunks
|
||||
|
||||
|
||||
def _expected_pcm(wav: bytes) -> tuple[int, int, int]:
|
||||
audio = decode_wav(wav, strict=True)
|
||||
processed = process_pcm(audio, output_bits=16)
|
||||
_require(processed.rate == WAV_RATE, "fixture rate mismatch")
|
||||
_require(processed.channels == WAV_CHANNELS, "fixture channel mismatch")
|
||||
_require(processed.frames == WAV_FRAMES, "fixture frame mismatch")
|
||||
_require(len(processed.payload) == WAV_FRAMES * 2, "fixture payload size mismatch")
|
||||
info = audio.info
|
||||
_require(info.input_used == len(wav), "fixture WAV is not fully consumed")
|
||||
return _crc32(processed.payload), len(processed.payload), processed.frames
|
||||
|
||||
|
||||
def _extract_mono16k_data(wav: bytes) -> bytes:
|
||||
_require(len(wav) >= 12 and wav[0:4] == b"RIFF" and wav[8:12] == b"WAVE", "KWS source is not RIFF/WAVE")
|
||||
fmt: bytes | None = None
|
||||
data: bytes | None = None
|
||||
offset = 12
|
||||
while offset + 8 <= len(wav):
|
||||
chunk_size = struct.unpack_from("<I", wav, offset + 4)[0]
|
||||
body = wav[offset + 8 : offset + 8 + chunk_size]
|
||||
_require(len(body) == chunk_size, "KWS source chunk is truncated")
|
||||
if wav[offset : offset + 4] == b"fmt ":
|
||||
fmt = body
|
||||
elif wav[offset : offset + 4] == b"data":
|
||||
data = body
|
||||
offset += 8 + chunk_size + (chunk_size & 1)
|
||||
_require(fmt is not None and len(fmt) >= 16 and data is not None, "KWS source lacks fmt/data")
|
||||
audio_format, channels, rate, _byte_rate, _align, bits = struct.unpack_from("<HHIIHH", fmt)
|
||||
_require(
|
||||
(audio_format, channels, rate, bits) == (1, 1, 16000, 16),
|
||||
"KWS source is not mono/16000/S16 PCM",
|
||||
)
|
||||
_require(len(data) <= KWS_WINDOW_BYTES and len(data) % 2 == 0, "KWS source exceeds one window")
|
||||
return data
|
||||
|
||||
|
||||
def _load_kws_pcm(corpus_root: Path) -> tuple[bytes, int]:
|
||||
source = corpus_root / KWS_CORPUS_RELPATH
|
||||
_require(source.is_file(), f"locked KWS corpus utterance missing: {source}")
|
||||
wav = source.read_bytes()
|
||||
_require(_sha256(wav) == KWS_WAV_SHA256, f"KWS source WAV SHA-256 mismatch: {source}")
|
||||
padded = _extract_mono16k_data(wav)
|
||||
padded = padded + bytes(KWS_WINDOW_BYTES - len(padded))
|
||||
_require(_sha256(padded) == KWS_PCM_SHA256, f"KWS padded PCM SHA-256 mismatch: {source}")
|
||||
return padded, _crc32(padded)
|
||||
|
||||
|
||||
def _format_array(name: str, data: bytes) -> str:
|
||||
lines = [f"static const uint8_t {name}[{len(data)}] __attribute__((aligned(64))) = {{"]
|
||||
for offset in range(0, len(data), 16):
|
||||
chunk = ",".join(f"0x{value:02X}" for value in data[offset : offset + 16])
|
||||
lines.append(f" {chunk},")
|
||||
lines.append("};")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _emit(args: argparse.Namespace) -> str:
|
||||
apumc, apumc_crc, apumc_payload_crc = _load_apumc(args.apumc)
|
||||
apum, apum_crc = _load_apum(args.apum)
|
||||
wav = _synthesize_wav()
|
||||
pcm_crc, pcm_bytes, pcm_frames = _expected_pcm(wav)
|
||||
kws_pcm, kws_crc = _load_kws_pcm(args.kws_corpus)
|
||||
|
||||
sections = [
|
||||
"/* Generated by scripts/build_apu_release_assets.py; do not edit. */",
|
||||
"#ifndef APU_RELEASE_ASSETS_H",
|
||||
"#define APU_RELEASE_ASSETS_H",
|
||||
"",
|
||||
"#include <stdint.h>",
|
||||
"",
|
||||
f"#define RS_APU_RELEASE_APUMC_SIZE UINT32_C({len(apumc)})",
|
||||
f"#define RS_APU_RELEASE_APUMC_CRC UINT32_C(0x{apumc_crc:08X})",
|
||||
f"#define RS_APU_RELEASE_APUMC_PAYLOAD_CRC UINT32_C(0x{apumc_payload_crc:08X})",
|
||||
f"#define RS_APU_RELEASE_APUM_SIZE UINT32_C({len(apum)})",
|
||||
f"#define RS_APU_RELEASE_APUM_CRC UINT32_C(0x{apum_crc:08X})",
|
||||
f"#define RS_APU_RELEASE_APUM_PAYLOAD_CRC UINT32_C(0x{APUM_PAYLOAD_CRC:08X})",
|
||||
f"#define RS_APU_RELEASE_WAV_SIZE UINT32_C({len(wav)})",
|
||||
f"#define RS_APU_RELEASE_WAV_CRC UINT32_C(0x{_crc32(wav):08X})",
|
||||
f"#define RS_APU_RELEASE_WAV_RATE UINT32_C({WAV_RATE})",
|
||||
f"#define RS_APU_RELEASE_WAV_CHANNELS UINT32_C({WAV_CHANNELS})",
|
||||
f"#define RS_APU_RELEASE_WAV_BITS UINT32_C({WAV_BITS})",
|
||||
f"#define RS_APU_RELEASE_PCM_CRC UINT32_C(0x{pcm_crc:08X})",
|
||||
f"#define RS_APU_RELEASE_PCM_BYTES UINT32_C({pcm_bytes})",
|
||||
f"#define RS_APU_RELEASE_PCM_FRAMES UINT32_C({pcm_frames})",
|
||||
f"#define RS_APU_RELEASE_KWS_SIZE UINT32_C({len(kws_pcm)})",
|
||||
f"#define RS_APU_RELEASE_KWS_CRC UINT32_C(0x{kws_crc:08X})",
|
||||
f"#define RS_APU_RELEASE_KWS_CLASS UINT32_C({KWS_EXPECTED_CLASS})",
|
||||
f"#define RS_APU_RELEASE_KWS_THRESHOLD UINT32_C({KWS_THRESHOLD})",
|
||||
f"#define RS_APU_RELEASE_KWS_DEBOUNCE UINT32_C({KWS_DEBOUNCE})",
|
||||
"",
|
||||
_format_array("rs_apu_release_apumc", apumc),
|
||||
"",
|
||||
_format_array("rs_apu_release_apum", apum),
|
||||
"",
|
||||
_format_array("rs_apu_release_wav", wav),
|
||||
"",
|
||||
_format_array("rs_apu_release_kws_pcm", kws_pcm),
|
||||
"",
|
||||
"#endif",
|
||||
"",
|
||||
]
|
||||
return "\n".join(sections)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--apumc", type=Path, required=True, help="P5 APUMC bundle path")
|
||||
parser.add_argument("--apum", type=Path, required=True, help="P7 APUM KWS model path")
|
||||
parser.add_argument(
|
||||
"--kws-corpus",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="locked apu-kws-pcm corpus root (run the P7 reference setup first)",
|
||||
)
|
||||
parser.add_argument("--output", type=Path, required=True, help="generated C header path")
|
||||
args = parser.parse_args()
|
||||
content = _emit(args)
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(content, encoding="ascii", newline="\n")
|
||||
print(f"apu_release assets: {args.output} ({len(content)} bytes)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,60 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build the hart-1 APU ownership-evidence payload used by HP RTL simulation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def build(args: argparse.Namespace) -> None:
|
||||
output = args.output.resolve()
|
||||
images = output / "images"
|
||||
images.mkdir(parents=True, exist_ok=True)
|
||||
elf = output / "hp_apu.elf"
|
||||
compiler = f"{args.cross}gcc"
|
||||
objcopy = f"{args.cross}objcopy"
|
||||
includes = [f"-I{path.resolve()}" for path in args.include]
|
||||
source_dir = args.source_dir.resolve()
|
||||
subprocess.run(
|
||||
[
|
||||
compiler,
|
||||
"-march=rv32imafdc_zicbom_zicsr_zifencei",
|
||||
"-mabi=ilp32d",
|
||||
"-nostdlib",
|
||||
"-nostartfiles",
|
||||
"-ffreestanding",
|
||||
"-O2",
|
||||
"-Wall",
|
||||
"-Wextra",
|
||||
"-Wl,--build-id=none",
|
||||
f"-Wl,-T,{source_dir / 'linker.ld'}",
|
||||
*includes,
|
||||
"-o",
|
||||
str(elf),
|
||||
str(source_dir / "start.S"),
|
||||
str(source_dir / "main.c"),
|
||||
],
|
||||
check=True,
|
||||
)
|
||||
subprocess.run(
|
||||
[objcopy, "-O", "binary", str(elf), str(images / "fw_jump.bin")],
|
||||
check=True,
|
||||
)
|
||||
(images / "retrosoc_hp.dtb").write_bytes(b"SMOK")
|
||||
(images / "Image").write_bytes(b"SMOK")
|
||||
(images / "rootfs.cpio.gz").write_bytes(b"SMOK")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--source-dir", type=Path, required=True)
|
||||
parser.add_argument("--output", type=Path, required=True)
|
||||
parser.add_argument("--cross", default="riscv32-unknown-elf-")
|
||||
parser.add_argument("--include", type=Path, action="append", default=[])
|
||||
build(parser.parse_args())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,856 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run the APU-P7 concurrent decode + continuous KWS zero-xrun RTL qualification.
|
||||
|
||||
Builds tests/rtl/apu_p7_concurrent_tb.sv once (sv2v + Icarus), then runs the
|
||||
frozen concurrency matrix: continuous RX KWS concurrent with looped WAV/FLAC
|
||||
decode streamed to I2S TX under the active P5 ready-memory qualification
|
||||
(64-word prefill, then ready memory without injected stalls). Each scenario
|
||||
requires zero codec underrun, zero I2S RX/KWS overrun, every full uninterrupted
|
||||
scheduled KWS window completed before its successor, and matching
|
||||
byte/frame/cycle/stall scoreboards. Writes the concurrent.json report and exits
|
||||
nonzero on any violation or missing input. Nothing is downloaded.
|
||||
|
||||
Prerequisites (no network access is used by this script):
|
||||
- Icarus Verilog (iverilog/vvp) and sv2v on PATH.
|
||||
- The deterministic P5 codec bundle `apu-p5.apumc`, built once with
|
||||
`make CONFIG=configs/ci/ihp130.mk apu-p5-bundle` (writes
|
||||
build/<variant>/apu/p5/apu-p5.apumc). Pass it with --apumc or let the newest
|
||||
build/*/apu/p5/apu-p5.apumc be auto-discovered.
|
||||
- The locked P7 TFLite model under .cache/retrosoc/sources/apu-mlperf-tiny
|
||||
(installed by the shared P7 reference setup); the APUM image is derived
|
||||
locally with scripts/apu_kws_convert.py. --apum accepts a prebuilt image.
|
||||
|
||||
The frozen release gate is 60 simulated seconds per scenario at 48 MHz PCLK
|
||||
(--seconds 60). The default is a short development run; --smoke selects the
|
||||
minimal single-scenario matrix for pytest.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from apu_codecs import crc8, crc16, decode_flac, decode_wav, process_pcm # noqa: E402
|
||||
from apu_kws import APUM_PAYLOAD_CRC, APUM_SHA256 # noqa: E402
|
||||
|
||||
PCLK_HZ = 48_000_000
|
||||
FULL_GATE_SECONDS = 60.0
|
||||
FULL_GATE_CYCLES = int(FULL_GATE_SECONDS * PCLK_HZ)
|
||||
FLAC_CONSTANT = -2
|
||||
CONTRACT_REVISION = "apu-kws-convert/1.0.0"
|
||||
DEFAULT_TFLITE = (
|
||||
ROOT
|
||||
/ ".cache/retrosoc/sources/apu-mlperf-tiny/benchmark/training/keyword_spotting"
|
||||
/ "trained_models/kws_ref_model.tflite"
|
||||
)
|
||||
BUNDLE_PREREQUISITE = "make CONFIG=configs/ci/ihp130.mk apu-p5-bundle"
|
||||
PROFILE = "configs/ci/ihp130.mk"
|
||||
|
||||
RESULT_PATTERN = re.compile(
|
||||
r"CONCURRENT_RESULT "
|
||||
r"rate=(?P<rate>\d+) "
|
||||
r"precision=(?P<precision>\d+) "
|
||||
r"seconds_ms=(?P<seconds_ms>\d+) "
|
||||
r"duration_cycles=(?P<duration_cycles>\d+) "
|
||||
r"jobs=(?P<jobs>\d+) "
|
||||
r"wav_jobs=(?P<wav_jobs>\d+) "
|
||||
r"flac_jobs=(?P<flac_jobs>\d+) "
|
||||
r"tx_words=(?P<tx_words>\d+) "
|
||||
r"consumed=(?P<consumed>\d+) "
|
||||
r"underrun=(?P<underrun>\d+) "
|
||||
r"rx_overrun=(?P<rx_overrun>\d+) "
|
||||
r"kws_overrun=(?P<kws_overrun>\d+) "
|
||||
r"kws_frames=(?P<kws_frames>\d+) "
|
||||
r"kws_windows=(?P<kws_windows>\d+) "
|
||||
r"rx_words=(?P<rx_words>\d+) "
|
||||
r"codec_input_bytes=(?P<codec_input_bytes>\d+) "
|
||||
r"codec_output_bytes=(?P<codec_output_bytes>\d+) "
|
||||
r"codec_frames=(?P<codec_frames>\d+) "
|
||||
r"codec_cycles=(?P<codec_cycles>\d+) "
|
||||
r"dma_read_stalls=(?P<dma_read_stalls>\d+) "
|
||||
r"dma_write_stalls=(?P<dma_write_stalls>\d+) "
|
||||
r"stream_stalls=(?P<stream_stalls>\d+) "
|
||||
r"active_cycles=(?P<active_cycles>\d+) "
|
||||
r"elapsed_cycles=(?P<elapsed_cycles>\d+) "
|
||||
r"scoreboard=(?P<scoreboard>\w+)"
|
||||
)
|
||||
|
||||
|
||||
def _sha256_bytes(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as source:
|
||||
for block in iter(lambda: source.read(1024 * 1024), b""):
|
||||
digest.update(block)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _write_json(path: Path, value: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
temporary.replace(path)
|
||||
|
||||
|
||||
def _hex_words(path: Path, payload: bytes, words: int) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
lines = []
|
||||
for index in range(words):
|
||||
chunk = payload[index * 4 : index * 4 + 4]
|
||||
lines.append(f"{int.from_bytes(chunk.ljust(4, bytes(1)), 'little'):08x}\n")
|
||||
path.write_text("".join(lines), encoding="utf-8")
|
||||
|
||||
|
||||
def _run_command(command: list[str], *, timeout: int | None = None) -> dict[str, Any]:
|
||||
started = time.monotonic()
|
||||
try:
|
||||
process = subprocess.run(
|
||||
command, cwd=ROOT, check=False, capture_output=True, text=True, timeout=timeout
|
||||
)
|
||||
return {
|
||||
"command": command,
|
||||
"duration_seconds": round(time.monotonic() - started, 3),
|
||||
"exit_code": process.returncode,
|
||||
"stdout": process.stdout,
|
||||
"stderr": process.stderr,
|
||||
"timed_out": False,
|
||||
}
|
||||
except subprocess.TimeoutExpired as error:
|
||||
stdout = error.stdout.decode("utf-8", errors="replace") if error.stdout else ""
|
||||
stderr = error.stderr.decode("utf-8", errors="replace") if error.stderr else ""
|
||||
return {
|
||||
"command": command,
|
||||
"duration_seconds": round(time.monotonic() - started, 3),
|
||||
"exit_code": 124,
|
||||
"stdout": stdout,
|
||||
"stderr": stderr,
|
||||
"timed_out": True,
|
||||
}
|
||||
|
||||
|
||||
def _tool_version(name: str, flag: str = "--version") -> dict[str, Any]:
|
||||
path = shutil.which(name)
|
||||
if path is None:
|
||||
return {"path": None, "version": None}
|
||||
probe = _run_command([path, flag], timeout=30)
|
||||
first_line = (probe["stdout"] or probe["stderr"]).splitlines()
|
||||
return {
|
||||
"path": path,
|
||||
"version": first_line[0].strip() if first_line else "unknown",
|
||||
"sha256": _sha256_file(Path(path)),
|
||||
}
|
||||
|
||||
|
||||
def _git_revision() -> tuple[str, bool]:
|
||||
revision = _run_command(["git", "rev-parse", "HEAD"], timeout=30)
|
||||
dirty = _run_command(["git", "status", "--porcelain"], timeout=30)
|
||||
return (
|
||||
revision["stdout"].strip() if revision["exit_code"] == 0 else "unknown",
|
||||
bool(dirty["stdout"].strip()) if dirty["exit_code"] == 0 else True,
|
||||
)
|
||||
|
||||
|
||||
def _production_sources() -> list[Path]:
|
||||
common = ROOT / "rtl/managed/clusterip/common/rtl"
|
||||
multimedia = ROOT / "rtl/ip/multimedia"
|
||||
names = (
|
||||
"apu_microcode_pkg.sv",
|
||||
"apu_dma.sv",
|
||||
"apu_ring_scheduler.sv",
|
||||
"apu_stream_router.sv",
|
||||
"apu_control_store.sv",
|
||||
"apu_microcode_loader.sv",
|
||||
"apu_local_sram.sv",
|
||||
"apu_kws_engine.sv",
|
||||
"apu_kws_sram_client.sv",
|
||||
"apu_kws_model_loader.sv",
|
||||
"apu_bitstream_engine.sv",
|
||||
"apu_entropy_engine.sv",
|
||||
"apu_reconstruction_engine.sv",
|
||||
"apu_transform_engine.sv",
|
||||
"apu_resampler.sv",
|
||||
"apu_kernel_engine.sv",
|
||||
"apu_primitive_dispatcher.sv",
|
||||
"apu_codec_sequencer.sv",
|
||||
"apu_codec_transport.sv",
|
||||
"apu_codec_controller.sv",
|
||||
"apu_reg.sv",
|
||||
"apb4_apu.sv",
|
||||
)
|
||||
return [
|
||||
common / "interface/apb4_if.sv",
|
||||
common / "interface/axi4_if.sv",
|
||||
common / "interface/axi4_stream_if.sv",
|
||||
common / "utils/register.sv",
|
||||
common / "utils/fifo.sv",
|
||||
ROOT / "rtl/tech/tc_sram.sv",
|
||||
ROOT / "rtl/ip/peripheral/dma_axi4_master.sv",
|
||||
*(multimedia / name for name in names),
|
||||
ROOT / "tests/rtl/apu_p7_concurrent_tb.sv",
|
||||
]
|
||||
|
||||
|
||||
def _verification_sources() -> dict[str, str]:
|
||||
inputs = [
|
||||
*_production_sources(),
|
||||
ROOT / "rtl/ip/multimedia/apu_define.svh",
|
||||
ROOT / "scripts/apu_codecs.py",
|
||||
ROOT / "scripts/apu_kws.py",
|
||||
ROOT / "scripts/apu_kws_convert.py",
|
||||
ROOT / "rtl/mini/script/convt_sv2v.py",
|
||||
Path(__file__).resolve(),
|
||||
]
|
||||
return {
|
||||
path.relative_to(ROOT).as_posix(): _sha256_file(path)
|
||||
for path in sorted(inputs)
|
||||
if path.is_file()
|
||||
}
|
||||
|
||||
|
||||
def _resolve_apumc(apumc: Path | None) -> Path:
|
||||
if apumc is not None:
|
||||
resolved = apumc.resolve()
|
||||
if not resolved.is_file():
|
||||
raise SystemExit(f"APUMC bundle not found: {resolved}\nprerequisite: {BUNDLE_PREREQUISITE}")
|
||||
return resolved
|
||||
candidates = sorted(
|
||||
ROOT.glob("build/*/apu/p5/apu-p5.apumc"), key=lambda path: path.stat().st_mtime
|
||||
)
|
||||
if not candidates:
|
||||
raise SystemExit(
|
||||
"no APUMC bundle found below build/*/apu/p5/\n"
|
||||
f"prerequisite: {BUNDLE_PREREQUISITE}"
|
||||
)
|
||||
return candidates[-1].resolve()
|
||||
|
||||
|
||||
def _resolve_apum(apum: Path | None) -> tuple[bytes, str]:
|
||||
if apum is not None:
|
||||
resolved = apum.resolve()
|
||||
if not resolved.is_file():
|
||||
raise SystemExit(f"APUM model image not found: {resolved}")
|
||||
payload = resolved.read_bytes()
|
||||
return payload, str(resolved)
|
||||
if not DEFAULT_TFLITE.is_file():
|
||||
raise SystemExit(
|
||||
f"locked P7 TFLite model not installed: {DEFAULT_TFLITE}\n"
|
||||
"prerequisite: make CONFIG=configs/ci/ihp130.mk setup-apu-kws-reference"
|
||||
)
|
||||
from apu_kws_convert import import_tflite
|
||||
|
||||
payload = import_tflite(DEFAULT_TFLITE)
|
||||
return payload, f"generated:{DEFAULT_TFLITE}"
|
||||
|
||||
|
||||
def _wav_payload(rate: int, bits: int, frames: int, seed: int) -> bytes:
|
||||
block_align = 2 * (bits // 8)
|
||||
state = seed | 1
|
||||
data = bytearray()
|
||||
for frame in range(frames):
|
||||
for channel in range(2):
|
||||
state = (1103515245 * state + 12345 + channel + frame) & 0x7FFFFFFF
|
||||
value = ((state >> 8) & ((1 << bits) - 1)) - (1 << (bits - 1))
|
||||
data.extend(value.to_bytes(bits // 8, "little", signed=True))
|
||||
header = bytearray()
|
||||
header.extend(b"RIFF")
|
||||
header.extend((36 + len(data)).to_bytes(4, "little"))
|
||||
header.extend(b"WAVE")
|
||||
header.extend(b"fmt ")
|
||||
header.extend((16).to_bytes(4, "little"))
|
||||
header.extend((1).to_bytes(2, "little"))
|
||||
header.extend((2).to_bytes(2, "little"))
|
||||
header.extend(rate.to_bytes(4, "little"))
|
||||
header.extend((rate * block_align).to_bytes(4, "little"))
|
||||
header.extend(block_align.to_bytes(2, "little"))
|
||||
header.extend(bits.to_bytes(2, "little"))
|
||||
header.extend(b"data")
|
||||
header.extend(len(data).to_bytes(4, "little"))
|
||||
return bytes(header + data)
|
||||
|
||||
|
||||
def _flac_payload(rate: int, bits: int, block: int) -> bytes:
|
||||
streaminfo = bytearray(34)
|
||||
streaminfo[0:2] = block.to_bytes(2, "big")
|
||||
streaminfo[2:4] = block.to_bytes(2, "big")
|
||||
streaminfo[10:18] = ((rate << 44) | ((bits - 1) << 36) | block).to_bytes(8, "big")
|
||||
size_code = 4 if bits == 16 else 6
|
||||
frame_header = ((0x3FFE << 18) | (7 << 12) | (size_code << 1)).to_bytes(4, "big")
|
||||
frame_header += b"\x00" + (block - 1).to_bytes(2, "big")
|
||||
frame_header += bytes([crc8(frame_header)])
|
||||
frame = frame_header + b"\x00" + FLAC_CONSTANT.to_bytes(bits // 8, "big", signed=True)
|
||||
frame += crc16(frame).to_bytes(2, "big")
|
||||
return b"fLaC" + bytes([0x80, 0, 0, 34]) + bytes(streaminfo) + frame
|
||||
|
||||
|
||||
def _expected_pcm(
|
||||
source: bytes, *, is_wav: bool, rate: int, bits: int, resample: bool
|
||||
) -> tuple[list[int], int]:
|
||||
decoded = decode_wav(source, strict=True) if is_wav else decode_flac(source, strict=True)
|
||||
processed = process_pcm(
|
||||
decoded, output_rate=rate, output_channels=2, output_bits=bits, resample=resample,
|
||||
i2s=True,
|
||||
)
|
||||
payload = processed.payload
|
||||
if len(payload) % 4 != 0:
|
||||
raise RuntimeError("BAM PCM payload is not word aligned")
|
||||
words = [
|
||||
int.from_bytes(payload[offset : offset + 4], "little")
|
||||
for offset in range(0, len(payload), 4)
|
||||
]
|
||||
return words, processed.frames
|
||||
|
||||
|
||||
def _write_pattern_hex(path: Path, words: list[int]) -> None:
|
||||
padded = list(words) + [0] * (16384 - len(words))
|
||||
path.write_text("".join(f"{word:08x}\n" for word in padded), encoding="utf-8")
|
||||
|
||||
|
||||
def _compile(build_dir: Path) -> dict[str, Any]:
|
||||
common = ROOT / "rtl/managed/clusterip/common/rtl"
|
||||
multimedia = ROOT / "rtl/ip/multimedia"
|
||||
filelist = build_dir / "apu_p7_concurrent.fl"
|
||||
filelist.write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"+define+SV_ASSRT_DISABLE",
|
||||
f"+incdir+{common}",
|
||||
f"+incdir+{common / 'interface'}",
|
||||
f"+incdir+{multimedia}",
|
||||
*(str(source) for source in _production_sources()),
|
||||
"",
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
converted = build_dir / "apu_p7_concurrent.v"
|
||||
conversion = _run_command(
|
||||
[
|
||||
sys.executable,
|
||||
str(ROOT / "rtl/mini/script/convt_sv2v.py"),
|
||||
"-f",
|
||||
str(filelist),
|
||||
"--output",
|
||||
str(converted),
|
||||
],
|
||||
timeout=900,
|
||||
)
|
||||
if conversion["exit_code"] != 0:
|
||||
raise RuntimeError(f"sv2v conversion failed: {conversion['stderr'][-2000:]}")
|
||||
binary = build_dir / "apu_p7_concurrent_sim"
|
||||
icarus = _run_command(
|
||||
[
|
||||
"iverilog",
|
||||
"-g2012",
|
||||
"-s",
|
||||
"apu_p7_concurrent_tb",
|
||||
"-o",
|
||||
str(binary),
|
||||
str(converted),
|
||||
],
|
||||
timeout=900,
|
||||
)
|
||||
if icarus["exit_code"] != 0:
|
||||
raise RuntimeError(f"Icarus compile failed: {icarus['stderr'][-2000:]}")
|
||||
result = {
|
||||
"status": "passed",
|
||||
"filelist": str(filelist),
|
||||
"converted": str(converted),
|
||||
"converted_sha256": _sha256_file(converted),
|
||||
"binary": str(binary),
|
||||
"sv2v": conversion,
|
||||
"icarus": icarus,
|
||||
}
|
||||
_write_json(build_dir / "compile.json", result)
|
||||
return result
|
||||
|
||||
|
||||
def _kws_schedule(rx_words: int, precision: int, rate: int) -> tuple[int, int, int]:
|
||||
words_per_frame = 2 if precision == 24 else 1
|
||||
decimation = 6 if rate == 96000 else 3
|
||||
samples_16k = (rx_words // words_per_frame) // decimation
|
||||
frames = (samples_16k - 480) // 320 + 1 if samples_16k >= 480 else 0
|
||||
windows = (samples_16k - 16000) // 1600 + 1 if samples_16k >= 16000 else 0
|
||||
return samples_16k, frames, windows
|
||||
|
||||
|
||||
def _check_scenario(
|
||||
scenario: dict[str, Any], parsed: dict[str, Any], workloads: dict[str, Any]
|
||||
) -> list[str]:
|
||||
errors: list[str] = []
|
||||
rate = scenario["rate_hz"]
|
||||
precision = scenario["precision_bits"]
|
||||
if parsed["scoreboard"] != "match":
|
||||
errors.append("testbench scoreboard did not match")
|
||||
for field in ("underrun", "rx_overrun", "kws_overrun"):
|
||||
if parsed[field] != 0:
|
||||
errors.append(f"{field}={parsed[field]} violates the zero-xrun gate")
|
||||
expected_rx_words = scenario["duration_cycles"] // scenario["frame_period"]
|
||||
if parsed["rx_words"] != expected_rx_words:
|
||||
errors.append(f"rx_words={parsed['rx_words']} != expected {expected_rx_words}")
|
||||
samples_16k, kws_frames, kws_windows = _kws_schedule(parsed["rx_words"], precision, rate)
|
||||
scenario["kws_16k_samples"] = samples_16k
|
||||
scenario["expected_kws_frames"] = kws_frames
|
||||
scenario["expected_kws_windows"] = kws_windows
|
||||
if parsed["kws_frames"] != kws_frames:
|
||||
errors.append(f"kws_frames={parsed['kws_frames']} != expected {kws_frames}")
|
||||
if parsed["kws_windows"] != kws_windows:
|
||||
errors.append(f"kws_windows={parsed['kws_windows']} != expected {kws_windows}")
|
||||
if parsed["jobs"] != parsed["wav_jobs"] + parsed["flac_jobs"]:
|
||||
errors.append("job accounting mismatch")
|
||||
if parsed["wav_jobs"] < 1 or parsed["flac_jobs"] < 1:
|
||||
errors.append("WAV and FLAC workloads must both run in every scenario")
|
||||
expected_tx = (
|
||||
parsed["wav_jobs"] * workloads["wav_out_words"]
|
||||
+ parsed["flac_jobs"] * workloads["flac_out_words"]
|
||||
)
|
||||
if parsed["tx_words"] != expected_tx or parsed["consumed"] != expected_tx:
|
||||
errors.append(f"tx_words/consumed != expected {expected_tx}")
|
||||
expected_input = (
|
||||
parsed["wav_jobs"] * workloads["wav_bytes"]
|
||||
+ parsed["flac_jobs"] * workloads["flac_bytes"]
|
||||
)
|
||||
if parsed["codec_input_bytes"] != expected_input:
|
||||
errors.append(f"codec_input_bytes != expected {expected_input}")
|
||||
if parsed["codec_output_bytes"] != expected_tx * 4:
|
||||
errors.append(f"codec_output_bytes != expected {expected_tx * 4}")
|
||||
expected_frames = (
|
||||
parsed["wav_jobs"] * workloads["wav_frames"]
|
||||
+ parsed["flac_jobs"] * workloads["flac_frames"]
|
||||
)
|
||||
if parsed["codec_frames"] != expected_frames:
|
||||
errors.append(f"codec_frames != expected {expected_frames}")
|
||||
return errors
|
||||
|
||||
|
||||
def _run_scenario(
|
||||
scenario: dict[str, Any],
|
||||
compile_result: dict[str, Any],
|
||||
artifacts: dict[str, str],
|
||||
workloads: dict[str, Any],
|
||||
timeout: int,
|
||||
) -> dict[str, Any]:
|
||||
scenario_dir = Path(artifacts["scenario_dir"])
|
||||
log_path = scenario_dir / "run.log"
|
||||
command = [
|
||||
"vvp",
|
||||
compile_result["binary"],
|
||||
f"+APUMC_HEX={artifacts['apumc_hex']}",
|
||||
f"+APUM_HEX={artifacts['apum_hex']}",
|
||||
f"+WAV_HEX={scenario_dir / 'wav.hex'}",
|
||||
f"+FLAC_HEX={scenario_dir / 'flac.hex'}",
|
||||
f"+WAV_PCM_HEX={scenario_dir / 'wav_pcm.hex'}",
|
||||
f"+FLAC_PCM_HEX={scenario_dir / 'flac_pcm.hex'}",
|
||||
f"+WAV_BYTES={workloads['wav_bytes']}",
|
||||
f"+WAV_FRAMES={workloads['wav_frames']}",
|
||||
f"+WAV_OUT_WORDS={workloads['wav_out_words']}",
|
||||
f"+WAV_SOURCE_RATE={workloads['wav_source_rate']}",
|
||||
f"+FLAC_BYTES={workloads['flac_bytes']}",
|
||||
f"+FLAC_FRAMES={workloads['flac_frames']}",
|
||||
f"+FLAC_OUT_WORDS={workloads['flac_out_words']}",
|
||||
f"+RATE={scenario['rate_hz']}",
|
||||
f"+PRECISION={scenario['precision_bits']}",
|
||||
f"+DURATION_CYCLES={scenario['duration_cycles']}",
|
||||
f"+SECONDS_MS={scenario['seconds_ms']}",
|
||||
f"+MAX_CYCLES={scenario['max_cycles']}",
|
||||
f"+SEED={scenario['seed']}",
|
||||
"+WORKLOAD_MASK=3",
|
||||
]
|
||||
execution = _run_command(command, timeout=timeout)
|
||||
log_path.write_text(execution["stdout"] + execution["stderr"], encoding="utf-8")
|
||||
result: dict[str, Any] = {
|
||||
"scenario": scenario["id"],
|
||||
"command": command,
|
||||
"log": str(log_path),
|
||||
"exit_code": execution["exit_code"],
|
||||
"timed_out": execution["timed_out"],
|
||||
"wall_seconds": execution["duration_seconds"],
|
||||
"timeout_seconds": timeout,
|
||||
"status": "failed",
|
||||
"errors": [],
|
||||
}
|
||||
match = RESULT_PATTERN.search(execution["stdout"])
|
||||
if execution["exit_code"] != 0 or match is None:
|
||||
tail = (execution["stdout"] + execution["stderr"]).strip().splitlines()[-8:]
|
||||
result["errors"].append(
|
||||
"simulator failed" if execution["exit_code"] != 0 else "no CONCURRENT_RESULT line"
|
||||
)
|
||||
result["log_tail"] = tail
|
||||
return result
|
||||
parsed: dict[str, Any] = {
|
||||
name: int(value) for name, value in match.groupdict().items() if name != "scoreboard"
|
||||
}
|
||||
parsed["scoreboard"] = match.group("scoreboard")
|
||||
result["result"] = parsed
|
||||
result["simulated_cycles_per_wall_second"] = (
|
||||
round(parsed["elapsed_cycles"] / execution["duration_seconds"], 3)
|
||||
if execution["duration_seconds"] > 0
|
||||
else None
|
||||
)
|
||||
errors = _check_scenario(scenario, parsed, workloads)
|
||||
result["errors"] = errors
|
||||
result["status"] = "passed" if not errors else "failed"
|
||||
return result
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--build-dir", type=Path, required=True)
|
||||
parser.add_argument("--apumc", type=Path, default=None)
|
||||
parser.add_argument("--apum", type=Path, default=None)
|
||||
parser.add_argument(
|
||||
"--seconds",
|
||||
type=float,
|
||||
default=0.5,
|
||||
help="simulated seconds per scenario; the frozen release gate is 60",
|
||||
)
|
||||
parser.add_argument("--smoke", action="store_true", help="minimal single-scenario matrix")
|
||||
parser.add_argument(
|
||||
"--rates",
|
||||
type=int,
|
||||
nargs="+",
|
||||
default=None,
|
||||
choices=(48000, 96000),
|
||||
help="I2S rates to run (default both, or 48000 in --smoke)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--precisions",
|
||||
type=int,
|
||||
nargs="+",
|
||||
default=None,
|
||||
choices=(16, 24),
|
||||
help="output precisions to run (default both, or 16 in --smoke)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--smoke-seconds",
|
||||
type=float,
|
||||
default=0.03,
|
||||
help="simulated seconds per scenario in --smoke mode (kept short for pytest)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--job-frames",
|
||||
type=int,
|
||||
default=None,
|
||||
help="stereo frames per WAV/FLAC job (default 512 in --smoke, else 4096)",
|
||||
)
|
||||
parser.add_argument("--timeout-seconds", type=int, default=14400)
|
||||
parser.add_argument("--seed", type=int, default=0)
|
||||
parser.add_argument("--output", type=Path, default=None)
|
||||
args = parser.parse_args()
|
||||
build_dir = args.build_dir.resolve()
|
||||
build_dir.mkdir(parents=True, exist_ok=True)
|
||||
output = args.output.resolve() if args.output else build_dir / "concurrent.json"
|
||||
|
||||
for tool in ("iverilog", "vvp", "sv2v"):
|
||||
if shutil.which(tool) is None:
|
||||
raise SystemExit(f"required tool not on PATH: {tool}")
|
||||
if args.seconds <= 0 or args.smoke_seconds <= 0:
|
||||
raise SystemExit("--seconds/--smoke-seconds must be positive")
|
||||
seconds = args.smoke_seconds if args.smoke else args.seconds
|
||||
job_frames = args.job_frames
|
||||
if job_frames is None:
|
||||
job_frames = 512 if args.smoke else 4096
|
||||
if job_frames < 64 or job_frames > 4096:
|
||||
raise SystemExit("--job-frames must be within 64..4096")
|
||||
apumc_path = _resolve_apumc(args.apumc)
|
||||
apumc_payload = apumc_path.read_bytes()
|
||||
apum_payload, apum_source = _resolve_apum(args.apum)
|
||||
if len(apum_payload) != 32768:
|
||||
raise SystemExit(f"APUM image must be 32768 bytes, got {len(apum_payload)}")
|
||||
apum_sha256 = _sha256_bytes(apum_payload)
|
||||
apum_matches_freeze = apum_sha256 == APUM_SHA256
|
||||
|
||||
if args.smoke and (args.rates is not None or args.precisions is not None):
|
||||
raise SystemExit("--smoke selects its own minimal matrix; do not combine with --rates/--precisions")
|
||||
rates = tuple(args.rates) if args.rates is not None else ((48000,) if args.smoke else (48000, 96000))
|
||||
precisions = (
|
||||
tuple(args.precisions) if args.precisions is not None else ((16,) if args.smoke else (16, 24))
|
||||
)
|
||||
scenarios: list[dict[str, Any]] = []
|
||||
for rate in rates:
|
||||
for precision in precisions:
|
||||
precision_id = "s16" if precision == 16 else "s24_32"
|
||||
duration_cycles = int(round(seconds * PCLK_HZ))
|
||||
frame_period = PCLK_HZ // rate // (2 if precision == 24 else 1)
|
||||
if duration_cycles % frame_period != 0:
|
||||
raise SystemExit(
|
||||
f"--seconds {seconds} does not divide the {rate}-Hz/{precision}-bit "
|
||||
"I2S frame cadence; choose a duration with millisecond resolution"
|
||||
)
|
||||
# WAV workload: the P5-qualified playback shapes. A 44.1 kHz
|
||||
# stereo S16 source resampled to 48 kHz for the 48 kHz I2S
|
||||
# scenarios, and an 8 kHz stereo S16 source resampled to 96 kHz
|
||||
# for the 96 kHz scenarios (the PLAYBACK_44100_48000 /
|
||||
# PLAYBACK_8000_96000 qualification shapes). Source frame counts
|
||||
# are multiples of 147 for the exact 160/147 output ratio.
|
||||
wav_source_rate = 44100 if rate == 48000 else 8000
|
||||
if wav_source_rate == 44100:
|
||||
wav_source_frames = max(147, 147 * (job_frames // 147))
|
||||
else:
|
||||
wav_source_frames = job_frames
|
||||
wav = _wav_payload(wav_source_rate, 16, wav_source_frames, args.seed)
|
||||
flac = _flac_payload(rate, precision, job_frames)
|
||||
wav_pcm, wav_frames = _expected_pcm(
|
||||
wav, is_wav=True, rate=rate, bits=precision, resample=True
|
||||
)
|
||||
flac_pcm, flac_frames = _expected_pcm(
|
||||
flac, is_wav=False, rate=rate, bits=precision, resample=False
|
||||
)
|
||||
if flac_frames != job_frames:
|
||||
raise SystemExit("FLAC BAM frame count mismatch")
|
||||
if len(wav_pcm) != wav_frames * (2 if precision == 24 else 1):
|
||||
raise SystemExit("WAV BAM PCM word count mismatch")
|
||||
if len(flac_pcm) != flac_frames * (2 if precision == 24 else 1):
|
||||
raise SystemExit("FLAC BAM PCM word count mismatch")
|
||||
pair_cycles = (wav_frames + flac_frames) * (PCLK_HZ // rate)
|
||||
if duration_cycles < pair_cycles:
|
||||
raise SystemExit(
|
||||
f"--seconds {seconds} at {rate}-Hz/{precision}-bit cannot fit one "
|
||||
f"WAV+FLAC job pair ({pair_cycles} playback cycles); increase "
|
||||
"--seconds or reduce --job-frames"
|
||||
)
|
||||
scenarios.append(
|
||||
{
|
||||
"id": f"{rate}-{precision_id}",
|
||||
"rate_hz": rate,
|
||||
"precision_bits": precision,
|
||||
"seconds": seconds,
|
||||
"seconds_ms": int(round(seconds * 1000)),
|
||||
"duration_cycles": duration_cycles,
|
||||
"max_cycles": duration_cycles + 400_000_000,
|
||||
"frame_period": frame_period,
|
||||
"seed": args.seed,
|
||||
"wav": wav,
|
||||
"flac": flac,
|
||||
"wav_pcm": wav_pcm,
|
||||
"flac_pcm": flac_pcm,
|
||||
"workloads": {
|
||||
"wav_bytes": len(wav),
|
||||
"flac_bytes": len(flac),
|
||||
"wav_sha256": _sha256_bytes(wav),
|
||||
"flac_sha256": _sha256_bytes(flac),
|
||||
"wav_source_rate": wav_source_rate,
|
||||
"wav_source_frames": wav_source_frames,
|
||||
"wav_frames": wav_frames,
|
||||
"flac_frames": flac_frames,
|
||||
"wav_out_words": len(wav_pcm),
|
||||
"flac_out_words": len(flac_pcm),
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
apumc_hex = build_dir / "apu-p5.hex"
|
||||
_hex_words(apumc_hex, apumc_payload, 16384)
|
||||
apum_hex = build_dir / "kws-apum.hex"
|
||||
_hex_words(apum_hex, apum_payload, 8192)
|
||||
artifacts = {"apumc_hex": str(apumc_hex), "apum_hex": str(apum_hex), "scenario_dir": ""}
|
||||
|
||||
started = time.monotonic()
|
||||
compile_result = _compile(build_dir)
|
||||
|
||||
scenario_results: list[dict[str, Any]] = []
|
||||
failures: list[str] = []
|
||||
for scenario in scenarios:
|
||||
scenario_dir = build_dir / "scenarios" / scenario["id"]
|
||||
scenario_dir.mkdir(parents=True, exist_ok=True)
|
||||
_hex_words(scenario_dir / "wav.hex", scenario["wav"], 8192)
|
||||
_hex_words(scenario_dir / "flac.hex", scenario["flac"], 1024)
|
||||
_write_pattern_hex(scenario_dir / "wav_pcm.hex", scenario["wav_pcm"])
|
||||
_write_pattern_hex(scenario_dir / "flac_pcm.hex", scenario["flac_pcm"])
|
||||
workloads = scenario["workloads"]
|
||||
artifacts["scenario_dir"] = str(scenario_dir)
|
||||
scenario_results.append(
|
||||
_run_scenario(scenario, compile_result, artifacts, workloads, args.timeout_seconds)
|
||||
)
|
||||
status = scenario_results[-1]["status"]
|
||||
print(f"APU-P7 concurrent: {scenario['id']} {status}", flush=True)
|
||||
_write_json(scenario_dir / "result.json", scenario_results[-1])
|
||||
if status != "passed":
|
||||
failures.extend(f"{scenario['id']}: {error}" for error in scenario_results[-1]["errors"])
|
||||
|
||||
git_revision, git_dirty = _git_revision()
|
||||
cycles_per_second = [
|
||||
item["simulated_cycles_per_wall_second"]
|
||||
for item in scenario_results
|
||||
if item.get("simulated_cycles_per_wall_second")
|
||||
]
|
||||
slowest = min(cycles_per_second) if cycles_per_second else None
|
||||
projected_per_config = (
|
||||
round(FULL_GATE_CYCLES / slowest, 1) if slowest else None
|
||||
)
|
||||
full_matrix_configs = 4
|
||||
projection = {
|
||||
"full_gate_seconds_per_config": FULL_GATE_SECONDS,
|
||||
"full_gate_cycles_per_config": FULL_GATE_CYCLES,
|
||||
"full_matrix_configs": full_matrix_configs,
|
||||
"measured_min_cycles_per_wall_second": slowest,
|
||||
"projected_wall_seconds_per_config": projected_per_config,
|
||||
"projected_wall_seconds_full_matrix": (
|
||||
round(projected_per_config * full_matrix_configs, 1) if projected_per_config else None
|
||||
),
|
||||
"projected_full_matrix_exceeds_12h": (
|
||||
bool(projected_per_config * full_matrix_configs > 12 * 3600)
|
||||
if projected_per_config
|
||||
else None
|
||||
),
|
||||
}
|
||||
aggregate = {
|
||||
"scenarios": len(scenario_results),
|
||||
"jobs": sum(item["result"]["jobs"] for item in scenario_results if "result" in item),
|
||||
"codec_underruns": sum(
|
||||
item["result"]["underrun"] for item in scenario_results if "result" in item
|
||||
),
|
||||
"rx_overruns": sum(
|
||||
item["result"]["rx_overrun"] for item in scenario_results if "result" in item
|
||||
),
|
||||
"kws_overruns": sum(
|
||||
item["result"]["kws_overrun"] for item in scenario_results if "result" in item
|
||||
),
|
||||
"codec_input_bytes": sum(
|
||||
item["result"]["codec_input_bytes"] for item in scenario_results if "result" in item
|
||||
),
|
||||
"codec_output_bytes": sum(
|
||||
item["result"]["codec_output_bytes"] for item in scenario_results if "result" in item
|
||||
),
|
||||
"codec_frames": sum(
|
||||
item["result"]["codec_frames"] for item in scenario_results if "result" in item
|
||||
),
|
||||
"codec_cycles": sum(
|
||||
item["result"]["codec_cycles"] for item in scenario_results if "result" in item
|
||||
),
|
||||
}
|
||||
report = {
|
||||
"schema_version": 1,
|
||||
"phase": "APU-P7",
|
||||
"report": "concurrent",
|
||||
"gate": "concurrent WAV/FLAC decode with continuous RX KWS zero-xrun",
|
||||
"status": "passed" if not failures else "failed",
|
||||
"success": not failures,
|
||||
"release_qualifying": bool(
|
||||
not failures
|
||||
and not args.smoke
|
||||
and seconds >= FULL_GATE_SECONDS
|
||||
and set(rates) == {48000, 96000}
|
||||
and set(precisions) == {16, 24}
|
||||
),
|
||||
"mode": "smoke" if args.smoke else "full-matrix",
|
||||
"seconds_per_scenario": seconds,
|
||||
"job_frames_per_workload": job_frames,
|
||||
"required_full_gate_seconds": FULL_GATE_SECONDS,
|
||||
"pclk_hz": PCLK_HZ,
|
||||
"git_revision": git_revision,
|
||||
"git_dirty": git_dirty,
|
||||
"command": [sys.executable, str(Path(__file__).resolve()), *sys.argv[1:]],
|
||||
"seed": args.seed,
|
||||
"profile": PROFILE,
|
||||
"bundle_prerequisite": BUNDLE_PREREQUISITE,
|
||||
"inputs": {
|
||||
"apumc": {
|
||||
"path": str(apumc_path),
|
||||
"bytes": len(apumc_payload),
|
||||
"sha256": _sha256_bytes(apumc_payload),
|
||||
},
|
||||
"apum": {
|
||||
"path": apum_source,
|
||||
"bytes": len(apum_payload),
|
||||
"sha256": apum_sha256,
|
||||
"payload_crc32": f"0x{APUM_PAYLOAD_CRC:08x}",
|
||||
"matches_locked_freeze": apum_matches_freeze,
|
||||
},
|
||||
"tflite": {
|
||||
"path": str(DEFAULT_TFLITE),
|
||||
"sha256": _sha256_file(DEFAULT_TFLITE) if DEFAULT_TFLITE.is_file() else None,
|
||||
},
|
||||
},
|
||||
"converter": {
|
||||
"path": "scripts/apu_kws_convert.py",
|
||||
"sha256": _sha256_file(ROOT / "scripts/apu_kws_convert.py"),
|
||||
"contract_revision": CONTRACT_REVISION,
|
||||
},
|
||||
"bam": {
|
||||
"path": "scripts/apu_codecs.py",
|
||||
"sha256": _sha256_file(ROOT / "scripts/apu_codecs.py"),
|
||||
},
|
||||
"rtl_revision": git_revision,
|
||||
"verification_sources": _verification_sources(),
|
||||
"simulator": {
|
||||
"name": "icarus",
|
||||
"iverilog": _tool_version("iverilog", "-V"),
|
||||
"vvp": _tool_version("vvp", "-V"),
|
||||
},
|
||||
"tools": {
|
||||
"sv2v": _tool_version("sv2v", "--version"),
|
||||
"python": platform.python_version(),
|
||||
},
|
||||
"compile": compile_result,
|
||||
"harness": {
|
||||
"axi_model": "ready memory, no injected stalls",
|
||||
"playback_prefill_words": 64,
|
||||
"playback_fifo_words": 1024,
|
||||
"router_fifo_words": 64,
|
||||
"rx_model": "one I2S frame word per cadence tick, never re-presented",
|
||||
"kws_config": "0x0180 (threshold 128, debounce 1)",
|
||||
},
|
||||
"matrix": [
|
||||
{
|
||||
"scenario": scenario["id"],
|
||||
"rate_hz": scenario["rate_hz"],
|
||||
"precision_bits": scenario["precision_bits"],
|
||||
"seconds": scenario["seconds"],
|
||||
"workloads": {
|
||||
"wav": {
|
||||
"shape": "resample" if scenario["workloads"]["wav_source_rate"]
|
||||
!= scenario["rate_hz"] else "passthrough",
|
||||
"source_rate_hz": scenario["workloads"]["wav_source_rate"],
|
||||
"source_precision_bits": 16,
|
||||
"source_channels": 2,
|
||||
"source_frames_per_job": scenario["workloads"]["wav_source_frames"],
|
||||
"output_frames_per_job": scenario["workloads"]["wav_frames"],
|
||||
"sha256": scenario["workloads"]["wav_sha256"],
|
||||
},
|
||||
"flac": {
|
||||
"shape": "mono-constant",
|
||||
"rate_hz": scenario["rate_hz"],
|
||||
"precision_bits": scenario["precision_bits"],
|
||||
"block_frames": scenario["workloads"]["flac_frames"],
|
||||
"output_frames_per_job": scenario["workloads"]["flac_frames"],
|
||||
"sha256": scenario["workloads"]["flac_sha256"],
|
||||
},
|
||||
},
|
||||
}
|
||||
for scenario in scenarios
|
||||
],
|
||||
"scenarios": scenario_results,
|
||||
"aggregate_counts": aggregate,
|
||||
"full_gate_projection": projection,
|
||||
"artifacts": {
|
||||
"build_dir": str(build_dir),
|
||||
"apumc_hex": str(apumc_hex),
|
||||
"apum_hex": str(apum_hex),
|
||||
"report": str(output),
|
||||
},
|
||||
"errors": failures,
|
||||
"wall_seconds": round(time.monotonic() - started, 3),
|
||||
}
|
||||
_write_json(output, report)
|
||||
print(f"APU-P7 concurrent report: {output} ({'passed' if not failures else 'failed'})")
|
||||
return 0 if not failures else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,625 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run the locked APU-P7 KWS 1000-window corpus through the production RTL path."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from apu_kws import ( # noqa: E402
|
||||
APUM_BYTES,
|
||||
APUM_PAYLOAD_CRC,
|
||||
APUM_SHA256,
|
||||
CORPUS_SHA256,
|
||||
TFLITE_SHA256,
|
||||
image_sha256,
|
||||
parse_apum,
|
||||
)
|
||||
from apu_kws_convert import TfliteError, import_tflite # noqa: E402
|
||||
from setup_apu_kws_reference import REQUIRED_ARCHIVE, REQUIRED_SOURCES # noqa: E402
|
||||
|
||||
MANIFEST = ROOT / "docs/ip/apu-kws-corpus.tsv"
|
||||
PCM_SOURCES = ROOT / ".cache/retrosoc/sources/apu-kws-pcm"
|
||||
KWS01 = ROOT / ".cache/retrosoc/sources/apu-kws-mfcc/datasets/kws01"
|
||||
Y_LABELS = KWS01 / "y_labels.csv"
|
||||
Y_LABELS_SHA256 = "6cb3709762d53fe64eb00eaf4a75796248bb3c1386f1d1f93ccdb07b071e9698"
|
||||
AGGREGATE_PCM_SHA256 = "abd1ff2c988ac168dd68c29f934d95114d1094d7818d79fe73adb166ac009bee"
|
||||
TFLITE = (
|
||||
ROOT
|
||||
/ ".cache/retrosoc/sources/apu-mlperf-tiny/benchmark/training/keyword_spotting"
|
||||
/ "trained_models/kws_ref_model.tflite"
|
||||
)
|
||||
FIXTURE = ROOT / "tests/rtl/apu_p7_accuracy_tb.sv"
|
||||
ROM = ROOT / "rtl/ip/multimedia/apu_kws_rom.svh"
|
||||
WINDOW_COUNT = 1000
|
||||
WINDOW_BYTES = 32000
|
||||
WINDOW_SAMPLES = 16000
|
||||
MINIMUM_TOP1 = 900
|
||||
CLASS_COUNT = 12
|
||||
SETUP_HINT = "install the locked P7 inputs with `make setup-apu-kws-reference`"
|
||||
RESULT_PATTERN = re.compile(r"^KWS_RESULT (\d+) (\d+) (\d+) (\d+)$", re.MULTILINE)
|
||||
COMPLETE_PATTERN = re.compile(r"^APU_P7_ACCURACY_COMPLETE windows=(\d+) first=(\d+)$", re.MULTILINE)
|
||||
VARIANT_PATTERN = re.compile(r"^(?P<profile>.+)-\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-[0-9a-f]{6,}$")
|
||||
|
||||
|
||||
class QualificationError(RuntimeError):
|
||||
"""A locked input, tool, or simulation step failed closed."""
|
||||
|
||||
|
||||
def _sha256_bytes(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as source:
|
||||
for block in iter(lambda: source.read(1024 * 1024), b""):
|
||||
digest.update(block)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _write_json(path: Path, value: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
temporary.replace(path)
|
||||
|
||||
|
||||
def _require_file(path: Path, description: str) -> Path:
|
||||
if not path.is_file():
|
||||
raise QualificationError(f"missing {description}: {path}; {SETUP_HINT}")
|
||||
return path
|
||||
|
||||
|
||||
def _git_identity(paths: list[Path] | None = None) -> dict[str, Any]:
|
||||
command = ["git", "-C", str(ROOT)]
|
||||
revision = subprocess.run(
|
||||
[*command, "rev-parse", "HEAD"], text=True, capture_output=True, check=False
|
||||
)
|
||||
status_command = [*command, "status", "--porcelain"]
|
||||
if paths is not None:
|
||||
status_command.extend(("--", *[str(path) for path in paths]))
|
||||
status = subprocess.run(status_command, text=True, capture_output=True, check=False)
|
||||
return {
|
||||
"git_revision": revision.stdout.strip() if revision.returncode == 0 else None,
|
||||
"git_dirty": (bool(status.stdout.strip()) if status.returncode == 0 else None),
|
||||
}
|
||||
|
||||
|
||||
def _tool_version(command: list[str]) -> str | None:
|
||||
result = subprocess.run(command, text=True, capture_output=True, check=False)
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
line = (result.stdout or result.stderr).splitlines()
|
||||
return line[0].strip() if line else None
|
||||
|
||||
|
||||
def _load_manifest() -> list[dict[str, Any]]:
|
||||
_require_file(MANIFEST, "locked PCM manifest")
|
||||
raw = MANIFEST.read_bytes()
|
||||
text = raw.replace(b"\r\n", b"\n") if b"\r\n" in raw else raw
|
||||
if _sha256_bytes(text) != CORPUS_SHA256:
|
||||
raise QualificationError(
|
||||
f"stale locked PCM manifest: {MANIFEST} canonical-text SHA-256 mismatch; {SETUP_HINT}"
|
||||
)
|
||||
lines = text.decode("ascii").split("\n")
|
||||
if lines[-1] != "":
|
||||
raise QualificationError("locked PCM manifest is missing its trailing LF")
|
||||
rows = lines[:-1]
|
||||
if len(rows) != WINDOW_COUNT:
|
||||
raise QualificationError(f"locked PCM manifest has {len(rows)} rows, expected 1000")
|
||||
records: list[dict[str, Any]] = []
|
||||
for index, row in enumerate(rows):
|
||||
fields = row.split("\t")
|
||||
if len(fields) != 6:
|
||||
raise QualificationError(f"locked PCM manifest row {index} does not have 6 columns")
|
||||
bin_name, relpath, class_text, wav_sha256, pcm_sha256 = fields[1:]
|
||||
if fields[0] != f"{index:06d}":
|
||||
raise QualificationError(f"locked PCM manifest row {index} is out of order")
|
||||
if relpath.startswith("/") or ".." in Path(relpath).parts:
|
||||
raise QualificationError(f"locked PCM manifest row {index} has an unsafe path")
|
||||
class_id = int(class_text)
|
||||
if not 0 <= class_id < CLASS_COUNT:
|
||||
raise QualificationError(f"locked PCM manifest row {index} class out of range")
|
||||
records.append(
|
||||
{
|
||||
"index": index,
|
||||
"bin_name": bin_name,
|
||||
"relpath": relpath,
|
||||
"class_id": class_id,
|
||||
"wav_sha256": wav_sha256,
|
||||
"pcm_sha256": pcm_sha256,
|
||||
}
|
||||
)
|
||||
return records
|
||||
|
||||
|
||||
def _load_labels(records: list[dict[str, Any]]) -> list[int]:
|
||||
_require_file(Y_LABELS, "official y_labels.csv")
|
||||
if _sha256_file(Y_LABELS) != Y_LABELS_SHA256:
|
||||
raise QualificationError(f"stale official y_labels.csv: {Y_LABELS}; {SETUP_HINT}")
|
||||
lines = Y_LABELS.read_text(encoding="ascii").splitlines()
|
||||
if len(lines) != WINDOW_COUNT:
|
||||
raise QualificationError(f"official y_labels.csv has {len(lines)} rows, expected 1000")
|
||||
labels: list[int] = []
|
||||
for record, line in zip(records, lines):
|
||||
name, total, label = line.split(",")
|
||||
if name != record["bin_name"] or int(total) != CLASS_COUNT:
|
||||
raise QualificationError(
|
||||
f"official y_labels.csv row {record['index']} does not match the PCM manifest"
|
||||
)
|
||||
if int(label) != record["class_id"]:
|
||||
raise QualificationError(
|
||||
f"official y_labels.csv row {record['index']} label disagrees with the PCM manifest"
|
||||
)
|
||||
labels.append(int(label))
|
||||
return labels
|
||||
|
||||
|
||||
def _padded_pcm(record: dict[str, Any]) -> bytes:
|
||||
source = _require_file(PCM_SOURCES / record["relpath"], "locked source WAV")
|
||||
wav = source.read_bytes()
|
||||
if _sha256_bytes(wav) != record["wav_sha256"]:
|
||||
raise QualificationError(f"source WAV SHA-256 mismatch: {source}; {SETUP_HINT}")
|
||||
if len(wav) < 12 or wav[0:4] != b"RIFF" or wav[8:12] != b"WAVE":
|
||||
raise QualificationError(f"source WAV is not RIFF/WAVE: {source}")
|
||||
fmt: bytes | None = None
|
||||
data: bytes | None = None
|
||||
offset = 12
|
||||
while offset + 8 <= len(wav):
|
||||
chunk_size = struct.unpack_from("<I", wav, offset + 4)[0]
|
||||
body = wav[offset + 8 : offset + 8 + chunk_size]
|
||||
if len(body) != chunk_size:
|
||||
raise QualificationError(f"source WAV chunk is truncated: {source}")
|
||||
if wav[offset : offset + 4] == b"fmt ":
|
||||
fmt = body
|
||||
elif wav[offset : offset + 4] == b"data":
|
||||
data = body
|
||||
offset += 8 + chunk_size + (chunk_size & 1)
|
||||
if fmt is None or len(fmt) < 16 or data is None:
|
||||
raise QualificationError(f"source WAV lacks fmt/data chunks: {source}")
|
||||
audio_format, channels, rate, _byte_rate, _align, bits = struct.unpack_from("<HHIIHH", fmt)
|
||||
if (audio_format, channels, rate, bits) != (1, 1, WINDOW_SAMPLES, 16):
|
||||
raise QualificationError(f"source WAV is not mono/16000/S16 PCM: {source}")
|
||||
if len(data) > WINDOW_BYTES or len(data) % 2 != 0:
|
||||
raise QualificationError(f"source WAV exceeds one 16000-sample window: {source}")
|
||||
padded = data + bytes(WINDOW_BYTES - len(data))
|
||||
if _sha256_bytes(padded) != record["pcm_sha256"]:
|
||||
raise QualificationError(f"padded PCM SHA-256 mismatch for row {record['index']}: {source}")
|
||||
return padded
|
||||
|
||||
|
||||
def _write_hex(path: Path, payload: bytes) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(
|
||||
"".join(
|
||||
f"{int.from_bytes(payload[offset : offset + 4], 'little'):08x}\n"
|
||||
for offset in range(0, len(payload), 4)
|
||||
),
|
||||
encoding="ascii",
|
||||
)
|
||||
|
||||
|
||||
def _production_sources() -> list[Path]:
|
||||
common = ROOT / "rtl/managed/clusterip/common/rtl"
|
||||
multimedia = ROOT / "rtl/ip/multimedia"
|
||||
return [
|
||||
common / "interface/axi4_stream_if.sv",
|
||||
multimedia / "apu_kws_sram_client.sv",
|
||||
multimedia / "apu_kws_engine.sv",
|
||||
FIXTURE,
|
||||
]
|
||||
|
||||
|
||||
def _rtl_inputs() -> list[Path]:
|
||||
return [
|
||||
*_production_sources(),
|
||||
ROOT / "rtl/ip/multimedia/apu_define.svh",
|
||||
ROM,
|
||||
Path(__file__).resolve(),
|
||||
]
|
||||
|
||||
|
||||
def _verification_sha256() -> str:
|
||||
digest = hashlib.sha256()
|
||||
for path in sorted(_rtl_inputs()):
|
||||
digest.update(path.relative_to(ROOT).as_posix().encode("utf-8"))
|
||||
digest.update(b"\0")
|
||||
digest.update(path.read_bytes())
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _subprocess_text(value: str | bytes | None) -> str:
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", errors="replace")
|
||||
return value or ""
|
||||
|
||||
|
||||
def _run_command(command: list[str], *, timeout: int | None = None) -> dict[str, Any]:
|
||||
started = time.monotonic()
|
||||
try:
|
||||
process = subprocess.run(command, check=False, capture_output=True, text=True, timeout=timeout)
|
||||
return {
|
||||
"command": command,
|
||||
"duration_seconds": round(time.monotonic() - started, 3),
|
||||
"exit_code": process.returncode,
|
||||
"stdout": process.stdout,
|
||||
"stderr": process.stderr,
|
||||
"timed_out": False,
|
||||
}
|
||||
except subprocess.TimeoutExpired as error:
|
||||
return {
|
||||
"command": command,
|
||||
"duration_seconds": round(time.monotonic() - started, 3),
|
||||
"exit_code": 124,
|
||||
"stdout": _subprocess_text(error.stdout),
|
||||
"stderr": _subprocess_text(error.stderr),
|
||||
"timed_out": True,
|
||||
}
|
||||
|
||||
|
||||
def _compile(build_dir: Path, simulator: str) -> dict[str, Any]:
|
||||
tools = {name: shutil.which(name) for name in ("sv2v", "iverilog", "vvp")}
|
||||
missing = [name for name, path in tools.items() if path is None]
|
||||
if missing:
|
||||
raise QualificationError(f"missing required simulators: {', '.join(missing)}")
|
||||
tools["verilator"] = shutil.which("verilator")
|
||||
if simulator == "verilator" and tools["verilator"] is None:
|
||||
raise QualificationError("--simulator verilator requires verilator on PATH")
|
||||
common = ROOT / "rtl/managed/clusterip/common/rtl"
|
||||
multimedia = ROOT / "rtl/ip/multimedia"
|
||||
filelist = build_dir / "apu_p7_accuracy.fl"
|
||||
filelist.write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"+define+SV_ASSRT_DISABLE",
|
||||
f"+incdir+{common}",
|
||||
f"+incdir+{common / 'interface'}",
|
||||
f"+incdir+{multimedia}",
|
||||
*(str(source) for source in _production_sources()),
|
||||
"",
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
converted = build_dir / "apu_p7_accuracy.v"
|
||||
conversion = _run_command(
|
||||
[
|
||||
sys.executable,
|
||||
str(ROOT / "rtl/mini/script/convt_sv2v.py"),
|
||||
"-f",
|
||||
str(filelist),
|
||||
"--output",
|
||||
str(converted),
|
||||
]
|
||||
)
|
||||
if conversion["exit_code"] != 0:
|
||||
raise QualificationError(f"APU-P7 accuracy sv2v failed: {conversion['stderr']}")
|
||||
simv = build_dir / "apu_p7_accuracy_simv"
|
||||
compile_result = _run_command(
|
||||
[tools["iverilog"], "-g2012", "-s", "apu_p7_accuracy_tb", "-o", str(simv), str(converted)]
|
||||
)
|
||||
if compile_result["exit_code"] != 0:
|
||||
raise QualificationError(f"APU-P7 accuracy Icarus compile failed: {compile_result['stderr']}")
|
||||
verilator_binary: Path | None = None
|
||||
verilator_log: dict[str, Any] | None = None
|
||||
if tools["verilator"] is not None:
|
||||
verilator_dir = build_dir / "verilator"
|
||||
ccache_dir = build_dir / "ccache"
|
||||
ccache_dir.mkdir(exist_ok=True)
|
||||
environment = {
|
||||
**os.environ,
|
||||
"CCACHE_DIR": str(ccache_dir),
|
||||
"CCACHE_TEMPDIR": str(ccache_dir),
|
||||
}
|
||||
started = time.monotonic()
|
||||
process = subprocess.run(
|
||||
[
|
||||
str(tools["verilator"]),
|
||||
"--binary",
|
||||
"--timing",
|
||||
"-O3",
|
||||
"-Wno-fatal",
|
||||
"--top-module",
|
||||
"apu_p7_accuracy_tb",
|
||||
"--Mdir",
|
||||
str(verilator_dir),
|
||||
str(converted),
|
||||
],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=environment,
|
||||
)
|
||||
verilator_log = {
|
||||
"command": list(process.args),
|
||||
"duration_seconds": round(time.monotonic() - started, 3),
|
||||
"exit_code": process.returncode,
|
||||
"stdout": process.stdout,
|
||||
"stderr": process.stderr,
|
||||
}
|
||||
if process.returncode != 0:
|
||||
if simulator == "verilator":
|
||||
raise QualificationError(
|
||||
f"APU-P7 accuracy Verilator compile failed: {process.stderr}"
|
||||
)
|
||||
else:
|
||||
verilator_binary = verilator_dir / "Vapu_p7_accuracy_tb"
|
||||
if simulator == "verilator" and verilator_binary is None:
|
||||
raise QualificationError("Verilator compile did not produce a simulation binary")
|
||||
return {
|
||||
"tools": tools,
|
||||
"filelist": filelist,
|
||||
"converted": converted,
|
||||
"simv": simv,
|
||||
"verilator": verilator_binary,
|
||||
"sv2v_log": conversion,
|
||||
"iverilog_log": compile_result,
|
||||
"verilator_log": verilator_log,
|
||||
}
|
||||
|
||||
|
||||
def _profile_name(build_dir: Path, override: str | None) -> str:
|
||||
if override:
|
||||
return override
|
||||
parts = build_dir.resolve().parts
|
||||
if len(parts) >= 4 and parts[-2:] == ("apu", "kws") and parts[-4] == "build":
|
||||
match = VARIANT_PATTERN.match(parts[-3])
|
||||
if match:
|
||||
return match.group("profile")
|
||||
return "standalone"
|
||||
|
||||
|
||||
def _parse_predictions(stdout: str, window_count: int) -> list[dict[str, int]]:
|
||||
predictions: dict[int, dict[str, int]] = {}
|
||||
for match in RESULT_PATTERN.finditer(stdout):
|
||||
index, class_id, score, hit = (int(value) for value in match.groups())
|
||||
if index in predictions:
|
||||
raise QualificationError(f"duplicated utterance index {index} in simulator output")
|
||||
if class_id >= CLASS_COUNT:
|
||||
raise QualificationError(f"simulator emitted invalid class {class_id}")
|
||||
predictions[index] = {"index": index, "predicted": class_id, "score": score, "hit": hit}
|
||||
complete = COMPLETE_PATTERN.search(stdout)
|
||||
if complete is None or int(complete.group(1)) != window_count or int(complete.group(2)) != 0:
|
||||
raise QualificationError("simulator did not emit the expected completion marker")
|
||||
expected = set(range(window_count))
|
||||
observed = set(predictions)
|
||||
if observed != expected:
|
||||
raise QualificationError(
|
||||
f"skipped utterances: missing {sorted(expected - observed)[:8]} "
|
||||
f"unexpected {sorted(observed - expected)[:8]}"
|
||||
)
|
||||
return [predictions[index] for index in sorted(predictions)]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--build-dir", type=Path, required=True)
|
||||
parser.add_argument("--max-windows", type=int, default=0,
|
||||
help="run only the first N locked windows (0 = all 1000)")
|
||||
parser.add_argument("--report", type=str, default="accuracy.json")
|
||||
parser.add_argument("--profile", type=str, default=None)
|
||||
parser.add_argument("--timeout-seconds", type=int, default=21600)
|
||||
parser.add_argument("--max-window-cycles", type=int, default=4000000)
|
||||
parser.add_argument(
|
||||
"--simulator",
|
||||
choices=("auto", "iverilog", "verilator"),
|
||||
default="auto",
|
||||
help="run simulator; auto prefers Verilator (Icarus is ~400x slower per window)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
if not 0 <= args.max_windows <= WINDOW_COUNT:
|
||||
raise SystemExit("--max-windows must be within 0..1000")
|
||||
window_count = WINDOW_COUNT if args.max_windows == 0 else args.max_windows
|
||||
build_dir = args.build_dir.resolve()
|
||||
started_utc = datetime.now(timezone.utc).isoformat()
|
||||
started = time.monotonic()
|
||||
report_path = build_dir / args.report
|
||||
|
||||
try:
|
||||
records = _load_manifest()
|
||||
labels = _load_labels(records)
|
||||
image = import_tflite(_require_file(TFLITE, "locked kws_ref_model.tflite"))
|
||||
parse_apum(image)
|
||||
pcm_dir = build_dir / "pcm"
|
||||
aggregate = hashlib.sha256()
|
||||
for record in records[:window_count]:
|
||||
padded = _padded_pcm(record)
|
||||
aggregate.update(padded)
|
||||
_write_hex(pcm_dir / f"pcm_{record['index']:06d}.hex", padded)
|
||||
aggregate_sha256 = aggregate.hexdigest()
|
||||
if window_count == WINDOW_COUNT and aggregate_sha256 != AGGREGATE_PCM_SHA256:
|
||||
raise QualificationError(
|
||||
"aggregate padded PCM SHA-256 does not match the frozen 1000-window value"
|
||||
)
|
||||
apum_hex = build_dir / "kws_apum.hex"
|
||||
_write_hex(apum_hex, image)
|
||||
built = _compile(build_dir, args.simulator)
|
||||
except (OSError, TfliteError, ValueError, QualificationError) as error:
|
||||
raise SystemExit(str(error)) from error
|
||||
|
||||
selected = (
|
||||
"verilator"
|
||||
if (args.simulator == "verilator" or (args.simulator == "auto" and built["verilator"]))
|
||||
else "icarus"
|
||||
)
|
||||
plusargs = [
|
||||
f"+APUM_HEX={apum_hex}",
|
||||
f"+PCM_DIR={pcm_dir}",
|
||||
f"+WINDOW_COUNT={window_count}",
|
||||
"+FIRST_INDEX=0",
|
||||
f"+MAX_WINDOW_CYCLES={args.max_window_cycles}",
|
||||
]
|
||||
command = (
|
||||
[str(built["verilator"]), *plusargs]
|
||||
if selected == "verilator"
|
||||
else [str(built["tools"]["vvp"]), str(built["simv"]), *plusargs]
|
||||
)
|
||||
execution = _run_command(command, timeout=args.timeout_seconds)
|
||||
log_path = build_dir / "sim.log"
|
||||
log_path.write_text(execution["stdout"] + execution["stderr"], encoding="utf-8")
|
||||
|
||||
failure: str | None = None
|
||||
predictions: list[dict[str, int]] = []
|
||||
if execution["timed_out"]:
|
||||
failure = f"simulator timeout after {args.timeout_seconds} seconds"
|
||||
elif execution["exit_code"] != 0:
|
||||
failure = f"simulator failed with exit code {execution['exit_code']}"
|
||||
else:
|
||||
try:
|
||||
predictions = _parse_predictions(execution["stdout"], window_count)
|
||||
except QualificationError as error:
|
||||
failure = str(error)
|
||||
|
||||
for prediction in predictions:
|
||||
prediction["expected"] = labels[prediction["index"]]
|
||||
correct = sum(1 for item in predictions if item["predicted"] == item["expected"])
|
||||
incorrect = [item["index"] for item in predictions if item["predicted"] != item["expected"]]
|
||||
per_class = [
|
||||
{"class_id": class_id, "total": 0, "correct": 0, "incorrect": 0}
|
||||
for class_id in range(CLASS_COUNT)
|
||||
]
|
||||
confusion = [[0] * CLASS_COUNT for _ in range(CLASS_COUNT)]
|
||||
for item in predictions:
|
||||
expected, predicted = item["expected"], item["predicted"]
|
||||
confusion[expected][predicted] += 1
|
||||
per_class[expected]["total"] += 1
|
||||
if expected == predicted:
|
||||
per_class[expected]["correct"] += 1
|
||||
else:
|
||||
per_class[expected]["incorrect"] += 1
|
||||
gate_applies = window_count == WINDOW_COUNT
|
||||
gate_passed = correct >= MINIMUM_TOP1
|
||||
success = failure is None and (not gate_applies or gate_passed)
|
||||
if failure is None and gate_applies and not gate_passed:
|
||||
failure = f"top-1 accuracy {correct}/1000 is below the required 900/1000"
|
||||
|
||||
rtl_inputs = _rtl_inputs()
|
||||
report: dict[str, Any] = {
|
||||
"schema_version": 1,
|
||||
"report": "accuracy",
|
||||
"target": "p7",
|
||||
"profile": _profile_name(build_dir, args.profile),
|
||||
"success": success,
|
||||
"failure": failure,
|
||||
"gate": {
|
||||
"window_count": WINDOW_COUNT,
|
||||
"minimum_top1": MINIMUM_TOP1,
|
||||
"applies": gate_applies,
|
||||
"passed": gate_passed if gate_applies else None,
|
||||
},
|
||||
"seed": 0,
|
||||
"command": [sys.executable, str(Path(__file__).resolve()), *sys.argv[1:]],
|
||||
"started_utc": started_utc,
|
||||
"duration_seconds": round(time.monotonic() - started, 3),
|
||||
"counts": {
|
||||
"windows_requested": window_count,
|
||||
"windows_completed": len(predictions),
|
||||
"top1_correct": correct,
|
||||
"top1_accuracy": round(correct / len(predictions), 6) if predictions else 0.0,
|
||||
"hits": sum(item["hit"] for item in predictions),
|
||||
"incorrect": len(incorrect),
|
||||
},
|
||||
"per_class": per_class,
|
||||
"confusion_matrix": confusion,
|
||||
"incorrect_indices": incorrect,
|
||||
"predictions": predictions,
|
||||
"inputs": {
|
||||
"manifest": {
|
||||
"path": str(MANIFEST.relative_to(ROOT)),
|
||||
"sha256": CORPUS_SHA256,
|
||||
},
|
||||
"y_labels": {"path": str(Y_LABELS.relative_to(ROOT)), "sha256": Y_LABELS_SHA256},
|
||||
"tflite": {"path": str(TFLITE.relative_to(ROOT)), "sha256": TFLITE_SHA256},
|
||||
"pcm_sources": {
|
||||
"path": str(PCM_SOURCES.relative_to(ROOT)),
|
||||
"windows": window_count,
|
||||
"source_sha256_verified": True,
|
||||
"padded_pcm_sha256_verified": True,
|
||||
"aggregate_padded_pcm_sha256": aggregate_sha256,
|
||||
},
|
||||
"reference_sources": REQUIRED_SOURCES,
|
||||
"reference_archives": REQUIRED_ARCHIVE,
|
||||
},
|
||||
"converter": {
|
||||
"path": "scripts/apu_kws_convert.py",
|
||||
"sha256": _sha256_file(ROOT / "scripts/apu_kws_convert.py"),
|
||||
"contract_revision": "apu-kws-convert/1.0.0",
|
||||
**_git_identity(),
|
||||
},
|
||||
"reference": {
|
||||
"path": "scripts/apu_kws.py",
|
||||
"sha256": _sha256_file(ROOT / "scripts/apu_kws.py"),
|
||||
},
|
||||
"rtl": {
|
||||
**_git_identity(rtl_inputs),
|
||||
"fixture": str(FIXTURE.relative_to(ROOT)),
|
||||
"verification_sha256": _verification_sha256(),
|
||||
"sources": [
|
||||
{"path": str(path.relative_to(ROOT)), "sha256": _sha256_file(path)}
|
||||
for path in rtl_inputs
|
||||
],
|
||||
},
|
||||
"model": {
|
||||
"apum_sha256": image_sha256(image),
|
||||
"payload_crc": f"0x{APUM_PAYLOAD_CRC:08x}",
|
||||
"bytes": APUM_BYTES,
|
||||
"frozen_sha256": APUM_SHA256,
|
||||
},
|
||||
"rom": {"path": str(ROM.relative_to(ROOT)), "sha256": _sha256_file(ROM)},
|
||||
"tools": {
|
||||
"python": sys.version.split()[0],
|
||||
"sv2v": _tool_version([str(built["tools"]["sv2v"]), "--version"]),
|
||||
"iverilog": _tool_version([str(built["tools"]["iverilog"]), "-V"]),
|
||||
"vvp": _tool_version([str(built["tools"]["vvp"]), "-V"]),
|
||||
"verilator": (
|
||||
_tool_version([str(built["tools"]["verilator"]), "--version"])
|
||||
if built["tools"]["verilator"] is not None
|
||||
else None
|
||||
),
|
||||
},
|
||||
"simulation": {
|
||||
"simulator": selected,
|
||||
"command": command,
|
||||
"exit_code": execution["exit_code"],
|
||||
"timed_out": execution["timed_out"],
|
||||
"duration_seconds": execution["duration_seconds"],
|
||||
"log": str(log_path),
|
||||
},
|
||||
"artifacts": {
|
||||
"build_dir": str(build_dir),
|
||||
"report": str(report_path),
|
||||
"apum_hex": str(apum_hex),
|
||||
"pcm_dir": str(pcm_dir),
|
||||
"filelist": str(built["filelist"]),
|
||||
"converted": str(built["converted"]),
|
||||
"simv": str(built["simv"]),
|
||||
"verilator": str(built["verilator"]) if built["verilator"] is not None else None,
|
||||
},
|
||||
}
|
||||
_write_json(report_path, report)
|
||||
if gate_applies:
|
||||
verdict = f"gate 900/1000 {'PASS' if gate_passed else 'FAIL'}"
|
||||
else:
|
||||
verdict = "subset, gate not applied"
|
||||
print(f"APU-P7 KWS accuracy: top-1 {correct}/{len(predictions)} ({verdict})")
|
||||
print(f"report: {report_path}")
|
||||
return 0 if success else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user