diff --git a/src/dvsim/sim/flow.py b/src/dvsim/sim/flow.py index 6dc400c4..1ebeea32 100644 --- a/src/dvsim/sim/flow.py +++ b/src/dvsim/sim/flow.py @@ -43,7 +43,7 @@ from dvsim.sim_results import BucketedFailures, SimResults from dvsim.test import Test from dvsim.testplan import Testplan -from dvsim.tool.utils import get_sim_tool_plugin +from dvsim.tool.utils import get_sim_tool_plugin, query_tool_version from dvsim.utils import TS_FORMAT, rm_path from dvsim.utils.fs import relative_to from dvsim.utils.git import git_https_url_with_commit @@ -716,7 +716,7 @@ def _gen_json_results( url=url, revision_info=self.revision, ) - tool = ToolMeta(name=self.tool.lower(), version="unknown") + tool = ToolMeta(name=self.tool.lower(), version=query_tool_version(self.tool) or "unknown") build_seed = self.build_seed if not self.run_only else None diff --git a/src/dvsim/sim/tool/base.py b/src/dvsim/sim/tool/base.py index aa19e4f4..b46a907f 100644 --- a/src/dvsim/sim/tool/base.py +++ b/src/dvsim/sim/tool/base.py @@ -5,8 +5,9 @@ """EDA simulation tool interface.""" from collections.abc import Mapping, Sequence +from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Protocol, runtime_checkable +from typing import TYPE_CHECKING, ClassVar, Protocol, runtime_checkable from dvsim.job.data import JobSpec from dvsim.sim.data import CoverageMetrics @@ -14,13 +15,32 @@ if TYPE_CHECKING: from dvsim.job.deploy import Deploy -__all__ = ("SimTool",) +__all__ = ("SimTool", "VersionQuery") + + +@dataclass(frozen=True) +class VersionQuery: + """Declarative description of how to query an EDA tool for its version. + + Attributes: + cmd: the command that makes the tool print its version. + pattern: a regex applied (in multiline mode) to the combined + stdout/stderr of ``cmd``. The first capture group is used as the + version string. + + """ + + cmd: str + pattern: str @runtime_checkable class SimTool(Protocol): """Simulation tool interface required by the Sim workflow.""" + version_query: ClassVar[VersionQuery | None] + """How to query this tool's version, or ``None`` if unsupported.""" + @staticmethod def get_cov_summary_table(cov_report_path: Path) -> tuple[Sequence[Sequence[str]], str]: """Get a coverage summary. diff --git a/src/dvsim/sim/tool/vcs.py b/src/dvsim/sim/tool/vcs.py index dc80714c..88090c90 100644 --- a/src/dvsim/sim/tool/vcs.py +++ b/src/dvsim/sim/tool/vcs.py @@ -7,10 +7,11 @@ import re from collections.abc import Mapping, Sequence from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar from dvsim.job.data import JobSpec from dvsim.sim.data import CodeCoverageMetrics, CoverageMetrics +from dvsim.sim.tool.base import VersionQuery if TYPE_CHECKING: from dvsim.job.deploy import Deploy @@ -21,6 +22,12 @@ class VCS: """Implement VCS tool support.""" + # `vcs -id` reports a line like: "Compiler version = VCS X-2025.06-SP2-1_Full64". + version_query: ClassVar[VersionQuery | None] = VersionQuery( + cmd="vcs -full64 -id", + pattern=r"^Compiler version\s*=\s*(?:VCS\s+)?(\S+)", + ) + @staticmethod def get_cov_summary_table(cov_report_path: Path) -> tuple[Sequence[Sequence[str]], str]: """Get a coverage summary. diff --git a/src/dvsim/sim/tool/xcelium.py b/src/dvsim/sim/tool/xcelium.py index a6401707..4745d53f 100644 --- a/src/dvsim/sim/tool/xcelium.py +++ b/src/dvsim/sim/tool/xcelium.py @@ -8,10 +8,11 @@ from collections import defaultdict from collections.abc import Mapping, Sequence from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar from dvsim.job.data import JobSpec from dvsim.sim.data import CodeCoverageMetrics, CoverageMetrics +from dvsim.sim.tool.base import VersionQuery if TYPE_CHECKING: from dvsim.job.deploy import Deploy @@ -22,6 +23,12 @@ class Xcelium: """Implement Xcelium tool support.""" + # `xrun -version` reports a line like: "TOOL: xrun(64) 24.03-s007". + version_query: ClassVar[VersionQuery | None] = VersionQuery( + cmd="xrun -version", + pattern=r"^TOOL:\s*xrun\(\d+\)\s+(\S+)", + ) + @staticmethod def get_cov_summary_table(cov_report_path: Path) -> tuple[Sequence[Sequence[str]], str]: """Get a coverage summary. diff --git a/src/dvsim/tool/utils.py b/src/dvsim/tool/utils.py index ca00d87c..fb3170c7 100644 --- a/src/dvsim/tool/utils.py +++ b/src/dvsim/tool/utils.py @@ -4,13 +4,19 @@ """EDA Tool base.""" +import re +import shlex +import subprocess +from collections.abc import Callable +from functools import cache + from dvsim.logging import log from dvsim.sim.tool.base import SimTool from dvsim.sim.tool.vcs import VCS from dvsim.sim.tool.xcelium import Xcelium from dvsim.sim.tool.z01x import Z01X -__all__ = ("get_sim_tool_plugin",) +__all__ = ("get_sim_tool_plugin", "query_tool_version") _SUPPORTED_SIM_TOOLS = { "vcs": VCS, @@ -18,6 +24,10 @@ "z01x": Z01X, } +# EDA tools should respond to a `--version`-style query near-instantly, but guard +# against a hung/misconfigured tool so report generation cannot block forever. +_VERSION_QUERY_TIMEOUT_S = 30 + def get_sim_tool_plugin(tool: str) -> SimTool: """Get a simulation tool plugin.""" @@ -31,3 +41,73 @@ def get_sim_tool_plugin(tool: str) -> SimTool: raise NotImplementedError(msg) return _SUPPORTED_SIM_TOOLS[tool] + + +@cache +def _run_version_command(cmd: str) -> str | None: + """Run a tool version-query command, returning its combined output or None. + + The tool's version may be printed to either stdout or stderr, so both are + captured and concatenated. Any failure to launch or run the command (the + tool is not on PATH, a non-EDA host, a timeout, ...) is treated as "version + unknown" rather than an error: querying the version must never break a run. + + Results are cached for the lifetime of the process since a tool's version is + invariant across a single dvsim invocation; this dedupes the query when + multiple configs share a tool. + """ + try: + result = subprocess.run( # noqa: S603 + shlex.split(cmd), + capture_output=True, + text=True, + timeout=_VERSION_QUERY_TIMEOUT_S, + check=False, + ) + except (OSError, subprocess.SubprocessError) as e: + log.debug("Failed to query tool version via '%s': %s", cmd, e) + return None + + return result.stdout + result.stderr + + +def query_tool_version( + tool: str, + *, + run: Callable[[str], str | None] = _run_version_command, +) -> str | None: + """Query the version of an EDA tool from the runtime environment. + + Runs the tool plugin's declared version-query command and parses the version + out of its output. This is a local, best-effort probe: it assumes the tool + is available on PATH of the host running dvsim. + + TODO: for farm launchers (LSF/SGE/SLURM) the tool may only exist on the + compute nodes. To be correct there, this should run as a lightweight + preflight job dispatched through the runtime backend so it inherits the same + environment (e.g. `module load`) as real jobs. + + Args: + tool: the name of the tool to query (as passed to `--tool`). + run: callable that executes the query command and returns its combined + output, or None on failure. Injectable for testing. + + Returns: + The parsed version string, or None if the tool declares no query, the + command failed, or the output did not match the expected pattern. + + """ + query = get_sim_tool_plugin(tool).version_query + if query is None: + return None + + output = run(query.cmd) + if output is None: + return None + + match = re.search(query.pattern, output, re.MULTILINE) + if match is None: + log.debug("Could not parse %s version from output of '%s'", tool, query.cmd) + return None + + return match.group(1).strip() diff --git a/tests/tool/test_utils.py b/tests/tool/test_utils.py index 28c211da..839105ab 100644 --- a/tests/tool/test_utils.py +++ b/tests/tool/test_utils.py @@ -5,12 +5,25 @@ """Test the EDA tool utilities.""" import pytest -from hamcrest import assert_that, equal_to, instance_of +from hamcrest import assert_that, equal_to, instance_of, none -from dvsim.sim.tool.base import SimTool -from dvsim.tool.utils import _SUPPORTED_SIM_TOOLS, get_sim_tool_plugin +from dvsim.sim.tool.base import SimTool, VersionQuery +from dvsim.tool.utils import _SUPPORTED_SIM_TOOLS, get_sim_tool_plugin, query_tool_version -__all__ = ("TestEDAToolPlugins",) +__all__ = ("TestEDAToolPlugins", "TestToolVersionQuery") + +# Representative output captured from the real tools' version-query commands. +_VCS_ID_OUTPUT = """\ +vcs script version : X-2025.06 +machine name = dab +machine type = linux64 +machine os = Linux 6.18.39 +The FLEXlm host ID of this machine is "1a2b3c4d5e6f 7g8h9i0j1k2l" +Compiler version = VCS X-2025.06-SP2-1_Full64 +VCS Build Date = Jan 29 2026 20:22:37 +""" + +_XRUN_VERSION_OUTPUT = "TOOL: xrun(64) 24.03-s007\n" class TestEDAToolPlugins: @@ -32,3 +45,36 @@ def test_plugins_implement_simtool_protocol(tool: str) -> None: plugin = get_sim_tool_plugin(tool) assert_that(plugin, instance_of(SimTool)) + + @staticmethod + @pytest.mark.parametrize("tool", _SUPPORTED_SIM_TOOLS.keys()) + def test_plugins_declare_version_query(tool: str) -> None: + """Test that every plugin declares a version query (inherited or not).""" + assert_that(get_sim_tool_plugin(tool).version_query, instance_of(VersionQuery)) + + +class TestToolVersionQuery: + """Test parsing of tool versions from version-query command output.""" + + @staticmethod + @pytest.mark.parametrize( + ("tool", "output", "expected"), + [ + ("vcs", _VCS_ID_OUTPUT, "X-2025.06-SP2-1_Full64"), + ("z01x", _VCS_ID_OUTPUT, "X-2025.06-SP2-1_Full64"), + ("xcelium", _XRUN_VERSION_OUTPUT, "24.03-s007"), + ], + ) + def test_parses_version(tool: str, output: str, expected: str) -> None: + """Test that the version is parsed from representative tool output.""" + assert_that(query_tool_version(tool, run=lambda _cmd: output), equal_to(expected)) + + @staticmethod + def test_returns_none_when_command_fails() -> None: + """Test that a failed query (runner returns None) yields None.""" + assert_that(query_tool_version("vcs", run=lambda _cmd: None), none()) + + @staticmethod + def test_returns_none_when_output_unrecognised() -> None: + """Test that unparseable output yields None rather than a bad version.""" + assert_that(query_tool_version("vcs", run=lambda _cmd: "totally unexpected"), none())