sync: SHANGHAI-U9 2026-07-27 三个 Simulink→RTL skill 转系统级 + 新增 simulink-rtl-redev
- simulink_rtl_dev / rtl-ip-replace: 从工程级 .claude/skills 迁到用户级并去工程耦合 (模型名/Notes 文档依赖/工程实测基线移出; addpath 与脚本路径改用户级) - rtl-ip-replace/scripts/gen_cmp_tb.py: Vivado 路径不再写死, 改读 VIVADO_PATH / XILINX_VIVADO (兼容 install root / bin / bin\vivado.bat 三种写法), 回落时提示 去设 skills_env.md 的 VIVADO_PATH - simulink-rtl-redev: 新增。原生模型改动后的仿真模型同步规范, 含两个实测过的工具 check_sync.py (三条时序链新鲜度闸门) 与 measure_k.m (时域/频域/包络三估计器互证) - sim-rtl-flow/SKILL.md: 加入 sync_sim:* 步骤, 引用上述三个 skill
This commit is contained in:
bowei
@@ -0,0 +1,27 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Step 0 of every rtl-ip-replace run (MANDATORY, user directive 2026-07-23):
|
||||
move ALL existing content of the fixed output dir into backup_<timestamp>/,
|
||||
so the run always generates everything fresh while keeping full history.
|
||||
|
||||
Usage: python backup_outdir.py <outdir>
|
||||
Existing backup_* subdirectories are kept in place (not re-nested).
|
||||
"""
|
||||
import os, shutil, sys, time
|
||||
|
||||
def main():
|
||||
if len(sys.argv) != 2:
|
||||
sys.exit('usage: backup_outdir.py <outdir>')
|
||||
d = os.path.abspath(sys.argv[1])
|
||||
os.makedirs(d, exist_ok=True)
|
||||
items = [x for x in os.listdir(d) if not x.startswith('backup_')]
|
||||
if not items:
|
||||
print('nothing to back up in', d)
|
||||
return
|
||||
bdir = os.path.join(d, 'backup_' + time.strftime('%Y%m%d_%H%M%S'))
|
||||
os.makedirs(bdir)
|
||||
for x in items:
|
||||
shutil.move(os.path.join(d, x), os.path.join(bdir, x))
|
||||
print(f'backed up {len(items)} item(s) -> {bdir}')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,215 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Filters -> Xilinx FIR Compiler (Task12 §六, requirement 4b).
|
||||
|
||||
GUIDED, by design: this script does the mechanical half (pull the integer taps
|
||||
out of the HDL Coder coefficient module, write the .coe, register the IP in
|
||||
ip_manifest, and write a surgery proposal with the exact latency-alignment
|
||||
arithmetic). It NEVER rewrites the RTL by itself -- the structural surgery and
|
||||
the delay alignment are reviewed with the user first (Task12 §6.3).
|
||||
|
||||
python fir_to_ip.py <HDLSRC> <OUT> [--manifest m.json]
|
||||
[--only MODULE ...] [--filter-type Single_Rate|Interpolation|Decimation]
|
||||
[--rate N] [--data-width W --data-frac F] [--channels N] [--paths N]
|
||||
[--reload] # switching between SEVERAL FIXED coefficient
|
||||
# sets; NOT an escape hatch for adaptive taps
|
||||
[--no-aclken]
|
||||
|
||||
Outputs in <OUT>:
|
||||
fir_<module>_<hash>.coe integer taps, radix 10 (no re-quantisation by the IP)
|
||||
ip_manifest.json/.tcl += {type fir_compiler ...}
|
||||
FIR_PROPOSAL_<module>.md what to cut, what to wire, how to align the latency
|
||||
"""
|
||||
import argparse, hashlib, json, os, sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def coe_path(outdir, module, coeffs, width, frac):
|
||||
h = hashlib.sha1((f'{width}/{frac}:' + ','.join(map(str, coeffs))).encode()).hexdigest()[:10]
|
||||
name = f'fir_{module}_{h}'
|
||||
p = os.path.join(outdir, name + '.coe')
|
||||
with open(p, 'w') as f: # ASCII only: .coe is read by Vivado, not by people
|
||||
f.write('; rtl-ip-replace: taps copied verbatim from the HDL Coder constant module\n')
|
||||
f.write('; (Task12 6.2.1 -- integer coefficients, the IP must not re-quantise)\n')
|
||||
f.write('radix = 10;\ncoefdata =\n')
|
||||
f.write(',\n'.join(str(c) for c in coeffs) + ';\n')
|
||||
return name, p
|
||||
|
||||
|
||||
PROPOSAL = """# FIR IP 替换方案 — `{module}` (Task12 §六,引导式,待确认)
|
||||
|
||||
来源:`{file}` Source Path:`{src}`
|
||||
IP:`{ipname}`(fir_compiler 7.2) — 系数 `.coe`:`{coe}`
|
||||
|
||||
## 1. 系数(契约 §6.2.1)
|
||||
|
||||
- 抽头数 **{taps}**,系数位宽 **{cw}**,小数位 **En{cf}**,对称:{sym}
|
||||
- `.coe` 为**整数**系数、逐个抄自 `{coef_signal}_i` 常量,IP 侧 `Quantization=Integer_Coefficients`,
|
||||
不做二次量化。改系数 = 改 .coe + 重生成 IP。
|
||||
|
||||
## 2. IP 配置(gen_ip.tcl 已按此写入)
|
||||
|
||||
| 项 | 值 | 理由 |
|
||||
|---|---|---|
|
||||
| Filter_Type | {ftype} | {rate_note} |
|
||||
| Data_Width | {dw}(原 RTL 二进制点 En{df}) | 位宽照抄;**小数位一律填 0** |
|
||||
| Coefficient_Width | {cw}(原 En{cf}) | 同上 |
|
||||
| Output_Rounding_Mode | Full_Precision | 契约 §6.2.2:截位放到 IP 外面做 |
|
||||
| RateSpecification / SamplePeriod | Input_Sample_Period / 1 | 禁止硬件折叠,延迟才是确定值 |
|
||||
| Has_ACLKEN | {aclken} | 用 `enb` 门控,与原 RTL 的使能节奏逐拍一致 |
|
||||
| Coefficient_Reload | {reload} | {reload_note} |
|
||||
| M_DATA_Has_TREADY / S_DATA_Has_FIFO | false / false | 纯流水,不引入可变延迟 |
|
||||
|
||||
**二进制点填 0 的理由**:core 做的是整数 MAC,二进制点只影响 GUI 显示;填 0 既与生成 RTL 的
|
||||
整数运算完全一致,又避开 `Data_Fractional_Bits` 的合法区间(0..16)限制。
|
||||
|
||||
**IP 硬限制(2026-07-24 在 Vivado 2018.1 实测)**:`Coefficient_Width∈[8,49]`、
|
||||
`Data_Width∈[2,47]`、`Data_Fractional_Bits∈[0,16]`。本例:系数 {cw} 位、数据 {dw} 位 → **{limits}**。
|
||||
超限时不得硬塞 IP,按 §一.12 双向调整原则回 Simulink 重新量化(HDL Coder 的 full-precision
|
||||
51 位抽头本来就没有工程意义),并复验性能。
|
||||
|
||||
## 3. 手术步骤(需人工确认后执行)
|
||||
|
||||
1. 在 `<OUT>/{file}` 中圈出滤波器边界:输入 `{din}`、输出 `{dout}`、有效 `{valid}`;
|
||||
复数({cplx})时每个实部/虚部各一路 IP(或 Number_Paths=2 走同一套系数)。
|
||||
2. 例化 `{ipname}`:`aclk=clk`、`aclken=enb`、`s_axis_data_tvalid=<原输入有效拍>`、
|
||||
`s_axis_data_tdata=<输入,符号扩展到 8 的倍数位>`。
|
||||
3. **延迟对齐(契约 §6.2.3,最容易翻车的一步)**:
|
||||
- 量 IP 延迟 L:综合后看 `m_axis_data_tvalid` 相对 `s_axis_data_tvalid` 的拍数
|
||||
(或 xsim 打一个脉冲量);
|
||||
- 量原 RTL 延迟 L0:数原来的 delay line + 乘加流水寄存器;
|
||||
- **L ≤ L0**:在 IP 输出后补 `L0-L` 拍移位寄存器;
|
||||
- **L > L0**:不许偷偷改节奏——按 §一.12 双向调整原则改 Simulink 源(整体延迟增加),
|
||||
并用模型仿真复验 MER/收敛不劣化,结论写进 Result 文档。
|
||||
4. 截位:把原 RTL 对乘加结果做的右移/饱和原样接到 `m_axis_data_tdata` 之后
|
||||
(IP 输出是 Full Precision,位宽 = {out_width_hint})。
|
||||
5. `m_axis_data_tvalid` 的翻转拍必须与原输出有效拍逐拍一致(契约 §6.2.4)。
|
||||
|
||||
## 4. 验收
|
||||
|
||||
`gen_cmp_tb.py` 双 DUT 逐拍对比,对齐后 **0 失配**方为 PASS(契约 §6.2.5);
|
||||
不 PASS 不得接入综合(§三)。
|
||||
"""
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('hdlsrc'); ap.add_argument('outdir')
|
||||
ap.add_argument('--manifest')
|
||||
ap.add_argument('--only', nargs='*', default=None, help='limit to these module names')
|
||||
ap.add_argument('--filter-type', default=None,
|
||||
choices=['Single_Rate', 'Interpolation', 'Decimation'])
|
||||
ap.add_argument('--rate', type=int, default=1)
|
||||
ap.add_argument('--data-width', type=int, default=None)
|
||||
ap.add_argument('--data-frac', type=int, default=None)
|
||||
ap.add_argument('--channels', type=int, default=1)
|
||||
ap.add_argument('--paths', type=int, default=1)
|
||||
ap.add_argument('--reload', action='store_true', help='adaptive taps via Coefficient_Reload')
|
||||
ap.add_argument('--no-aclken', action='store_true')
|
||||
a = ap.parse_args()
|
||||
|
||||
out = os.path.abspath(a.outdir)
|
||||
man = json.load(open(a.manifest or os.path.join(out, 'manifest.json')))
|
||||
filters = man.get('filter', [])
|
||||
banks = [f for f in filters if f['kind'] == 'fir_coef_bank']
|
||||
blocks = [f for f in filters if f['kind'].startswith('fir_block')]
|
||||
adaptive = [f for f in filters if f['kind'] == 'adaptive_fir_candidate']
|
||||
if a.only:
|
||||
banks = [b for b in banks if b['module'] in a.only]
|
||||
|
||||
if not banks:
|
||||
print('no fixed-coefficient FIR coefficient bank in the manifest '
|
||||
'-- nothing in scope for fir_compiler.')
|
||||
for f in adaptive:
|
||||
print(f"INVENTORY ONLY (not replaced): {f['module']}.{f['tap_regs']} "
|
||||
f"({f['taps']} runtime taps, delay line {f['delay_line']} deep {f['delay_depth']}). "
|
||||
f"Per SKILL scope: Farrow/LMS/correlator filters keep their RTL.")
|
||||
for f in [x for x in filters if x['kind'] == 'fir_block_candidate']:
|
||||
print(f"INVENTORY ONLY (not replaced): {f['module']} [{f['file']}] -- "
|
||||
f"coefficients are not constants ({len(f.get('coef_ports', []))} coef port(s)).")
|
||||
|
||||
ipman_p = os.path.join(out, 'ip_manifest.json')
|
||||
ipman = json.load(open(ipman_p)) if os.path.exists(ipman_p) else {'ips': []}
|
||||
done = 0
|
||||
for b in banks:
|
||||
# the FIR block that consumes this coefficient module (same Simulink parent path)
|
||||
parent = next((x for x in blocks
|
||||
if b['source_path'].startswith(x['source_path'])
|
||||
and x['source_path'] and x['module'] != b['module']), None)
|
||||
din = sorted(parent['data_in']) if parent else []
|
||||
dout = sorted(parent['data_out']) if parent else []
|
||||
dinfo = (parent['data_in'][din[0]] if din else {})
|
||||
dw = a.data_width or dinfo.get('width') or 16
|
||||
df = a.data_frac if a.data_frac is not None else (dinfo.get('frac') or 0)
|
||||
ftype = a.filter_type or ({'interpolation': 'Interpolation',
|
||||
'decimation': 'Decimation'}.get(
|
||||
parent['rate_hint'] if parent else '', 'Single_Rate'))
|
||||
if ftype != 'Single_Rate' and a.rate == 1:
|
||||
print(f" !! {b['module']}: rate change inferred from the module name but no --rate "
|
||||
f"given; falling back to Single_Rate. Confirm the factor in the Simulink block.")
|
||||
ftype = 'Single_Rate'
|
||||
ipname, coe = coe_path(out, b['module'], b['coeffs'], b['coef_width'], b['coef_frac'] or 0)
|
||||
|
||||
# FIR Compiler 7.2 hard limits (probed on 2026-07-24, Vivado 2018.1):
|
||||
# Coefficient_Width 8..49, Data_Width 2..47, Data_Fractional_Bits 0..16.
|
||||
# The binary point is documentation only -- the core does integer MACs, and
|
||||
# feeding frac=0 keeps us inside the legal range AND matches the generated
|
||||
# RTL, which is plain integer arithmetic on the raw bits.
|
||||
limits = []
|
||||
if not 8 <= b['coef_width'] <= 49:
|
||||
limits.append(f"coefficient width {b['coef_width']} outside 8..49")
|
||||
if not 2 <= dw <= 47:
|
||||
limits.append(f"data width {dw} outside 2..47")
|
||||
if limits:
|
||||
print(' !! ' + '; '.join(limits) +
|
||||
' -- fir_compiler cannot take this as-is: re-quantise the Simulink source '
|
||||
'(Task12 rule 12, bidirectional adjustment) or keep the RTL.')
|
||||
|
||||
entry = {'type': 'fir_compiler', 'name': ipname, 'coe': os.path.abspath(coe),
|
||||
'coef_width': b['coef_width'], 'coef_frac': 0,
|
||||
'width': dw, 'frac': 0,
|
||||
'coef_binpt_rtl': b['coef_frac'], 'data_binpt_rtl': df,
|
||||
'limits_ok': 'no' if limits else 'yes', 'filter_type': ftype,
|
||||
'interp': a.rate if ftype == 'Interpolation' else 1,
|
||||
'decim': a.rate if ftype == 'Decimation' else 1,
|
||||
'zero_pack': 1, 'channels': a.channels, 'paths': a.paths,
|
||||
'aclken': 'false' if a.no_aclken else 'true',
|
||||
'reload': 'true' if a.reload else 'false',
|
||||
'taps': b['taps']}
|
||||
ipman['ips'] = [i for i in ipman['ips'] if i.get('name') != ipname]
|
||||
if limits:
|
||||
print(' -> NOT registered in ip_manifest (would fail create_ip); '
|
||||
'proposal written so the source can be fixed first.')
|
||||
else:
|
||||
ipman['ips'].append(entry)
|
||||
done += 1
|
||||
|
||||
prop = os.path.join(out, f"FIR_PROPOSAL_{b['module']}_{b['coef_signal']}.md")
|
||||
open(prop, 'w', encoding='utf-8').write(PROPOSAL.format(
|
||||
module=b['module'], file=(parent or b)['file'], src=b['source_path'],
|
||||
ipname=ipname, coe=os.path.basename(coe), taps=b['taps'],
|
||||
cw=b['coef_width'], cf=b['coef_frac'], sym='是' if b['symmetric'] else '否',
|
||||
coef_signal=b['coef_signal'], ftype=ftype,
|
||||
rate_note=('抽取/内插比必须与 Simulink 块一致(本次取 %d)' % a.rate)
|
||||
if ftype != 'Single_Rate' else '单速率',
|
||||
dw=dw, df=df, aclken='false' if a.no_aclken else 'true',
|
||||
reload='true' if a.reload else 'false',
|
||||
reload_note='系数运行时更新(AXI-Stream config 通道)' if a.reload else '定系数,无需重载',
|
||||
din=', '.join(din) or '(未定位,人工确认)', dout=', '.join(dout) or '(未定位,人工确认)',
|
||||
valid='validIn/validOut' if (parent and parent.get('valid_in')) else '(人工确认)',
|
||||
cplx='是' if (parent and parent.get('complex')) else '否',
|
||||
limits='; '.join(limits) + ' —— 超限,先改源头' if limits else '均在范围内',
|
||||
out_width_hint=f"{dw}+{b['coef_width']}+ceil(log2({b['taps']})) 位左右"))
|
||||
print(f"FIR {b['module']}: {b['taps']} taps x {b['coef_width']}b (En{b['coef_frac']}) "
|
||||
f"-> {ipname} [{os.path.basename(coe)}]\n proposal: {os.path.basename(prop)}")
|
||||
done += 1
|
||||
|
||||
json.dump(ipman, open(ipman_p, 'w'), indent=1)
|
||||
with open(os.path.join(out, 'ip_manifest.tcl'), 'w') as f:
|
||||
for ip in ipman['ips']:
|
||||
f.write('lappend ips {%s}\n' % ' '.join(f'{k} {{{v}}}' for k, v in ip.items()))
|
||||
print(f'done: {done} FIR IP(s) registered; RTL untouched (guided surgery, Task12 6.3)')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,168 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate the dual-DUT cycle-true comparison testbench + xsim run script.
|
||||
|
||||
python gen_cmp_tb.py <top_module> <hdlsrc_dir> [--outdir DIR] [--cycles N]
|
||||
[--ip-simdir DIR] [--compare-at PORT=SIGNAL ...]
|
||||
|
||||
- DUT0 = <top> compiled from <hdlsrc>/*.v (original HDL Coder RTL)
|
||||
- DUT1 = <top>_IP compiled from <hdlsrc>/rtl_ip/*.v (+ IP simulation netlists)
|
||||
- identical stimulus into both: clk; reset (asserted RST_CYCLES then deasserted);
|
||||
enable-like inputs (enb/clk_enable/*_en names) held 1; every other input driven
|
||||
from a per-port LFSR (deterministic, seeded by port name hash).
|
||||
!! For control-heavy tops, replace the LFSR block with replayed vectors from
|
||||
the HDL Coder <top>_tb.v -- see SKILL.md.
|
||||
- compare: every output port, every posedge after both-ready; mismatches counted
|
||||
and logged; result file first line = PASS or "FAIL <n>".
|
||||
- --compare-at OUT=QUALIFIER limits comparison of OUT to cycles where the named
|
||||
qualifier signal (an output of the DUTs, e.g. a done/valid strobe) is 1 --
|
||||
the CORDIC consumption-point contract.
|
||||
"""
|
||||
import argparse, hashlib, os, re, sys
|
||||
|
||||
# Vivado install root, for the generated run_cmp.bat (xvlog/xelab/xsim + glbl.v).
|
||||
# Machine-specific, so it comes from the environment, never from a literal path.
|
||||
# VIVADO_PATH may point at the install root, its bin/, or bin/vivado.bat --
|
||||
# all three are accepted. XILINX_VIVADO (set by Vivado's settings64) also works.
|
||||
VIVADO_FALLBACK = r'D:\Xilinx\Vivado\2018.1'
|
||||
|
||||
|
||||
def _vivado_root():
|
||||
v = (os.environ.get('VIVADO_PATH') or os.environ.get('XILINX_VIVADO') or '').strip().strip('"')
|
||||
if not v:
|
||||
print(f'NOTE: neither VIVADO_PATH nor XILINX_VIVADO is set -- falling back to\n'
|
||||
f' {VIVADO_FALLBACK}\n'
|
||||
f' That literal is one machine\'s install. Set VIVADO_PATH to this\n'
|
||||
f' machine\'s Vivado (see the VIVADO_PATH entry in ~/skills_env.md)\n'
|
||||
f' or run_cmp.bat may silently use the wrong version.')
|
||||
return VIVADO_FALLBACK
|
||||
p = v.rstrip('\\/')
|
||||
if p.lower().endswith(('.bat', '.exe')) or os.path.isfile(p):
|
||||
p = os.path.dirname(p) # ...\bin\vivado.bat -> ...\bin
|
||||
if os.path.basename(p).lower() == 'bin':
|
||||
p = os.path.dirname(p) # ...\bin -> install root
|
||||
return p or VIVADO_FALLBACK
|
||||
|
||||
PORT_RE = re.compile(r'^\s*(input|output)\s+(signed\s+)?(\[[^\]]+\]\s*)?(\w+)\s*[,;)]', re.M)
|
||||
|
||||
def parse_ports(vfile, top):
|
||||
txt = open(vfile, encoding='utf-8', errors='replace').read()
|
||||
m = re.search(r'module\s+%s\b(.*?)endmodule' % re.escape(top), txt, re.S)
|
||||
if not m:
|
||||
sys.exit(f'module {top} not found in {vfile}')
|
||||
body = m.group(1)
|
||||
ports = []
|
||||
for pm in PORT_RE.finditer(body):
|
||||
d, signed, rng, name = pm.group(1), pm.group(2), (pm.group(3) or '').strip(), pm.group(4)
|
||||
ports.append({'dir': d, 'signed': bool(signed), 'range': rng, 'name': name})
|
||||
return ports
|
||||
|
||||
def lfsr_seed(name):
|
||||
return int(hashlib.sha1(name.encode()).hexdigest()[:8], 16) | 1
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('top'); ap.add_argument('hdlsrc')
|
||||
ap.add_argument('--rtl-ip', required=True, help='FIXED rtl_ip dir (vivado/rtl_ip/<model>)')
|
||||
ap.add_argument('--outdir', default=None)
|
||||
ap.add_argument('--cycles', type=int, default=200000)
|
||||
ap.add_argument('--rst-cycles', type=int, default=20)
|
||||
ap.add_argument('--ip-simdir', default='', help='dir with generated IP sim .v files (globbed recursively)')
|
||||
ap.add_argument('--compare-at', nargs='*', default=[], metavar='OUT=QUAL')
|
||||
a = ap.parse_args()
|
||||
hdlsrc = os.path.abspath(a.hdlsrc)
|
||||
rtl_ip_dir = os.path.abspath(a.rtl_ip)
|
||||
outdir = a.outdir or os.path.join(rtl_ip_dir, 'cmp')
|
||||
os.makedirs(outdir, exist_ok=True)
|
||||
qual = dict(kv.split('=') for kv in a.compare_at)
|
||||
|
||||
src = None
|
||||
for fn in os.listdir(hdlsrc):
|
||||
if fn.endswith('.v') and re.search(r'^\s*module\s+%s\b' % re.escape(a.top),
|
||||
open(os.path.join(hdlsrc, fn), encoding='utf-8', errors='replace').read(), re.M):
|
||||
src = os.path.join(hdlsrc, fn); break
|
||||
if not src:
|
||||
sys.exit(f'{a.top} not declared in {hdlsrc}/*.v')
|
||||
ports = parse_ports(src, a.top)
|
||||
ins = [p for p in ports if p['dir'] == 'input']
|
||||
outs = [p for p in ports if p['dir'] == 'output']
|
||||
|
||||
def is_clk(n): return n in ('clk', 'clock')
|
||||
def is_rst(n): return 'reset' in n or n == 'rst'
|
||||
def is_en(n): return n in ('enb', 'clk_enable') or n.endswith('_en') or n.endswith('_enable')
|
||||
|
||||
tb = ['`timescale 1 ns / 1 ns', 'module tb_cmp;',
|
||||
' reg clk = 0; always #2 clk = ~clk;',
|
||||
f' reg rst = 1; initial begin repeat ({a.rst_cycles}) @(posedge clk); rst <= 0; end',
|
||||
' integer mismatches = 0; integer fd;']
|
||||
conn0, conn1 = [], []
|
||||
for p in ins:
|
||||
n, rng = p['name'], p['range']
|
||||
if is_clk(n):
|
||||
conn0.append(f'.{n}(clk)'); conn1.append(f'.{n}(clk)'); continue
|
||||
if is_rst(n):
|
||||
conn0.append(f'.{n}(rst)'); conn1.append(f'.{n}(rst)'); continue
|
||||
if is_en(n):
|
||||
tb.append(f' wire {rng} {n} = 1\'b1;')
|
||||
else:
|
||||
w = 32
|
||||
m = re.match(r'\[(\d+):0\]', rng or '')
|
||||
if m: w = int(m.group(1)) + 1
|
||||
tb.append(f' reg [31:0] lfsr_{n} = 32\'h{lfsr_seed(n):08x};')
|
||||
tb.append(f' always @(posedge clk) lfsr_{n} <= {{lfsr_{n}[30:0], lfsr_{n}[31]^lfsr_{n}[21]^lfsr_{n}[1]^lfsr_{n}[0]}};')
|
||||
tb.append(f' wire {rng} {n} = lfsr_{n}[{min(w,32)-1}:0];')
|
||||
conn0.append(f'.{n}({n})'); conn1.append(f'.{n}({n})')
|
||||
for p in outs:
|
||||
n, rng = p['name'], p['range']
|
||||
tb.append(f' wire {rng} o0_{n}, o1_{n};')
|
||||
conn0.append(f'.{n}(o0_{n})'); conn1.append(f'.{n}(o1_{n})')
|
||||
tb += [f' {a.top} dut0 ({", ".join(conn0)});',
|
||||
f' {a.top}_IP dut1 ({", ".join(conn1)});',
|
||||
f' reg armed = 0; initial begin repeat ({a.rst_cycles + 5}) @(posedge clk); armed <= 1; end',
|
||||
' always @(posedge clk) if (armed) begin']
|
||||
for p in outs:
|
||||
n = p['name']
|
||||
g = f' && (o0_{qual[n]} === 1\'b1)' if n in qual else ''
|
||||
tb.append(f' if ((o0_{n} !== o1_{n}){g}) begin')
|
||||
tb.append(f' mismatches = mismatches + 1;')
|
||||
tb.append(f' if (mismatches <= 50) $display("MISMATCH t=%0t port={n} orig=%h ip=%h", $time, o0_{n}, o1_{n});')
|
||||
tb.append(' end')
|
||||
tb += [ ' end',
|
||||
f' initial begin repeat ({a.cycles}) @(posedge clk);',
|
||||
f' fd = $fopen("cmp_result.txt", "w");',
|
||||
' if (mismatches == 0) begin $fdisplay(fd, "PASS"); $display("CMP PASS"); end',
|
||||
' else begin $fdisplay(fd, "FAIL %0d", mismatches); $display("CMP FAIL %0d", mismatches); end',
|
||||
' $fclose(fd); $finish;',
|
||||
' end', 'endmodule']
|
||||
open(os.path.join(outdir, 'tb_cmp.v'), 'w').write('\n'.join(tb) + '\n')
|
||||
|
||||
vroot = _vivado_root()
|
||||
xbin = os.path.join(vroot, 'bin')
|
||||
glbl = os.path.join(vroot, 'data', 'verilog', 'src', 'glbl.v')
|
||||
if not os.path.isfile(glbl):
|
||||
print(f'WARNING: glbl.v not found under {vroot}\n'
|
||||
f' set VIVADO_PATH (or XILINX_VIVADO) to your Vivado install, '
|
||||
f'its bin/, or bin/vivado.bat; run_cmp.bat will fail as written.')
|
||||
bat = [f'@echo off', f'cd /d %~dp0',
|
||||
f'call "{xbin}\\xvlog.bat" "{glbl}"',
|
||||
f'for %%f in ("{hdlsrc}\\*.v") do if /I not "%%~nxf"=="{a.top}_tb.v" call "{xbin}\\xvlog.bat" "%%f"',
|
||||
f'for %%f in ("{rtl_ip_dir}\\*.v") do call "{xbin}\\xvlog.bat" "%%f"']
|
||||
if a.ip_simdir:
|
||||
bat.append(f'for /R "{a.ip_simdir}" %%f in (*.v) do call "{xbin}\\xvlog.bat" "%%f"')
|
||||
# dist_mem_gen/blk_mem_gen sim models load .mif by BARE filename relative to
|
||||
# the xsim run directory -- without this copy every ROM reads 0.
|
||||
bat.append(f'for /R "{a.ip_simdir}" %%f in (*.mif) do copy /Y "%%f" . >nul')
|
||||
bat += [f'call "{xbin}\\xvlog.bat" tb_cmp.v',
|
||||
f'call "{xbin}\\xelab.bat" tb_cmp glbl -s cmp_sim -timescale 1ns/1ns '
|
||||
f'-L dist_mem_gen_v8_0 -L blk_mem_gen_v8_4 -L fifo_generator_v13_2 '
|
||||
f'-L cordic_v6_0_14 -L fir_compiler_v7_2_11 -L cic_compiler_v4_0_13 '
|
||||
f'-L xbip_utils_v3_0_8 -L xbip_utils_v3_0_9 -L xbip_pipe_v3_0_5 '
|
||||
f'-L unisims_ver -L unimacro_ver -L secureip',
|
||||
f'call "{xbin}\\xsim.bat" cmp_sim -R',
|
||||
f'type cmp_result.txt']
|
||||
open(os.path.join(outdir, 'run_cmp.bat'), 'w').write('\r\n'.join(bat) + '\r\n')
|
||||
print(f'generated {outdir}\\tb_cmp.v ({len(ins)} in / {len(outs)} out ports) and run_cmp.bat')
|
||||
print('NOTE: exclude the original *_tb.v from compile; for control-heavy tops replace the')
|
||||
print(' LFSR stimulus with replayed HDL Coder testbench vectors (see SKILL.md).')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,169 @@
|
||||
# =============================================================================
|
||||
# gen_ip.tcl -- create the Vivado IPs listed in rtl_ip/ip_manifest.tcl
|
||||
# usage: vivado -mode batch -source gen_ip.tcl -tclargs <ip_manifest.tcl> <ip_out_dir> [part]
|
||||
# Managed-IP style: .xci + synthesis + simulation targets are generated so the
|
||||
# cores can be (a) added to the synthesis project and (b) compiled for xsim.
|
||||
# =============================================================================
|
||||
set manifest [lindex $argv 0]
|
||||
set ipdir [file normalize [lindex $argv 1]]
|
||||
set part xc7z100ffg900-2
|
||||
if {[llength $argv] > 2} { set part [lindex $argv 2] }
|
||||
|
||||
file mkdir $ipdir
|
||||
create_project -in_memory -part $part
|
||||
set_property target_language Verilog [current_project]
|
||||
|
||||
set ips {}
|
||||
source $manifest ;# fills $ips: list of {type {..} name {..} width {..} depth {..} coe {..} mode {..}}
|
||||
|
||||
# manifest field with a default (older manifests do not carry the newer keys)
|
||||
proc opt {arrname key def} {
|
||||
upvar 1 $arrname c
|
||||
if {[info exists c($key)] && $c($key) ne {}} { return $c($key) }
|
||||
return $def
|
||||
}
|
||||
|
||||
foreach ip $ips {
|
||||
array unset c; array set c $ip
|
||||
set name $c(name)
|
||||
if {[file exists $ipdir/$name/$name.xci]} {
|
||||
puts "###### SKIP $name (exists)"
|
||||
continue
|
||||
}
|
||||
switch -- $c(type) {
|
||||
dist_mem_gen {
|
||||
create_ip -name dist_mem_gen -vendor xilinx.com -library ip -version 8.0 \
|
||||
-module_name $name -dir $ipdir
|
||||
set_property -dict [list \
|
||||
CONFIG.memory_type {rom} \
|
||||
CONFIG.data_width $c(width) \
|
||||
CONFIG.depth $c(depth) \
|
||||
CONFIG.coefficient_file $c(coe) \
|
||||
] [get_ips $name]
|
||||
}
|
||||
blk_mem_gen {
|
||||
# SDP, common clock, READ_FIRST on the write port, no output register,
|
||||
# ENA/ENB pins -- matches SimpleDualPortRAM_generic semantics exactly.
|
||||
create_ip -name blk_mem_gen -vendor xilinx.com -library ip -version 8.4 \
|
||||
-module_name $name -dir $ipdir
|
||||
set_property -dict [list \
|
||||
CONFIG.Memory_Type {Simple_Dual_Port_RAM} \
|
||||
CONFIG.Assume_Synchronous_Clk {true} \
|
||||
CONFIG.Use_Byte_Write_Enable {false} \
|
||||
CONFIG.Write_Width_A $c(width) \
|
||||
CONFIG.Write_Depth_A $c(depth) \
|
||||
CONFIG.Read_Width_A $c(width) \
|
||||
CONFIG.Write_Width_B $c(width) \
|
||||
CONFIG.Read_Width_B $c(width) \
|
||||
CONFIG.Operating_Mode_A {READ_FIRST} \
|
||||
CONFIG.Enable_A {Use_ENA_Pin} \
|
||||
CONFIG.Enable_B {Use_ENB_Pin} \
|
||||
CONFIG.Register_PortB_Output_of_Memory_Primitives {false} \
|
||||
CONFIG.Register_PortB_Output_of_Memory_Core {false} \
|
||||
] [get_ips $name]
|
||||
}
|
||||
fifo_generator {
|
||||
# guided: mode field carries standard|fwft; depth/width from manifest
|
||||
create_ip -name fifo_generator -vendor xilinx.com -library ip -version 13.2 \
|
||||
-module_name $name -dir $ipdir
|
||||
set rm {Standard_FIFO}
|
||||
if {[info exists c(mode)] && $c(mode) eq "fwft"} { set rm {First_Word_Fall_Through} }
|
||||
set_property -dict [list \
|
||||
CONFIG.Fifo_Implementation {Common_Clock_Block_RAM} \
|
||||
CONFIG.Performance_Options $rm \
|
||||
CONFIG.Input_Data_Width $c(width) \
|
||||
CONFIG.Input_Depth $c(depth) \
|
||||
CONFIG.Output_Data_Width $c(width) \
|
||||
CONFIG.Output_Depth $c(depth) \
|
||||
] [get_ips $name]
|
||||
}
|
||||
cordic {
|
||||
# guided: Arc_Tan phase-seed kernel. Coarse rotation = full-quadrant atan2
|
||||
# (bidirectional-adjustment principle: better than the FSM's RHP quirk);
|
||||
# Radians 3QN phase; Truncate rounding; ACLKEN so enb gating stays exact;
|
||||
# parallel/maximum pipelining -> latency ~input_width+4, well under the
|
||||
# 64-cycle commit budget.
|
||||
create_ip -name cordic -vendor xilinx.com -library ip -version 6.0 \
|
||||
-module_name $name -dir $ipdir
|
||||
set_property -dict [list \
|
||||
CONFIG.Functional_Selection {Arc_Tan} \
|
||||
CONFIG.Input_Width $c(width) \
|
||||
CONFIG.Output_Width $c(out_width) \
|
||||
CONFIG.Data_Format {SignedFraction} \
|
||||
CONFIG.Phase_Format {Radians} \
|
||||
CONFIG.Coarse_Rotation {true} \
|
||||
CONFIG.Round_Mode {Truncate} \
|
||||
CONFIG.ACLKEN {true} \
|
||||
CONFIG.Pipelining_Mode {Maximum} \
|
||||
] [get_ips $name]
|
||||
}
|
||||
fir_compiler {
|
||||
# guided (Task12 §六). Cycle-true-critical settings:
|
||||
# Integer_Coefficients : taps go in as integers, the IP must NOT re-quantise
|
||||
# Full_Precision output : truncation stays OUTSIDE, done the way the RTL did it
|
||||
# Input_Sample_Period/1 : one sample per clock, no hardware folding
|
||||
# -> the latency is a fixed number we can align to
|
||||
# ACLKEN : gated by enb exactly like the generated RTL
|
||||
# no TREADY / no input FIFO: pure pipeline, no variable-latency handshake
|
||||
create_ip -name fir_compiler -vendor xilinx.com -library ip -version 7.2 \
|
||||
-module_name $name -dir $ipdir
|
||||
set_property -dict [list \
|
||||
CONFIG.Filter_Type [opt c filter_type {Single_Rate}] \
|
||||
CONFIG.CoefficientSource {COE_File} \
|
||||
CONFIG.Coefficient_File $c(coe) \
|
||||
CONFIG.Quantization {Integer_Coefficients} \
|
||||
CONFIG.Coefficient_Width $c(coef_width) \
|
||||
CONFIG.Coefficient_Fractional_Bits [opt c coef_frac 0] \
|
||||
CONFIG.Coefficient_Sign {Signed} \
|
||||
CONFIG.Coefficient_Structure [opt c coef_structure {Inferred}] \
|
||||
CONFIG.Data_Width $c(width) \
|
||||
CONFIG.Data_Fractional_Bits [opt c frac 0] \
|
||||
CONFIG.Data_Sign {Signed} \
|
||||
CONFIG.Output_Rounding_Mode {Full_Precision} \
|
||||
CONFIG.RateSpecification {Input_Sample_Period} \
|
||||
CONFIG.SamplePeriod 1 \
|
||||
CONFIG.HardwareOversamplingRate 1 \
|
||||
CONFIG.Interpolation_Rate [opt c interp 1] \
|
||||
CONFIG.Decimation_Rate [opt c decim 1] \
|
||||
CONFIG.Zero_Pack_Factor [opt c zero_pack 1] \
|
||||
CONFIG.Number_Channels [opt c channels 1] \
|
||||
CONFIG.Number_Paths [opt c paths 1] \
|
||||
CONFIG.Coefficient_Reload [opt c reload {false}] \
|
||||
CONFIG.Num_Reload_Slots [opt c reload_slots 1] \
|
||||
CONFIG.Has_ACLKEN [opt c aclken {true}] \
|
||||
CONFIG.Has_ARESETn [opt c aresetn {false}] \
|
||||
CONFIG.M_DATA_Has_TREADY {false} \
|
||||
CONFIG.S_DATA_Has_FIFO {false} \
|
||||
] [get_ips $name]
|
||||
}
|
||||
cic_compiler {
|
||||
# guided: only for large integer rate changes (CIC), fixed rate so the
|
||||
# decimation/interpolation cadence is static and comparable cycle by cycle.
|
||||
create_ip -name cic_compiler -vendor xilinx.com -library ip -version 4.0 \
|
||||
-module_name $name -dir $ipdir
|
||||
set rate [opt c rate 1]
|
||||
set_property -dict [list \
|
||||
CONFIG.Filter_Type [opt c filter_type {Decimation}] \
|
||||
CONFIG.Number_Of_Stages [opt c stages 3] \
|
||||
CONFIG.Differential_Delay [opt c diff_delay 1] \
|
||||
CONFIG.Sample_Rate_Changes {Fixed} \
|
||||
CONFIG.Fixed_Or_Initial_Rate $rate \
|
||||
CONFIG.Minimum_Rate [opt c min_rate $rate] \
|
||||
CONFIG.Maximum_Rate [opt c max_rate $rate] \
|
||||
CONFIG.Input_Data_Width $c(width) \
|
||||
CONFIG.Quantization {Full_Precision} \
|
||||
CONFIG.RateSpecification {Sample_Period} \
|
||||
CONFIG.SamplePeriod 1 \
|
||||
CONFIG.Number_Of_Channels [opt c channels 1] \
|
||||
CONFIG.Use_Xtreme_DSP_Slice {false} \
|
||||
CONFIG.HAS_ACLKEN [opt c aclken {true}] \
|
||||
CONFIG.HAS_ARESETN [opt c aresetn {false}] \
|
||||
CONFIG.HAS_DOUT_TREADY {false} \
|
||||
] [get_ips $name]
|
||||
}
|
||||
default { puts "###### UNKNOWN IP TYPE: $c(type) ($name)"; continue }
|
||||
}
|
||||
generate_target {instantiation_template synthesis simulation} [get_ips $name]
|
||||
puts "###### GENERATED $name ($c(type))"
|
||||
}
|
||||
puts "###### DONE [llength $ips] ip(s) in $ipdir"
|
||||
@@ -0,0 +1,227 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Incremental mode for rtl-ip-replace (Task12 rules 13-16 / Task12 §5).
|
||||
|
||||
Baseline = the newest backup_<ts>/ under the fixed dir, i.e. the previous run's
|
||||
ORIGINAL MATLAB RTL (src_orig/) + rewritten RTL (*.v) + ip/ + cmp/.
|
||||
|
||||
plan <HDLSRC> <OUT> [--apply] [--full] [--baseline DIR]
|
||||
Run right after backup_outdir.py. Classifies every source file against the
|
||||
baseline snapshot; --apply copies back the rewritten .v of files whose
|
||||
MATLAB source is byte-identical (nothing to redo for them).
|
||||
|
||||
finalize <HDLSRC> <OUT> [--baseline DIR]
|
||||
Run after the transform scripts (step 2), BEFORE create_ip / cmp sim.
|
||||
Writes TRANSFORM.diff, compares it + ip_manifest + .coe + scripts_sha1
|
||||
against the baseline, and decides whether step 3 (IP gen) and step 4
|
||||
(cycle-true sim) may be skipped. Decisions land in incr_plan.json and are
|
||||
appended to PROVENANCE.log.
|
||||
|
||||
Skipping the sim is only allowed when the rewrite this run performed is
|
||||
byte-for-byte the same rewrite the baseline PASS was proven on (Task12 §5.4).
|
||||
"""
|
||||
import argparse, json, os, shutil, sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from ipflow_common import (SRC_ORIG, TRANSFORM_DIFF, append_provenance, # noqa: E402
|
||||
scripts_sha1, sha1_file, snapshot_sources, src_files)
|
||||
|
||||
PLAN = 'incr_plan.json'
|
||||
|
||||
|
||||
def find_baseline(out, explicit=None):
|
||||
if explicit:
|
||||
return os.path.abspath(explicit)
|
||||
cands = sorted(d for d in os.listdir(out)
|
||||
if d.startswith('backup_') and os.path.isdir(os.path.join(out, d)))
|
||||
for d in reversed(cands): # newest usable one
|
||||
if os.path.isdir(os.path.join(out, d, SRC_ORIG)):
|
||||
return os.path.join(out, d)
|
||||
return None
|
||||
|
||||
|
||||
def read_sha(p):
|
||||
return sha1_file(p) if os.path.exists(p) else None
|
||||
|
||||
|
||||
def baseline_cmp_pass(base):
|
||||
p = os.path.join(base, 'cmp', 'cmp_result.txt')
|
||||
if not os.path.exists(p):
|
||||
return False, 'no cmp_result.txt in baseline'
|
||||
first = open(p, errors='replace').readline().strip()
|
||||
return first.startswith('PASS'), first
|
||||
|
||||
|
||||
def baseline_scripts_sha1(base):
|
||||
p = os.path.join(base, 'PROVENANCE.log')
|
||||
if os.path.exists(p):
|
||||
for ln in open(p, errors='replace'):
|
||||
if ln.startswith('scripts_sha1:'):
|
||||
return ln.split(':', 1)[1].strip()
|
||||
return None
|
||||
|
||||
|
||||
def save(out, plan):
|
||||
json.dump(plan, open(os.path.join(out, PLAN), 'w'), indent=2)
|
||||
|
||||
|
||||
def load(out):
|
||||
p = os.path.join(out, PLAN)
|
||||
return json.load(open(p)) if os.path.exists(p) else {}
|
||||
|
||||
|
||||
def cmd_plan(a):
|
||||
hdlsrc, out = os.path.abspath(a.hdlsrc), os.path.abspath(a.outdir)
|
||||
os.makedirs(out, exist_ok=True)
|
||||
snapshot_sources(hdlsrc, out) # this run's baseline-to-be (rule 14)
|
||||
base = None if a.full else find_baseline(out, a.baseline)
|
||||
plan = {'baseline': base, 'incremental': bool(base), 'files': {}, 'reasons': []}
|
||||
if not base:
|
||||
plan['reasons'].append('forced --full' if a.full else 'no usable baseline (backup_*/src_orig)')
|
||||
save(out, plan)
|
||||
print('FULL run: ' + plan['reasons'][0])
|
||||
return
|
||||
|
||||
bsrc = os.path.join(base, SRC_ORIG)
|
||||
new, old = src_files(hdlsrc), src_files(bsrc)
|
||||
for fn in new:
|
||||
if fn not in old:
|
||||
plan['files'][fn] = 'new'
|
||||
elif read_sha(os.path.join(hdlsrc, fn)) == read_sha(os.path.join(bsrc, fn)):
|
||||
plan['files'][fn] = 'same'
|
||||
else:
|
||||
plan['files'][fn] = 'changed'
|
||||
for fn in old:
|
||||
if fn not in new:
|
||||
plan['files'][fn] = 'removed'
|
||||
|
||||
same = [f for f, c in plan['files'].items() if c == 'same']
|
||||
all_same = bool(same) and all(c == 'same' for c in plan['files'].values())
|
||||
plan['all_same'] = all_same
|
||||
applied = []
|
||||
if a.apply:
|
||||
for fn in same:
|
||||
src = os.path.join(base, fn) # the REWRITTEN file
|
||||
if os.path.exists(src):
|
||||
shutil.copy(src, os.path.join(out, fn))
|
||||
applied.append(fn)
|
||||
# .coe / manifests are only safe to take wholesale when NOTHING changed;
|
||||
# otherwise the transform scripts must re-derive them (a stale .coe of a
|
||||
# table that moved would silently survive).
|
||||
extra = (['ip_manifest.tcl', 'ip_manifest.json', 'manifest.json']
|
||||
+ [f for f in sorted(os.listdir(base)) if f.endswith('.coe')]) if all_same else []
|
||||
for fn in extra:
|
||||
s, d = os.path.join(base, fn), os.path.join(out, fn)
|
||||
if os.path.exists(s) and not os.path.exists(d):
|
||||
shutil.copy(s, d)
|
||||
applied.append(fn)
|
||||
plan['reused_from_baseline'] = applied
|
||||
save(out, plan)
|
||||
|
||||
ok, first = baseline_cmp_pass(base)
|
||||
print(f'baseline: {base}')
|
||||
print(' cmp_result: ' + (first if first else 'MISSING'))
|
||||
for c in ('same', 'changed', 'new', 'removed'):
|
||||
fs = [f for f, k in plan['files'].items() if k == c]
|
||||
if fs:
|
||||
print(f' {c:8s}: {" ".join(sorted(fs))}')
|
||||
if applied:
|
||||
print(f' applied {len(applied)} file(s) from baseline (skip transform for them)')
|
||||
if plan['all_same']:
|
||||
print(' ALL MATLAB sources unchanged -> step 2 (transform scripts) can be skipped entirely')
|
||||
if not ok:
|
||||
print(' NOTE: baseline has no PASS -- cycle-true sim CANNOT be inherited')
|
||||
|
||||
|
||||
def cmd_finalize(a):
|
||||
from ipflow_common import write_transform_diff
|
||||
hdlsrc, out = os.path.abspath(a.hdlsrc), os.path.abspath(a.outdir)
|
||||
write_transform_diff(out)
|
||||
plan = load(out)
|
||||
if plan and not plan.get('incremental'):
|
||||
base = None # a full run was decided in `plan` -- honour it
|
||||
else:
|
||||
base = plan.get('baseline') or find_baseline(out, a.baseline)
|
||||
blockers, ip_reuse, sim_inherit = [], False, False
|
||||
|
||||
if not base:
|
||||
blockers.append('full run (no baseline / --full): everything regenerated')
|
||||
else:
|
||||
if any(c in ('new', 'removed') for c in plan.get('files', {}).values()):
|
||||
blockers.append('source file set changed (new/removed)')
|
||||
d_new = read_sha(os.path.join(out, TRANSFORM_DIFF))
|
||||
d_old = read_sha(os.path.join(base, TRANSFORM_DIFF))
|
||||
if d_old is None:
|
||||
blockers.append('baseline has no TRANSFORM.diff')
|
||||
elif d_new != d_old:
|
||||
blockers.append('TRANSFORM.diff differs from baseline (rewrite is not the same one)')
|
||||
m_new = read_sha(os.path.join(out, 'ip_manifest.tcl'))
|
||||
m_old = read_sha(os.path.join(base, 'ip_manifest.tcl'))
|
||||
coe_new = {f: read_sha(os.path.join(out, f)) for f in os.listdir(out) if f.endswith('.coe')}
|
||||
coe_old = {f: read_sha(os.path.join(base, f)) for f in os.listdir(base) if f.endswith('.coe')}
|
||||
ip_same = (m_new is not None and m_new == m_old and coe_new == coe_old)
|
||||
if not ip_same:
|
||||
blockers.append('ip_manifest/.coe differ from baseline')
|
||||
have_ip = os.path.isdir(os.path.join(out, 'ip')) or os.path.isdir(os.path.join(base, 'ip'))
|
||||
if ip_same and not have_ip: # Task12 §5.5: an incomplete baseline may not be inherited
|
||||
ip_same = False
|
||||
blockers.append('baseline has no ip/ directory')
|
||||
s_now, s_old = scripts_sha1(), baseline_scripts_sha1(base)
|
||||
if s_old != s_now:
|
||||
blockers.append(f'skill scripts changed ({s_old} -> {s_now})')
|
||||
ok, first = baseline_cmp_pass(base)
|
||||
if not ok:
|
||||
blockers.append('baseline cmp_result is not PASS')
|
||||
|
||||
ip_reuse = ip_same and not any(b.startswith(('no baseline', 'source file set')) for b in blockers)
|
||||
sim_inherit = not blockers
|
||||
if ip_reuse and not os.path.isdir(os.path.join(out, 'ip')) \
|
||||
and os.path.isdir(os.path.join(base, 'ip')):
|
||||
shutil.copytree(os.path.join(base, 'ip'), os.path.join(out, 'ip'))
|
||||
for f in ('ip_manifest.json',):
|
||||
if os.path.exists(os.path.join(base, f)) and not os.path.exists(os.path.join(out, f)):
|
||||
shutil.copy(os.path.join(base, f), os.path.join(out, f))
|
||||
|
||||
plan.update({'ip_reuse': ip_reuse, 'sim_inherit': sim_inherit, 'blockers': blockers})
|
||||
save(out, plan)
|
||||
|
||||
files = plan.get('files', {})
|
||||
reused = sorted(f for f, c in files.items() if c == 'same')
|
||||
regen = sorted(f for f, c in files.items() if c != 'same')
|
||||
append_provenance(out, [
|
||||
'',
|
||||
'# --- incremental run (Task12 rules 13-16) ---',
|
||||
'incremental: ' + ('on' if base else 'off'),
|
||||
'baseline: ' + (os.path.basename(base) + '/' if base else '-'),
|
||||
'reused_files: ' + (' '.join(reused) or '-'),
|
||||
'regenerated_files: ' + (' '.join(regen) or '-'),
|
||||
'ip_reuse: ' + ('full' if ip_reuse else 'none'),
|
||||
'cmp_sim: ' + (f'INHERITED PASS from {os.path.basename(base)}/cmp/cmp_result.txt'
|
||||
if sim_inherit else 'MUST RUN'),
|
||||
'blockers: ' + ('; '.join(blockers) or '-'),
|
||||
])
|
||||
print('ip_reuse :', 'YES (skip step 3 create_ip)' if ip_reuse else 'no (run step 3)')
|
||||
print('sim_inherit :', 'YES (skip step 4 cycle-true sim, PASS inherited from %s)'
|
||||
% os.path.basename(base) if sim_inherit else 'no (run step 4)')
|
||||
for b in blockers:
|
||||
print(' blocker:', b)
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
sub = p.add_subparsers(dest='cmd', required=True)
|
||||
for name in ('plan', 'finalize'):
|
||||
s = sub.add_parser(name)
|
||||
s.add_argument('hdlsrc')
|
||||
s.add_argument('outdir')
|
||||
s.add_argument('--baseline', help='explicit baseline dir (default: newest backup_*)')
|
||||
if name == 'plan':
|
||||
s.add_argument('--apply', action='store_true',
|
||||
help='copy back rewritten files whose source is unchanged')
|
||||
s.add_argument('--full', action='store_true', help='force a full run')
|
||||
a = p.parse_args()
|
||||
(cmd_plan if a.cmd == 'plan' else cmd_finalize)(a)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,197 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replace the HDL Coder `persistent isempty-init` table idiom with dist ROM IP.
|
||||
|
||||
Idiom (MATLAB `persistent T; if isempty(T), T = <const>; end`):
|
||||
|
||||
wire [..] P [0:D-1]; + D const assigns (the constant image)
|
||||
wire [..] T_1 [0:D-1]; generate: T_1[i] = P[i];
|
||||
reg [..] T_reg_reg [0:D-1]; always@(posedge clk/reset):
|
||||
reset -> 0; enb -> T_reg_reg <= T_reg_reg_next
|
||||
wire [..] T_reg_reg_next [0:D-1]; generate: T[i] = T_reg_reg[i];
|
||||
T_reg_reg_next[i] = TMP[i];
|
||||
reg [..] TMP [0:D-1]; always@*: TMP = (S) ? T_1 : T; // S = first-call flag
|
||||
|
||||
Net semantics: T reads 0 until the first (enb && S) tick, then the constant
|
||||
image forever. TMP reads (S || loaded) ? const : 0.
|
||||
|
||||
Replacement (bit- and cycle-true):
|
||||
dist_mem_gen ROM (content = P) per read site
|
||||
reg T__ipinited: reset->0, (enb && S)->1
|
||||
read T[e] -> (T__ipinited ? ROM[e] : 0)
|
||||
read TMP[e] -> ((S || T__ipinited) ? ROM[e] : 0)
|
||||
all five arrays + three blocks deleted.
|
||||
|
||||
Usage: python initrom_to_ip.py <hdlsrc> (operates on <hdlsrc>/rtl_ip copies;
|
||||
run AFTER rtl_ip_scan + source copy [rom_to_ip.py --copy-only or manual],
|
||||
BEFORE rom_to_ip's rewriting of any remaining plain ROMs, BEFORE namespace_ip.)
|
||||
"""
|
||||
import argparse, hashlib, json, os, re, shutil, sys
|
||||
|
||||
CONST_ASSIGN = re.compile(r"^\s*assign\s+(\w+)\[(\d+)\]\s*=\s*(\d+)'s?([bdh])([0-9a-fA-F_]+)\s*;\s*$", re.M)
|
||||
|
||||
def to_bits(radix, lit, width):
|
||||
lit = lit.replace('_', '')
|
||||
v = int(lit, {'b': 2, 'd': 10, 'h': 16}[radix]) & ((1 << width) - 1)
|
||||
return format(v, '0%db' % width)
|
||||
|
||||
def find_block(lines, i0, opener):
|
||||
"""Return (start,end) line span of the statement block whose header line index
|
||||
is i0 (an `always`/`generate` line); tracks begin/end + generate/endgenerate."""
|
||||
depth, i = 0, i0
|
||||
started = False
|
||||
while i < len(lines):
|
||||
t = lines[i]
|
||||
depth += len(re.findall(r'\bbegin\b|\bgenerate\b|\bcase\b', t))
|
||||
depth -= len(re.findall(r'\bend\b(?!module|generate|case)|\bendgenerate\b|\bendcase\b', t))
|
||||
if depth > 0:
|
||||
started = True
|
||||
if started and depth <= 0:
|
||||
return i0, i
|
||||
i += 1
|
||||
sys.exit(f'unbalanced block at line {i0}: {lines[i0]}')
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('hdlsrc')
|
||||
ap.add_argument('outdir', help='FIXED output dir (e.g. <repo>/vivado/rtl_ip/<model>)')
|
||||
a = ap.parse_args()
|
||||
from ipflow_common import copy_sources
|
||||
hdlsrc = os.path.abspath(a.hdlsrc)
|
||||
rtl_ip = os.path.abspath(a.outdir)
|
||||
copy_sources(hdlsrc, rtl_ip)
|
||||
|
||||
ipman_p = os.path.join(rtl_ip, 'ip_manifest.json')
|
||||
ipman = json.load(open(ipman_p)) if os.path.exists(ipman_p) else {'ips': []}
|
||||
coe_of_hash = {ip.get('content_hash'): ip['name'] for ip in ipman['ips'] if 'content_hash' in ip}
|
||||
|
||||
for fn in [f for f in os.listdir(rtl_ip) if f.endswith('.v')]:
|
||||
path = os.path.join(rtl_ip, fn)
|
||||
txt = open(path, encoding='utf-8', errors='replace').read()
|
||||
|
||||
# locate idiom instances: assign T_1[gv] = P[gv]; inside a generate loop
|
||||
pairs = []
|
||||
for m in re.finditer(r'assign\s+(\w+)_1\[(\w+)\]\s*=\s*(\w+)\[\2\]\s*;', txt):
|
||||
T, P = m.group(1), m.group(3)
|
||||
if f'{T}_reg_reg' in txt:
|
||||
pairs.append((T, P))
|
||||
if not pairs:
|
||||
continue
|
||||
print(f'[{fn}] isempty-init tables: {pairs}')
|
||||
|
||||
for T, P in pairs:
|
||||
# ---- table content ----
|
||||
vals, W = {}, None
|
||||
for m in CONST_ASSIGN.finditer(txt):
|
||||
if m.group(1) == P:
|
||||
W = W or len(to_bits(m.group(4), m.group(5), int(m.group(3))))
|
||||
vals[int(m.group(2))] = to_bits(m.group(4), m.group(5), int(m.group(3)))
|
||||
D = len(vals)
|
||||
if D == 0 or set(vals) != set(range(D)):
|
||||
print(f' SKIP {T}: const image {P} incomplete'); continue
|
||||
AW = max(1, (D - 1).bit_length())
|
||||
content = '\n'.join(vals[i] for i in range(D))
|
||||
h = hashlib.sha1((f'{W}x{D}\n' + content).encode()).hexdigest()[:10]
|
||||
if h not in coe_of_hash:
|
||||
ipname = f'rom_{W}x{D}_{h}'
|
||||
with open(os.path.join(rtl_ip, ipname + '.coe'), 'w') as f:
|
||||
f.write('memory_initialization_radix=2;\nmemory_initialization_vector=\n')
|
||||
f.write(',\n'.join(vals[i] for i in range(D)) + ';\n')
|
||||
ipman['ips'].append({'type': 'dist_mem_gen', 'name': ipname, 'width': W, 'depth': D,
|
||||
'coe': os.path.abspath(os.path.join(rtl_ip, ipname + '.coe')),
|
||||
'content_hash': h})
|
||||
coe_of_hash[h] = ipname
|
||||
ipname = coe_of_hash[h]
|
||||
|
||||
lines = txt.split('\n')
|
||||
|
||||
# ---- discover TMP name + guard signal from the mux always block ----
|
||||
mm = re.search(r'(\w+)\[\w+\]\s*=\s*%s_1\[\w+\]\s*;' % re.escape(T), txt)
|
||||
if not mm:
|
||||
print(f' SKIP {T}: mux TMP not found'); continue
|
||||
TMP = mm.group(1)
|
||||
gm = None
|
||||
for i, ln in enumerate(lines):
|
||||
if re.search(r'\bif\s*\(\s*(\w+)\s*==\s*1\'b0\s*\)', ln) is not None:
|
||||
# candidate: block must contain TMP writes
|
||||
s2, e2 = find_block(lines, max(0, i - 1), 'always')
|
||||
body = '\n'.join(lines[s2:e2 + 1])
|
||||
if f'{TMP}[' in body and f'{T}_1[' in body:
|
||||
gm = re.search(r'\bif\s*\(\s*(\w+)\s*==\s*1\'b0\s*\)', ln).group(1)
|
||||
break
|
||||
if not gm:
|
||||
print(f' SKIP {T}: guard signal not found'); continue
|
||||
S = gm
|
||||
|
||||
# ---- delete the three blocks (reg process, copy generate, next generate, mux always) ----
|
||||
kill = set()
|
||||
for i, ln in enumerate(lines):
|
||||
if (re.search(r'begin\s*:\s*%s_reg_process' % re.escape(T), ln)
|
||||
or re.search(r'begin\s*:\s*%s_1_gen' % re.escape(T), ln)
|
||||
or re.search(r'begin\s*:\s*%s_reg_reg_next_gen' % re.escape(T), ln)):
|
||||
j = i
|
||||
while j > 0 and not re.search(r'^\s*(always|generate|genvar|\s*for)', lines[j]):
|
||||
j -= 1
|
||||
# generate blocks start at 'generate'/'genvar'; always at 'always'
|
||||
while j > 0 and re.match(r'^\s*for\b', lines[j]):
|
||||
j -= 1
|
||||
if re.match(r'^\s*genvar', lines[j - 1] if j else ''):
|
||||
j -= 1
|
||||
s2, e2 = find_block(lines, j, None)
|
||||
# extend to endgenerate if present
|
||||
while e2 + 1 < len(lines) and re.match(r'^\s*endgenerate', lines[e2 + 1]):
|
||||
e2 += 1
|
||||
kill.update(range(s2, e2 + 1))
|
||||
if re.match(r'^\s*always\s*@\*', ln):
|
||||
s2, e2 = find_block(lines, i, None)
|
||||
body = '\n'.join(lines[s2:e2 + 1])
|
||||
if f'{TMP}[' in body and f'{T}_1[' in body and S in body:
|
||||
kill.update(range(s2, e2 + 1))
|
||||
# delete array declarations
|
||||
for i, ln in enumerate(lines):
|
||||
if re.search(r'\b(%s|%s_1|%s_reg_reg|%s_reg_reg_next|%s)\s*\[0:%d\]\s*;'
|
||||
% tuple(map(re.escape, (T, T, T, T, TMP))) % () if False else
|
||||
r'\b(' + '|'.join(map(re.escape, (T, T + '_1', T + '_reg_reg',
|
||||
T + '_reg_reg_next', TMP))) + r')\s+\[0:' + str(D - 1) + r'\]\s*;', ln):
|
||||
kill.add(i)
|
||||
if re.match(r'^\s*assign\s+%s\[\d+\]' % re.escape(P), ln):
|
||||
kill.add(i)
|
||||
if re.search(r'\b%s\s+\[0:%d\]\s*;' % (re.escape(P), D - 1), ln):
|
||||
kill.add(i)
|
||||
survivors = [ln for i, ln in enumerate(lines) if i not in kill]
|
||||
txt = '\n'.join(survivors)
|
||||
|
||||
# ---- rewrite remaining real read sites ----
|
||||
inited = f'{T}__ipinited'
|
||||
inst, k = [], 0
|
||||
zeros = f"{W}'sb" + '0' * W
|
||||
for arr, gate in ((T, inited), (TMP, f'({S} || {inited})')):
|
||||
for m in list(re.finditer(re.escape(arr) + r'\[([^\]\n]+)\]', txt)):
|
||||
expr = m.group(1)
|
||||
if expr.isdigit():
|
||||
continue
|
||||
aq, qq = f'{T}_ipa{k}', f'{T}_ipq{k}'
|
||||
inst.append(f' wire [{AW-1}:0] {aq};\n wire [{W-1}:0] {qq};\n'
|
||||
f' assign {aq} = {expr};\n'
|
||||
f' {ipname} u_{T}_ip{k} (.a({aq}), .spo({qq}));\n')
|
||||
txt = txt.replace(f'{arr}[{expr}]', f'({gate} ? $signed({qq}) : {zeros})')
|
||||
k += 1
|
||||
flag = (f' reg {inited};\n'
|
||||
f' always @(posedge clk or posedge reset)\n'
|
||||
f' if (reset) {inited} <= 1\'b0;\n'
|
||||
f' else if (enb && ({S})) {inited} <= 1\'b1;\n')
|
||||
txt = txt.replace('endmodule',
|
||||
f' // ---- rtl-ip-replace(initrom): {T} -> {ipname} ({k} read site(s)) ----\n'
|
||||
+ flag + ''.join(inst) + '\nendmodule', 1)
|
||||
print(f' {T} ({W}x{D}) -> {ipname}, {k} read site(s), guard={S}')
|
||||
|
||||
open(path, 'w', encoding='utf-8').write(txt)
|
||||
|
||||
json.dump(ipman, open(ipman_p, 'w'), indent=1)
|
||||
with open(os.path.join(rtl_ip, 'ip_manifest.tcl'), 'w') as f:
|
||||
for ip in ipman['ips']:
|
||||
f.write('lappend ips {%s}\n' % ' '.join(f'{k} {{{v}}}' for k, v in ip.items()
|
||||
if k != 'content_hash'))
|
||||
print('ip_manifest updated')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,110 @@
|
||||
"""Shared helpers: source copying + PROVENANCE.log (mandatory traceability).
|
||||
|
||||
RULE (user directive 2026-07-23): script-processed RTL lives ONLY in the fixed
|
||||
directory <repo>/vivado/rtl_ip/<model>/ ; the Vivado project may reference RTL
|
||||
ONLY from there, NEVER from the MATLAB hdlsrc directly. PROVENANCE.log in the
|
||||
fixed directory records the absolute path of the original MATLAB RTL.
|
||||
Generated Xilinx IPs live in <fixed_dir>/ip/.
|
||||
"""
|
||||
import difflib, hashlib, os, re, shutil, time
|
||||
|
||||
SRC_ORIG = 'src_orig' # verbatim snapshot of the MATLAB RTL this run is based on
|
||||
TRANSFORM_DIFF = 'TRANSFORM.diff'
|
||||
|
||||
|
||||
def src_files(d):
|
||||
return [fn for fn in sorted(os.listdir(d))
|
||||
if fn.endswith('.v') and not fn.endswith('_tb.v')]
|
||||
|
||||
|
||||
def sha1_file(p):
|
||||
return hashlib.sha1(open(p, 'rb').read()).hexdigest()
|
||||
|
||||
|
||||
def copy_sources(hdlsrc, outdir, force=False):
|
||||
os.makedirs(outdir, exist_ok=True)
|
||||
copied = []
|
||||
for fn in src_files(hdlsrc):
|
||||
dst = os.path.join(outdir, fn)
|
||||
if force or not os.path.exists(dst):
|
||||
shutil.copy(os.path.join(hdlsrc, fn), dst)
|
||||
copied.append(fn)
|
||||
snapshot_sources(hdlsrc, outdir)
|
||||
write_provenance(outdir, hdlsrc)
|
||||
return copied
|
||||
|
||||
|
||||
def snapshot_sources(hdlsrc, outdir):
|
||||
"""Task12 rule 14: keep the ORIGINAL MATLAB RTL this run is based on, so the
|
||||
next run can diff against it and reuse unchanged work (incremental mode)."""
|
||||
d = os.path.join(outdir, SRC_ORIG)
|
||||
os.makedirs(d, exist_ok=True)
|
||||
for fn in src_files(hdlsrc):
|
||||
dst = os.path.join(d, fn)
|
||||
if not os.path.exists(dst) or sha1_file(dst) != sha1_file(os.path.join(hdlsrc, fn)):
|
||||
shutil.copy(os.path.join(hdlsrc, fn), dst)
|
||||
return d
|
||||
|
||||
|
||||
_HUNK = re.compile(r'^@@ -\d+(,\d+)? \+\d+(,\d+)? @@')
|
||||
|
||||
|
||||
def write_transform_diff(outdir):
|
||||
"""src_orig/*.v -> rewritten *.v, as removed/added lines only (n=0) with hunk
|
||||
line numbers normalised away: two runs whose scripts performed the SAME
|
||||
rewrite produce byte-identical files, even if the MATLAB source changed
|
||||
elsewhere (that is exactly the class-B case of Task12 §5.2)."""
|
||||
orig = os.path.join(outdir, SRC_ORIG)
|
||||
out = []
|
||||
for fn in src_files(orig):
|
||||
mod = os.path.join(outdir, fn)
|
||||
if not os.path.exists(mod):
|
||||
out.append(f'=== {fn}: MISSING in outdir')
|
||||
continue
|
||||
a = open(os.path.join(orig, fn), errors='replace').read().splitlines()
|
||||
b = open(mod, errors='replace').read().splitlines()
|
||||
d = [_HUNK.sub('@@', ln) for ln in
|
||||
difflib.unified_diff(a, b, fromfile=f'orig/{fn}', tofile=f'ip/{fn}',
|
||||
n=0, lineterm='')]
|
||||
if d:
|
||||
out.extend(d)
|
||||
p = os.path.join(outdir, TRANSFORM_DIFF)
|
||||
open(p, 'w').write('\n'.join(out) + '\n')
|
||||
return p
|
||||
|
||||
|
||||
def scripts_sha1(script_dir=None):
|
||||
"""Fingerprint of the skill itself -- changed scripts forbid inheriting a
|
||||
previous cycle-true PASS (Task12 §5.4/5.5)."""
|
||||
d = script_dir or os.path.dirname(os.path.abspath(__file__))
|
||||
h = hashlib.sha1()
|
||||
for fn in sorted(os.listdir(d)):
|
||||
if fn.endswith(('.py', '.tcl')):
|
||||
h.update(fn.encode())
|
||||
h.update(open(os.path.join(d, fn), 'rb').read())
|
||||
return h.hexdigest()[:12]
|
||||
|
||||
def write_provenance(outdir, hdlsrc):
|
||||
hdlsrc = os.path.abspath(hdlsrc)
|
||||
lines = ['# PROVENANCE.log -- generated by rtl-ip-replace, do not edit',
|
||||
'# RULE: the Vivado project must use RTL from THIS directory only;',
|
||||
'# direct use of the MATLAB-generated hdlsrc below is FORBIDDEN.',
|
||||
'source_hdlsrc: ' + hdlsrc,
|
||||
'updated_at: ' + time.strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'ip_subdir: ip/',
|
||||
'files:']
|
||||
for fn in src_files(hdlsrc):
|
||||
p = os.path.join(hdlsrc, fn)
|
||||
h = sha1_file(p)[:12]
|
||||
mt = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(os.path.getmtime(p)))
|
||||
lines.append(f' {fn} src_sha1={h} src_mtime={mt}')
|
||||
lines.append('src_snapshot: ' + SRC_ORIG + '/ # original MATLAB RTL kept for the next incremental run')
|
||||
lines.append('scripts_sha1: ' + scripts_sha1())
|
||||
open(os.path.join(outdir, 'PROVENANCE.log'), 'w').write('\n'.join(lines) + '\n')
|
||||
|
||||
|
||||
def append_provenance(outdir, lines):
|
||||
"""Append the incremental-run block (Task12 §5.6). Called last, after the
|
||||
transform scripts have rewritten PROVENANCE.log."""
|
||||
with open(os.path.join(outdir, 'PROVENANCE.log'), 'a') as f:
|
||||
f.write('\n'.join(lines) + '\n')
|
||||
@@ -0,0 +1,38 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Suffix every module in rtl_ip/ with _IP so original and IP-based RTL can
|
||||
coexist in one xsim `work` library for the dual-DUT cycle-true comparison.
|
||||
|
||||
- renames `module X` -> `module X_IP` for every module DECLARED in rtl_ip/*.v
|
||||
- rewrites instantiations of those modules accordingly (word-boundary match)
|
||||
- IP core instances (dist_mem_gen/blk_mem_gen/... names from ip_manifest) are
|
||||
left untouched -- they are not declared here.
|
||||
|
||||
Run LAST, after rom_to_ip.py / ram_to_ip.py / any hand edits (CORDIC, FIFO).
|
||||
Idempotent: modules already ending in _IP are skipped.
|
||||
"""
|
||||
import argparse, os, re
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('rtl_ip_dir')
|
||||
a = ap.parse_args()
|
||||
d = os.path.abspath(a.rtl_ip_dir)
|
||||
files = [os.path.join(d, f) for f in os.listdir(d) if f.endswith('.v')]
|
||||
|
||||
mods = []
|
||||
for p in files:
|
||||
for m in re.finditer(r'^\s*module\s+(\w+)', open(p, encoding='utf-8', errors='replace').read(), re.M):
|
||||
if not m.group(1).endswith('_IP'):
|
||||
mods.append(m.group(1))
|
||||
mods = sorted(set(mods), key=len, reverse=True)
|
||||
print('modules:', mods)
|
||||
|
||||
for p in files:
|
||||
txt = open(p, encoding='utf-8', errors='replace').read()
|
||||
for mod in mods:
|
||||
txt = re.sub(r'\b%s\b(?!_IP)' % re.escape(mod), mod + '_IP', txt)
|
||||
open(p, 'w', encoding='utf-8').write(txt)
|
||||
print(f'suffixed {len(mods)} modules across {len(files)} files')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replace SimpleDualPortRAM_generic (HDL Coder) with blk_mem_gen SDP instances.
|
||||
|
||||
Semantics of the original (must be preserved exactly):
|
||||
- single clock, `enb` gates BOTH write and read registration
|
||||
- read latency 1 cycle; write/read same address -> read returns OLD data (read-first)
|
||||
=> blk_mem_gen: Simple Dual Port, common clock, Port-A write mode READ_FIRST,
|
||||
ENA/ENB pins used, NO output register.
|
||||
|
||||
The wrapper keeps the module name/ports/params; a generate-if selects the
|
||||
per-(AddrWidth,DataWidth) IP; unlisted combos fall back to the original
|
||||
behavioral body (so simulation still works, and synthesis of an unlisted combo
|
||||
is loudly visible in the utilization report instead of silently wrong).
|
||||
|
||||
Run AFTER rom_to_ip.py (needs rtl_ip/ copies), BEFORE namespace_ip.py.
|
||||
"""
|
||||
import argparse, json, os, re
|
||||
|
||||
WRAP = '''// regenerated by ram_to_ip.py -- blk_mem_gen SDP wrappers (read-first, 1-cycle, common clk)
|
||||
`timescale 1 ns / 1 ns
|
||||
module {mod}
|
||||
(clk, enb, wr_din, wr_addr, wr_en, rd_addr, dout);
|
||||
parameter integer AddrWidth = 1;
|
||||
parameter integer DataWidth = 1;
|
||||
input clk;
|
||||
input enb;
|
||||
input signed [DataWidth - 1:0] wr_din;
|
||||
input [AddrWidth - 1:0] wr_addr;
|
||||
input wr_en;
|
||||
input [AddrWidth - 1:0] rd_addr;
|
||||
output signed [DataWidth - 1:0] dout;
|
||||
|
||||
generate
|
||||
{cases} begin: g_behav // fallback: original behavioral RAM (unlisted param combo)
|
||||
reg [DataWidth - 1:0] ram [2**AddrWidth - 1:0];
|
||||
reg [DataWidth - 1:0] data_int;
|
||||
integer i;
|
||||
initial begin
|
||||
for (i=0; i<=2**AddrWidth - 1; i=i+1) ram[i] = 0;
|
||||
data_int = 0;
|
||||
end
|
||||
always @(posedge clk)
|
||||
if (enb == 1'b1) begin
|
||||
if (wr_en == 1'b1) ram[wr_addr] <= wr_din;
|
||||
data_int <= ram[rd_addr];
|
||||
end
|
||||
assign dout = data_int;
|
||||
end
|
||||
endgenerate
|
||||
endmodule
|
||||
'''
|
||||
|
||||
CASE = ''' if (AddrWidth == {aw} && DataWidth == {dw}) begin: g_a{aw}d{dw}
|
||||
wire [{dwm}:0] doutb_i;
|
||||
{ip} u_bram (
|
||||
.clka(clk), .ena(enb), .wea(wr_en), .addra(wr_addr), .dina(wr_din),
|
||||
.clkb(clk), .enb(enb), .addrb(rd_addr), .doutb(doutb_i));
|
||||
assign dout = doutb_i;
|
||||
end else
|
||||
'''
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('hdlsrc')
|
||||
ap.add_argument('outdir', help='FIXED output dir (e.g. <repo>/vivado/rtl_ip/<model>)')
|
||||
ap.add_argument('--manifest', default=None)
|
||||
a = ap.parse_args()
|
||||
hdlsrc = os.path.abspath(a.hdlsrc)
|
||||
rtl_ip = os.path.abspath(a.outdir)
|
||||
man = json.load(open(a.manifest or os.path.join(rtl_ip, 'manifest.json')))
|
||||
|
||||
combos = sorted({(r['AddrWidth'], r['DataWidth']) for r in man['ram']
|
||||
if r['AddrWidth'] and r['DataWidth']})
|
||||
if not combos:
|
||||
print('no SimpleDualPortRAM instantiations found'); return
|
||||
|
||||
ipman_p = os.path.join(rtl_ip, 'ip_manifest.json')
|
||||
ipman = json.load(open(ipman_p)) if os.path.exists(ipman_p) else {'ips': []}
|
||||
have = {ip['name'] for ip in ipman['ips']}
|
||||
|
||||
cases = ''
|
||||
for aw, dw in combos:
|
||||
ip = f'bram_sdp_a{aw}d{dw}'
|
||||
cases += CASE.format(aw=aw, dw=dw, dwm=dw - 1, ip=ip)
|
||||
if ip not in have:
|
||||
ipman['ips'].append({'type': 'blk_mem_gen', 'name': ip, 'width': dw,
|
||||
'depth': 2 ** aw, 'mode': 'sdp_readfirst_commonclk'})
|
||||
|
||||
# rewrite every SimpleDualPortRAM_*.v module file present in rtl_ip
|
||||
n = 0
|
||||
for fn in os.listdir(rtl_ip):
|
||||
if re.match(r'SimpleDualPortRAM_\w+\.v$', fn):
|
||||
mod = fn[:-2]
|
||||
open(os.path.join(rtl_ip, fn), 'w').write(WRAP.format(mod=mod, cases=cases))
|
||||
print('wrapped', mod, 'combos:', combos)
|
||||
n += 1
|
||||
|
||||
json.dump(ipman, open(ipman_p, 'w'), indent=1)
|
||||
with open(os.path.join(rtl_ip, 'ip_manifest.tcl'), 'w') as f:
|
||||
for ip in ipman['ips']:
|
||||
f.write('lappend ips {%s}\n' % ' '.join(f'{k} {{{v}}}' for k, v in ip.items()))
|
||||
print(f'done: {n} RAM module(s) wrapped, {len(combos)} blk_mem_gen config(s)')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,108 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replace fully-constant wire-array ROMs with Vivado dist_mem_gen instances.
|
||||
|
||||
- Copies every non-tb .v from <hdlsrc> into <hdlsrc>/rtl_ip/ (first run only, or --force).
|
||||
- For each fully-constant ROM in the manifest:
|
||||
* extracts the table into a .coe (dedup by content hash: identical tables share one IP config)
|
||||
* deletes the wire-array declaration and the constant assigns from the copy
|
||||
* per READ SITE inserts: wire [AW-1:0] <name>_a<k>; wire [W-1:0] <name>_q<k>;
|
||||
assign <name>_a<k> = <read_expr>;
|
||||
<ip> u_<name>_r<k> (.a(<name>_a<k>), .spo(<name>_q<k>));
|
||||
and rewrites `<name>[<read_expr>]` -> `$signed(<name>_q<k>)` (or unsigned).
|
||||
- Appends dist_mem_gen entries to ip_manifest.json / ip_manifest.tcl.
|
||||
|
||||
dist_mem_gen is ASYNC read => bit- and cycle-true against the original combinational read.
|
||||
Run AFTER rtl_ip_scan.py, BEFORE namespace_ip.py.
|
||||
"""
|
||||
import argparse, hashlib, json, os, re, shutil, sys
|
||||
|
||||
def to_bits(radix, lit, width):
|
||||
lit = lit.replace('_', '')
|
||||
base = {'b': 2, 'd': 10, 'h': 16}[radix]
|
||||
v = int(lit, base) & ((1 << width) - 1)
|
||||
return format(v, '0%db' % width)
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('hdlsrc')
|
||||
ap.add_argument('outdir', help='FIXED output dir (e.g. <repo>/vivado/rtl_ip/<model>)')
|
||||
ap.add_argument('--manifest', default=None)
|
||||
ap.add_argument('--force', action='store_true')
|
||||
a = ap.parse_args()
|
||||
from ipflow_common import copy_sources
|
||||
hdlsrc = os.path.abspath(a.hdlsrc)
|
||||
rtl_ip = os.path.abspath(a.outdir)
|
||||
os.makedirs(rtl_ip, exist_ok=True)
|
||||
man = json.load(open(a.manifest or os.path.join(rtl_ip, 'manifest.json')))
|
||||
|
||||
# 1. copy sources + PROVENANCE.log
|
||||
copy_sources(hdlsrc, rtl_ip, force=a.force)
|
||||
|
||||
ipman_p = os.path.join(rtl_ip, 'ip_manifest.json')
|
||||
ipman = json.load(open(ipman_p)) if os.path.exists(ipman_p) else {'ips': []}
|
||||
coe_of_hash, roms_done = {}, 0
|
||||
|
||||
CONST_ASSIGN = re.compile(r"^\s*assign\s+(\w+)\[(\d+)\]\s*=\s*(\d+)'s?([bdh])([0-9a-fA-F_]+)\s*;\s*$", re.M)
|
||||
|
||||
by_file = {}
|
||||
for rom in man['const_rom']:
|
||||
if not rom.get('fully_const'):
|
||||
print('SKIP (partial const):', rom['name']); continue
|
||||
by_file.setdefault(rom['file'], []).append(rom)
|
||||
|
||||
for fn, roms in by_file.items():
|
||||
path = os.path.join(rtl_ip, fn)
|
||||
txt = open(path, encoding='utf-8', errors='replace').read()
|
||||
for rom in roms:
|
||||
name, W, D, signed = rom['name'], rom['width'], rom['depth'], rom['signed']
|
||||
AW = max(1, (D - 1).bit_length())
|
||||
# collect table
|
||||
bits = {}
|
||||
for m in CONST_ASSIGN.finditer(txt):
|
||||
if m.group(1) == name:
|
||||
bits[int(m.group(2))] = to_bits(m.group(4), m.group(5), W)
|
||||
if len(bits) != D:
|
||||
print(f'ERROR {name}: found {len(bits)}/{D} const assigns -- skipped'); continue
|
||||
content = '\n'.join(bits[i] for i in range(D))
|
||||
h = hashlib.sha1((f'{W}x{D}\n' + content).encode()).hexdigest()[:10]
|
||||
if h not in coe_of_hash:
|
||||
ipname = f'rom_{W}x{D}_{h}'
|
||||
coe = os.path.join(rtl_ip, ipname + '.coe')
|
||||
with open(coe, 'w') as f:
|
||||
f.write('memory_initialization_radix=2;\nmemory_initialization_vector=\n')
|
||||
f.write(',\n'.join(bits[i] for i in range(D)) + ';\n')
|
||||
coe_of_hash[h] = ipname
|
||||
ipman['ips'].append({'type': 'dist_mem_gen', 'name': ipname, 'width': W,
|
||||
'depth': D, 'coe': os.path.abspath(coe)})
|
||||
ipname = coe_of_hash[h]
|
||||
|
||||
# remove decl + assigns
|
||||
txt = re.sub(r'^\s*wire\s+(signed\s+)?\[%d:0\]\s+%s\s+\[0:%d\]\s*;.*\n'
|
||||
% (W - 1, re.escape(name), D - 1), '', txt, flags=re.M)
|
||||
txt = re.sub(r"^\s*assign\s+%s\[\d+\]\s*=\s*[^;]+;\s*\n" % re.escape(name), '', txt, flags=re.M)
|
||||
|
||||
# rewrite read sites (unique exprs; every occurrence of same expr shares one instance)
|
||||
inst_lines, k = [], 0
|
||||
for expr in dict.fromkeys(rom['read_exprs']): # preserve order, dedup
|
||||
aq, qq = f'{name}_a{k}', f'{name}_q{k}'
|
||||
inst_lines.append(f' wire [{AW-1}:0] {aq};\n wire [{W-1}:0] {qq};\n'
|
||||
f' assign {aq} = {expr};\n'
|
||||
f' {ipname} u_{name}_r{k} (.a({aq}), .spo({qq}));\n')
|
||||
rd = f'$signed({qq})' if signed else qq
|
||||
txt = txt.replace(f'{name}[{expr}]', rd)
|
||||
k += 1
|
||||
# place instances right before `endmodule`
|
||||
txt = txt.replace('endmodule', ' // ---- rtl-ip-replace: %s -> %s (%d read site(s)) ----\n%s\nendmodule'
|
||||
% (name, ipname, k, ''.join(inst_lines)), 1)
|
||||
roms_done += 1
|
||||
print(f'ROM {name} ({W}x{D}, {k} reads) -> {ipname} [{fn}]')
|
||||
open(path, 'w', encoding='utf-8').write(txt)
|
||||
|
||||
json.dump(ipman, open(ipman_p, 'w'), indent=1)
|
||||
with open(os.path.join(rtl_ip, 'ip_manifest.tcl'), 'w') as f:
|
||||
for ip in ipman['ips']:
|
||||
f.write('lappend ips {%s}\n' % ' '.join(f'{k} {{{v}}}' for k, v in ip.items()))
|
||||
print(f'done: {roms_done} ROMs, {len(coe_of_hash)} unique tables; ip_manifest updated')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,266 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Scan HDL Coder generated Verilog for IP-replaceable structures.
|
||||
|
||||
Emits a JSON manifest of:
|
||||
const_rom : `wire [W:0] name [0:D];` fully populated by constant assigns -> dist/blk ROM IP
|
||||
reg_table : `reg` arrays (runtime-written tables) -> review by hand (RAM or leave)
|
||||
ram : SimpleDualPortRAM_generic instantiations (AddrWidth/DataWidth combos) -> blk_mem_gen
|
||||
fifo : modules/instances whose name contains 'fifo' -> fifo_generator (guided)
|
||||
cordic : files with the cx/cy/cz + atanT iterative kernel -> cordic IP (guided)
|
||||
filter : FIR filters (Task12 §六) -> fir_compiler / cic_compiler (guided)
|
||||
fir_coef_bank : HDL Coder <Name>Coef module, constant CoefOut_i taps
|
||||
fir_block : module whose header says "Discrete FIR Filter"
|
||||
adaptive_fir_candidate: hand-rolled tap regs + delay line + MAC (LMS etc.)
|
||||
|
||||
Usage: python rtl_ip_scan.py <hdlsrc_dir> [<outdir>] [-o manifest.json]
|
||||
"""
|
||||
import argparse, json, os, re, sys
|
||||
|
||||
ROM_DECL = re.compile(r'^\s*wire\s+(signed\s+)?\[(\d+):0\]\s+(\w+)\s+\[0:(\d+)\]\s*;', re.M)
|
||||
REG_DECL = re.compile(r'^\s*reg\s+(signed\s+)?\[(\d+):0\]\s+(\w+)\s+\[0:(\d+)\]\s*;', re.M)
|
||||
CONST_ASSIGN = re.compile(r"^\s*assign\s+(\w+)\[(\d+)\]\s*=\s*(\d+)'s?([bdh])([0-9a-fA-F_xzXZ]+)\s*;", re.M)
|
||||
RAM_INST = re.compile(r'SimpleDualPortRAM_\w+\s*#\s*\(((?:[^()]|\([^()]*\))*)\)\s*(\w+)\s*\(', re.S)
|
||||
PARAM = re.compile(r'\.(\w+)\s*\(\s*(\d+)\s*\)')
|
||||
|
||||
# ---- filter (Task12 §六) ----------------------------------------------------
|
||||
MODULE = re.compile(r'^[ \t]*module\s+(\w+)\b(.*?)^\s*endmodule', re.M | re.S)
|
||||
DECL = re.compile(r'^\s*(input|output|inout|wire|reg)\s+(signed\s+)?(?:\[(\d+):0\]\s+)?(\w+)\s*;'
|
||||
r'(?:[ \t]*//[ \t]*([usq]fix\d+(?:_En-?\d+)?|\w+))?', re.M)
|
||||
CONST_SIG = re.compile(r"^\s*assign\s+(\w+)\s*=\s*(\d+)'s?([bdh])([0-9a-fA-F_]+)\s*;", re.M)
|
||||
BASE_IDX = re.compile(r'^(.*?)_?(\d+)$')
|
||||
DLY_CHAIN = re.compile(r'^\s*(\w*?)(\d+)_next(_re|_im)?\s*=\s*\1(\d+)(?:_re|_im)?\s*;', re.M)
|
||||
FIXPT = re.compile(r'([us])fix(\d+)(?:_En(-?\d+))?')
|
||||
|
||||
|
||||
def to_signed(width, radix, digits):
|
||||
v = int(digits.replace('_', ''), {'b': 2, 'd': 10, 'h': 16}[radix])
|
||||
return v - (1 << width) if v >> (width - 1) else v
|
||||
|
||||
|
||||
def split_modules(txt):
|
||||
"""[(name, body, header_comment_block)] -- HDL Coder writes the Simulink block
|
||||
kind ('Discrete FIR Filter') and Source Path in the comment above the module."""
|
||||
out = []
|
||||
for m in MODULE.finditer(txt):
|
||||
out.append((m.group(1), m.group(2), txt[max(0, m.start() - 1200):m.start()]))
|
||||
return out
|
||||
|
||||
|
||||
def decls_of(body):
|
||||
d = {}
|
||||
for m in DECL.finditer(body):
|
||||
d[m.group(4)] = {'dir': m.group(1), 'signed': m.group(2) is not None,
|
||||
'width': int(m.group(3)) + 1 if m.group(3) else 1,
|
||||
'type': m.group(5) or ''}
|
||||
return d
|
||||
|
||||
|
||||
def fixpt_of(type_comment, width):
|
||||
m = FIXPT.search(type_comment or '')
|
||||
if not m:
|
||||
return {'width': width, 'frac': None, 'signed': None}
|
||||
return {'width': int(m.group(2)), 'frac': int(m.group(3) or 0), 'signed': m.group(1) == 's'}
|
||||
|
||||
|
||||
def const_groups(body, decls):
|
||||
"""constant `assign base_<i> = W'sh...;` families, index-complete from 0."""
|
||||
vals = {}
|
||||
for m in CONST_SIG.finditer(body):
|
||||
name, w, radix, digits = m.group(1), int(m.group(2)), m.group(3), m.group(4)
|
||||
vals[name] = (w, to_signed(w, radix, digits))
|
||||
groups = {}
|
||||
for name, (w, v) in vals.items():
|
||||
bm = BASE_IDX.match(name)
|
||||
if bm:
|
||||
groups.setdefault((bm.group(1), w), {})[int(bm.group(2))] = v
|
||||
# HDL Coder names the first element without an index (CoefData, CoefData_1, ...)
|
||||
for (base, w), members in groups.items():
|
||||
if 0 not in members and base in vals and vals[base][0] == w:
|
||||
members[0] = vals[base][1]
|
||||
out = []
|
||||
for (base, w), members in groups.items():
|
||||
n = len(members)
|
||||
if n < 4 or sorted(members) != list(range(n)):
|
||||
continue
|
||||
coeffs = [members[i] for i in range(n)]
|
||||
ex = decls.get(base + '_0') or decls.get(base + '0') or decls.get(base) or {}
|
||||
out.append({'base': base, 'n': n, 'width': w, 'coeffs': coeffs,
|
||||
'frac': fixpt_of(ex.get('type', ''), w)['frac'],
|
||||
'symmetric': coeffs == coeffs[::-1]})
|
||||
return out
|
||||
|
||||
|
||||
def scan_filters(path, txt):
|
||||
"""fir_coef_bank / fir_block / adaptive_fir_candidate (guided replacement)."""
|
||||
found = []
|
||||
fn = os.path.basename(path)
|
||||
for name, body, hdr in split_modules(txt):
|
||||
decls = decls_of(body)
|
||||
is_fir_hdr = 'Discrete FIR Filter' in hdr
|
||||
srcpath = (re.search(r'//\s*Source Path:\s*(\S+)', hdr) or [None, ''])[1] \
|
||||
if re.search(r'//\s*Source Path:\s*(\S+)', hdr) else ''
|
||||
ports = {k: v for k, v in decls.items() if v['dir'] in ('input', 'output')}
|
||||
coef_in = [p for p in ports if re.match(r'coef', p, re.I) and ports[p]['dir'] == 'input']
|
||||
|
||||
for g in const_groups(body, decls):
|
||||
# A constant family is only a TAP BANK if it is named/exported as one, or
|
||||
# if it is fed to multipliers inside a FIR module. Without this guard every
|
||||
# HDL Coder state/parameter constant family (state_1, mu_2, ...) looks like
|
||||
# a filter -- 5 false positives on RxAlignCore alone.
|
||||
exports = [p for p in ports if ports[p]['dir'] == 'output'
|
||||
and re.search(r'coef|tap', p, re.I)]
|
||||
named = re.search(r'coef|tap', name, re.I) or re.search(r'coef|tap', g['base'], re.I)
|
||||
multiplied = any(re.search(r'\b%s(?:_\d+)?\s*\*|\*\s*%s(?:_\d+)?\b'
|
||||
% (re.escape(g['base']), re.escape(g['base'])), body)
|
||||
for _ in (0,)) and (is_fir_hdr or re.search(r'fir|filt', name, re.I))
|
||||
if not (exports or named or multiplied):
|
||||
continue
|
||||
found.append({'file': fn, 'module': name, 'kind': 'fir_coef_bank',
|
||||
'source_path': srcpath, 'coef_signal': g['base'],
|
||||
'taps': g['n'], 'coef_width': g['width'], 'coef_frac': g['frac'],
|
||||
'symmetric': g['symmetric'], 'coeffs': g['coeffs'],
|
||||
'exports_taps': bool(exports),
|
||||
'note': 'fixed coefficients -> fir_compiler (.coe), guided (Task12 §6.2)'})
|
||||
|
||||
din = [p for p in ports if re.match(r'(data|d)?in', p, re.I) and ports[p]['dir'] == 'input'
|
||||
and ports[p]['width'] > 1]
|
||||
dout = [p for p in ports if re.search(r'out', p, re.I) and ports[p]['dir'] == 'output'
|
||||
and ports[p]['width'] > 1]
|
||||
if is_fir_hdr or (re.search(r'fir|filt', name, re.I) and din and dout):
|
||||
e = {'file': fn, 'module': name,
|
||||
'kind': 'fir_block' if is_fir_hdr else 'fir_block_candidate',
|
||||
'source_path': srcpath,
|
||||
'complex': any(p.endswith('_re') for p in din),
|
||||
'data_in': {p: fixpt_of(ports[p]['type'], ports[p]['width']) for p in sorted(din)},
|
||||
'data_out': {p: fixpt_of(ports[p]['type'], ports[p]['width']) for p in sorted(dout)},
|
||||
'valid_in': any(re.match(r'valid|.*vld', p, re.I) and ports[p]['dir'] == 'input'
|
||||
for p in ports),
|
||||
'valid_out': any(re.match(r'valid|.*vld', p, re.I) and ports[p]['dir'] == 'output'
|
||||
for p in ports),
|
||||
'coef_ports': sorted(coef_in),
|
||||
'adaptive': bool(coef_in),
|
||||
'rate_hint': ('interpolation' if re.search(r'interp|up', name, re.I) else
|
||||
'decimation' if re.search(r'decim|down', name, re.I) else 'single_rate?'),
|
||||
'note': 'rate/phase must be confirmed against the Simulink block before IP config'}
|
||||
found.append(e)
|
||||
|
||||
# hand-rolled adaptive FIR: tap regs + delay-line chain + multiplies
|
||||
chains = {}
|
||||
for m in DLY_CHAIN.finditer(body):
|
||||
if int(m.group(2)) == int(m.group(4)) + 1:
|
||||
chains.setdefault(m.group(1), set()).add(int(m.group(2)))
|
||||
chain = max(((b, len(s)) for b, s in chains.items()), key=lambda x: x[1], default=None)
|
||||
# tap registers: indexed reg group named like weights/taps. Do NOT require the
|
||||
# reg itself to sit next to a `*` -- HDL Coder feeds multipliers through operand
|
||||
# temporaries (da0_re * db0_re), so that test misses every real case (EqCore).
|
||||
tapregs = {}
|
||||
for rn, d in decls.items():
|
||||
bm = BASE_IDX.match(re.sub(r'_(re|im)$', '', rn))
|
||||
if d['dir'] == 'reg' and d['width'] > 1 and bm \
|
||||
and re.fullmatch(r'w\w*|tap\w*|coef\w*|h', bm.group(1), re.I):
|
||||
tapregs.setdefault(bm.group(1), set()).add(int(bm.group(2)))
|
||||
taps = max(((b, len(s)) for b, s in tapregs.items()), key=lambda x: x[1], default=None)
|
||||
if chain and chain[1] >= 3 and taps and taps[1] >= 3 and '*' in body:
|
||||
found.append({'file': fn, 'module': name, 'kind': 'adaptive_fir_candidate',
|
||||
'source_path': srcpath, 'tap_regs': taps[0], 'taps': taps[1],
|
||||
'delay_line': chain[0], 'delay_depth': chain[1] + 1,
|
||||
'note': 'runtime-updated taps: fir_compiler Reloadable, or KEEP the RTL '
|
||||
'with a written reason (Task12 §6.1)'})
|
||||
return found
|
||||
|
||||
KINDS = ('const_rom', 'reg_table', 'ram', 'fifo', 'cordic', 'filter')
|
||||
|
||||
|
||||
def scan_file(path):
|
||||
txt = open(path, encoding='utf-8', errors='replace').read()
|
||||
out = {k: [] for k in KINDS}
|
||||
|
||||
# constant assigns grouped by array name
|
||||
assigns = {}
|
||||
for m in CONST_ASSIGN.finditer(txt):
|
||||
assigns.setdefault(m.group(1), {})[int(m.group(2))] = (m.group(3), m.group(4), m.group(5))
|
||||
|
||||
for m in ROM_DECL.finditer(txt):
|
||||
signed, msb, name, dmax = m.group(1) is not None, int(m.group(2)), m.group(3), int(m.group(4))
|
||||
depth = dmax + 1
|
||||
vals = assigns.get(name, {})
|
||||
# read sites: name[expr] not on its own assign-constant lines
|
||||
reads = []
|
||||
for rm in re.finditer(re.escape(name) + r'\[([^\]\n]+)\]', txt):
|
||||
idx = rm.group(1)
|
||||
if idx.isdigit(): # constant index = the table-fill assigns
|
||||
continue
|
||||
reads.append(idx.strip())
|
||||
entry = {'file': os.path.basename(path), 'name': name, 'width': msb + 1,
|
||||
'depth': depth, 'signed': signed, 'n_const': len(vals),
|
||||
'fully_const': len(vals) == depth, 'read_exprs': reads}
|
||||
if len(vals) == depth and reads:
|
||||
out['const_rom'].append(entry)
|
||||
elif len(vals) > 0:
|
||||
entry['note'] = 'PARTIAL const fill -- inspect by hand'
|
||||
out['const_rom'].append(entry)
|
||||
|
||||
for m in REG_DECL.finditer(txt):
|
||||
out['reg_table'].append({'file': os.path.basename(path), 'name': m.group(3),
|
||||
'width': int(m.group(2)) + 1, 'depth': int(m.group(4)) + 1,
|
||||
'note': 'runtime-written reg array: treat as RAM or leave as-is'})
|
||||
|
||||
for m in RAM_INST.finditer(txt):
|
||||
params = dict((k, int(v)) for k, v in PARAM.findall(m.group(1)))
|
||||
out['ram'].append({'file': os.path.basename(path), 'inst': m.group(2),
|
||||
'AddrWidth': params.get('AddrWidth'), 'DataWidth': params.get('DataWidth')})
|
||||
|
||||
for m in re.finditer(r'^\s*module\s+(\w*[fF][iI][fF][oO]\w*)', txt, re.M):
|
||||
out['fifo'].append({'file': os.path.basename(path), 'module': m.group(1)})
|
||||
|
||||
if re.search(r'\bcx\b', txt) and re.search(r'\bcz\b', txt) and 'atanT' in txt:
|
||||
out['cordic'].append({'file': os.path.basename(path),
|
||||
'note': 'iterative CORDIC kernel (cx/cy/cz + atanT) -- guided replacement'})
|
||||
|
||||
out['filter'] = scan_filters(path, txt)
|
||||
return out
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('hdlsrc')
|
||||
ap.add_argument('outdir', nargs='?', default=None,
|
||||
help='FIXED output dir; manifest goes to <outdir>/manifest.json')
|
||||
ap.add_argument('-o', '--out', default=None)
|
||||
a = ap.parse_args()
|
||||
if a.outdir and not a.out:
|
||||
a.out = os.path.join(a.outdir, 'manifest.json')
|
||||
manifest = dict({'hdlsrc': os.path.abspath(a.hdlsrc), 'files': {}}, **{k: [] for k in KINDS})
|
||||
for fn in sorted(os.listdir(a.hdlsrc)):
|
||||
if not fn.endswith('.v') or fn.endswith('_tb.v'):
|
||||
continue
|
||||
r = scan_file(os.path.join(a.hdlsrc, fn))
|
||||
for k in KINDS:
|
||||
manifest[k].extend(r[k])
|
||||
|
||||
# a Discrete FIR Filter block expands into a tree of helper modules (Filter,
|
||||
# subFilter, FilterTapSystolic*): keep the block, drop its own guts.
|
||||
tops = [f['source_path'] for f in manifest['filter']
|
||||
if f['kind'] == 'fir_block' and f.get('source_path')]
|
||||
kept = []
|
||||
for f in manifest['filter']:
|
||||
inner = any(f.get('source_path', '').startswith(t + '/') for t in tops)
|
||||
if inner and f['kind'] == 'fir_block_candidate':
|
||||
continue # internal plumbing of a detected FIR block
|
||||
if inner and f['kind'] == 'fir_coef_bank':
|
||||
f['belongs_to'] = next(t for t in tops if f['source_path'].startswith(t + '/'))
|
||||
kept.append(f)
|
||||
manifest['filter'] = kept
|
||||
outp = a.out or os.path.join(a.hdlsrc, 'rtl_ip', 'manifest.json')
|
||||
os.makedirs(os.path.dirname(outp), exist_ok=True)
|
||||
json.dump(manifest, open(outp, 'w'), indent=1)
|
||||
print('manifest ->', outp)
|
||||
for k in KINDS:
|
||||
print(f' {k}: {len(manifest[k])}')
|
||||
for e in manifest[k]:
|
||||
brief = {kk: (f'[{len(vv)} values]' if kk == 'coeffs' else vv)
|
||||
for kk, vv in e.items() if kk != 'read_exprs'}
|
||||
print(' ', brief,
|
||||
f"reads={len(e.get('read_exprs', []))}" if 'read_exprs' in e else '')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user