refactor: add JobSpec common abstraction

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.
@@ -416,27 +417,36 @@ def deploy_objects(self) -> Mapping[str, CompletedJobStatus]:
             log.error("Nothing to run!")
             sys.exit(1)
 
+        jobs = [d.get_job_spec() for d in deploy]
+
         if os.environ.get("DVSIM_DEPLOY_DUMP", "true"):
             filename = f"deploy_{self.branch}_{self.timestamp}.json"
             (Path(self.scratch_root) / filename).write_text(
                 json.dumps(
                     # Sort on full name to ensure consistent ordering
                     sorted(
-                        [d.dump() for d in deploy],
-                        key=lambda d: d["full_name"],
+                        [
+                            j.model_dump(
+                                # callback functions can't be serialised
+                                exclude={"pre_launch", "post_finish"},
+                                mode="json",
+                            )
+                            for j in jobs
+                        ],
+                        key=lambda j: j["full_name"],
                     ),
                     indent=2,
                 ),
             )
 
         return Scheduler(
-            items=deploy,
+            items=jobs,
             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 +456,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 +486,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."""