OpenROAD (#46)

* reorganize + cleanup + fixes

improved yosys ASIC synthesis for OpenROAD flow

* fix yosys VHDL parameters skip chparam

* fix abc clock

* 'scrub' cli command and '--scrub' run option for cleaning up run directories

* Adding OpenROAD flow 🎉

examples/vhdl/pipeline/pipelined_adder.toml

@@ -11,3 +11,8 @@ clock_port = 'clock'
 sources = ['pipelined_adder_tb.vhdl']
 top = 'pipelined_adder_tb'
 uut = 'uut'
+
+[flow.openroad]
+platform = "sky130hd"
+clocks.main_clock.freq = "200 MHz"
+core_utilization = 45

src/__init__.py

@@ -3,15 +3,19 @@
 from .design import Design
 from .flow import FPGA, Flow, SimFlow, SynthFlow
 from .tool import Tool
+from .flow_runner import FlowRunner, DefaultRunner, Dse
 from .version import __version__
 
 __all__ = [
     "__version__",
     "cli",
     "Cocotb",
+    "DefaultRunner",
     "design",
     "Design",
+    "Dse",
     "flow_runner",
+    "FlowRunner",
     "flows",
     "Flow",
     "FPGA",

src/cli.py

@@ -3,6 +3,7 @@
 import inspect
 import logging
 import os
+import re
 import sys
 from pathlib import Path
 from typing import Any, Dict, Iterable, Optional, Tuple, Union
@@ -30,10 +31,13 @@
     registered_flows,
 )
 from .flow_runner import (
+    DIR_NAME_HASH_LEN,
     DefaultRunner,
     XedaOptions,
     add_file_logger,
+    get_flow_class,
     prepare,
+    scrub_runs,
     settings_to_dict,
 )
 from .flow_runner.dse import Dse
@@ -125,6 +129,11 @@ def cli(ctx: click.Context, **kwargs):
     default=True,
     help="Incremental build. Useful during development. Flows run under a <design_name>/<flow_name>_<flow_settings_hash> subfolder in incremental mode.",
 )
+@click.option(
+    "--incremental-fresh",
+    default=False,
+    help="In incremental mode, cleanup folder contents before running.",
+)
 @click.option(
     "--xedaproject",
     type=click.Path(
@@ -177,19 +186,38 @@ def cli(ctx: click.Context, **kwargs):
 )
 @click.option("--detailed-logs/--no-detailed-logs", show_envvar=True, default=True)
 @click.option("--log-level", show_envvar=True, type=int, default=None)
+@click.option(
+    "--post-cleanup",
+    is_flag=True,
+    help="Remove flow files except settings.json, results.json, and artifacts _after_ running the flow.",
+)
+@click.option(
+    "--post-cleanup-purge",
+    is_flag=True,
+    help="Remove flow run_dir and all of its content _after_ running the flow.",
+)
+@click.option(
+    "--scrub",
+    is_flag=True,
+    help="Remove all previous flow directories of the same flow withing the current 'xeda_run_dir' _before_ running the flow. Requires user confirmation.",
+)
 @click.pass_context
 def run(
     ctx: click.Context,
     flow: str,
     cached_dependencies: bool,
     flow_settings: Union[None, str, Iterable[str]],
     incremental: bool = True,
+    incremental_fresh: bool = False,
     xeda_run_dir: Optional[Path] = None,
     xedaproject: Optional[str] = None,
     design_name: Optional[str] = None,
     design_file: Optional[str] = None,
     log_level: Optional[int] = None,
     detailed_logs: bool = False,
+    post_cleanup: bool = False,
+    post_cleanup_purge: bool = False,
+    scrub: bool = False,
 ):
     """`run` command"""
     assert ctx
@@ -223,8 +251,11 @@ def run(
             cached_dependencies=cached_dependencies,
         )
         launcher.settings.incremental = incremental
+        launcher.settings.incremental_fresh = incremental_fresh
+        launcher.settings.post_cleanup = post_cleanup
+        launcher.settings.post_cleanup_purge = post_cleanup_purge
+        launcher.settings.scrub_old_runs = scrub
         launcher.settings.debug = options.debug
-        launcher.settings.incremental = True
         launcher.run_flow(flow_class, design, accum_flow_settings)
     except FlowFatalError as e:
         log.critical(
@@ -476,6 +507,58 @@ def dse(
     )
 
 
+@cli.command(
+    context_settings=CONTEXT_SETTINGS,
+    short_help="Remove all flow runs in the specified xeda_run_dir",
+)
+@click.argument(
+    "flow",
+    metavar="FLOW_NAME",
+    type=click.Choice(all_flows),
+    required=True,
+)
+@click.argument(
+    "design_name",
+    metavar="DESIGN_NAME",
+    required=True,
+)
+@click.option(
+    "--xeda-run-dir",
+    type=click.Path(
+        file_okay=False,
+        dir_okay=True,
+        writable=True,
+        readable=True,
+        resolve_path=True,
+        allow_dash=True,
+        path_type=Path,
+    ),
+    envvar="XEDA_RUN_DIR",
+    help="Parent folder for execution of xeda commands.",
+    default="xeda_run",
+    show_default=True,
+    show_envvar=True,
+)
+@click.option(
+    "--incremental/--no-incremental",
+    default=True,
+    help="Include incremental build directories (<design_name>/<flow_name>_<flow_settings_hash> name pattern)",
+)
+@click.pass_context
+def scrub(ctx: click.Context, flow, design_name, xeda_run_dir, incremental):
+    xeda_run_dir = Path(xeda_run_dir).resolve()
+    # just to make sure flow exists and name is canonical
+    flow_class = get_flow_class(flow)
+
+    regex = re.compile(f"^{design_name}_" + (r"[a-z0-9]" * DIR_NAME_HASH_LEN) + r"$")
+    design_dirs = [p for p in xeda_run_dir.glob("design_") if p.is_dir() and regex.match(p.name)]
+    if incremental and (xeda_run_dir / design_name).exists():
+        design_dirs.append(xeda_run_dir / design_name)
+
+    for dd in design_dirs:
+        scrub_runs(flow_class.name, dd)
+
+
 SHELLS: Dict[str, Dict[str, Any]] = {
     "bash": {
         "eval_file": "~/.bashrc",

src/cli_utils.py

@@ -16,8 +16,8 @@
 
 from .console import console
 from .design import Design
-from .flow_runner import FlowNotFoundError, XedaOptions, get_flow_class
 from .flow import Flow
+from .flow_runner import FlowNotFoundError, XedaOptions, get_flow_class
 from .xedaproject import XedaProject
 
 __all__ = [

src/flow/_flow.py

@@ -84,9 +84,9 @@ def _validate_quiet(cls, value, values):
             description="Number of physical CPUs to use.",
         )
         no_console: bool = Field(False, hidden_from_schema=True)
-        reports_dir: str = Field("reports", hidden_from_schema=True)
-        checkpoints_dir: str = Field("checkpoints", hidden_from_schema=True)
-        outputs_dir: str = Field("outputs", hidden_from_schema=True)
+        reports_dir: Path = Field(Path("reports"), hidden_from_schema=True)
+        checkpoints_dir: Path = Field(Path("checkpoints"), hidden_from_schema=True)
+        outputs_dir: Path = Field(Path("outputs"), hidden_from_schema=True)
         clean: bool = False  # TODO remove!
         lib_paths: List[
             Union[
@@ -208,9 +208,11 @@ def __init__(self, settings: Settings, design: Design, run_path: Path):
 
         # generated artifacts as a dict of category to list of file paths
         self.artifacts = Box()
-        self.reports: DictStrPath = {}
+        self.reports: DictStrPath = {}  # DEPRECATED! TODO: remove
         # TODO deprecate and use self.reports
-        self.reports_dir = run_path / self.settings.reports_dir
+        if self.settings.reports_dir:
+            self.settings.reports_dir.mkdir(exist_ok=True, parents=True)
+        self.reports_dir = self.settings.reports_dir  # DEPRECATED! use settings.reports_dir
         self.results = self.Results(
             success=False,
             # "Time of the execution of run() in fractional seconds.
@@ -241,12 +243,13 @@ def parse_reports(self) -> bool:
         return True
 
     def copy_from_template(
-        self, resource_name, lstrip_blocks=False, trim_blocks=False, **kwargs
+        self, resource_name, lstrip_blocks=False, trim_blocks=False, script_filename=None, **kwargs
     ) -> Path:
         template = self.jinja_env.get_template(resource_name)
         template.environment.lstrip_blocks = lstrip_blocks
         template.environment.trim_blocks = trim_blocks
-        script_path: Path = self.run_path / resource_name
+        script_path = Path(script_filename) if script_filename else self.run_path / resource_name
+
         log.debug("generating %s from template.", str(script_path.resolve()))
         rendered_content = template.render(
             settings=self.settings,
@@ -256,11 +259,25 @@ def copy_from_template(
         )
         with open(script_path, "w") as f:
             f.write(rendered_content)
-        return script_path.relative_to(self.run_path)  # resource_name
+        return script_path.resolve().relative_to(self.run_path)  # resource_name
 
     def add_template_filter(self, filter_name: str, func) -> None:
+        assert filter_name
+        if filter_name in self.jinja_env.filters:
+            raise ValueError(f"Template filter with name {filter_name} already exists!")
         self.jinja_env.filters[filter_name] = func
 
+    def add_template_filter_func(self, func) -> None:
+        self.add_template_filter(func.__name__, func)
+
+    def add_template_global_func(self, func, filter_name=None) -> None:
+        if not filter_name:
+            filter_name = func.__name__
+        assert filter_name
+        if filter_name in self.jinja_env.globals:
+            raise ValueError(f"Template global with name {filter_name} already exists!")
+        self.jinja_env.globals[filter_name] = func
+
     def add_template_test(self, filter_name: str, func) -> None:
         self.jinja_env.tests[filter_name] = func