Address review: register metadata engine, coverage wiring, virtual seqr fix, Xcelium wrapper, generic RAL access

src/api/server.py

@@ -46,7 +46,7 @@ class HealthResponse(BaseModel):
     status: str = "ok"
     version: str = "0.3.0"
     api_version: str = "v1"
-    simulators: List[str] = ["stub", "icarus"]
+    simulators: List[str] = ["stub", "icarus", "vcs", "questa", "xcelium"]
 
 
 class VersionInfo(BaseModel):

src/config.py

@@ -125,7 +125,7 @@ class AutoTrainConfig(BaseModel):
     max_iterations: int = Field(default=5, ge=1, le=50)
     coverage_target: float = Field(default=90.0, ge=0.0, le=100.0)
     coverage_gain_min: float = Field(default=2.0, ge=0.0, description="Min % gain per iteration to continue")
-    simulator: str = Field(default="stub", pattern=r"^(stub|icarus|vcs|questa)$")
+    simulator: str = Field(default="stub", pattern=r"^(stub|icarus|vcs|questa|xcelium|xrun)$")
     sim_timeout: int = Field(default=300, ge=10)
     num_seeds: int = Field(default=3, ge=1, le=20, description="Number of regression seeds per iteration")
     generate_regression_test: bool = True

src/generation/templates/register_info_pkg.sv.j2

@@ -0,0 +1,129 @@
+// UVM Register Metadata Package for {{ spec.design_name }}
+// Auto-generated from YAML spec — drives metadata-driven sequences
+// Enables generic register access without protocol-specific knowledge
+{% set regs = spec.registers if spec.registers else [] %}
+{% set addr_ints = regs|map(attribute='address')|map('replace', '0x', '')|map('replace', 'h', '')|map('int', 16)|list %}
+{% set max_addr_val = addr_ints|max if addr_ints else 0 %}
+{% set addr_bits = [max_addr_val.bit_length(), 3]|max %}
+{% set sizes = regs|map(attribute='size')|select('number')|list %}
+{% set data_width = sizes|max if sizes else 8 %}
+
+package {{ spec.design_name }}_reg_info_pkg;
+
+  // -----------------------------------------------------------------------
+  // Access Policy Enum
+  // -----------------------------------------------------------------------
+  typedef enum bit [2:0] {
+    RO  = 3'b001,  // Read-only
+    WO  = 3'b010,  // Write-only
+    RW  = 3'b011,  // Read-Write
+    RC  = 3'b100,  // Read-Clear
+    W1C = 3'b101,  // Write-1-to-Clear
+    NA  = 3'b000   // Not applicable
+  } access_policy_e;
+
+  // -----------------------------------------------------------------------
+  // Field Information
+  // -----------------------------------------------------------------------
+  typedef struct {
+    string        name;
+    int           msb;
+    int           lsb;
+    int           width;
+    access_policy_e access;
+    logic [63:0]  reset_val;
+  } field_info_t;
+
+  // -----------------------------------------------------------------------
+  // Register Information
+  // -----------------------------------------------------------------------
+  typedef struct {
+    string        name;
+    logic [{{ addr_bits-1 }}:0] address;
+    int           size;
+    access_policy_e access;
+    logic [63:0]  reset_val;
+    int           num_fields;
+    field_info_t  fields[16];
+  } register_info_t;
+
+  // -----------------------------------------------------------------------
+  // Register Database
+  // -----------------------------------------------------------------------
+  function automatic register_info_t get_register_by_name(string name);
+    register_info_t r;
+    r.name = "";
+    {% for reg in regs %}
+    if (name == "{{ reg.name|lower }}") begin
+      r.name = "{{ reg.name }}";
+      r.address = {{ addr_bits }}'h{{ reg.address.replace('0x', '').replace('h', '') }};
+      r.size = {{ reg.size if reg.size else 8 }};
+      r.reset_val = {% if reg.reset is defined and reg.reset %}{{ reg.size if reg.size else 8 }}'h{{ reg.reset|string|replace('0x', '')|replace('h', '') }}{% else %}'h0{% endif %};
+      {% if reg.access is defined and reg.access %}
+      {% set ra = reg.access|lower %}
+      {% if ra == 'ro' %}r.access = RO;
+      {% elif ra == 'wo' %}r.access = WO;
+      {% elif ra == 'rw' %}r.access = RW;
+      {% elif ra == 'rc' %}r.access = RC;
+      {% elif ra == 'w1c' %}r.access = W1C;
+      {% else %}r.access = RW;
+      {% endif %}
+      {% else %}r.access = RW;
+      {% endif %}
+      {% if reg.fields %}
+      r.num_fields = {{ reg.fields|length }};
+      {% for f in reg.fields %}
+      {% if ':' in f.bits %}{% set parts = f.bits.split(':') %}{% set msb = parts[0]|int %}{% set lsb = parts[1]|int %}{% else %}{% set msb = f.bits|int - 1 %}{% set lsb = 0 %}{% endif %}
+      r.fields[{{ loop.index0 }}].name = "{{ f.name }}";
+      r.fields[{{ loop.index0 }}].msb = {{ msb }};
+      r.fields[{{ loop.index0 }}].lsb = {{ lsb }};
+      r.fields[{{ loop.index0 }}].width = {{ msb - lsb + 1 }};
+      r.fields[{{ loop.index0 }}].reset_val = {% if f.reset is defined and f.reset %}{{ msb - lsb + 1 }}'h{{ f.reset|string|replace('0x', '')|replace('h', '') }}{% else %}'h0{% endif %};
+      {% endfor %}
+      {% else %}
+      r.num_fields = 0;
+      {% endif %}
+      return r;
+    end
+    {% endfor %}
+    return r;
+  endfunction
+
+  function automatic register_info_t get_register_by_offset(logic [{{ addr_bits-1 }}:0] addr);
+    {% for reg in regs %}
+    if (addr == {{ addr_bits }}'h{{ reg.address.replace('0x', '').replace('h', '') }})
+      return get_register_by_name("{{ reg.name|lower }}");
+    {% endfor %}
+    register_info_t r;
+    r.name = "";
+    return r;
+  endfunction
+
+  // -----------------------------------------------------------------------
+  // Register Count
+  // -----------------------------------------------------------------------
+  function automatic int get_num_registers();
+    return {{ regs|length }};
+  endfunction
+
+  // -----------------------------------------------------------------------
+  // Register Name List (for iterating in sequences)
+  // -----------------------------------------------------------------------
+  function automatic void get_all_register_names(ref string names[$]);
+    {% for reg in regs %}
+    names.push_back("{{ reg.name|lower }}");
+    {% endfor %}
+  endfunction
+
+  // -----------------------------------------------------------------------
+  // Access Policy Helpers
+  // -----------------------------------------------------------------------
+  function automatic bit is_readable(access_policy_e ap);
+    return (ap inside {RO, RW, RC});
+  endfunction
+
+  function automatic bit is_writable(access_policy_e ap);
+    return (ap inside {WO, RW, W1C});
+  endfunction
+
+endpackage

src/generation/templates/scoreboard.sv.j2

@@ -282,12 +282,14 @@ class {{ spec.design_name }}_scoreboard extends uvm_scoreboard;
     end
 {% endif %}
 
-    if (tr.addr inside {3'h3, 3'h4, 3'h7}) begin
+    if (tr.addr inside { {% for r in regs %}{{ addr_bits }}'h{{ r.address.replace('0x', '').replace('h', '') }}{% if not loop.last %}, {% endif %}{% endfor %} }) begin
       if (tr.data === expected) begin
         match_count++;
+        reg_match_count[tr.addr]++;
         `uvm_info("SB", $sformatf("READ MATCH: addr=0x%0h data=0x%02h", tr.addr, tr.data), UVM_HIGH)
       end else begin
         mismatch_count++;
+        reg_mismatch_count[tr.addr]++;
         log_error($sformatf("READ MISMATCH: addr=0x%0h expected=0x%02h got=0x%02h",
                   tr.addr, expected, tr.data));
       end
@@ -636,6 +638,10 @@ class {{ spec.design_name }}_scoreboard extends uvm_scoreboard;
     end
     cov_wordlen_seen[dbits - 5] = 1;
     cov_parity_seen[pen + eps*2] = 1;
+{% if p == "uart" %}
+    if (cov_handle != null)
+      cov_handle.sample_uart_config(baud, dbits, sbits, pen + eps*2);
+{% endif %}
     `uvm_info("SB_REC_CFG", $sformatf("record_config: baud=%0d dbits=%0d sbits=%0d pen=%0d eps=%0d",
               baud, dbits, sbits, pen, eps), UVM_MEDIUM)
   endfunction

src/generation/templates/virtual_sequencer.sv.j2

@@ -13,8 +13,17 @@ class {{ spec.design_name }}_virtual_sequencer extends uvm_sequencer;
 
 {% if p == "uart" %}
   // Additional UART sub-sequencers for multi-port / layered protocols
+  // Number of sub-seqrs = number of interfaces in spec
   uvm_sequencer #({{ spec.design_name }}_seq_item) uart0_seqr;
+{% if spec.interfaces|length > 1 %}
   uvm_sequencer #({{ spec.design_name }}_seq_item) uart1_seqr;
+{% endif %}
+{% if spec.interfaces|length > 2 %}
+  uvm_sequencer #({{ spec.design_name }}_seq_item) uart2_seqr;
+{% endif %}
+{% if spec.interfaces|length > 3 %}
+  uvm_sequencer #({{ spec.design_name }}_seq_item) uart3_seqr;
+{% endif %}
 {% elif p == "apb" %}
   uvm_sequencer #({{ spec.design_name }}_seq_item) apb_seqr;
 {% elif p == "axi4lite" %}
@@ -36,7 +45,15 @@ class {{ spec.design_name }}_virtual_sequencer extends uvm_sequencer;
 {% if p == "uart" %}
     void'(uvm_config_db #(uvm_sequencer #({{ spec.design_name }}_seq_item))::get(this, "", "bus_seqr", bus_seqr));
     void'(uvm_config_db #(uvm_sequencer #({{ spec.design_name }}_seq_item))::get(this, "", "uart0_seqr", uart0_seqr));
+{% if spec.interfaces|length > 1 %}
     void'(uvm_config_db #(uvm_sequencer #({{ spec.design_name }}_seq_item))::get(this, "", "uart1_seqr", uart1_seqr));
+{% endif %}
+{% if spec.interfaces|length > 2 %}
+    void'(uvm_config_db #(uvm_sequencer #({{ spec.design_name }}_seq_item))::get(this, "", "uart2_seqr", uart2_seqr));
+{% endif %}
+{% if spec.interfaces|length > 3 %}
+    void'(uvm_config_db #(uvm_sequencer #({{ spec.design_name }}_seq_item))::get(this, "", "uart3_seqr", uart3_seqr));
+{% endif %}
 {% else %}
     void'(uvm_config_db #(uvm_sequencer #({{ spec.design_name }}_seq_item))::get(this, "", "bus_seqr", bus_seqr));
 {% endif %}

src/main.py

@@ -34,7 +34,7 @@ Examples:
     parser.add_argument("--auto-train", action="store_true", help="Enable coverage-driven auto-training loop")
     parser.add_argument("--max-iterations", type=int, default=5, help="Max auto-training iterations (default: 5)")
     parser.add_argument("--coverage-target", type=float, default=90.0, help="Coverage target %% (default: 90)")
-    parser.add_argument("--simulator", default="stub", choices=["stub", "icarus", "vcs", "questa"],
+    parser.add_argument("--simulator", default="stub", choices=["stub", "icarus", "vcs", "questa", "xcelium", "xrun"],
                         help="Simulator backend (default: stub)")
     return parser
 

src/models/template_model.py

@@ -21,19 +21,21 @@ class TemplateModel(GenerationModel):
     }
 
     TEMPLATE_MAP = {
+        "env/{name}_env.sv": "env.sv.j2",
+        "env/{name}_scoreboard.sv": "scoreboard.sv.j2",
+        "env/{name}_coverage_collector.sv": "coverage_collector.sv.j2",
+        "agent/{name}_driver.sv": "driver.sv.j2",
+        "agent/{name}_monitor.sv": "monitor.sv.j2",
+        "agent/{name}_sequencer.sv": "sequencer.sv.j2",
+        "agent/{name}_sequence_item.sv": "sequence_item.sv.j2",
+        "sequences/{name}_sequence.sv": "sequence.sv.j2",
+        "tests/{name}_test.sv": "test.sv.j2",
+        "sequences/{name}_reg_model_adapter.sv": "reg_model_adapter.sv.j2",
+        "sequences/{name}_register_info_pkg.sv": "register_info_pkg.sv.j2",
+        "{name}_reg_block.sv": "reg_block.sv.j2",
+        "{name}_interface.sv": "interface.sv.j2",
         "testbench.sv": "testbench.sv.j2",
-        "interface_{name}.sv": "interface.sv.j2",
-        "sequence_item_{name}.sv": "sequence_item.sv.j2",
-        "driver_{name}.sv": "driver.sv.j2",
-        "monitor_{name}.sv": "monitor.sv.j2",
-        "serial_monitor_{name}.sv": "serial_monitor.sv.j2",
-        "agent_{name}.sv": "agent.sv.j2",
-        "scoreboard_{name}.sv": "scoreboard.sv.j2",
-        "coverage_collector_{name}.sv": "coverage_collector.sv.j2",
-        "ral_model_{name}.sv": "ral_model.sv.j2",
-        "base_sequence_{name}.sv": "sequence.sv.j2",
-        "test_{name}.sv": "test.sv.j2",
-        "environment_{name}.sv": "env.sv.j2",
+        "rtl/protocol_core.v": "rtl/protocol_core.v.j2",
     }
 
     RTL_MAP = {

src/pipeline.py

@@ -364,6 +364,9 @@ class TBPipeline:
         if sim_type == "questa":
             from src.simulation.questa import QuestaSimulator
             return QuestaSimulator(work_dir=sim_output_path(self.cfg))
+        if sim_type == "xcelium" or sim_type == "xrun":
+            from src.simulation.xcelium import XceliumSimulator
+            return XceliumSimulator(work_dir=sim_output_path(self.cfg))
         return StubSimulator(work_dir=sim_output_path(self.cfg))
 
     def _merge_cfg(self, loaded: PipelineConfig) -> None:

src/simulation/__init__.py

@@ -1,6 +1,10 @@
 from src.simulation.base import CoverageBin, CoverageDB, SimResult, Simulator
 from src.simulation.icarus import IcarusSimulator
 from src.simulation.stub_sim import StubSimulator
+from src.simulation.vcs import VcsSimulator
+from src.simulation.questa import QuestaSimulator
+from src.simulation.xcelium import XceliumSimulator
 
 __all__ = ["Simulator", "SimResult", "CoverageBin", "CoverageDB",
-           "IcarusSimulator", "StubSimulator"]
+           "IcarusSimulator", "StubSimulator", "VcsSimulator",
+           "QuestaSimulator", "XceliumSimulator"]

src/simulation/xcelium.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import List, Optional
+
+from src.simulation.base import CoverageBin, SimResult, Simulator
+
+
+class XceliumSimulator(Simulator):
+    """Cadence Xcelium simulation backend.
+
+    Uses xrun for compile+elaborate+simulate in one step.
+    Detects xrun at runtime; falls back gracefully if not installed.
+    """
+
+    def __init__(
+        self,
+        work_dir: str = "sim_output",
+        xrun_path: str = "xrun",
+        compile_flags: Optional[List[str]] = None,
+        run_flags: Optional[List[str]] = None,
+    ):
+        super().__init__(work_dir)
+        self.xrun_path = xrun_path
+        self.compile_flags = compile_flags or [
+            "-64bit",
+            "-sv",
+            "-access",
+            "+rwc",
+            "-coverage",
+            "all",
+            "-nowarn",
+            "NONPORTTOPCONN",
+        ]
+        self.run_flags = run_flags or [
+            "-coverage",
+            "all",
+            "-exit",
+        ]
+
+    def run(
+        self,
+        files: List[str],
+        top: str = "testbench",
+        plusargs: Optional[List[str]] = None,
+    ) -> SimResult:
+        work_dir = Path(self.work_dir)
+        work_dir.mkdir(parents=True, exist_ok=True)
+
+        # Check if xrun is available
+        try:
+            subprocess.run(
+                [self.xrun_path, "-version"],
+                capture_output=True,
+                timeout=10,
+                check=True,
+            )
+        except (FileNotFoundError, subprocess.TimeoutExpired, subprocess.CalledProcessError):
+            return SimResult(
+                passed=False,
+                errors=[f"Xcelium ({self.xrun_path}) not found or not executable"],
+            )
+
+        # Write file list
+        flist_path = work_dir / "xrun_files.f"
+        flist_path.write_text("\n".join(str(f) for f in files), encoding="utf-8")
+
+        plusargs = plusargs or []
+        seed = None
+        for pa in plusargs:
+            if "+seed=" in pa:
+                seed = pa.split("=")[1]
+                break
+
+        xrun_cmd = [
+            self.xrun_path,
+            *self.compile_flags,
+            "-f",
+            str(flist_path),
+            "-top",
+            top,
+            "-l",
+            str(work_dir / "xrun_sim.log"),
+        ]
+        if seed:
+            xrun_cmd += [f"+ntb_random_seed={seed}"]
+        xrun_cmd += self.run_flags
+
+        try:
+            result = subprocess.run(
+                xrun_cmd,
+                cwd=str(work_dir),
+                capture_output=True,
+                timeout=600,
+            )
+            log = result.stdout + "\n" + result.stderr
+            log_file = work_dir / "xrun_sim.log"
+            if log_file.exists():
+                log = log_file.read_text(encoding="utf-8", errors="replace")
+
+            if result.returncode != 0:
+                errors = self._parse_errors(log)
+                return SimResult(
+                    passed=False,
+                    errors=errors or ["Xcelium simulation failed"],
+                    log_output=log[:5000],
+                )
+
+            return self.parse_coverage(log)
+
+        except subprocess.TimeoutExpired:
+            return SimResult(
+                passed=False,
+                errors=["Xcelium simulation timed out (>600s)"],
+            )
+        except Exception as e:
+            return SimResult(
+                passed=False,
+                errors=[f"Xcelium simulation error: {e}"],
+            )
+
+    def parse_coverage(self, log: str) -> SimResult:
+        bins: List[CoverageBin] = []
+        passed = True
+        errors: List[str] = []
+
+        xrun_cov = re.compile(
+            r"COVERAGE:\s+(\S+)\s+(\d+)/(\d+)\s+\[(HIT|MISS)\]"
+        )
+        for match in xrun_cov.finditer(log):
+            name = match.group(1)
+            hit = int(match.group(2))
+            goal = int(match.group(3))
+            bins.append(CoverageBin(name=name, hit_count=hit, goal=goal))
+            if hit < goal:
+                passed = False
+
+        # Xcelium coverage summary
+        total_cov = re.compile(
+            r"(Line|Toggle|Functional|Assertion)\s+Coverage:\s+([\d.]+)%"
+        )
+        for match in total_cov.finditer(log):
+            cov_type = match.group(1)
+            pct = float(match.group(2))
+            bins.append(CoverageBin(
+                name=f"xrun_{cov_type.lower()}_total",
+                hit_count=int(pct),
+                goal=100,
+            ))
+            if pct < 100:
+                passed = False
+
+        # Error extraction
+        err_patterns = re.compile(
+            r"\*[WE]|(Error|Warning|Fatal):\s*(.*)", re.IGNORECASE
+        )
+        for match in err_patterns.finditer(log):
+            err_text = match.group(0).strip()
+            if len(err_text) > 20:
+                errors.append(err_text[:200])
+
+        if not bins:
+            # No structured coverage; check for pass/fail
+            if re.search(r"TEST.*PASSED|UVM.*PASSED", log, re.IGNORECASE):
+                passed = True
+            elif re.search(r"TEST.*FAILED|UVM.*FAILED|#\s*FAIL", log, re.IGNORECASE):
+                passed = False
+                errors.append("Simulation FAILED (no structured coverage found)")
+
+        return SimResult(
+            passed=passed,
+            total_bins=len(bins),
+            covered_bins=sum(1 for b in bins if b.covered),
+            bins=bins,
+            errors=errors[:20],
+            log_output=log[:5000],
+        )
+
+    def _parse_errors(self, log: str) -> List[str]:
+        errors = []
+        for line in log.split("\n"):
+            if any(kw in line for kw in ["*E", "*W", "Error:", "Fatal:", "FAIL"]):
+                errors.append(line.strip()[:200])
+        return errors[:20]