name: "kmh_dut"
desc: "所有昆明湖DUT"
children:
  - name: "frontend"
    desc: "前端模块"
    children:
      - name: "ifu"
        desc: "指令单元 (Instruction Fetch Unit)"
        children:
          - name: "rvc_expander"
            desc: "RVC指令扩充器"

scripts/build_<顶层模块>_<下层模块名>_..._<目标模块名>.py

def build(cfg) -> bool:
    """编译DUT
    Args:
        cfg: 运行时配置，可通过它访问配置项，例如 cfg.rtl.version
    Return:
        返回 True 或者 False，表明该函数是否完成预期目标
    """

import os
from comm import warning, info


def build(cfg):
    # import 相关依赖
    from toffee_test.markers import match_version
    from comm import is_all_file_exist, get_rtl_dir, exe_cmd, get_root_dir
    # 检查RTL版本（version参数为空，表示所有版本都支持）
    if not match_version(cfg.rtl.version, "openxiangshan-kmh-*"):
        warning("ifu frontend rvc expander: %s" % f"Unsupported RTL version {cfg.rtl.version}")
        return False
    # 检查在当前RTL中，目标文件是否存在
    f = is_all_file_exist(["rtl/RVCExpander.sv"], get_rtl_dir(cfg=cfg))
    assert f is True, f"File {f} not found"
    # 如果dut中不存在RVCExpander，则调用picker进行Python打包
    if not os.path.exists(get_root_dir("dut/RVCExpander")):
        info("Exporting RVCExpander.sv")
        s, out, err = exe_cmd(f'picker export --cp_lib false {get_rtl_dir("rtl/RVCExpander.sv", cfg=cfg)} --lang python --tdir {get_root_dir("dut")}/ -w rvc.fst -c')
        assert s, "Failed to export RVCExpander.sv: %s\n%s" % (out, err)
    # 如果tools中不存在disasm/build，则需要编译disasm
    if not os.path.exists(get_root_dir("tools/disasm/build")):
        info("Building disasm")
        s, _, _ = exe_cmd("make -C %s" % get_root_dir("tools/disasm"))
        assert s, "Failed to build disasm"
    # 编译成功
    return True

def line_coverage_files(cfg):
    return ["RVCExpander.v"]

def line_coverage_files(cfg)-> list[str]:
    """指定需要覆盖的文件
    Args:
        cfg: 运行时配置，可通过它访问配置项，例如 cfg.rtl.version
    Return:
        返回统计代码行覆盖率的目标RTL文件名
    """

def line_coverage_files(cfg):
    return ["RVCExpander.v"]

ut_frontend/ifu/rvc_expander
├── classical_version
│   ├── env
│   │   ├── __init__.py
│   │   └── rvc_expander_wrapper.py
│   ├── __init__.py
│   └── test_rvc_expander.py
├── __init__.py
├── README.md
└── toffee_version
    ├── agent
    │   └── __init__.py
    ├── bundle
    │   └── __init__.py
    ├── env
    │   ├── __init__.py
    │   └── ref_rvc_expand.py
    ├── __init__.py
    └── test
        ├── __init__.py
        ├── rvc_expander_fixture.py
        └── test_rvc.py

class RVCExpander(toffee.Bundle):
    def __init__(self, cover_group, **kwargs):
        super().__init__()
        self.cover_group = cover_group
        self.dut = DUTRVCExpander(**kwargs) # 创建DUT
        self.io = toffee.Bundle.from_prefix("io_", self.dut) # 通过 Bundle 使用前缀关联引脚
        self.bind(self.dut)                 # 把 Bundle 与 DUT 进行绑定

    def expand(self, instr, fsIsOff):
        self.io["in"].value = instr         # 给DUT引脚赋值
        self.io["fsIsOff"].value = fsIsOff  # 给DUT引脚赋值
        self.dut.RefreshComb()              # 推动组合电路
        self.cover_group.sample()           # 调用sample对功能覆盖率进行统计
        return self.io["out_bits"].value, self.io["ill"].value  # 返回结果 和 是否是非法指令

    def stat(self):                         # 获取当前状态
        return {
            "instr": self.io["in"].value,         # 输入指令
            "decode": self.io["out_bits"].value,  # 返回展开结果
            "ilegal": self.io["ill"].value != 0,  # 输入是否非法
        }

import toffee.funcov as fc
# 创建功能覆盖率组
g = fc.CovGroup(UT_FCOV("../../../CLASSIC"))

def init_rvc_expander_funcov(expander, g: fc.CovGroup):
    """Add watch points to the RVCExpander module to collect function coverage information"""

    # 1. Add point RVC_EXPAND_RET to check expander return value:
    #    - bin ERROR. The instruction is not illegal
    #    - bin SUCCE. The instruction is not expanded
    g.add_watch_point(expander, {
                                "ERROR": lambda x: x.stat()["ilegal"] == False,
                                "SUCCE": lambda x: x.stat()["ilegal"] != False,
                          }, name = "RVC_EXPAND_RET")
    ...
    # 5. Reverse mark function coverage to the check point
    def _M(name):
        # get the module name
        return module_name_with(name, "../../test_rv_decode")

    #  - mark RVC_EXPAND_RET
    g.mark_function("RVC_EXPAND_RET",     _M(["test_rvc_expand_16bit_full",
                                              "test_rvc_expand_32bit_full",
                                              "test_rvc_expand_32bit_randomN"]), bin_name=["ERROR", "SUCCE"])
    ...

version_check = get_version_checker("openxiangshan-kmh-*")             # 指定满足要的RTL版本
@pytest.fixture()
def rvc_expander(request):
    version_check()                                                    # 进行版本检查
    fname = request.node.name                                          # 获取调用该fixture的测试用例
    wave_file = get_out_dir("decoder/rvc_expander_%s.fst" % fname)     # 设置波形文件路径
    coverage_file = get_out_dir("decoder/rvc_expander_%s.dat" % fname) # 设置代码覆盖率文件路径
    coverage_dir = os.path.dirname(coverage_file)
    os.makedirs(coverage_dir, exist_ok=True)                           # 目标目录不存在则创建目录
    expander = RVCExpander(g, coverage_filename=coverage_file, waveform_filename=wave_file)
                                                                       # 创建RVCExpander
    expander.dut.io_in.AsImmWrite()                                    # 设置io_in引脚的写入时机为立即写入             
    expander.dut.io_fsIsOff.AsImmWrite()                               # 设置io_fsIsOff引脚的写入时机为立即写入  
    init_rvc_expander_funcov(expander, g)                              # 初始化功能检查点
    yield expander                                                     # 返回创建好的 RVCExpander 给 Test Case
    expander.dut.Finish()                                              # Tests Case运行完成后，结束DUT
    set_line_coverage(request, coverage_file)                          # 把生成的代码覆盖率文件告诉 toffee-report
    set_func_coverage(request, g)                                      # 把生成的功能覆盖率数据告诉 toffee-report
    g.clear()                                                          # 清空功能覆盖统计

N = 10
T = 1<<16
@pytest.mark.toffee_tags(TAG_LONG_TIME_RUN)
@pytest.mark.parametrize("start,end",
                         [(r*(T//N), (r+1)*(T//N) if r < N-1 else T) for r in range(N)])
def test_rvc_expand_16bit_full(rvc_expander, start, end):
    """Test the RVC expand function with a full compressed instruction set

    Description:
        Perform an expand check on 16-bit compressed instructions within the range from 'start' to 'end'.
    """
    # Add check point: RVC_EXPAND_RANGE to check expander input range.
    #   When run to here, the range[start, end] is covered
    covered = -1
    g.add_watch_point(rvc_expander, {
                                "RANGE[%d-%d]"%(start, end): lambda _: covered == end
                          }, name = "RVC_EXPAND_ALL_16B", dynamic_bin=True)
    # Reverse mark function to the check point
    g.mark_function("RVC_EXPAND_ALL_16B", test_rvc_expand_16bit_full, bin_name="RANGE[%d-%d]"%(start, end))
    # Drive the expander and check the result
    rvc_expand(rvc_expander, generate_rvc_instructions(start, end))
    # When go to here, the range[start, end] is covered
    covered = end
    g.sample()                                                              # 覆盖率采样

module RVCExpander(
  input  [31:0] io_in,
  input         io_fsIsOff,
  output [31:0] io_out_bits,
  output        io_ill
);

class RVCExpanderIOBundle(Bundle):
	_in, _fsIsOff ,_out_bits,_ill = Signals(4)

self.agent = RVCExpanderAgent(RVCExpanderIOBundle.from_prefix("io").bind(dut))

self.io = toffee.Bundle.from_prefix("io_", self.dut) # 通过 Bundle 使用前缀关联引脚
self.bind(self.dut)   

self.io["in"].value = instr
self.io["fsIsOff"].value = False

def gen_bundle_code_from_dict(bundle_name: str, dut, dict: dict, max_width: int = 120)
def gen_bundle_code_from_prefix(bundle_name: str, dut, prefix: str = "", max_width: int = 120):
def gen_bundle_code_from_regex(bundle_name: str, dut, regex: str, max_width: int = 120):

python bundle_code_intel_gen.py [signal] [target]

class RVCExpanderAgent(Agent):
    def __init__(self, bundle:RVCExpanderIOBundle):
        super().__init__(bundle)
        self.bundle = bundle
    
    @driver_method()
    async def expand(self, instr, fsIsOff):             # 传入参数：RVC指令和fs.status使能情况
        self.bundle._in.value = instr                   # 引脚赋值
        self.bundle._fsIsOff.value = fsIsOff            # 引脚赋值
        
        await self.bundle.step()                        # 推动时钟
        return self.bundle._out_bits.value,             # 返回值：扩展后指令
                self.bundle._ill.value                  # 返回值：指令合法校验

class RVCExpanderEnv(Env):
    def __init__(self, dut:DUTRVCExpander):
        super().__init__()
        dut.io_in.xdata.AsImmWrite()        
        dut.io_fsIsOff.xdata.AsImmWrite()   # 设置引脚写入时机
        self.agent = RVCExpanderAgent(RVCExpanderIOBundle.from_prefix("io").bind(dut)) # 补全前缀，绑定DUT

@toffee_test.fixture
async def rvc_expander(toffee_request: toffee_test.ToffeeRequest):
    import asyncio
    version_check()
    dut = toffee_request.create_dut(DUTRVCExpander)
    start_clock(dut)
    init_rvc_expander_funcov(dut, gr)
    
    toffee_request.add_cov_groups([gr])
    expander = RVCExpanderEnv(dut)
    yield expander

    cur_loop = asyncio.get_event_loop()
    for task in asyncio.all_tasks(cur_loop):
        if task.get_name() == "__clock_loop":
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                break

def test_a(): # 不合理，无法通过a判断测试目标
    pass

def test_rvc_expand_16bit_full(): # 合理，可以通过用例名称大体知道测试内容
    pass

def rvc_expand(rvc_expander, ref_insts, is_32bit=False, fsIsOff=False):
    """compare the RVC expand result with the reference

    Args:
        rvc_expander (warpper): the fixture of the RVC expander
        ref_insts (list[int]]): the reference instruction list
    """
    find_error = 0
    for insn in ref_insts:
        insn_disasm = disasmbly(insn)
        value, instr_ex = rvc_expander.expand(insn, fsIsOff)
        if is_32bit:
            assert value == insn, "RVC expand error, 32bit instruction need to be the same"
        if (insn_disasm == "unknown") and  (instr_ex == 0):
            debug(f"find bad inst:{insn}, ref: 1, dut: 0")
            find_error +=1
        elif (insn_disasm != "unknown") and  (instr_ex == 1):
            if (instr_filter(insn_disasm) != 1): 
                debug(f"find bad inst:{insn},disasm:{insn_disasm}, ref: 0, dut: 1")
                find_error +=1
    assert 0 == find_error, "RVC expand error (%d errros)" % find_error

def test_<name>(a: type_a, b: type_b):
    """Test abstract

    Args:
        a (type_a): description of arg a.
        b (type_b): description of arg b.

    Detailed test description here (if need).
    """
    ...

def rvc_expand(rvc_expander, ref_insts, is_32bit=False, fsIsOff=False):
    """compare the RVC expand result with the reference

    Args:
        rvc_expander (warpper): the fixture of the RVC expander
        ref_insts (list[int]]): the reference instruction list
    """
    find_error = 0
    for insn in ref_insts:
        insn_disasm = disasmbly(insn)
        value, instr_ex = rvc_expander.expand(insn, fsIsOff)
        if is_32bit:
            assert value == insn, "RVC expand error, 32bit instruction need to be the same"
        if (insn_disasm == "unknown") and  (instr_ex == 0):
            debug(f"find bad inst:{insn}, ref: 1, dut: 0")
            find_error +=1
        elif (insn_disasm != "unknown") and  (instr_ex == 1):
            if (instr_filter(insn_disasm) != 1): 
                debug(f"find bad inst:{insn},disasm:{insn_disasm}, ref: 0, dut: 1")
                find_error +=1
    assert 0 == find_error, "RVC expand error (%d errros)" % find_error

@pytest.mark.toffee_tags(TAG_SMOKE)
def test_rvc_expand_16bit_smoke(rvc_expander):
    """Test the RVC expand function with 1 compressed instruction"""
    rvc_expand(rvc_expander, generate_rvc_instructions(start=100, end=101))

N = 10
T = 1<<16
@pytest.mark.toffee_tags(TAG_LONG_TIME_RUN)
@pytest.mark.parametrize("start,end",
                         [(r*(T//N), (r+1)*(T//N) if r < N-1 else T) for r in range(N)])
def test_rvc_expand_16bit_full(rvc_expander, start, end):
    """Test the RVC expand function with a full compressed instruction set

    Description:
        Perform an expand check on 16-bit compressed instructions within the range from 'start' to 'end'.
    """
    # Add check point: RVC_EXPAND_RANGE to check expander input range.
    #   When run to here, the range[start, end] is covered
    g.add_watch_point(rvc_expander, {
                                "RANGE[%d-%d]"%(start, end): lambda _: True
                          }, name = "RVC_EXPAND_ALL_16B").sample()

    # Reverse mark function to the check point
    g.mark_function("RVC_EXPAND_ALL_16B", test_rvc_expand_16bit_full, bin_name="RANGE[%d-%d]"%(start, end))

    # Drive the expander and check the result
    rvc_expand(rvc_expander, generate_rvc_instructions(start, end))

# 省略前面
    if not os.path.exists(get_root_dir("dut/RVCExpander")):
        info("Exporting RVCExpander.sv")
        s, out, err = exe_cmd(f'picker export --cp_lib false {get_rtl_dir("rtl/RVCExpander.sv", cfg=cfg)
                                                              } --lang python --tdir {get_root_dir("dut")}/ -w rvc.fst -c')
        assert s, "Failed to export RVCExpander.sv: %s\n%s" % (out, err)
# 省略后面

set_line_coverage(request, coverage_file)                          # 把生成的代码覆盖率文件告诉 toffee-report

// *verilator coverage_off*
// 忽略统计的代码段
...
// *verilator coverage_on*

module example;
    always @(posedge clk) begin
        // *verilator coverage_off*
        if (debug_signal) begin
            $display("This is for debugging only");
        end
        // *verilator coverage_on*
        if (enable) begin
            do_something();
        end
    end
endmodule

def set_line_coverage(request, datfile, ignore=[]):
    """Pass

    Args:
        request (pytest.Request): Pytest的默认fixture，
        datfile (string): DUT生成的
        ignore (list[str]): 覆盖率过滤文件/或者文件夹
    """

...
set_line_coverage(request, coverage_file,
                  get_root_dir("scripts/frontend_ifu_rvc_expander"))

from comm import UT_FCOV
# 本模块名为：ut_frontend.ifu.rvc_expander.classical_version.env.rvc_expander_wrapper
# 通过../../../去掉了classical_version和上级模块env，rvc_expander_wrapper
# UT_FCOV会默认去掉前缀 ut_
g = fc.CovGroup(UT_FCOV("../../../CLASSIC"))
# name = UT_FCOV("../../../CLASSIC")

import toffee.funcov as fc
# 使用上面指定的GROUP名字
g = fc.CovGroup(name)

def init_rvc_expander_funcov(expander, g: fc.CovGroup):
    """Add watch points to the RVCExpander module to collect function coverage information"""
    # 1. Add point RVC_EXPAND_RET to check expander return value:
    #    - bin ERROR. The instruction is not illegal
    #    - bin SUCCE. The instruction is not expanded
    g.add_watch_point(expander, {
                                "ERROR": lambda x: x.stat()["ilegal"] == False,
                                "SUCCE": lambda x: x.stat()["ilegal"] != False,
                          }, name = "RVC_EXPAND_RET")
    # 5. Reverse mark function coverage to the check point
    def _M(name):
        # get the module name
        return module_name_with(name, "../../test_rv_decode")

    #  - mark RVC_EXPAND_RET
    g.mark_function("RVC_EXPAND_RET",_M(["test_rvc_expand_16bit_full",
                                              "test_rvc_expand_32bit_full",
                                              "test_rvc_expand_32bit_randomN"]), bin_name=["ERROR", "SUCCE"])

    # The End                                                                              
    return None 

N=10
T=1<<32
@pytest.mark.toffee_tags([TAG_LONG_TIME_RUN, TAG_RARELY_USED])
@pytest.mark.parametrize("start,end",
                         [(r*(T//N), (r+1)*(T//N) if r < N-1 else T) for r in range(N)])
def test_rvc_expand_32bit_full(rvc_expander, start, end):
    """Test the RVC expand function with a full 32 bit instruction set

    Description:
        Randomly generate N 32-bit instructions for each check, and repeat the process K times.
    """
    # Add check point: RVC_EXPAND_ALL_32B to check instr bits.
    covered = -1
    g.add_watch_point(rvc_expander, {"RANGE[%d-%d]"%(start, end): lambda _: covered == end},
                      name = "RVC_EXPAND_ALL_32B", dynamic_bin=True)
    # Reverse mark function to the check point
    g.mark_function("RVC_EXPAND_ALL_32B", test_rvc_expand_32bit_full)
    # Drive the expander and check the result
    rvc_expand(rvc_expander, list([_ for _ in range(start, end)]))
    # When go to here, the range[start, end] is covered
    covered = end
    g.sample()

@pytest.fixture()
def decoder(request):
    # before test
    init_rv_decoder_funcov(g)
    func_name = request.node.name
    # If the output directory does not exist, create it
    output_dir_path = get_out_dir("decoder/log")
    os.makedirs(output_dir_path, exist_ok=True)
    decoder = Decode(DUTDecodeStage(
        waveform_filename=get_out_dir("decoder/decode_%s.fst"%func_name),
        coverage_filename=get_out_dir("decoder/decode_%s.dat"%func_name),
    ))
    decoder.dut.InitClock("clock")
    decoder.dut.StepRis(lambda x: g.sample())
    yield decoder
    # after test
    decoder.dut.Finish()
    coverage_file = get_out_dir("decoder/decode_%s.dat"%func_name)
    if not os.path.exists(coverage_file):
        raise FileNotFoundError(f"File not found: {coverage_file}")
    set_line_coverage(request, coverage_file, get_root_dir("scripts/backend_ctrlblock_decode"))
    set_func_coverage(request, g)
    g.clear()