simu.py

#!/usr/bin/env python3
import argparse
import sys
import os
import re
import traceback
import subprocess
import types
import inspect
IVERILOG_PATH = 'iverilog'
ROOT_DIR = '.' + os.path.sep
TEST_DIR = ROOT_DIR + 'tests'
TMP_DIR  = ROOT_DIR + '.tmp'
sys.path.append(ROOT_DIR)

from polyphony.compiler.__main__ import compile_main, logging_setting
from polyphony.compiler.common.env import env
from polyphony.compiler.common.common import read_source
from polyphony.compiler.__main__ import setup, compile_plan, output_hdl, output_plan
from polyphony.compiler.__main__ import compile as compile_polyphony
from polyphony.simulator import Simulator, SimulationModelBuilder, HDLAssertionError, SimulationObserver, Model, Net, Reg, Port

def parse_options():
    if not os.path.exists(TMP_DIR):
        os.mkdir(TMP_DIR)
    parser = argparse.ArgumentParser(prog='simu')
    parser.add_argument('-C', dest='compile_only', action='store_true')
    parser.add_argument('-P', '--python', dest='enable_python', action='store_true',
                        default=False, help='enable python simulation')
    parser.add_argument('-c', '--config', dest='config',
                        metavar='CONFIG', help='set configration(json literal or file)')
    parser.add_argument('-dd', '--disable_debug', dest='debug_mode',
                        action='store_false', default=True, help='disable debug mode')
    parser.add_argument('-vd', '--verilog_dump', dest='verilog_dump',
                        action='store_true', help='output vcd file in testbench')
    parser.add_argument('-vm', '--verilog_monitor', dest='verilog_monitor',
                        action='store_true', help='enable $monitor in testbench')
    parser.add_argument('-hd', '--hdl_debug', dest='hdl_debug_mode',
                        action='store_true', default=False, help='enable HDL debug mode')
    parser.add_argument('-p', dest='with_path_name', action='store_true')
    parser.add_argument('-t', '--targets', nargs='+', dest='targets', default=list())
    parser.add_argument('--watch', dest='watch_signals', default='',
                        help='comma-separated signal names to watch (e.g. "m.i,m.o")')
    parser.add_argument('source', help='Python source file')
    return parser.parse_args()


def case_name_from_path(casefile_path):
    casefile = os.path.basename(casefile_path)
    casename, _ = os.path.splitext(casefile)
    return casename


def exec_test(casefile_path, simu_options):
    casename = case_name_from_path(casefile_path)
    if simu_options.with_path_name:
        p, _ = os.path.splitext(casefile_path)
        simu_options.output_prefix = p.replace('.', '').replace(os.path.sep, '_')
        casename = f'{simu_options.output_prefix}_{casename}'
    else:
        simu_options.output_prefix = ''

    compiler_options = setup_compiler(casefile_path, casename, simu_options)
    source_text = read_source(casefile_path)

    scopes = exec_compile(casefile_path, source_text, compiler_options, simu_options)
    if not scopes:
        return ['Compile Error'], []

    hdl_finishes = simulate_on_verilog(casename, casefile_path, simu_options)
    if simu_options.enable_python:
        py_finishes = simulate_on_python(casefile_path, source_text, scopes, simu_options)
    else:
        py_finishes = []
    return hdl_finishes, py_finishes


def targets_from_source_comment(filepath):
    targets = []
    file = open(filepath, 'r')
    for line in file.readlines():
        if not line.startswith('# TEST '):
            continue
        terms = line[:-1].split(' ')
        if len(terms) <= 2:
            continue
        targets.append((terms[2], terms[3:]))
    return targets


def setup_compiler(casefile_path, casename, simu_options):
    assert simu_options
    compiler_options = types.SimpleNamespace()
    compiler_options.output_name = casename
    compiler_options.output_prefix = simu_options.output_prefix
    compiler_options.output_dir = TMP_DIR
    compiler_options.verbose_level = 0
    compiler_options.quiet_level = 0 if simu_options.debug_mode else 3
    if hasattr(simu_options, 'targets'):
        compiler_options.targets = simu_options.targets
    else:
        compiler_options.targets = targets_from_source_comment(casefile_path)
    compiler_options.config = simu_options.config
    compiler_options.debug_mode = simu_options.debug_mode
    compiler_options.hdl_debug_mode = simu_options.hdl_debug_mode
    compiler_options.verilog_dump = simu_options.verilog_dump
    compiler_options.verilog_monitor = simu_options.verilog_monitor
    compiler_options.watch_signals = getattr(simu_options, 'watch_signals', '')
    setup(casefile_path, compiler_options)
    return compiler_options


def exec_compile(casefile_path, source_text, compiler_options, simu_options):
    try:
        plan = compile_plan()
        scopes = compile_polyphony(plan, source_text, casefile_path)
        output_hdl(output_plan(), scopes, compiler_options, stage_offset=len(plan))
    except Exception as e:
        print(f'[COMPILE PYTHON] FAILED: {casefile_path}')
        if env.dev_debug_mode:
            traceback.print_exc()
        print(e)
        return None
    if simu_options.compile_only:
        return None
    return scopes


def simulate_on_verilog(casename, casefile_path, simu_options):
    hdl_finishes = []
    for testbench in env.testbenches:
        result_lines = call_iverilog(testbench.base_name, casename, casefile_path, simu_options)
        if result_lines:
            # result_lines[-2] == '***:finish'
            hdl_finishes.append(result_lines[-2])
        else:
            hdl_finishes.append('FAIL')
    return hdl_finishes


def call_iverilog(testname, casename, casefile_path, options):
    if options.output_prefix:
        test_filename = f'{options.output_prefix}_{testname}'
    else:
        test_filename = testname
    hdl_files = [
        f'{TMP_DIR}{os.path.sep}{casename}.v',
        f'{TMP_DIR}{os.path.sep}{test_filename}.v',
    ]
    exec_name = f'{TMP_DIR}{os.path.sep}{test_filename}'
    args = (f'{IVERILOG_PATH} -I {TMP_DIR} -W all -Wno-implicit-dimensions -o {exec_name} -s {testname}').split(' ')
    args += hdl_files
    try:
        subprocess.check_call(args)
    except Exception as e:
        print(f'[COMPILE HDL] FAILED: {casefile_path}')
        return
    try:
        out = subprocess.check_output([exec_name], timeout=3)
        lines = out.decode('utf-8').split('\n')
        for line in lines:
            if options.debug_mode:
                print(line)
            if 'FAILED' in line:
                raise Exception()
        return lines
    except Exception as e:
        print(f'[HDL SIMULATION] FAILED: {casefile_path}')
        print(e)
    return None


def model_selector_with_argv(models):
    def model_selector(*args, **kwargs):
        args_str = []
        for a in args:
            if type(a).__name__ == 'type' or inspect.isfunction(a):
                args_str.append(a.__name__)
            else:
                args_str.append(str(a))
        for model, hdlmodule in models.values():
            if args_str == hdlmodule.scope._bound_args:
                return model
        raise ValueError('model not found')
    return model_selector


def model_selector_with_argtypes(models, name):
    def model_selector(*args, **kwargs):
        arg_types = []
        for a in args:
            if isinstance(a, bool):
                arg_types.append('b')
            elif isinstance(a, int):
                arg_types.append('i')
            else:
                assert False, f'unsupported arg type: {type(a)}'
        kwarg_types = {}
        for k, v in kwargs.items():
            if isinstance(v, bool):
                kwarg_types[k] = 'b'
            elif isinstance(v, int):
                kwarg_types[k] = 'i'
            else:
                assert False, f'unsupported arg type: {type(v)}'
        for model, hdlmodule in models.values():
            if hdlmodule.name == name and not arg_types and not kwarg_types:
                return model()
            names = hdlmodule.name.rsplit('_', 1)
            orig_name = names[0]
            if orig_name != name:
                continue
            if len(names) == 1 and not arg_types:
                return model(*args, **kwargs)
            a_types = arg_types[:]
            module_arg_types = re.findall(r'i|b', names[-1])
            if kwarg_types:
                for param_name in hdlmodule.scope.param_names():
                    if param_name in kwarg_types:
                        a_types.append(kwarg_types[param_name])
            if len(a_types) < len(module_arg_types):
                values = hdlmodule.scope.param_default_values()
                offs = len(a_types)
                for v in values[offs:]:
                    if isinstance(a, bool):
                        a_types.append('b')
                    elif isinstance(a, int):
                        a_types.append('i')
                    else:
                        # no default value
                        break
            if a_types == module_arg_types:
                return model(*args, **kwargs)
        raise ValueError('model not found')
    return model_selector


import sys
from io import StringIO
import contextlib


def resolve_watch_signals(simulator, watch_str):
    """Resolve --watch signal names to (name, value) pairs.

    Uses the name table built from the core model tree. Signal names like "m.i"
    are matched by trying:
    1. Direct lookup by name in the name table (hierarchical match)
    2. Suffix match: look for any signal ending with the signal's leaf name,
       where the prefix matches a subscope or instance name
    """
    if not watch_str:
        return
    from polyphony.simulator import _build_name_table
    name_table = _build_name_table(simulator)
    # Reverse: name -> (id, value)
    reverse_table = {}
    for core_model in simulator.models:
        _build_reverse_table(core_model, reverse_table)
    signal_names = [s.strip() for s in watch_str.split(',')]
    for sig_name in signal_names:
        # Try exact match in name table values
        found = False
        for val_id, hier_name in name_table.items():
            if hier_name == sig_name:
                value = reverse_table.get(val_id)
                if value is not None:
                    simulator.observer.add_watch(sig_name, value)
                    found = True
                    break
        if found:
            continue
        # Walk the dot-separated path on each core model
        parts = sig_name.split('.')
        # Skip first part (instance variable name in testbench, e.g. "s")
        attr_path = parts[1:] if len(parts) > 1 else parts
        for core_model in simulator.models:
            obj = _walk_model_path(core_model, attr_path)
            if obj is not None:
                if isinstance(obj, Port):
                    simulator.observer.add_watch(sig_name, obj.value)
                    found = True
                elif isinstance(obj, (Net, Reg)):
                    simulator.observer.add_watch(sig_name, obj)
                    found = True
                break
        if not found:
            print(f"Warning: signal '{sig_name}' not found, skipping")


def _walk_model_path(core_model, path_parts):
    """Walk a dot-path like ['i', 'tvalid'] on the core model tree.

    Intermediate Model objects are unwrapped to their core model.
    Returns the final attribute (Port/Net/Reg) or None.
    """
    obj = core_model
    for part in path_parts:
        attr = getattr(obj, part, None)
        if attr is None:
            return None
        if isinstance(attr, Model):
            obj = super(Model, attr).__getattribute__("__model")
        else:
            return attr
    return None


def _build_reverse_table(core_model, table):
    """Build {id(value): value} mapping for all signals in the model tree."""
    for attr_name, val in vars(core_model).items():
        if attr_name.startswith('_') or attr_name in ('hdlmodule', 'clk', 'rst'):
            continue
        if isinstance(val, (Net, Reg)):
            table[id(val)] = val
        elif isinstance(val, Port):
            if val.value is not None:
                table[id(val.value)] = val.value
        elif isinstance(val, Model):
            sub_core = super(Model, val).__getattribute__("__model")
            _build_reverse_table(sub_core, table)


def simulate_on_python(casefile_path, source_text, scopes, simu_options):
    finishes = []
    casename = case_name_from_path(casefile_path)
    main_py_module = types.ModuleType('__main__')
    try:
        code_obj = compile(source_text, casefile_path, 'exec')
        casefile_dir = os.path.dirname(casefile_path)
        sys.path.append(casefile_dir)
        exec(code_obj, main_py_module.__dict__)
    except Exception as e:
        print(e)
        finishes.append('FAIL')
        return finishes
    py_objects = []
    for key, value in vars(main_py_module).items():
        if inspect.isclass(value) and getattr(value, '_is_module', False):
            py_objects.append(value)
        elif inspect.isfunction(value) and key.startswith(casename):
            py_objects.append(value)

    for testbench in env.testbenches:
        test = getattr(main_py_module, testbench.base_name)
        if not test:
            continue
        models = {}
        test_hdlmodule = env.hdlscope(testbench)
        for _, sub_hdlmodule, _, _ in test_hdlmodule.sub_modules.values():
            if sub_hdlmodule.name in models:
                continue
            model = SimulationModelBuilder().build_model(sub_hdlmodule, main_py_module, is_top=True)
            models[sub_hdlmodule.name] = (model, sub_hdlmodule)
        # Replacing classes/functions under test with a model selector
        for obj in py_objects:
            if inspect.isclass(obj):
                test._orig_func.__globals__[obj.__name__] = model_selector_with_argv(models)
            elif inspect.isfunction(obj):
                test._orig_func.__globals__[obj.__name__] = model_selector_with_argtypes(models, obj.__name__)
            else:
                assert False
        # exec test
        try:
            simulate_models = [model for model, _ in models.values()]
            test._orig_func._execute_on_simu = True
            simulator = Simulator(simulate_models)
            simulator.case_name = casename
            watch_signals = getattr(simu_options, 'watch_signals', '')
            if simu_options.verilog_dump or watch_signals:
                vcd_path = f"{TMP_DIR}{os.sep}{casename}_py.vcd"
                log_path = f"{TMP_DIR}{os.sep}{casename}_py.log"
                simulator.observer = SimulationObserver(vcd_file=vcd_path, log_file=log_path)
            with simulator:
                if watch_signals and simulator.observer:
                    resolve_watch_signals(simulator, watch_signals)
                    simulator.observer.dump_initial(simulator.clock_time)
                test()
            finishes.append('OK')
        except HDLAssertionError as e:  # from hdlmodule code
            print(f'ASSERTION FAILED: {e.args[0]}')
            print('[PYTHON SIMULATION] FAILED:' + casefile_path)
            finishes.append('FAIL')
        except AssertionError as e:  # from python test code
            _, _, tb = sys.exc_info()
            tb_info = traceback.extract_tb(tb)
            filename, line, _, code = tb_info[-1]
            if filename == casefile_path:
                print(f'ASSERTION FAILED: {casename}.py:{line} {code}')
                print(f'[PYTHON SIMULATION] FAILED: {filename}')
            else:
                print(f'[PYTHON SIMULATION] INTERNAL ERROR: {filename}:{line} {code}')
                traceback.print_exc()
            finishes.append('FAIL')
        except Exception as e:
            _, _, tb = sys.exc_info()
            tb_info = traceback.extract_tb(tb)
            filename, line, _, code = tb_info[-1]
            print(f'[PYTHON SIMULATION] INTERNAL ERROR: {filename}:{line} {code}')
            traceback.print_exc()
            finishes.append('FAIL')
    return finishes


if __name__ == '__main__':
    options = parse_options()
    hdl, py = exec_test(options.source, simu_options=options)
    hdl_success = True if '' not in hdl and 'FAIL' not in hdl else False
    py_success = True if not options.enable_python or ('' not in py and 'FAIL' not in py) else False
    if hdl_success and py_success:
        print('OK')