style: linting, docstrings and typing

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

Author: machshev

src/dvsim/flow/base.py

@@ -445,16 +445,19 @@ def _gen_results(self, results: Mapping[str, CompletedJobStatus]) -> str:
         prints the full list of failures for debug / triage to the final
         report, which is in markdown format.
 
-        Results:
-            dictionary mapping deployed item names to job status.
+        Args:
+            results: dictionary mapping deployed item names to completed job status.
+
+        Returns:
+            Results as a formatted string
 
         """
 
     def gen_results(self, results: Mapping[str, CompletedJobStatus]) -> None:
-        """Public facing API for _gen_results().
+        """Generate flow results.
 
         Args:
-            results: dictionary mapping deployed item names to job status.
+            results: dictionary mapping deployed item names to completed job status.
 
         """
         for item in self.cfgs:

src/dvsim/job/deploy.py

@@ -293,7 +293,7 @@ def pre_launch(self, launcher: Launcher) -> None:
         This is invoked by launcher::_pre_launch().
         """
 
-    def post_finish(self, status) -> None:
+    def post_finish(self, status: str) -> None:
         """Perform additional post-finish activities (callback).
 
         This is invoked by launcher::_post_finish().

src/dvsim/launcher/base.py

@@ -67,7 +67,7 @@ class Launcher(ABC):
     poll_freq = 1
 
     # Points to the python virtual env area.
-    pyvenv = None
+    pyvenv: Path | None = None
 
     # If a history of previous invocations is to be maintained, then keep no
     # more than this many directories.
@@ -250,6 +250,10 @@ def poll(self) -> str | None:
         """Poll the launched job for completion.
 
         Invokes _check_status() and _post_finish() when the job completes.
+
+        Returns:
+            status of the job or None
+
         """
 
     @abstractmethod

src/dvsim/scheduler.py

@@ -91,7 +91,14 @@ def __init__(
         *,
         interactive: bool,
     ) -> None:
-        """Initialise a job scheduler."""
+        """Initialise a job scheduler.
+
+        Args:
+            items: sequence of jobs to deploy.
+            launcher_cls: Launcher class to use to deploy the jobs.
+            interactive: launch the tools in interactive mode.
+
+        """
         self.items: Sequence[Deploy] = items
 
         # 'scheduled[target][cfg]' is a list of Deploy objects for the chosen
@@ -148,7 +155,7 @@ def __init__(
             msg = self.msg_fmt.format(0, 0, 0, 0, 0, self._total[target])
             self.status_printer.init_target(target=target, msg=msg)
 
-        # A map from the Deploy object names tracked by this class to their
+        # A map from the Deployment names tracked by this class to their
         # current status. This status is 'Q', 'D', 'P', 'F' or 'K',
         # corresponding to membership in the dicts above. This is not
         # per-target.
@@ -271,8 +278,13 @@ def _enqueue_successors(self, item: Deploy | None = None) -> None:
         them to _queued.
         """
         for next_item in self._get_successors(item):
-            assert next_item.full_name not in self.item_status
-            assert next_item not in self._queued[next_item.target]
+            if (
+                next_item.full_name in self.item_status
+                or next_item in self._queued[next_item.target]
+            ):
+                msg = f"Job {next_item.full_name} already scheduled"
+                raise RuntimeError(msg)
+
             self.item_status[next_item.full_name] = "Q"
             self._queued[next_item.target].append(next_item)
             self._unschedule_item(next_item)
@@ -281,6 +293,10 @@ def _cancel_successors(self, item: Deploy) -> None:
         """Cancel an item's successors.
 
         Recursively move them from _scheduled or _queued to _killed.
+
+        Args:
+            item: job whose successors are to be canceled.
+
         """
         items = list(self._get_successors(item))
         while items:
@@ -291,14 +307,17 @@ def _cancel_successors(self, item: Deploy) -> None:
     def _get_successors(self, item: Deploy | None = None) -> Sequence[Deploy]:
         """Find immediate successors of an item.
 
-        'item' is a job that has completed. We choose the target that follows
-        the 'item''s current target and find the list of successors whose
-        dependency list contains 'item'. If 'item' is None, we pick successors
-        from all cfgs, else we pick successors only from the cfg to which the
-        item belongs.
+        We choose the target that follows the 'item''s current target and find
+        the list of successors whose dependency list contains 'item'. If 'item'
+        is None, we pick successors from all cfgs, else we pick successors only
+        from the cfg to which the item belongs.
+
+        Args:
+            item: is a job that has completed.
+
+        Returns:
+            list of item's successors, or an empty list if there are none.
 
-        Returns the list of item's successors, or an empty list if there are
-        none.
         """
         if item is None:
             target = next(iter(self._scheduled))
@@ -320,8 +339,10 @@ def _get_successors(self, item: Deploy | None = None) -> Sequence[Deploy]:
             while not found:
                 if target == item.target:
                     found = True
+
                 try:
                     target = next(target_iterator)
+
                 except StopIteration:
                     return []
 
@@ -349,7 +370,15 @@ def _get_successors(self, item: Deploy | None = None) -> Sequence[Deploy]:
         return successors
 
     def _ok_to_enqueue(self, item: Deploy) -> bool:
-        """Return true if ALL dependencies of item are complete."""
+        """Check if all dependencies jobs are completed.
+
+        Args:
+            item: is a deployment job.
+
+        Returns:
+            true if ALL dependencies of item are complete.
+
+        """
         for dep in item.dependencies:
             # Ignore dependencies that were not scheduled to run.
             if dep not in self.items:
@@ -366,11 +395,18 @@ def _ok_to_enqueue(self, item: Deploy) -> bool:
         return True
 
     def _ok_to_run(self, item: Deploy) -> bool:
-        """Return true if the required dependencies have passed.
+        """Check if a job is ready to start.
 
         The item's needs_all_dependencies_passing setting is used to figure
         out whether we can run this item or not, based on its dependent jobs'
         statuses.
+
+        Args:
+            item: is a deployment job.
+
+        Returns:
+            true if the required dependencies have passed.
+
         """
         # 'item' can run only if its dependencies have passed (their results
         # should already show up in the item to status map).
@@ -395,7 +431,9 @@ def _ok_to_run(self, item: Deploy) -> bool:
     def _poll(self, hms: str) -> bool:
         """Check for running items that have finished.
 
-        Returns True if something changed.
+        Returns:
+            True if something changed.
+
         """
         max_poll = min(
             self.launcher_cls.max_poll,
@@ -615,6 +653,11 @@ def _cancel_item(self, item: Deploy, *, cancel_successors: bool = True) -> None:
 
         Supplied item may be in _scheduled list or the _queued list. From
         either, we move it straight to _killed.
+
+        Args:
+            item: is a deployment job.
+            cancel_successors: if set then cancel successors as well (True).
+
         """
         self.item_status[item.full_name] = "K"
         self._killed[item.target].add(item)
@@ -627,7 +670,12 @@ def _cancel_item(self, item: Deploy, *, cancel_successors: bool = True) -> None:
             self._cancel_successors(item)
 
     def _kill_item(self, item: Deploy) -> None:
-        """Kill a running item and cancel all of its successors."""
+        """Kill a running item and cancel all of its successors.
+
+        Args:
+            item: is a deployment job.
+
+        """
         self._launchers[item.full_name].kill()
         self.item_status[item.full_name] = "K"
         self._killed[item.target].add(item)

src/dvsim/sim_utils.py

@@ -6,15 +6,19 @@
 
 import re
 from collections import OrderedDict
+from collections.abc import Sequence
+from io import TextIOWrapper
 from pathlib import Path
 
 
-def get_cov_summary_table(cov_report_txt: Path, tool: str):
+def get_cov_summary_table(
+    txt_cov_report: Path,
+    tool: str,
+) -> tuple[Sequence[Sequence[str]], str]:
     """Capture the summary results as a list of lists.
 
     The text coverage report is passed as input to the function, in addition to
-    the tool used. The tool returns a 2D list if the coverage report file was read
-    and the coverage was extracted successfully.
+    the tool used.
 
     Returns:
         tuple of, List of metrics and values, and final coverage total
@@ -23,7 +27,7 @@ def get_cov_summary_table(cov_report_txt: Path, tool: str):
         the appropriate exception if the coverage summary extraction fails.
 
     """
-    with Path(cov_report_txt).open() as f:
+    with Path(txt_cov_report).open() as f:
         if tool == "xcelium":
             return xcelium_cov_summary_table(f)
 
@@ -35,7 +39,8 @@ def get_cov_summary_table(cov_report_txt: Path, tool: str):
 
 
 # Same desc as above, but specific to Xcelium and takes an opened input stream.
-def xcelium_cov_summary_table(buf):
+def xcelium_cov_summary_table(buf: TextIOWrapper) -> tuple[Sequence[Sequence[str]], str]:
+    """Capture the summary results as a list of lists from Xcelium."""
     for line in buf:
         if "name" in line:
             # Strip the line and remove the unwanted "* Covered" string.
@@ -87,7 +92,8 @@ def xcelium_cov_summary_table(buf):
 
 
 # Same desc as above, but specific to VCS and takes an opened input stream.
-def vcs_cov_summary_table(buf):
+def vcs_cov_summary_table(buf: TextIOWrapper) -> tuple[Sequence[Sequence[str]], str]:
+    """Capture the summary results as a list of lists from VCS."""
     for line in buf:
         match = re.match("total coverage summary", line, re.IGNORECASE)
         if match:
@@ -124,7 +130,7 @@ def get_job_runtime(log_text: list, tool: str) -> tuple[float, str]:
         tool: is the EDA tool used to run the job.
 
     Returns:
-        the runtime, units as a tuple.
+        a tuple of (runtime, units).
 
     Raises:
         NotImplementedError: exception if the EDA tool is not supported.
@@ -184,8 +190,7 @@ def xcelium_job_runtime(log_text: list) -> tuple[float, str]:
     """
     pattern = r"^TOOL:\s*xrun.*: Exiting on .*\(total:\s*(\d+):(\d+):(\d+)\)\s*$"
     for line in reversed(log_text):
-        m = re.search(pattern, line)
-        if m:
+        if m := re.search(pattern, line):
             t = int(m.group(1)) * 3600 + int(m.group(2)) * 60 + int(m.group(3))
             return t, "s"
     msg = "Job runtime not found in the log."
@@ -265,9 +270,10 @@ def vcs_simulated_time(log_text: list) -> tuple[float, str]:
     next_line = ""
 
     for line in reversed(log_text):
-        if "V C S   S i m u l a t i o n   R e p o r t" in line and (
-            m := re.search(pattern, next_line)
-        ):
+        if "V C S   S i m u l a t i o n   R e p o r t" not in line:
+            continue
+
+        if m := re.search(pattern, next_line):
             return float(m.group(1)), m.group(2).lower()
 
         next_line = line