feat: replace the rom of apu into sram

This commit is contained in:
maksyuki committed 2026-09-25 08:35:07 +08:00
1 parent 9298a582a4
commit 92830d963d
74 files changed
+5126 -697

No files matched your search

+29
View File
@@ -26,6 +26,35 @@ Scripts are part of the build contract. Prefer existing helpers over ad-hoc
shell behavior, preserve structured JSON results, and keep setup/download
behavior controlled by `dependencies/dependencies.lock.json`.
`generate_apu_kws_rtl_constants.py` also emits the frozen APU-P9 APUC image
and 15-bank layout manifest. `apu_kws_coeff.py` owns the independent APUC 1.0
schema and release validator; generated assets remain below the selected
variant's `apu/coefficients/` directory.
`apu_p9_evidence.py` initializes or assembles the six frozen P9 evidence files;
missing qualifying runs stay explicitly `unrun` rather than becoming passes.
`apu_p9_memory_ab.py` performs the like-for-like inferred-memory, macro-count,
synthesis-time and process-tree peak-RSS comparison from two explicit variant
roots; it never creates or switches Git worktrees itself.
The frozen P9 qualification entry points are explicit and do not turn smoke
runs into release evidence:
```sh
make CONFIG=configs/ci/ihp130.mk setup-apu-kws-reference apu-p9-coefficients apu-p9-evidence
python3 scripts/run_apu_p7_kws_rtl.py \
--build-dir build/<candidate>/apu/p9/accuracy --profile ihp130-p9
python3 scripts/run_apu_p7_concurrent_rtl.py \
--build-dir build/<candidate>/apu/p9/concurrent --seconds 60
make CONFIG=configs/ci/ihp130.mk apu-p9-memory-ab \
APU_P9_BASELINE_ROOT=build/<baseline> APU_P9_CANDIDATE_ROOT=build/<candidate>
```
Both A/B roots must come from clean committed revisions; the baseline is the
frozen pre-P9 revision and both configurations must enable P7 with the same
IHP130 tool, PDK, recipe, and dependency inputs. Assemble the six final reports
with `scripts/apu_p9_evidence.py assemble`; a `passed` input is accepted only
when its report-specific checks and named source artifacts are complete.
`setup_npu_reference.py` prepares the locked KWS/VWW inputs and the existing
TensorFlow/gemmlowp reference sources. `npu_framework_reference.py` builds the
host-only adapter in `tests/cpp/npu_framework_reference.cc`; its integer kernels
+63 -8
View File
@@ -51,6 +51,7 @@ ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import apu_kws # noqa: E402
import apu_kws_coeff # noqa: E402
from apu_isa import ( # noqa: E402
APUMC_ABI_V2,
ControlOpcode,
@@ -73,6 +74,7 @@ SECTION_ORDER = (
"formats",
"apumc",
"apum",
"apuc",
"isa",
"mp3_stub",
)
@@ -147,16 +149,26 @@ 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),
"IP_VERSION_VALUE": _hex_literal(
r"localparam logic \[31:0\] IpVersion = EnableP7 \? 32'h([0-9a-fA-F_]+) :",
text,
),
"CAPABILITY0_IMPLEMENTED": _hex_literal(
r"localparam logic \[31:0\] Capability0 = EnableP7 \? 32'h([0-9a-fA-F_]+) :",
text,
),
"CAPABILITY1_IMPLEMENTED": _hex_literal(
r"localparam logic \[31:0\] Capability1 = \(EnableP5 \|\| EnableP7\) \? "
r"32'h([0-9a-fA-F_]+) :",
text,
),
"IRQ_ALL": _hex_literal(
r"localparam logic \[10:0\] IrqMask = 11'h([0-9a-fA-F_]+);", text
r"localparam logic \[11:0\] IrqMask = EnableP7 \? 12'h([0-9a-fA-F_]+) :",
text,
),
}
@@ -350,6 +362,32 @@ def canonical_inventory() -> dict[str, dict[str, int]]:
{name: apb[name] for name in ("APUM_MAGIC", "APUM_ABI", "APUM_IMAGE_BYTES",
"APUM_PAYLOAD_CRC")},
)
_require_equal(
"APUC container/SVH",
{
"APUC_MAGIC": apu_kws_coeff.APUC_MAGIC,
"APUC_ABI": apu_kws_coeff.APUC_ABI,
"APUC_IMAGE_BYTES": apu_kws_coeff.APUC_BYTES,
"APUC_HEADER_BYTES": apu_kws_coeff.APUC_HEADER_BYTES,
"APUC_PAYLOAD_BYTES": apu_kws_coeff.APUC_PAYLOAD_BYTES,
"APUC_PAYLOAD_CRC": apu_kws_coeff.APUC_PAYLOAD_CRC,
"APUC_COEFFICIENT_ID_LO": apu_kws_coeff.APUC_COEFFICIENT_ID[0],
"APUC_COEFFICIENT_ID_HI": apu_kws_coeff.APUC_COEFFICIENT_ID[1],
},
{
name: apb[name]
for name in (
"APUC_MAGIC",
"APUC_ABI",
"APUC_IMAGE_BYTES",
"APUC_HEADER_BYTES",
"APUC_PAYLOAD_BYTES",
"APUC_PAYLOAD_CRC",
"APUC_COEFFICIENT_ID_LO",
"APUC_COEFFICIENT_ID_HI",
)
},
)
discovery_c = _c_discovery_defines()
discovery_rtl = _rtl_discovery_values()
_require_equal("discovery C/RTL", discovery_rtl,
@@ -400,6 +438,22 @@ def canonical_inventory() -> dict[str, dict[str, int]]:
"parameter_end": apu_kws.APUM_PARAMETER_END,
"payload_crc": apu_kws.APUM_PAYLOAD_CRC,
}
apuc = {
"magic": apu_kws_coeff.APUC_MAGIC,
"abi": apu_kws_coeff.APUC_ABI,
"image_bytes": apu_kws_coeff.APUC_BYTES,
"header_bytes": apu_kws_coeff.APUC_HEADER_BYTES,
"payload_bytes": apu_kws_coeff.APUC_PAYLOAD_BYTES,
"profile": apu_kws_coeff.APUC_PROFILE,
"layout_id": apu_kws_coeff.APUC_LAYOUT_ID,
"bank_count": apu_kws_coeff.APUC_BANK_COUNT,
"table_count": apu_kws_coeff.APUC_TABLE_COUNT,
"payload_crc": apu_kws_coeff.APUC_PAYLOAD_CRC,
"coefficient_id_low": apu_kws_coeff.APUC_COEFFICIENT_ID[0],
"coefficient_id_high": apu_kws_coeff.APUC_COEFFICIENT_ID[1],
"associated_apum_crc": apu_kws_coeff.APUC_APUM_PAYLOAD_CRC,
"ln2_q24": apu_kws_coeff.APUC_LN2_Q24,
}
discovery = {
"ip_id": discovery_c["IP_ID_VALUE"],
@@ -425,6 +479,7 @@ def canonical_inventory() -> dict[str, dict[str, int]]:
"formats": format_section,
"apumc": apumc,
"apum": apum,
"apuc": apuc,
"isa": isa,
"mp3_stub": _mp3_stub_section(apb),
}
@@ -465,16 +520,16 @@ def rtl_implemented_digest() -> int:
def c_implemented_digest() -> int:
"""Extract the handwritten P7 release digest from apu_regs.h."""
"""Extract the handwritten P9 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*$",
r"^#define\s+RS_APU_DIGEST_P9_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}")
raise ValueError(f"RS_APU_DIGEST_P9_IMPLEMENTED not found in {C_HEADER.name}")
return int(match.group(1), 16)
+228
View File
@@ -0,0 +1,228 @@
"""Frozen APUC 1.0 coefficient-container definitions and validator."""
from __future__ import annotations
import hashlib
import struct
import zlib
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Final
APUC_MAGIC: Final = 0x43555041
APUC_ABI: Final = 0x00010000
APUC_HEADER_BYTES: Final = 64
APUC_PAYLOAD_BYTES: Final = 15 * 1024 * 4
APUC_BYTES: Final = APUC_HEADER_BYTES + APUC_PAYLOAD_BYTES
APUC_PROFILE: Final = 1
APUC_LAYOUT_ID: Final = 1
APUC_BANK_COUNT: Final = 15
APUC_TABLE_COUNT: Final = 10
APUC_PAYLOAD_CRC: Final = 0x25E7C27D
APUC_COEFFICIENT_ID: Final = (0x8A0B038D, 0x806C780D)
APUC_APUM_PAYLOAD_CRC: Final = 0xB9034B22
APUC_LN2_Q24: Final = 11629080
APUC_LOGICAL_SHA256: Final = "f416585119e488c82e5cea1dabb17424e33a16dd8041393135903a613edced46"
APUC_PAYLOAD_SHA256: Final = "8d030b8a0d786c80a7fa072439700a4485711971d79a6a62e78fcafb1657d83a"
APUC_SHA256: Final = "d670688442ea6efe1e638108157f55fc55c75218c8edbee931aaca3497567818"
APUC_TABLES: Final = {
"hann": (480, "5fb8748b60ff53c721d3f97c2516b9f77aebd8bb15103a072219505769d4af61"),
"twiddle_real": (256, "49faa57826aa487032eef985d7d25f29b5346692f22ab9780209cb118c9c5b8a"),
"twiddle_imag": (256, "3552f5ed86694388ee674dcb988509e1f8fec5c61c7435e289a27591a2c817ec"),
"mel": (10280, "a13dc9eaf00e6e2be9710c4a1bd432239c0181b5d2ee9903250c667d906f16be"),
"dct": (400, "93a71d890110064feb515cb7527740ce3ff0116451931e27925ac1c36339153e"),
"log": (1025, "20b0b265de2c9e319dc70b2fa0ee136add8693937108c5f0c65ccc0ca146050a"),
"fir3": (63, "77ccbd59d8346f709d299ce762c1976dd8abbba3263d47ed65ef394db570870a"),
"fir6": (63, "a00ceab9105dddbed5e6d2426753dd5dbc99c9b0c166b1f19f43c530ceb9faac"),
"softmax": (125, "410f11dd7f10d5516b41efdec861afe6a7f2a34dd116163f7758ac5110b65ae7"),
"apum_profile": (
1496,
"4c30d9a5e054436bab5acfb4651ea361d0211956c9211e9d36d182b9fdc88991",
),
}
def _expected_locations() -> dict[str, list[tuple[int, int]]]:
locations: dict[str, list[tuple[int, int]]] = {
name: [] for name in APUC_TABLES
}
locations["hann"] = [(0, index) for index in range(480)]
locations["dct"] = [(0, 480 + index) for index in range(400)]
locations["fir3"] = [(0, 880 + index) for index in range(63)]
locations["fir6"] = [(0, 943 + index) for index in range(63)]
locations["twiddle_real"] = [(1, index) for index in range(256)]
locations["twiddle_imag"] = [(2, index) for index in range(256)]
locations["log"] = [(1 + (index & 1), 256 + (index >> 1)) for index in range(1025)]
locations["mel"] = [
(3 + (index >> 10), index & 1023) if index < 10240 else (13, index - 10240)
for index in range(10280)
]
locations["apum_profile"] = [
(13, 40 + index) if index < 984 else (14, index - 984)
for index in range(1496)
]
locations["softmax"] = [(14, 512 + index) for index in range(125)]
return locations
def _expected_profile_offsets() -> list[int]:
offsets = list(range(0, 0x500, 4))
for start in (0x1000, 0x1540, 0x2840, 0x2D80, 0x4080, 0x45C0, 0x58C0, 0x5E00, 0x7100):
offsets.extend(range(start, start + 0x200, 4))
offsets.extend(range(0x7630, 0x7690, 4))
return offsets
def validate_layout_manifest(image: bytes, manifest: dict[str, Any]) -> None:
"""Validate the generated layout with a frozen oracle independent of its packer."""
validate_apuc(image)
if manifest.get("schema_version") != 1:
raise ValueError("APUC layout manifest schema version is not 1")
generator = manifest.get("generator")
if not isinstance(generator, dict):
raise ValueError("APUC layout manifest lacks generator identity")
generator_path = generator.get("path")
generator_sha = generator.get("sha256")
generator_revision = generator.get("revision")
repository_commit = generator.get("repository_commit")
if generator_path != "scripts/generate_apu_kws_rtl_constants.py" or not isinstance(
generator_sha, str
):
raise ValueError("APUC generator path/SHA identity is incomplete")
if generator_revision != f"sha256:{generator_sha}":
raise ValueError("APUC generator revision does not match its source SHA")
if (
not isinstance(repository_commit, str)
or len(repository_commit) != 40
or any(character not in "0123456789abcdef" for character in repository_commit.lower())
):
raise ValueError("APUC generator repository revision is not a full Git commit")
source = Path(__file__).resolve().parents[1] / generator_path
if not source.is_file() or hashlib.sha256(source.read_bytes()).hexdigest() != generator_sha:
raise ValueError("APUC generator source does not match the recorded revision")
expected_container = {
"magic": f"0x{APUC_MAGIC:08x}",
"abi": f"0x{APUC_ABI:08x}",
"bytes": APUC_BYTES,
"header_bytes": APUC_HEADER_BYTES,
"payload_bytes": APUC_PAYLOAD_BYTES,
"payload_crc32": f"0x{APUC_PAYLOAD_CRC:08x}",
"coefficient_id": [f"0x{word:08x}" for word in APUC_COEFFICIENT_ID],
"associated_apum_crc32": f"0x{APUC_APUM_PAYLOAD_CRC:08x}",
"logical_sha256": APUC_LOGICAL_SHA256,
"payload_sha256": APUC_PAYLOAD_SHA256,
"sha256": APUC_SHA256,
}
if manifest.get("container") != expected_container:
raise ValueError("APUC container manifest does not match the frozen identity")
if manifest.get("layout") != {
"geometry": "tc_sram_1024x32",
"banks": APUC_BANK_COUNT,
"words_per_bank": 1024,
"populated_words": 14444,
"zero_padding_words": 916,
}:
raise ValueError("APUC physical layout summary does not match the frozen geometry")
if manifest.get("apum_profile_offsets") != _expected_profile_offsets():
raise ValueError("APUC APUM-profile offsets do not match the independent oracle")
table_records = manifest.get("tables")
if not isinstance(table_records, dict) or set(table_records) != set(APUC_TABLES):
raise ValueError("APUC layout manifest does not contain the exact ten tables")
payload_words = struct.unpack(f"<{APUC_BANK_COUNT * 1024}I", image[APUC_HEADER_BYTES:])
expected_locations = _expected_locations()
occupied: set[tuple[int, int]] = set()
for name, (word_count, frozen_sha) in APUC_TABLES.items():
record = table_records.get(name)
locations = expected_locations[name]
expected_record = {
"words": word_count,
"sha256_le32": frozen_sha,
"locations": [[bank, row] for bank, row in locations],
}
if record != expected_record:
raise ValueError(f"APUC table {name} does not match the independent map/hash oracle")
if any(location in occupied for location in locations):
raise ValueError(f"APUC table {name} overlaps another table")
occupied.update(locations)
table_bytes = b"".join(
struct.pack("<I", payload_words[(bank * 1024) + row]) for bank, row in locations
)
if hashlib.sha256(table_bytes).hexdigest() != frozen_sha:
raise ValueError(f"APUC payload words for {name} do not match the frozen hash")
if len(occupied) != 14444:
raise ValueError("APUC independent populated-position count is not 14444")
for bank in range(APUC_BANK_COUNT):
for row in range(1024):
if (bank, row) not in occupied and payload_words[(bank * 1024) + row] != 0:
raise ValueError(f"APUC padding is nonzero at bank {bank} row {row}")
@dataclass(frozen=True)
class ApucHeader:
magic: int
abi: int
total_bytes: int
payload_offset: int
payload_bytes: int
profile: int
layout_id: int
bank_count: int
table_count: int
payload_crc: int
coefficient_id: tuple[int, int]
associated_apum_crc: int
ln2_q24: int
def validate_apuc(image: bytes) -> ApucHeader:
"""Validate exact APUC 1.0 bytes and return the frozen header."""
if len(image) != APUC_BYTES:
raise ValueError(f"APUC image must be exactly {APUC_BYTES} bytes")
words = struct.unpack_from("<16I", image)
expected = (
APUC_MAGIC,
APUC_ABI,
APUC_BYTES,
APUC_HEADER_BYTES,
APUC_PAYLOAD_BYTES,
APUC_PROFILE,
APUC_LAYOUT_ID,
APUC_BANK_COUNT,
APUC_TABLE_COUNT,
APUC_PAYLOAD_CRC,
*APUC_COEFFICIENT_ID,
APUC_APUM_PAYLOAD_CRC,
APUC_LN2_Q24,
0,
0,
)
if words != expected:
raise ValueError("APUC header does not match the frozen profile")
payload = image[APUC_HEADER_BYTES:]
if zlib.crc32(payload) & 0xFFFFFFFF != APUC_PAYLOAD_CRC:
raise ValueError("APUC payload CRC mismatch")
if hashlib.sha256(payload).hexdigest() != APUC_PAYLOAD_SHA256:
raise ValueError("APUC payload SHA-256 mismatch")
if hashlib.sha256(image).hexdigest() != APUC_SHA256:
raise ValueError("APUC image SHA-256 mismatch")
return ApucHeader(
magic=words[0],
abi=words[1],
total_bytes=words[2],
payload_offset=words[3],
payload_bytes=words[4],
profile=words[5],
layout_id=words[6],
bank_count=words[7],
table_count=words[8],
payload_crc=words[9],
coefficient_id=(words[10], words[11]),
associated_apum_crc=words[12],
ln2_q24=words[13],
)
+342
View File
@@ -0,0 +1,342 @@
#!/usr/bin/env python3
"""Initialize or assemble fail-closed APU-P9 evidence reports."""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
from typing import Any
from apu_kws_coeff import validate_layout_manifest
REPORTS = (
"coefficient-layout.json",
"coefficient-lifecycle.json",
"memo-equivalence.json",
"numerical-latency.json",
"memory-synthesis-ab.json",
"physical.json",
)
ROOT = Path(__file__).resolve().parents[1]
REQUIRED_CHECKS = {
REPORTS[1]: (
"lp_admission",
"unsupported_discovery",
"cache_acl",
"dma_backpressure",
"payload_corruption",
"readback_corruption",
"partial_retry",
"lock",
"ownership",
"hard_reset",
"soft_reset",
"resource_reset",
"quiesce_load",
"abort_every_state",
),
REPORTS[2]: (
"valid_v1_v2",
"mutated_v1_v2",
"duplicate_keys",
"collisions",
"full_wrap",
"full_table",
"deeper_call_bypass",
"key_boundaries",
"bitmap_boundaries",
"interrupted_clear_insert",
"verdict_equivalence",
"first_error_equivalence",
"visit_count_equivalence",
),
REPORTS[3]: (
"all_coefficients",
"scalar_table_boundaries",
"variable_backpressure",
"no_double_commit",
"no_stale_response",
"mfcc_bit_exact",
"layers_bit_exact",
"full_1000_windows",
"complete_window_before_successor",
"concurrent_60s_zero_xrun",
),
REPORTS[4]: (
"configuration_matches",
"revision_qualified",
"candidate_synthesis_completed",
"target_inferred_bits_zero",
"apu_reduction_at_least_90_percent",
"candidate_apu_bits_at_most_65536",
"candidate_macro_inventory_76",
"pass_timing_reported",
"post_memory_inventory_reported",
"peak_rss_measured_both",
"candidate_peak_rss_lower",
"synthesis_duration_measured_both",
"candidate_duration_lower",
),
REPORTS[5]: (
"revision_qualified",
"macro_inventory_76",
"logic_macro_area_separate",
"wns_nonnegative",
"tns_zero",
"hold_reported",
"no_unconstrained_paths",
"macro_views_complete",
"netlist_models_complete",
),
}
BASELINE_REVISION = "0b73994772068e5fac00fb1a98f87f88ed121cdf"
def _read_json(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise ValueError(f"cannot read JSON evidence {path}: {error}") from error
if not isinstance(value, dict):
raise ValueError(f"JSON evidence is not an object: {path}")
return value
def _write_json(path: Path, value: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="ascii")
def _full_revision(value: object) -> bool:
return isinstance(value, str) and len(value) == 40 and all(
character in "0123456789abcdef" for character in value.lower()
)
def _require_artifacts(report: dict[str, Any], name: str) -> None:
artifacts = report.get("artifacts")
if not isinstance(artifacts, dict) or not artifacts:
raise ValueError(f"{name} passed report lacks named artifacts")
missing = [
key
for key, value in artifacts.items()
if not isinstance(value, str) or not Path(value).is_file()
]
if missing:
raise ValueError(f"{name} passed report has missing artifacts: {missing}")
def _require_passed_tests(report: dict[str, Any], name: str) -> None:
tests = report.get("tests")
if not isinstance(tests, list) or not tests or any(
not isinstance(test, dict) or test.get("status") != "passed" for test in tests
):
raise ValueError(f"{name} passed report lacks passed test records")
def _validate_memory_ab(report: dict[str, Any]) -> None:
baseline = report.get("baseline")
candidate = report.get("candidate")
tool = report.get("tool_identity")
if not isinstance(baseline, dict) or not isinstance(candidate, dict) or not isinstance(tool, dict):
raise ValueError("memory-synthesis-ab.json lacks baseline/candidate/tool identity")
if (
baseline.get("revision") != BASELINE_REVISION
or baseline.get("dirty") is not False
or candidate.get("revision") != report.get("revision")
or candidate.get("dirty") is not False
):
raise ValueError("memory-synthesis-ab.json revisions are not frozen and clean")
memory = candidate.get("memory")
macros = candidate.get("macros")
pass_timing = candidate.get("pass_timing")
post_memory = candidate.get("post_memory")
if (
not isinstance(memory, dict)
or memory.get("target_bits") != 0
or not isinstance(memory.get("apu_bits"), int)
or memory["apu_bits"] > 65536
or not isinstance(macros, dict)
or macros.get("apu_wrappers") != 76
or not isinstance(pass_timing, dict)
or any(pass_timing.get(name) is None for name in ("memory", "opt_dff", "abc", "total_ns"))
or not isinstance(post_memory, dict)
or not isinstance(post_memory.get("cell_types"), dict)
):
raise ValueError("memory-synthesis-ab.json lacks the frozen memory/pass inventory")
yosys_path = Path(str(tool.get("yosys_path", "")))
yosys_sha = tool.get("yosys_sha256")
if not yosys_path.is_file() or not isinstance(yosys_sha, str):
raise ValueError("memory-synthesis-ab.json lacks the Yosys executable identity")
if hashlib.sha256(yosys_path.read_bytes()).hexdigest() != yosys_sha:
raise ValueError("memory-synthesis-ab.json Yosys executable SHA-256 does not match")
if tool.get("read_slang_provider") != "yosys-executable":
raise ValueError("memory-synthesis-ab.json does not identify the Slang provider")
def _validate_numerical(report: dict[str, Any]) -> None:
accuracy = report.get("accuracy")
concurrency = report.get("concurrency")
latency = report.get("latency")
if (
not isinstance(accuracy, dict)
or accuracy.get("windows") != 1000
or not isinstance(accuracy.get("top1_correct"), int)
or accuracy["top1_correct"] < 900
or not isinstance(concurrency, dict)
or concurrency.get("seconds_per_scenario", 0) < 60
or sorted(concurrency.get("rates", [])) != [48000, 96000]
or sorted(concurrency.get("precisions", [])) != [16, 24]
or concurrency.get("xrun_count") != 0
or not isinstance(latency, dict)
or not isinstance(latency.get("maximum_window_cycles"), int)
or latency["maximum_window_cycles"] > 4_800_000
):
raise ValueError("numerical-latency.json lacks the frozen accuracy/latency/concurrency result")
def _validate_physical(report: dict[str, Any]) -> None:
inventory = report.get("macro_inventory")
area = report.get("area")
timing = report.get("timing")
if (
not isinstance(inventory, dict)
or inventory.get("wrappers") != 76
or not isinstance(area, dict)
or any(not isinstance(area.get(name), (int, float)) for name in ("logic_um2", "macro_um2", "total_um2"))
or not isinstance(timing, dict)
or not isinstance(timing.get("wns_ns"), (int, float))
or timing["wns_ns"] < 0
or timing.get("tns_ns") != 0
or timing.get("unconstrained_paths") != 0
or not isinstance(timing.get("hold"), dict)
):
raise ValueError("physical.json lacks the frozen macro/area/timing result")
def _validate_passed(name: str, report: dict[str, Any]) -> None:
if name == REPORTS[0]:
if report.get("profile") != "configs/ci/ihp130.mk":
raise ValueError("coefficient-layout.json does not use configs/ci/ihp130.mk")
layout_source = Path(str(report.get("source", "")))
apuc_source = Path(str(report.get("apuc_source", "")))
if not layout_source.is_file() or not apuc_source.is_file():
raise ValueError("coefficient-layout.json lacks its layout/APUC source artifacts")
validate_layout_manifest(apuc_source.read_bytes(), _read_json(layout_source))
return
if report.get("schema_version") != 1 or report.get("phase") != "APU-P9":
raise ValueError(f"{name} passed report lacks the APU-P9 schema identity")
if report.get("profile") != "configs/ci/ihp130.mk":
raise ValueError(f"{name} passed report does not use configs/ci/ihp130.mk")
if not _full_revision(report.get("revision")) or report.get("dirty") is not False:
raise ValueError(f"{name} passed report is not revision-qualified")
checks = report.get("checks")
required = REQUIRED_CHECKS[name]
if not isinstance(checks, dict) or any(checks.get(check) is not True for check in required):
raise ValueError(f"{name} passed report lacks one or more frozen checks")
if name in (REPORTS[1], REPORTS[2]):
_require_passed_tests(report, name)
if name == REPORTS[2] and (
report.get("baseline_revision") != BASELINE_REVISION
or report.get("candidate_revision") != report.get("revision")
):
raise ValueError("memo-equivalence.json revisions do not match the frozen comparison")
if name == REPORTS[3]:
_validate_numerical(report)
elif name == REPORTS[4]:
_validate_memory_ab(report)
elif name == REPORTS[5]:
_validate_physical(report)
_require_artifacts(report, name)
def _initialize(args: argparse.Namespace) -> None:
layout = _read_json(args.layout)
try:
apuc = args.apuc.read_bytes()
except OSError as error:
raise ValueError(f"cannot read APUC evidence {args.apuc}: {error}") from error
validate_layout_manifest(apuc, layout)
profile_path = args.profile.resolve()
try:
profile = profile_path.relative_to(ROOT).as_posix()
except ValueError:
profile = str(profile_path)
layout["phase"] = "APU-P9"
layout["profile"] = profile
layout["status"] = "passed"
layout["source"] = str(args.layout.resolve())
layout["apuc_source"] = str(args.apuc.resolve())
_write_json(args.output_dir / REPORTS[0], layout)
requirements = {
REPORTS[1]: "APUC admission, corruption, retry, lock, reset, quiesce and abort matrix",
REPORTS[2]: "baseline versus macro memo verdict, first-error and visit-count equivalence",
REPORTS[3]: "bit-exact coefficient, MFCC/layer/window and bounded-cycle qualification",
REPORTS[4]: "same-input baseline/candidate inferred memory, runtime and peak-RSS comparison",
REPORTS[5]: "76-wrapper area, WNS/TNS, unconstrained-path and macro-view accounting",
}
for name, requirement in requirements.items():
destination = args.output_dir / name
if not destination.exists():
_write_json(
destination,
{
"schema_version": 1,
"phase": "APU-P9",
"status": "unrun",
"requirement": requirement,
"reason": "required qualifying run has not been supplied",
},
)
def _assemble(args: argparse.Namespace) -> None:
sources: dict[str, Path] = {}
for item in args.report:
name, separator, raw_path = item.partition("=")
if not separator or name not in REPORTS:
raise ValueError(f"report must be NAME=PATH for one of {REPORTS}: {item}")
sources[name] = Path(raw_path)
missing = [name for name in REPORTS if name not in sources]
if missing:
raise ValueError(f"missing required P9 reports: {missing}")
for name, source in sources.items():
report = _read_json(source)
status = report.get("status")
if status not in ("passed", "unrun", "failed", "timed_out"):
raise ValueError(f"{name} lacks a recognized fail-closed status")
if status == "passed":
_validate_passed(name, report)
elif not isinstance(report.get("reason", report.get("failure")), str):
raise ValueError(f"{name} {status} report lacks a reason/failure")
report["source"] = str(source.resolve())
_write_json(args.output_dir / name, report)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
initialize = subparsers.add_parser("initialize")
initialize.add_argument("--layout", type=Path, required=True)
initialize.add_argument("--apuc", type=Path, required=True)
initialize.add_argument("--profile", type=Path, required=True)
initialize.add_argument("--output-dir", type=Path, required=True)
initialize.set_defaults(handler=_initialize)
assemble = subparsers.add_parser("assemble")
assemble.add_argument("--report", action="append", default=[])
assemble.add_argument("--output-dir", type=Path, required=True)
assemble.set_defaults(handler=_assemble)
args = parser.parse_args()
try:
args.handler(args)
except ValueError as error:
parser.error(str(error))
return 0
if __name__ == "__main__":
raise SystemExit(main())
+306
View File
@@ -0,0 +1,306 @@
#!/usr/bin/env python3
"""Compare revision-qualified APU-P9 baseline/candidate synthesis artifacts."""
from __future__ import annotations
import argparse
import hashlib
import json
import shutil
from pathlib import Path
from typing import Any
BASELINE_REVISION = "0b73994772068e5fac00fb1a98f87f88ed121cdf"
TARGET_MODULES = (
"apu_kws_engine",
"apu_kws_model_loader",
"apu_microcode_loader",
"apu_kws_coeff_store",
"apu_proof_memo",
)
APU_PATH_MARKER = ".u_apb4_apu."
REQUIRED_CONFIGURATION = {
"PDK": "IHP130",
"APU_ENABLE_P7": "YES",
"HAVE_SRAM_MACRO": "YES",
"SYNTH": "YOSYS",
}
def _one(root: Path, pattern: str) -> Path:
matches = sorted(root.glob(pattern))
if len(matches) != 1:
raise ValueError(f"expected one {pattern} below {root}, found {len(matches)}")
return matches[0]
def _json(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise ValueError(f"cannot read {path}: {error}") from error
if not isinstance(value, dict):
raise ValueError(f"expected JSON object: {path}")
return value
def _memory_inventory(path: Path) -> dict[str, Any]:
report = _json(path)
modules = report.get("modules")
if not isinstance(modules, dict):
raise ValueError(f"Yosys stat report lacks modules: {path}")
rows: list[dict[str, Any]] = []
for name, values in modules.items():
if not isinstance(values, dict):
continue
bits = int(values.get("num_memory_bits", 0))
memories = int(values.get("num_memories", 0))
if bits == 0 and memories == 0:
continue
rows.append({"module": name, "memories": memories, "bits": bits})
apu_rows = [row for row in rows if APU_PATH_MARKER in row["module"]]
target_rows = [
row for row in apu_rows if any(name in row["module"] for name in TARGET_MODULES)
]
return {
"artifact": str(path.resolve()),
"whole_soc_bits": sum(row["bits"] for row in rows),
"apu_bits": sum(row["bits"] for row in apu_rows),
"target_bits": sum(row["bits"] for row in target_rows),
"apu_rows": sorted(apu_rows, key=lambda row: (-row["bits"], row["module"])),
"target_rows": sorted(target_rows, key=lambda row: (-row["bits"], row["module"])),
}
def _logic_inventory(path: Path) -> dict[str, Any]:
report = _json(path)
design = report.get("design")
if not isinstance(design, dict):
raise ValueError(f"post-memory report lacks design statistics: {path}")
cell_types = design.get("num_cells_by_type")
if not isinstance(cell_types, dict):
raise ValueError(f"post-memory report lacks cell types: {path}")
normalized = {str(name): int(count) for name, count in cell_types.items()}
return {
"artifact": str(path.resolve()),
"cells": int(design.get("num_cells", sum(normalized.values()))),
"flops": sum(count for name, count in normalized.items() if "DFF" in name.upper()),
"muxes": sum(count for name, count in normalized.items() if "MUX" in name.upper()),
"cell_types": normalized,
}
def _pass_timing(path: Path) -> dict[str, Any]:
report = _json(path)
passes = report.get("passes")
if not isinstance(passes, dict) or not isinstance(report.get("total_ns"), int):
raise ValueError(f"Yosys performance report is incomplete: {path}")
timings: dict[str, Any] = {
"artifact": str(path.resolve()),
"total_ns": report["total_ns"],
"generator": report.get("generator"),
}
for name in ("memory", "opt_dff", "abc"):
entry = passes.get(name)
timings[name] = int(entry["runtime_ns"]) if isinstance(entry, dict) and "runtime_ns" in entry else None
return timings
def _macro_inventory(path: Path) -> dict[str, Any]:
instances = [line.strip().replace("/", ".") for line in path.read_text().splitlines() if line.strip()]
apu = [instance for instance in instances if APU_PATH_MARKER in instance]
classes = {
"control_store": sum(".u_control_store." in value for value in apu),
"codec_common_data": sum(".u_local_sram." in value for value in apu),
"kws_model_scratch": sum(".u_kws_sram_client." in value for value in apu),
"loader_path_stack": sum(
(".u_microcode_loader." in value) and (".u_proof_memo." not in value) for value in apu
),
"verifier_memo": sum(".u_proof_memo." in value for value in apu),
"coefficient_store": sum(".u_kws_coeff_store." in value for value in apu),
}
return {
"artifact": str(path.resolve()),
"apu_wrappers": len(apu),
"classes": classes,
"instances": apu,
}
def _variant(root: Path) -> dict[str, Any]:
pre_memory_path = _one(root, "syn/yosys/rpt/*_pre_memory.json")
pre_tech_path = _one(root, "syn/yosys/rpt/*_pre_tech.json")
macro_path = _one(root, "syn/yosys/rpt/*_macros.rpt")
perf_path = _one(root, "syn/yosys/yosys-perf.json")
result_path = _one(root, "syn/yosys/result-synth.json")
manifest_path = _one(root, "meta/manifest.json")
memory = _memory_inventory(pre_memory_path)
result = _json(result_path)
manifest = _json(manifest_path)
repository = manifest.get("repository")
if not isinstance(repository, dict):
raise ValueError(f"manifest lacks repository identity: {manifest_path}")
return {
"root": str(root.resolve()),
"memory": memory,
"post_memory": _logic_inventory(pre_tech_path),
"pass_timing": _pass_timing(perf_path),
"synthesis": {
"status": result.get("status"),
"exit_code": result.get("exit_code"),
"duration_seconds": result.get("duration_seconds"),
"peak_rss_kib": result.get("peak_rss_kib"),
"result": str(result_path.resolve()),
},
"macros": _macro_inventory(macro_path),
"manifest": manifest,
"revision": repository.get("commit"),
"dirty": repository.get("dirty"),
"artifacts": {
"pre_memory": str(pre_memory_path.resolve()),
"pre_tech": str(pre_tech_path.resolve()),
"macros": str(macro_path.resolve()),
"yosys_perf": str(perf_path.resolve()),
"synthesis_result": str(result_path.resolve()),
"manifest": str(manifest_path.resolve()),
},
}
def _configuration_digest(variant: dict[str, Any]) -> Any:
manifest = variant.get("manifest")
if not isinstance(manifest, dict):
return None
comparable = {
"configuration": manifest.get("configuration"),
"dependency_lock": manifest.get("dependency_lock"),
"tools": manifest.get("tools"),
}
if any(value is None for value in comparable.values()):
return None
return json.dumps(comparable, sort_keys=True, separators=(",", ":"))
def _configuration_is_frozen(variant: dict[str, Any]) -> bool:
configuration = variant["manifest"].get("configuration")
return (
variant["manifest"].get("profile") == "ihp130"
and isinstance(configuration, dict)
and all(configuration.get(name) == value for name, value in REQUIRED_CONFIGURATION.items())
)
def _sha256(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 main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--baseline-root", type=Path, required=True)
parser.add_argument("--candidate-root", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--yosys", type=Path, default=None)
args = parser.parse_args()
try:
baseline = _variant(args.baseline_root)
candidate = _variant(args.candidate_root)
except ValueError as error:
parser.error(str(error))
baseline_bits = baseline["memory"]["apu_bits"]
candidate_bits = candidate["memory"]["apu_bits"]
reduction = 0.0 if baseline_bits == 0 else 1.0 - (candidate_bits / baseline_bits)
expected_macros = {
"control_store": 8,
"codec_common_data": 12,
"kws_model_scratch": 16,
"loader_path_stack": 8,
"verifier_memo": 17,
"coefficient_store": 15,
}
baseline_digest = _configuration_digest(baseline)
candidate_digest = _configuration_digest(candidate)
baseline_duration = baseline["synthesis"]["duration_seconds"]
candidate_duration = candidate["synthesis"]["duration_seconds"]
yosys_path = args.yosys
if yosys_path is None:
resolved_yosys = shutil.which("yosys")
if resolved_yosys is None:
parser.error("yosys executable is required to record its SHA-256")
yosys_path = Path(resolved_yosys)
if not yosys_path.is_file():
parser.error(f"yosys executable does not exist: {yosys_path}")
baseline_passes = baseline["pass_timing"]
candidate_passes = candidate["pass_timing"]
revision_qualified = (
baseline["revision"] == BASELINE_REVISION
and candidate["revision"] != BASELINE_REVISION
and isinstance(candidate["revision"], str)
and len(candidate["revision"]) == 40
and baseline["dirty"] is False
and candidate["dirty"] is False
)
checks = {
"configuration_matches": baseline_digest is not None
and baseline_digest == candidate_digest
and _configuration_is_frozen(baseline)
and _configuration_is_frozen(candidate),
"revision_qualified": revision_qualified,
"candidate_synthesis_completed": candidate["synthesis"]["status"] == "passed",
"target_inferred_bits_zero": candidate["memory"]["target_bits"] == 0,
"apu_reduction_at_least_90_percent": reduction >= 0.90,
"candidate_apu_bits_at_most_65536": candidate_bits <= 65536,
"candidate_macro_inventory_76": candidate["macros"]["apu_wrappers"] == 76
and candidate["macros"]["classes"] == expected_macros,
"pass_timing_reported": all(candidate_passes[name] is not None for name in ("memory", "opt_dff", "abc"))
and baseline_passes["memory"] is not None
and baseline_passes["opt_dff"] is not None,
"post_memory_inventory_reported": baseline["post_memory"]["cells"] > 0
and candidate["post_memory"]["cells"] > 0,
"peak_rss_measured_both": baseline["synthesis"]["peak_rss_kib"] is not None
and candidate["synthesis"]["peak_rss_kib"] is not None,
"candidate_peak_rss_lower": baseline["synthesis"]["peak_rss_kib"] is not None
and candidate["synthesis"]["peak_rss_kib"] is not None
and candidate["synthesis"]["peak_rss_kib"] < baseline["synthesis"]["peak_rss_kib"],
"synthesis_duration_measured_both": baseline_duration is not None
and candidate_duration is not None,
"candidate_duration_lower": baseline_duration is not None
and candidate_duration is not None
and candidate_duration < baseline_duration,
}
report = {
"schema_version": 1,
"phase": "APU-P9",
"profile": "configs/ci/ihp130.mk",
"revision": candidate["revision"],
"dirty": candidate["dirty"],
"status": "passed" if all(checks.values()) else "failed",
"reason": None if all(checks.values()) else "one or more frozen APU-P9 A/B checks failed",
"baseline": baseline,
"candidate": candidate,
"apu_inferred_reduction": reduction,
"expected_candidate_macro_classes": expected_macros,
"tool_identity": {
"yosys_path": str(yosys_path.resolve()),
"yosys_sha256": _sha256(yosys_path),
"read_slang_provider": "yosys-executable",
},
"artifacts": {
**{f"baseline_{name}": path for name, path in baseline["artifacts"].items()},
**{f"candidate_{name}": path for name, path in candidate["artifacts"].items()},
},
"checks": checks,
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="ascii")
print(f"APU-P9 memory A/B: {report['status']} ({args.output})")
return 0 if report["status"] == "passed" else 1
if __name__ == "__main__":
raise SystemExit(main())
+17 -2
View File
@@ -1,8 +1,9 @@
#!/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
Reads the deterministic P5 APUMC microcode bundle, the frozen P9 APUC
coefficient image, the 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
@@ -24,6 +25,7 @@ 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
from apu_kws_coeff import APUC_PAYLOAD_CRC, validate_apuc # noqa: E402
APUMC_MAGIC = 0x41504D43
APUMC_ABI_VERSIONS = (0x00010000, 0x00020000)
@@ -82,6 +84,12 @@ def _load_apum(path: Path) -> tuple[bytes, int]:
return image, _crc32(image)
def _load_apuc(path: Path) -> tuple[bytes, int]:
image = path.read_bytes()
validate_apuc(image)
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)
@@ -150,6 +158,7 @@ def _format_array(name: str, data: bytes) -> str:
def _emit(args: argparse.Namespace) -> str:
apumc, apumc_crc, apumc_payload_crc = _load_apumc(args.apumc)
apuc, apuc_crc = _load_apuc(args.apuc)
apum, apum_crc = _load_apum(args.apum)
wav = _synthesize_wav()
pcm_crc, pcm_bytes, pcm_frames = _expected_pcm(wav)
@@ -165,6 +174,9 @@ def _emit(args: argparse.Namespace) -> str:
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_APUC_SIZE UINT32_C({len(apuc)})",
f"#define RS_APU_RELEASE_APUC_CRC UINT32_C(0x{apuc_crc:08X})",
f"#define RS_APU_RELEASE_APUC_PAYLOAD_CRC UINT32_C(0x{APUC_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})",
@@ -184,6 +196,8 @@ def _emit(args: argparse.Namespace) -> str:
"",
_format_array("rs_apu_release_apumc", apumc),
"",
_format_array("rs_apu_release_apuc", apuc),
"",
_format_array("rs_apu_release_apum", apum),
"",
_format_array("rs_apu_release_wav", wav),
@@ -199,6 +213,7 @@ def _emit(args: argparse.Namespace) -> str:
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--apumc", type=Path, required=True, help="P5 APUMC bundle path")
parser.add_argument("--apuc", type=Path, required=True, help="P9 APUC coefficient image path")
parser.add_argument("--apum", type=Path, required=True, help="P7 APUM KWS model path")
parser.add_argument(
"--kws-corpus",
+197 -4
View File
@@ -1,20 +1,41 @@
#!/usr/bin/env python3
"""Generate the frozen APU-P7 frontend ROM and APUM validation constants."""
"""Generate frozen APU-P7 references and the APU-P9 coefficient image."""
from __future__ import annotations
import argparse
import hashlib
import json
import struct
import subprocess
import zlib
from decimal import Decimal, ROUND_HALF_EVEN, getcontext
from pathlib import Path
from apu_kws_coeff import (
APUC_ABI,
APUC_APUM_PAYLOAD_CRC,
APUC_BANK_COUNT,
APUC_BYTES,
APUC_COEFFICIENT_ID,
APUC_HEADER_BYTES,
APUC_LAYOUT_ID,
APUC_LOGICAL_SHA256,
APUC_MAGIC,
APUC_PAYLOAD_BYTES,
APUC_PAYLOAD_CRC,
APUC_PAYLOAD_SHA256,
APUC_PROFILE,
APUC_SHA256,
APUC_TABLE_COUNT,
)
getcontext().prec = 100
ZERO = Decimal(0)
ONE = Decimal(1)
TWO = Decimal(2)
def _atan(value: Decimal) -> Decimal:
square = value * value
term = value
@@ -224,6 +245,160 @@ def _fixed_apum_words(image: bytes) -> tuple[list[int], list[int]]:
return offsets, [struct.unpack_from("<I", image, offset)[0] for offset in offsets]
def _word_bytes(words: list[int]) -> bytes:
return b"".join(struct.pack("<I", value & 0xFFFFFFFF) for value in words)
def _coefficient_tables(image: bytes) -> tuple[list[tuple[str, list[int]]], list[int], list[int]]:
fixed_offsets, fixed_words = _fixed_apum_words(image)
twiddle_real, twiddle_imag = _twiddle()
tables = [
("hann", _hann()),
("twiddle_real", twiddle_real),
("twiddle_imag", twiddle_imag),
("mel", _mel_weights()),
("dct", _dct()),
("log", _log_rom()),
("fir3", _fir(3)),
("fir6", _fir(6)),
("softmax", _softmax_exp()),
("apum_profile", fixed_words),
]
return tables, fixed_offsets, fixed_words
def _pack_coefficient_banks(image: bytes) -> tuple[list[list[int]], dict[str, list[tuple[int, int]]]]:
tables, _fixed_offsets, fixed_words = _coefficient_tables(image)
by_name = dict(tables)
banks = [[0] * 1024 for _ in range(APUC_BANK_COUNT)]
inverse: dict[str, list[tuple[int, int]]] = {name: [] for name, _values in tables}
def place(name: str, index: int, bank: int, row: int) -> None:
banks[bank][row] = by_name[name][index] & 0xFFFFFFFF
inverse[name].append((bank, row))
for index in range(480):
place("hann", index, 0, index)
for index in range(400):
place("dct", index, 0, 480 + index)
for index in range(63):
place("fir3", index, 0, 880 + index)
place("fir6", index, 0, 943 + index)
for index in range(256):
place("twiddle_real", index, 1, index)
place("twiddle_imag", index, 2, index)
for index in range(1025):
bank = 1 + (index & 1)
place("log", index, bank, 256 + (index >> 1))
for index in range(10280):
if index < 10240:
place("mel", index, 3 + (index >> 10), index & 1023)
else:
place("mel", index, 13, index - 10240)
for index in range(len(fixed_words)):
if index < 984:
place("apum_profile", index, 13, 40 + index)
else:
place("apum_profile", index, 14, index - 984)
for index in range(125):
place("softmax", index, 14, 512 + index)
return banks, inverse
def build_apuc(image: bytes) -> tuple[bytes, dict[str, object]]:
"""Build the exact APUC 1.0 image and its reviewable layout manifest."""
if len(image) != 32768:
raise ValueError("APUM image must be exactly 32768 bytes")
tables, fixed_offsets, _fixed_words = _coefficient_tables(image)
banks, inverse = _pack_coefficient_banks(image)
logical = b"".join(_word_bytes(values) for _name, values in tables)
payload = b"".join(_word_bytes(bank) for bank in banks)
payload_crc = zlib.crc32(payload) & 0xFFFFFFFF
ln2_q24 = _rne(TWO.ln() * Decimal(1 << 24))
header = struct.pack(
"<16I",
APUC_MAGIC,
APUC_ABI,
APUC_BYTES,
APUC_HEADER_BYTES,
APUC_PAYLOAD_BYTES,
APUC_PROFILE,
APUC_LAYOUT_ID,
APUC_BANK_COUNT,
APUC_TABLE_COUNT,
payload_crc,
APUC_COEFFICIENT_ID[0],
APUC_COEFFICIENT_ID[1],
APUC_APUM_PAYLOAD_CRC,
ln2_q24,
0,
0,
)
apuc = header + payload
logical_sha = hashlib.sha256(logical).hexdigest()
payload_sha = hashlib.sha256(payload).hexdigest()
apuc_sha = hashlib.sha256(apuc).hexdigest()
if payload_crc != APUC_PAYLOAD_CRC:
raise ValueError(f"APUC payload CRC drift: 0x{payload_crc:08x}")
if logical_sha != APUC_LOGICAL_SHA256:
raise ValueError(f"APUC logical SHA-256 drift: {logical_sha}")
if payload_sha != APUC_PAYLOAD_SHA256:
raise ValueError(f"APUC payload SHA-256 drift: {payload_sha}")
if apuc_sha != APUC_SHA256:
raise ValueError(f"APUC image SHA-256 drift: {apuc_sha}")
populated = sum(len(locations) for locations in inverse.values())
generator_path = Path(__file__).resolve()
generator_sha = hashlib.sha256(generator_path.read_bytes()).hexdigest()
revision = subprocess.run(
["git", "-C", str(generator_path.parents[1]), "rev-parse", "HEAD"],
check=False,
capture_output=True,
text=True,
)
repository_commit = revision.stdout.strip() if revision.returncode == 0 else ""
manifest: dict[str, object] = {
"schema_version": 1,
"generator": {
"path": "scripts/generate_apu_kws_rtl_constants.py",
"revision": f"sha256:{generator_sha}",
"sha256": generator_sha,
"repository_commit": repository_commit,
},
"container": {
"magic": f"0x{APUC_MAGIC:08x}",
"abi": f"0x{APUC_ABI:08x}",
"bytes": len(apuc),
"header_bytes": APUC_HEADER_BYTES,
"payload_bytes": len(payload),
"payload_crc32": f"0x{payload_crc:08x}",
"coefficient_id": [f"0x{word:08x}" for word in APUC_COEFFICIENT_ID],
"associated_apum_crc32": f"0x{APUC_APUM_PAYLOAD_CRC:08x}",
"logical_sha256": logical_sha,
"payload_sha256": payload_sha,
"sha256": apuc_sha,
},
"layout": {
"geometry": "tc_sram_1024x32",
"banks": APUC_BANK_COUNT,
"words_per_bank": 1024,
"populated_words": populated,
"zero_padding_words": APUC_BANK_COUNT * 1024 - populated,
},
"tables": {
name: {
"words": len(values),
"sha256_le32": hashlib.sha256(_word_bytes(values)).hexdigest(),
"locations": [[bank, row] for bank, row in inverse[name]],
}
for name, values in tables
},
"apum_profile_offsets": fixed_offsets,
}
return apuc, manifest
def _emit_profile_words(offsets: list[int], values: list[int]) -> str:
ranges: list[tuple[int, int, int]] = []
range_start = offsets[0]
@@ -284,10 +459,18 @@ def _emit_profile_words(offsets: list[int], values: list[int]) -> str:
return "\n".join(lines)
def generate(apum: Path, output: Path, profile_output: Path) -> None:
def generate(
apum: Path,
output: Path,
profile_output: Path,
apuc_output: Path | None = None,
manifest_output: Path | None = None,
) -> None:
image = apum.read_bytes()
if len(image) != 32768:
raise ValueError("APUM image must be exactly 32768 bytes")
output.parent.mkdir(parents=True, exist_ok=True)
profile_output.parent.mkdir(parents=True, exist_ok=True)
fixed_offsets, fixed_words = _fixed_apum_words(image)
text = [
"// Copyright (c) 2026 Yuchi Miao <miaoyuchi@ict.ac.cn>",
@@ -328,6 +511,14 @@ def generate(apum: Path, output: Path, profile_output: Path) -> None:
),
encoding="ascii",
)
if (apuc_output is None) != (manifest_output is None):
raise ValueError("APUC output and manifest output must be requested together")
if apuc_output is not None and manifest_output is not None:
apuc, manifest = build_apuc(image)
apuc_output.parent.mkdir(parents=True, exist_ok=True)
manifest_output.parent.mkdir(parents=True, exist_ok=True)
apuc_output.write_bytes(apuc)
manifest_output.write_text(json.dumps(manifest, indent=2) + "\n", encoding="ascii")
def main() -> int:
@@ -335,11 +526,13 @@ def main() -> int:
parser.add_argument("--apum", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--profile-output", type=Path)
parser.add_argument("--apuc-output", type=Path)
parser.add_argument("--manifest-output", type=Path)
args = parser.parse_args()
profile_output = args.profile_output
if profile_output is None:
profile_output = args.output.with_name("apu_kws_apum_profile.svh")
generate(args.apum, args.output, profile_output)
generate(args.apum, args.output, profile_output, args.apuc_output, args.manifest_output)
return 0
+3
View File
@@ -112,8 +112,11 @@ def _production_sources() -> list[Path]:
"apu_ring_scheduler.sv",
"apu_stream_router.sv",
"apu_control_store.sv",
"apu_proof_memo.sv",
"apu_microcode_loader.sv",
"apu_local_sram.sv",
"apu_kws_coeff_store.sv",
"apu_kws_coeff_loader.sv",
"apu_kws_engine.sv",
"apu_kws_sram_client.sv",
"apu_kws_model_loader.sv",
+14 -1
View File
@@ -45,6 +45,7 @@ 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
from generate_apu_kws_rtl_constants import build_apuc # noqa: E402
PCLK_HZ = 48_000_000
FULL_GATE_SECONDS = 60.0
@@ -175,8 +176,11 @@ def _production_sources() -> list[Path]:
"apu_ring_scheduler.sv",
"apu_stream_router.sv",
"apu_control_store.sv",
"apu_proof_memo.sv",
"apu_microcode_loader.sv",
"apu_local_sram.sv",
"apu_kws_coeff_store.sv",
"apu_kws_coeff_loader.sv",
"apu_kws_engine.sv",
"apu_kws_sram_client.sv",
"apu_kws_model_loader.sv",
@@ -452,6 +456,7 @@ def _run_scenario(
compile_result["binary"],
f"+APUMC_HEX={artifacts['apumc_hex']}",
f"+APUM_HEX={artifacts['apum_hex']}",
f"+APUC_HEX={artifacts['apuc_hex']}",
f"+WAV_HEX={scenario_dir / 'wav.hex'}",
f"+FLAC_HEX={scenario_dir / 'flac.hex'}",
f"+WAV_PCM_HEX={scenario_dir / 'wav_pcm.hex'}",
@@ -658,7 +663,15 @@ def main() -> int:
_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": ""}
apuc_hex = build_dir / "kws-apuc.hex"
apuc_payload, _layout = build_apuc(apum_payload)
_hex_words(apuc_hex, apuc_payload, 15376)
artifacts = {
"apumc_hex": str(apumc_hex),
"apum_hex": str(apum_hex),
"apuc_hex": str(apuc_hex),
"scenario_dir": "",
}
started = time.monotonic()
compile_result = _compile(build_dir)
+7
View File
@@ -30,6 +30,7 @@ from apu_kws import ( # noqa: E402
parse_apum,
)
from apu_kws_convert import TfliteError, import_tflite # noqa: E402
from generate_apu_kws_rtl_constants import build_apuc # noqa: E402
from setup_apu_kws_reference import REQUIRED_ARCHIVE, REQUIRED_SOURCES # noqa: E402
MANIFEST = ROOT / "docs/ip/apu-kws-corpus.tsv"
@@ -220,7 +221,9 @@ def _production_sources() -> list[Path]:
return [
common / "interface/axi4_stream_if.sv",
multimedia / "apu_kws_sram_client.sv",
multimedia / "apu_kws_coeff_store.sv",
multimedia / "apu_kws_engine.sv",
ROOT / "tests/rtl/apu_kws_coeff_fixture.sv",
FIXTURE,
]
@@ -448,6 +451,9 @@ def main() -> int:
)
apum_hex = build_dir / "kws_apum.hex"
_write_hex(apum_hex, image)
apuc_hex = build_dir / "kws_apuc.hex"
apuc, _layout = build_apuc(image)
_write_hex(apuc_hex, apuc)
built = _compile(build_dir, args.simulator)
except (OSError, TfliteError, ValueError, QualificationError) as error:
raise SystemExit(str(error)) from error
@@ -459,6 +465,7 @@ def main() -> int:
)
plusargs = [
f"+APUM_HEX={apum_hex}",
f"+APUC_HEX={apuc_hex}",
f"+PCM_DIR={pcm_dir}",
f"+WINDOW_COUNT={window_count}",
"+FIRST_INDEX=0",