feat!: make substitutions more predictable, strictly class-level (#648)

## Misc. Enhancements/Bugfixes

* `openlane.flows`

  * `SequentialFlow`
    * Substitutions are now to be strictly consumed by the subclass initializer,
      i.e., it can no longer be done on the object-level and only on the class
      level. Additionally, it can provided as a list of tuples instead of a
      dictionary so the same key may be reused multiple times.
    * Step IDs are re-normalized after every substitution, so a substitution for
      `OpenROAD.DetailedPlacement-1` for example would always refer to the
      second `OpenROAD.DetailedPlacement` AFTER applying all previous
      substitutions, instead of the second "original"
      `OpenROAD.DetailedPlacement` in the flow.

* `openlane.config`

  * `meta.substituting_steps` now only apply to the sequential flow declared in
    `meta.flow` and not all flows.

Changelog.md

@@ -36,8 +36,8 @@
     names, these are mappings from corners to layer/via RC values.
     * `PNR_CORNERS`, `RSZ_CORNERS`, and `CTS_CORNERS` all now support multiple
       corners to have the same set of liberty files (as RC values may differ.)
-  * Added `SET_RC_VERBOSE`, which (very noisily) logs set-RC-related commands
-    to logs.
+  * Added `SET_RC_VERBOSE`, which (very noisily) logs set-RC-related commands to
+    logs.
   * Always read libs before reading odb.
   * Added `log_cmd` from OpenROAD-flow-scripts -- neat idea for consistency
   * Lib files are now *always* read BEFORE reading database files.
@@ -148,6 +148,24 @@
 
 ## Misc. Enhancements/Bugfixes
 
+* `openlane.flows`
+
+  * `SequentialFlow`
+    * Substitutions are now to be strictly consumed by the subclass initializer,
+      i.e., it can no longer be done on the object-level and only on the class
+      level. Additionally, it can provided as a list of tuples instead of a
+      dictionary so the same key may be reused multiple times.
+    * Step IDs are re-normalized after every substitution, so a substitution for
+      `OpenROAD.DetailedPlacement-1` for example would always refer to the
+      second `OpenROAD.DetailedPlacement` AFTER applying all previous
+      substitutions, instead of the second "original"
+      `OpenROAD.DetailedPlacement` in the flow.
+
+* `openlane.config`
+
+  * `meta.substituting_steps` now only apply to the sequential flow declared in
+    `meta.flow` and not all flows.
+
 * `openlane.state`
 
   * `DesignFormat`
@@ -210,6 +228,13 @@
   * `VIAS_RC` removed and replaced by `VIAS_R` with a format similar to
     `LAYERS_RC`.
 
+* `openlane.flows`
+
+  * Step IDs are re-normalized after every substitution, so a substitution for
+    `OpenROAD.DetailedPlacement-1` for example would always refer to the second
+    `OpenROAD.DetailedPlacement` AFTER applying all previous substitutions,
+    instead of the second "original" `OpenROAD.DetailedPlacement` in the flow.
+
 * `openlane.steps`
 
   * `TclStep` now uses the IDs uppercased for `CURRENT_` and `SAVE_`.
@@ -229,6 +254,9 @@
 
 * `openlane.config`
 
+  * `meta.substituting_steps` now only apply to the sequential flow declared in
+    `meta.flow` and not all flows.
+
   * `WIRE_LENGTH_THRESHOLD`, `GPIO_PAD_*`, `FP_TRACKS_INFO`, `FP_TAPCELL_DIST`
     are no longer global variables.
 

openlane/__main__.py

@@ -21,7 +21,7 @@
 import subprocess
 from textwrap import dedent
 from functools import partial
-from typing import Any, Dict, Sequence, Tuple, Type, Optional, List, Union
+from typing import Any, Dict, Sequence, Tuple, Type, Optional, List
 
 import click
 from cloup import (
@@ -77,21 +77,32 @@ def run(
             err("No config file(s) have been provided.")
             ctx.exit(1)
 
-        flow_description: Optional[Union[str, List[str]]] = None
-        substitutions: Union[None, Dict[str, Union[str, None]]] = None
+        TargetFlow: Optional[Type[Flow]] = Flow.factory.get("Classic")
 
         for config_file in config_files:
             if meta := Config.get_meta(config_file):
                 if meta.flow is not None:
-                    flow_description = meta.flow
-                    if meta.substituting_steps is not None:
-                        substitutions = meta.substituting_steps
+                    if isinstance(meta.flow, str):
+                        if found := Flow.factory.get(meta.flow):
+                            TargetFlow = found
+                        else:
+                            err(
+                                f"Unknown flow '{meta.flow}' specified in configuration file's 'meta' object."
+                            )
+                            ctx.exit(1)
+                    elif isinstance(meta.flow, list):
+                        TargetFlow = SequentialFlow.Make(meta.flow)
+                    if meta.substituting_steps is not None and issubclass(
+                        TargetFlow, SequentialFlow
+                    ):
+                        TargetFlow = TargetFlow.Substitute(meta.substituting_steps)  # type: ignore  # Type checker is being rowdy with this one
 
         if flow_name is not None:
-            flow_description = flow_name
-
-        if flow_description is None:
-            flow_description = "Classic"
+            if found := Flow.factory.get(flow_name):
+                TargetFlow = found
+            else:
+                err(f"Unknown flow '{flow_name}' passed to initialization function.")
+                ctx.exit(1)
 
         if len(initial_state_element_override):
             if with_initial_state is None:
@@ -114,18 +125,9 @@ def run(
                 overrides=overrides,
             )
 
-        TargetFlow: Type[Flow]
-
-        if isinstance(flow_description, str):
-            if FlowClass := Flow.factory.get(flow_description):
-                TargetFlow = FlowClass
-            else:
-                err(
-                    f"Unknown flow '{flow_description}' specified in configuration file's 'meta' object."
-                )
-                ctx.exit(1)
-        else:
-            TargetFlow = SequentialFlow.make(flow_description)
+        assert (
+            TargetFlow is not None
+        ), "TargetFlow is unexpectedly None. Please report this as a bug."
 
         kwargs: Dict[str, Any] = {
             "pdk_root": pdk_root,
@@ -134,8 +136,6 @@ def run(
             "config_override_strings": config_override_strings,
             "design_dir": design_dir,
         }
-        if issubclass(TargetFlow, SequentialFlow):
-            kwargs["Substitute"] = substitutions
         flow = TargetFlow(config_files, **kwargs)
     except PassedDirectoryError as e:
         err(e)

openlane/config/config.py

@@ -166,7 +166,11 @@ class Meta:
 
     version: int = 1
     flow: Union[None, str, List[str]] = None
-    substituting_steps: Union[None, Dict[str, Union[str, None]]] = None
+    substituting_steps: Union[
+        None,
+        Dict[str, Union[str, None]],
+        List[Tuple[str, Union[str, None]]],
+    ] = None
     step: Union[None, str] = None
     openlane_version: Union[None, str] = __version__
 

openlane/flows/sequential.py

@@ -26,6 +26,7 @@
     Union,
 )
 
+from deprecated.sphinx import deprecated
 from rapidfuzz import process, fuzz, utils
 
 from .flow import Flow, FlowException, FlowError
@@ -39,6 +40,12 @@
     DeferredStepError,
 )
 
+Substitution = Union[str, Type[Step], None]
+SubstitutionsObject = Union[
+    Dict[str, Substitution],
+    List[Tuple[str, Substitution]],
+]
+
 
 class SequentialFlow(Flow):
     """
@@ -67,20 +74,53 @@ class SequentialFlow(Flow):
     :cvar gating_config_vars: A mapping from step ID (wildcards) to lists of
         Boolean variable names. All Boolean variables must be True for a step with
         a specific ID to execute.
+
+    :cvar Substitutions: Consumed by the subclass initializer - allows for a
+        quick interface where steps can be removed, replaced, appended or
+        prepended. After subclass intialization, it is set to ``None``, and
+        a new subclass must be created to modify substitutions further.
+
+        The list of substitutions may be specified as either a dictionary
+        or a list of tuples: while the former is more terse, the latter is
+        allows using the same key more than once.
+
+        The substitutions are mappings from Step IDs or objects to Step IDs,
+        objects, or ``None``. In case of ``None``, the step in question is
+        removed.
+
+        In case the key is specified as a string, you may add ``-`` as a prefix
+        to the Step ID to indicate you want to insert a step before the step
+        in question, and similary ``+`` indicates you want to insert a step
+        after the step in question. You may not prepend or append ``None``.
+
+        Step IDs are made unique after every substitution, i.e., whenever the
+        substitution occurs, the first instance of a Step in a sequential flow
+        shall have its ID unadulterated, while the next instance will have
+        ``-1``, the one after ``-2``, etc. If ``-1`` is removed, the previous
+        ``-2`` shall become ``-1``, for example.
     """
 
-    Substitutions: Optional[Dict[str, Union[str, Type[Step], None]]] = None
+    Substitutions: Optional[SubstitutionsObject] = None
     gating_config_vars: Dict[str, List[str]] = {}
 
+    def __init__(
+        self,
+        *args,
+        **kwargs,
+    ):
+        self.Steps = self.Steps.copy()  # Break global reference
+        super().__init__(*args, **kwargs)
+
+    # ---
+
     def __init_subclass__(Self, scm_type=None, name=None, **kwargs):
         Self.Steps = Self.Steps.copy()  # Break global reference
         Self.config_vars = Self.config_vars.copy()
         Self.gating_config_vars = Self.gating_config_vars.copy()
-        if substitute := Self.Substitutions:
-            for key, item in substitute.items():
-                Self.__substitute_step(Self, key, item)
-
         Self.__normalize_step_ids(Self)
+        if Self.Substitutions:
+            Self.__substitute_in_place(Self, Self.Substitutions)
+            Self.Substitutions = None
 
         # Validate Gating Config Vars
         variables_by_name = {}
@@ -106,7 +146,7 @@ def __init_subclass__(Self, scm_type=None, name=None, **kwargs):
                     )
 
     @classmethod
-    def make(Self, step_ids: List[str]) -> Type[SequentialFlow]:
+    def Make(Self, step_ids: List[str]) -> Type[SequentialFlow]:
         Step_list = []
         for name in step_ids:
             step = Step.factory.get(name)
@@ -120,20 +160,40 @@ class CustomSequentialFlow(SequentialFlow):
 
         return CustomSequentialFlow
 
-    def __init__(
-        self,
-        *args,
-        Substitute: Optional[Dict[str, Union[str, Type[Step], None]]] = None,
-        **kwargs,
-    ):
-        self.Steps = self.Steps.copy()  # Break global reference
+    @classmethod
+    @deprecated(
+        "use .Make",
+        version="3.0.0",
+        action="once",
+    )
+    def make(Self, step_ids: List[str]) -> Type[SequentialFlow]:
+        return Self.Make(step_ids)
+
+    @classmethod
+    def Substitute(Self, Substitutions: SubstitutionsObject) -> Type[SequentialFlow]:
+        """
+        Convenience method to quickly subclass a sequential flow and add
+        Substitutions to it.
 
-        if substitute := Substitute:
-            for key, item in substitute.items():
-                self.__substitute_step(self, key, item)
-            self.__normalize_step_ids(self)
+        The new flow shall be named ``{previous_flow_name}'``.
 
-        super().__init__(*args, **kwargs)
+        :param Substitutions: The substitutions to use for the new subclass.
+        """
+        return type(Self.__name__ + "'", (Self,), {"Substitutions": Substitutions})
+
+    @staticmethod
+    def __substitute_in_place(
+        target: Type[SequentialFlow],
+        Substitutions: Optional[SubstitutionsObject],
+    ):
+        if Substitutions is None:
+            return Substitutions
+        if isinstance(Substitutions, dict):
+            for key, item in Substitutions.items():
+                target.__substitute_step(target, key, item)
+        else:
+            for key, item in Substitutions:
+                target.__substitute_step(target, key, item)
 
     @staticmethod
     def __substitute_step(
@@ -173,38 +233,39 @@ def __substitute_step(
         if with_step is None:
             for index in reversed(step_indices):
                 del target.Steps[index]
-            return
-
-        if isinstance(with_step, str):
-            with_step_opt = Step.factory.get(with_step)
-            if with_step_opt is None:
-                raise FlowException(
-                    f"Could not {mode} '{id}' with '{with_step}': no replacement step with ID '{with_step}' found."
-                )
-            with_step = with_step_opt
+        else:
+            if isinstance(with_step, str):
+                with_step_opt = Step.factory.get(with_step)
+                if with_step_opt is None:
+                    raise FlowException(
+                        f"Could not {mode} '{id}' with '{with_step}': no replacement step with ID '{with_step}' found."
+                    )
+                with_step = with_step_opt
 
-        for i in step_indices:
-            if mode == "replace":
-                target.Steps[i] = with_step
-            elif mode == "append":
-                target.Steps.insert(i + 1, with_step)
-            elif mode == "prepend":
-                target.Steps.insert(i, with_step)
+            for i in step_indices:
+                if mode == "replace":
+                    target.Steps[i] = with_step
+                elif mode == "append":
+                    target.Steps.insert(i + 1, with_step)
+                elif mode == "prepend":
+                    target.Steps.insert(i, with_step)
+        target.__normalize_step_ids(target)
 
     @staticmethod
     def __normalize_step_ids(target: Union[SequentialFlow, Type[SequentialFlow]]):
         ids_used: Set[str] = set()
 
         for i, step in enumerate(target.Steps):
             counter = 0
-            id = step.id
+            imp_id = step.get_implementation_id()
+            id = imp_id
             if (
                 id == NotImplemented
             ):  # Will be validated later by initialization: ignore for now
                 continue
             while id in ids_used:
                 counter += 1
-                id = f"{step.id}-{counter}"
+                id = f"{imp_id}-{counter}"
             if id != step.id:
                 target.Steps[i] = step.with_id(id)
             ids_used.add(id)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "openlane"
-version = "3.0.0.dev12"
+version = "3.0.0.dev13"
 description = "An infrastructure for implementing chip design flows"
 authors = ["Efabless Corporation and Contributors <[email protected]>"]
 readme = "Readme.md"