refactor: add JobSpec common abstraction

The `Deploy` classes are coupled to the scheduler and also the flow
objects in a way that is not clear. It's also not clear which of the
public attributes are depended on externally and which are internal
working out.

The intention is to make clear what the interface is that the scheduler
depends on and what the flow object depends on as a result of the job
runs. Initially I tried to refactor the `Deploy` objects themselves to
make the existing objects clear, however this failed due to too many
moving parts and the level of coupling.

This commit introduces a `JobSpec` pydantic data model that gives a fully
typed and runtime validated model containing everything the scheduler
needs to run the job. The scheduler no longer receives `Deploy` objects,
but instead `JobSpec` objects created from `Deploy` objects. Which means
`Deploy` objects become `JobSpec` factories.

In addition, the obvious dependencies on `Deploy` objects within the
flow objects have been replaced with an expanded `CompletedJobStatus`
pydantic model. There may still be dependencies that exist that have
been missed, but the main ones have been removed in the results
processing path.

This opens the way for writing tests for the scheduler, and potentially
the launchers, against clear interfaces using data classes as inputs and
outputs.

There is room for improvement in the data classes, as some attributes can
potentially be removed with some refactoring.

Signed-off-by: James McCorrie <[email protected]>

Author: machshev

ruff-ci.toml

@@ -92,7 +92,6 @@ ignore = [
     "S103",
     "S202",
     "S311",
-    "S602",
     "S605",
     "SIM115",
     "SLF001",

src/dvsim/cli.py

@@ -82,7 +82,7 @@ def resolve_scratch_root(arg_scratch_root, proj_root):
         arg_scratch_root = os.path.realpath(arg_scratch_root)
 
     try:
-        os.makedirs(arg_scratch_root, exist_ok=True)
+        Path(arg_scratch_root).mkdir(exist_ok=True, parents=True)
     except PermissionError as e:
         log.fatal(f"Failed to create scratch root {arg_scratch_root}:\n{e}.")
         sys.exit(1)
@@ -238,7 +238,7 @@ def copy_repo(src, dest) -> None:
     log.verbose("[copy_repo] [cmd]: \n%s", " ".join(cmd))
 
     # Make sure the dest exists first.
-    os.makedirs(dest, exist_ok=True)
+    Path(dest).mkdir(exist_ok=True, parents=True)
     try:
         subprocess.run(cmd, check=True, capture_output=True)
     except subprocess.CalledProcessError as e:

src/dvsim/flow/base.py

@@ -16,9 +16,10 @@
 import hjson
 
 from dvsim.flow.hjson import set_target_attribute
+from dvsim.job.data import CompletedJobStatus
 from dvsim.launcher.factory import get_launcher_cls
 from dvsim.logging import log
-from dvsim.scheduler import CompletedJobStatus, Scheduler
+from dvsim.scheduler import Scheduler
 from dvsim.utils import (
     find_and_substitute_wildcards,
     md_results_to_html,
@@ -403,7 +404,7 @@ def create_deploy_objects(self) -> None:
         for item in self.cfgs:
             item._create_deploy_objects()
 
-    def deploy_objects(self) -> Mapping[str, CompletedJobStatus]:
+    def deploy_objects(self) -> Sequence[CompletedJobStatus]:
         """Public facing API for deploying all available objects.
 
         Runs each job and returns a map from item to status.
@@ -430,13 +431,13 @@ def deploy_objects(self) -> Mapping[str, CompletedJobStatus]:
             )
 
         return Scheduler(
-            items=deploy,
+            items=[d.get_job_spec() for d in deploy],
             launcher_cls=get_launcher_cls(),
             interactive=self.interactive,
         ).run()
 
     @abstractmethod
-    def _gen_results(self, results: Mapping[str, CompletedJobStatus]) -> str:
+    def _gen_results(self, results: Sequence[CompletedJobStatus]) -> str:
         """Generate flow results.
 
         The function is called after the flow has completed. It collates
@@ -446,18 +447,18 @@ def _gen_results(self, results: Mapping[str, CompletedJobStatus]) -> str:
         report, which is in markdown format.
 
         Args:
-            results: dictionary mapping deployed item names to completed job status.
+            results: completed job status objects.
 
         Returns:
             Results as a formatted string
 
         """
 
-    def gen_results(self, results: Mapping[str, CompletedJobStatus]) -> None:
+    def gen_results(self, results: Sequence[CompletedJobStatus]) -> None:
         """Generate flow results.
 
         Args:
-            results: dictionary mapping deployed item names to completed job status.
+            results: completed job status objects.
 
         """
         for item in self.cfgs:
@@ -476,7 +477,7 @@ def gen_results(self, results: Mapping[str, CompletedJobStatus]) -> None:
             self.write_results(self.results_html_name, self.results_summary_md)
 
     @abstractmethod
-    def gen_results_summary(self) -> None:
+    def gen_results_summary(self) -> str:
         """Public facing API to generate summary results for each IP/cfg file."""
 
     def write_results(self, html_filename: str, text_md: str, json_str: str | None = None) -> None:

src/dvsim/flow/formal.py

@@ -228,7 +228,7 @@ def _gen_results(self, results):
                 "results.hjson",
             )
             try:
-                with open(result_data) as results_file:
+                with Path(result_data).open() as results_file:
                     self.result = hjson.load(results_file, use_decimal=True)
             except OSError as err:
                 log.warning("%s", err)

src/dvsim/flow/one_shot.py

@@ -4,7 +4,7 @@
 
 """Class describing a one-shot build configuration object."""
 
-import os
+import pathlib
 from collections import OrderedDict
 
 from dvsim.flow.base import FlowCfg
@@ -135,7 +135,7 @@ def _create_dirs(self) -> None:
         """Create initial set of directories."""
         for link in self.links:
             rm_path(self.links[link])
-            os.makedirs(self.links[link])
+            pathlib.Path(self.links[link]).mkdir(parents=True)
 
     def _create_deploy_objects(self) -> None:
         """Create deploy objects from build modes."""

src/dvsim/flow/sim.py

@@ -6,25 +6,24 @@
 
 import fnmatch
 import json
-import os
 import re
 import sys
 from collections import OrderedDict, defaultdict
-from collections.abc import Mapping
+from collections.abc import Sequence
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import ClassVar
 
 from tabulate import tabulate
 
 from dvsim.flow.base import FlowCfg
+from dvsim.job.data import CompletedJobStatus, JobSpec
 from dvsim.job.deploy import (
     CompileSim,
     CovAnalyze,
     CovMerge,
     CovReport,
     CovUnr,
-    Deploy,
     RunTest,
 )
 from dvsim.logging import log
@@ -423,7 +422,7 @@ def _create_dirs(self) -> None:
         """Create initial set of directories."""
         for link in self.links:
             rm_path(self.links[link])
-            os.makedirs(self.links[link])
+            Path(self.links[link]).mkdir(parents=True)
 
     def _expand_run_list(self, build_map):
         """Generate a list of tests to be run.
@@ -540,8 +539,10 @@ def _create_deploy_objects(self) -> None:
         self._create_dirs()
 
     def _cov_analyze(self) -> None:
-        """Use the last regression coverage data to open up the GUI tool to
-        analyze the coverage.
+        """Open GUI tool for coverage analysis.
+
+        Use the last regression coverage data to open up the GUI tool to analyze
+        the coverage.
         """
         # Create initial set of directories, such as dispatched, passed etc.
         self._create_dirs()
@@ -555,8 +556,10 @@ def cov_analyze(self) -> None:
             item._cov_analyze()
 
     def _cov_unr(self) -> None:
-        """Use the last regression coverage data to generate unreachable
-        coverage exclusions.
+        """Generate unreachable coverage exclusions.
+
+        Use the last regression coverage data to generate unreachable coverage
+        exclusions.
         """
         # TODO, Only support VCS
         if self.tool not in ["vcs", "xcelium"]:
@@ -573,7 +576,7 @@ def cov_unr(self) -> None:
         for item in self.cfgs:
             item._cov_unr()
 
-    def _gen_json_results(self, run_results: Mapping[Deploy, str]) -> str:
+    def _gen_json_results(self, run_results: Sequence[CompletedJobStatus]) -> str:
         """Return the run results as json-formatted dictionary."""
 
         def _pct_str_to_float(s: str) -> float | None:
@@ -589,14 +592,8 @@ def _pct_str_to_float(s: str) -> float | None:
 
         def _test_result_to_dict(tr) -> dict:
             """Map a test result entry to a dict."""
-            job_time_s = (
-                tr.job_runtime.with_unit("s").get()[0] if tr.job_runtime is not None else None
-            )
-            sim_time_us = (
-                tr.simulated_time.with_unit("us").get()[0]
-                if tr.simulated_time is not None
-                else None
-            )
+            job_time_s = tr.job_runtime
+            sim_time_us = tr.simulated_time
             pass_rate = tr.passing * 100.0 / tr.total if tr.total > 0 else 0
             return {
                 "name": tr.name,
@@ -644,7 +641,7 @@ def _test_result_to_dict(tr) -> dict:
 
         # If the testplan does not yet have test results mapped to testpoints,
         # map them now.
-        sim_results = SimResults(self.deploy, run_results)
+        sim_results = SimResults(results=run_results)
         if not self.testplan.test_results_mapped:
             self.testplan.map_test_results(test_results=sim_results.table)
 
@@ -707,12 +704,14 @@ def _test_result_to_dict(tr) -> dict:
         # Extract failure buckets.
         if sim_results.buckets:
             by_tests = sorted(sim_results.buckets.items(), key=lambda i: len(i[1]), reverse=True)
+
             for bucket, tests in by_tests:
                 unique_tests = defaultdict(list)
                 for test, line, context in tests:
-                    if not isinstance(test, RunTest):
+                    if test.job_type != "RunTest":
                         continue
                     unique_tests[test.name].append((test, line, context))
+
                 fts = []
                 for test_name, test_runs in unique_tests.items():
                     frs = []
@@ -721,12 +720,13 @@ def _test_result_to_dict(tr) -> dict:
                             {
                                 "seed": str(test.seed),
                                 "failure_message": {
-                                    "log_file_path": test.get_log_path(),
+                                    "log_file_path": str(test.log_path),
                                     "log_file_line_num": line,
                                     "text": "".join(context),
                                 },
                             },
                         )
+
                     fts.append(
                         {
                             "name": test_name,
@@ -747,7 +747,7 @@ def _test_result_to_dict(tr) -> dict:
         # Return the `results` dictionary as json string.
         return json.dumps(self.results_dict)
 
-    def _gen_results(self, results: Mapping[Deploy, str]) -> str:
+    def _gen_results(self, results: Sequence[CompletedJobStatus]) -> str:
         """Generate simulation results.
 
         The function is called after the regression has completed. It collates the
@@ -761,12 +761,12 @@ def _gen_results(self, results: Mapping[Deploy, str]) -> str:
         def indent_by(level: int) -> str:
             return " " * (4 * level)
 
-        def create_failure_message(test, line, context):
+        def create_failure_message(test: JobSpec, line, context):
             message = [f"{indent_by(2)}* {test.qual_name}\\"]
             if line:
-                message.append(f"{indent_by(2)}  Line {line}, in log " + test.get_log_path())
+                message.append(f"{indent_by(2)}  Line {line}, in log {test.log_path}")
             else:
-                message.append(f"{indent_by(2)} Log {test.get_log_path()}")
+                message.append(f"{indent_by(2)} Log {test.log_path}")
             if context:
                 message.append("")
                 lines = [f"{indent_by(4)}{c.rstrip()}" for c in context]
@@ -817,8 +817,7 @@ def create_bucket_report(buckets):
             fail_msgs.append("")
             return fail_msgs
 
-        deployed_items = self.deploy
-        sim_results = SimResults(deployed_items, results)
+        sim_results = SimResults(results=results)
 
         # Generate results table for runs.
         results_str = "## " + self.results_title + "\n"
@@ -887,8 +886,17 @@ def create_bucket_report(buckets):
 
             # Append coverage results if coverage was enabled.
             if self.cov_report_deploy is not None:
-                report_status = results[self.cov_report_deploy.full_name]
-                if report_status == "P":
+                report_status: CompletedJobStatus | None = None
+                for job_status in results:
+                    if job_status.full_name == self.cov_report_deploy.full_name:
+                        report_status = job_status
+                        break
+
+                if report_status is None:
+                    msg = f"Coverage report not found for {self.cov_report_deploy.full_name}"
+                    raise KeyError(msg)
+
+                if report_status.status == "P":
                     results_str += "\n## Coverage Results\n"
                     # Link the dashboard page using "cov_report_page" value.
                     if hasattr(self, "cov_report_page"):
@@ -907,7 +915,7 @@ def create_bucket_report(buckets):
         self.results_md = results_str
         return results_str
 
-    def gen_results_summary(self):
+    def gen_results_summary(self) -> str:
         """Generate the summary results table.
 
         This method is specific to the primary cfg. It summarizes the results