From c470bc79e4d5a1880794b477b4c7a13578739756 Mon Sep 17 00:00:00 2001 From: Rupert Swarbrick Date: Thu, 2 Apr 2026 17:42:31 +0100 Subject: [PATCH] refactor: Make the typing clearer in Deploy._process_exports Firstly, the stricter return type is needed because the CovMerge constructor updates self.exports (which is generated with that function). We also know the key and value types, so we may as well give them as well. For the rest of the change, we avoid making the type of Deploy.exports change through the run, by defining a new "merged_exports" field to use instead. This allows the type checker to reason more sensibly about what's going on. Signed-off-by: Rupert Swarbrick --- src/dvsim/job/deploy.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/dvsim/job/deploy.py b/src/dvsim/job/deploy.py index 5996088e..95c31ee7 100644 --- a/src/dvsim/job/deploy.py +++ b/src/dvsim/job/deploy.py @@ -7,7 +7,7 @@ import pprint import random import shlex -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterable, Mapping from pathlib import Path from typing import TYPE_CHECKING, ClassVar @@ -96,6 +96,7 @@ def __init__(self, sim_cfg: "FlowCfg") -> None: self.dry_run = False self.flow_makefile = "" self.name = "" + self.exports: Iterable[Mapping[str, str]] = [] # Declare attributes that need to be extracted from the HJSon cfg. self._define_attrs() @@ -109,8 +110,10 @@ def __init__(self, sim_cfg: "FlowCfg") -> None: # Do variable substitutions. self._subst_vars() - # List of vars required to be exported to sub-shell, as a dict. - self.exports = self._process_exports() + # List of vars required to be exported to sub-shell, as a dict. This + # has been loaded from the hjson as a list of dicts and we want to + # flatten it to a single dictionary. + self.merged_exports: dict[str, str] = self._process_exports(self.exports) # Construct the job's command. self.cmd = self._construct_cmd() @@ -148,7 +151,7 @@ def get_job_spec(self) -> "JobSpec": weight=self.weight, timeout_mins=(None if self.gui else self.get_timeout_mins()), cmd=self.cmd, - exports=self.exports, + exports=self.merged_exports, dry_run=self.dry_run, interactive=self.sim_cfg.interactive, odir=self.odir, @@ -287,7 +290,7 @@ def _subst_vars(self, ignored_subst_vars=None) -> None: ignore_error=False, ) - def _process_exports(self) -> Mapping: + def _process_exports(self, exports_list: Iterable[Mapping[str, str]]) -> dict[str, str]: """Convert 'exports' as a list of dicts in the HJson to a dict. Exports is a list of key-value pairs that are to be exported to the @@ -297,7 +300,7 @@ def _process_exports(self) -> Mapping: into a dict variable, which makes it easy to merge the list of exports with the subprocess' env where the ASIC tool is invoked. """ - return {k: str(v) for item in self.exports for k, v in item.items()} + return {k: str(v) for item in exports_list for k, v in item.items()} def _construct_cmd(self) -> str: """Construct the command that will eventually be launched.""" @@ -336,12 +339,12 @@ def is_equivalent_job(self, item: "Deploy") -> bool: return False # Check if exports have identical set of keys. - if self.exports.keys() != item.exports.keys(): + if self.merged_exports.keys() != item.merged_exports.keys(): return False # Check if exports have identical values. - for key, val in self.exports.items(): - item_val = item.exports[key] + for key, val in self.merged_exports.items(): + item_val = item.merged_exports[key] if type(item_val) is str: item_val = item_val.replace(item.name, self.name) if val != item_val: @@ -867,7 +870,7 @@ def __init__(self, run_items, sim_cfg) -> None: self.needs_all_dependencies_passing = False # Append cov_db_dirs to the list of exports. - self.exports["cov_db_dirs"] = shlex.quote(" ".join(self.cov_db_dirs)) + self.merged_exports["cov_db_dirs"] = shlex.quote(" ".join(self.cov_db_dirs)) def _define_attrs(self) -> None: super()._define_attrs()