refactor: add WorkspaceCfg

Reduce dependency on sim_cfg by pulling out some of the data and putting
it on a new WorkspaceCfg pydantic object.

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

Author: machshev

src/dvsim/job/deploy.py

@@ -11,6 +11,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, ClassVar
 
+from pydantic import BaseModel
+from pydantic.config import ConfigDict
 from tabulate import tabulate
 
 from dvsim.job.time import JobTime
@@ -27,6 +29,20 @@
 if TYPE_CHECKING:
     from dvsim.flow.sim import SimCfg
 
+
+class WorkspaceConfig(BaseModel):
+    """Workspace configuration."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    project: str
+    timestamp: str
+
+    project_root: Path
+    scratch_root: Path
+    scratch_path: Path
+
+
 __all__ = (
     "CompileSim",
     "Deploy",
@@ -101,6 +117,14 @@ def __init__(self, sim_cfg: "SimCfg") -> None:
         # Job's wall clock time (a.k.a CPU time, or runtime).
         self.job_runtime = JobTime()
 
+        self.workspace_cfg = WorkspaceConfig(
+            project=sim_cfg.name,
+            project_root=sim_cfg.proj_root,
+            scratch_root=Path(sim_cfg.scratch_root),
+            scratch_path=Path(sim_cfg.scratch_path),
+            timestamp=sim_cfg.args.timestamp,
+        )
+
     def _define_attrs(self) -> None:
         """Define the attributes this instance needs to have.
 

src/dvsim/launcher/base.py

@@ -19,7 +19,7 @@
 from dvsim.utils import clean_odirs, mk_symlink, rm_path
 
 if TYPE_CHECKING:
-    from dvsim.job.deploy import Deploy
+    from dvsim.job.deploy import Deploy, WorkspaceConfig
 
 
 class LauncherError(Exception):
@@ -97,18 +97,18 @@ def __init__(self, deploy: "Deploy") -> None:
             deploy: deployment object that will be launched.
 
         """
-        cfg = deploy.sim_cfg
+        workspace_cfg = deploy.workspace_cfg
 
         # One-time preparation of the workspace.
         if not Launcher.workspace_prepared:
-            # TODO: CLI args should be processed far earlier than this
-            self.prepare_workspace(cfg.project, cfg.proj_root, cfg.args)
+            self.prepare_workspace(workspace_cfg)
             Launcher.workspace_prepared = True
 
         # One-time preparation of the workspace, specific to the cfg.
-        if cfg not in Launcher.workspace_prepared_for_cfg:
-            self.prepare_workspace_for_cfg(cfg)
-            Launcher.workspace_prepared_for_cfg.add(cfg)
+        project = workspace_cfg.project
+        if project not in Launcher.workspace_prepared_for_cfg:
+            self.prepare_workspace_for_cfg(workspace_cfg)
+            Launcher.workspace_prepared_for_cfg.add(project)
 
         # Store the deploy object handle.
         self.deploy = deploy
@@ -155,34 +155,40 @@ def set_pyvenv(project: str) -> None:
         # The code below allows each launcher variant to set its own virtualenv
         # because the loading / activating mechanism could be different between
         # them.
-        Launcher.pyvenv = os.environ.get(
-            f"{project.upper()}_PYVENV_{Launcher.variant.upper()}",
-        )
+        common_venv = f"{project.upper()}_PYVENV"
+        variant = Launcher.variant.upper()
+
+        venv_path = os.environ.get(f"{common_venv}_{variant}")
+
+        if not venv_path:
+            venv_path = os.environ.get(common_venv)
 
         if not Launcher.pyvenv:
             Launcher.pyvenv = os.environ.get(f"{project.upper()}_PYVENV")
 
     @staticmethod
     @abstractmethod
-    def prepare_workspace(project: str, repo_top: str, args: Mapping) -> None:
+    def prepare_workspace(cfg: "WorkspaceConfig") -> None:
         """Prepare the workspace based on the chosen launcher's needs.
 
         This is done once for the entire duration for the flow run.
 
         Args:
-            project: the name of the project.
-            repo_top: the path to the repository.
-            args: command line args passed to dvsim.
+            cfg: workspace configuration
 
         """
 
     @staticmethod
     @abstractmethod
-    def prepare_workspace_for_cfg(cfg: Mapping) -> None:
+    def prepare_workspace_for_cfg(cfg: "WorkspaceConfig") -> None:
         """Prepare the workspace for a cfg.
 
         This is invoked once for each cfg.
         'cfg' is the flow configuration object.
+
+        Args:
+            cfg: workspace configuration
+
         """
 
     def __str__(self) -> str:

src/dvsim/launcher/fake.py

@@ -4,14 +4,14 @@
 
 """Fake Launcher that returns random results."""
 
-from collections.abc import Mapping
 from random import choice, random
 from typing import TYPE_CHECKING
 
 from dvsim.launcher.base import ErrorMessage, Launcher
 
 if TYPE_CHECKING:
-    from dvsim.job.deploy import CovReport, Deploy, RunTest
+    from dvsim.job.deploy import CovReport, Deploy, RunTest, WorkspaceConfig
+
 
 __all__ = ("FakeLauncher",)
 
@@ -75,22 +75,23 @@ def kill(self) -> None:
         )
 
     @staticmethod
-    def prepare_workspace(project: str, repo_top: str, args: Mapping) -> None:
+    def prepare_workspace(cfg: "WorkspaceConfig") -> None:
         """Prepare the workspace based on the chosen launcher's needs.
 
         This is done once for the entire duration for the flow run.
 
         Args:
-            project: the name of the project.
-            repo_top: the path to the repository.
-            args: command line args passed to dvsim.
+            cfg: workspace configuration
 
         """
 
     @staticmethod
-    def prepare_workspace_for_cfg(cfg: Mapping) -> None:
+    def prepare_workspace_for_cfg(cfg: "WorkspaceConfig") -> None:
         """Prepare the workspace for a cfg.
 
         This is invoked once for each cfg.
-        'cfg' is the flow configuration object.
+
+        Args:
+            cfg: workspace configuration
+
         """

src/dvsim/launcher/local.py

@@ -8,14 +8,13 @@
 import os
 import shlex
 import subprocess
-from collections.abc import Mapping
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 from dvsim.launcher.base import ErrorMessage, Launcher, LauncherBusyError, LauncherError
 
 if TYPE_CHECKING:
-    from dvsim.job.deploy import Deploy
+    from dvsim.job.deploy import Deploy, WorkspaceConfig
 
 
 class LocalLauncher(Launcher):
@@ -184,22 +183,23 @@ def _close_job_log_file(self) -> None:
             self._log_file.close()
 
     @staticmethod
-    def prepare_workspace(project: str, repo_top: str, args: Mapping) -> None:
+    def prepare_workspace(cfg: "WorkspaceConfig") -> None:
         """Prepare the workspace based on the chosen launcher's needs.
 
         This is done once for the entire duration for the flow run.
 
         Args:
-            project: the name of the project.
-            repo_top: the path to the repository.
-            args: command line args passed to dvsim.
+            cfg: workspace configuration
 
         """
 
     @staticmethod
-    def prepare_workspace_for_cfg(cfg: Mapping) -> None:
+    def prepare_workspace_for_cfg(cfg: "WorkspaceConfig") -> None:
         """Prepare the workspace for a cfg.
 
         This is invoked once for each cfg.
-        'cfg' is the flow configuration object.
+
+        Args:
+            cfg: workspace configuration
+
         """