fix yosys VHDL parameters skip chparam

cleanup and fix/supress lint errors

src/flow/_flow.py

@@ -24,7 +24,7 @@
     validator,
 )
 from ..design import Design
-from ..utils import camelcase_to_snakecase, regex_match, try_convert, unique
+from ..utils import camelcase_to_snakecase, regex_match, try_convert_to_primitives, unique
 
 log = logging.getLogger(__name__)
 
@@ -300,7 +300,7 @@ def match_pattern(pat: str, content: str) -> Tuple[bool, str]:
                     return False, content
                 match_dict = match.groupdict()
                 for k, v in match_dict.items():
-                    self.results[k] = try_convert(v)
+                    self.results[k] = try_convert_to_primitives(v)
                     log.debug("%s: %s", k, self.results.get(k))
                 if sequential:
                     content = content[match.span(0)[1] :]

src/flow_runner/__init__.py

@@ -42,7 +42,7 @@
     dump_json,
     set_hierarchy,
     snakecase_to_camelcase,
-    try_convert,
+    try_convert_to_primitives,
 )
 from ..version import __version__
 from ..xedaproject import XedaProject
@@ -523,7 +523,7 @@ def settings_to_dict(
                     f"Settings should be in KEY=VALUE format! (value given: {override})"
                 )
             key, val = sp
-            set_hierarchy(res, key, try_convert(val, convert_lists=True))
+            set_hierarchy(res, key, try_convert_to_primitives(val, convert_lists=True))
         return res
     if isinstance(settings, Flow.Settings):
         return asdict(settings)
@@ -532,7 +532,7 @@ def settings_to_dict(
             return settings
         expanded: DictStrHier = {}
         for k, v in settings.items():
-            set_hierarchy(expanded, k, try_convert(v, convert_lists=True))
+            set_hierarchy(expanded, k, try_convert_to_primitives(v, convert_lists=True))
         return expanded
     raise TypeError(f"overrides is of unsupported type: {type(settings)}")
 

src/flows/dc/__init__.py

@@ -7,7 +7,7 @@
 from box import Box
 
 from ...tool import Tool
-from ...utils import dict_merge, toml_load, try_convert
+from ...utils import dict_merge, toml_load, try_convert_to_primitives
 from ...flow import AsicSynthFlow
 
 log = logging.getLogger(__name__)
@@ -95,7 +95,7 @@ def parse_reports(self) -> bool:
         def parse_kvs(kvs):
             kvs = re.split(r"\s*\n\s*", kvs)
             kvs = [re.split(r"\s*:\s*", s.strip()) for s in kvs if s.strip()]
-            return {s[0].strip(): try_convert(s[1]) for s in kvs}
+            return {s[0].strip(): try_convert_to_primitives(s[1]) for s in kvs}
 
         path_group_re = re.compile(
             r"^\s*Timing Path Group\s+'(?P<path_group_name>\w+)'\n\s*\-+\s*\n(?P<kv>(?:^.*\n)+)",

src/flows/ise/__init__.py

@@ -3,7 +3,7 @@
 from typing import Mapping, Union
 
 from ...tool import Tool
-from ...utils import try_convert
+from ...utils import try_convert_to_primitives
 from ...flow import FpgaSynthFlow
 
 logger = logging.getLogger(__name__)
@@ -66,6 +66,6 @@ def parse_reports(self) -> bool:
             r"Slack:\s+(?P<wns>\-?\d+(?:\.\d+)?)ns",
         )
         if "wns" in self.results:
-            wns = try_convert(self.results["wns"])
+            wns = try_convert_to_primitives(self.results["wns"])
             fail |= not isinstance(wns, (float, int)) or wns < 0
         return not fail

src/flows/openroad/openroad_scripts/utils/def2stream.py

@@ -1,3 +1,4 @@
+# flake8: noqa
 import pya
 import re
 import json

src/flows/quartus/__init__.py

@@ -7,7 +7,7 @@
 from ...dataclass import Field
 from ...tool import Docker, Tool
 from ...types import PathLike
-from ...utils import try_convert
+from ...utils import try_convert_to_primitives
 from ...flow import FpgaSynthFlow
 
 log = logging.getLogger(__name__)
@@ -313,12 +313,12 @@ def parse_reports(self) -> bool:
             print("wns:", wns)
             whs = worst_slacks.get("Hold")
             if wns is not None:
-                wns = try_convert(wns)
+                wns = try_convert_to_primitives(wns)
                 self.results["wns"] = wns
                 if isinstance(wns, (int, float)):
                     failed |= wns < 0
             if whs is not None:
-                whs = try_convert(whs)
+                whs = try_convert_to_primitives(whs)
                 self.results["whs"] = whs
                 if isinstance(whs, (int, float)):
                     failed |= whs < 0

src/flows/yosys/__init__.py

@@ -171,6 +171,10 @@ class Settings(Yosys.Settings, SynthFlow.Settings):
         abc_script: Union[None, Path, List[str]] = None
         hilomap: Optional[HiLoMap] = None
         insbuf: Optional[Tuple[str, str, str]] = None
+        top_is_vhdl: Optional[bool] = Field(
+            None,
+            description="set to `true` to specify top module is VHDL, or `false` to override detection based on last source.",
+        )
 
         @validator("liberty")
         def _str_to_list(value):
@@ -265,7 +269,14 @@ def run(self) -> None:
                 abc_script_file = str(ss.abc_script)
         clock_period_ps: Optional[float] = None
         if ss.clocks:
-            clock_period_ps = list(ss.clocks.values())[0].period_ps
+            assert ss.main_clock
+            clock_period_ps = ss.main_clock.period_ps
+        if ss.top_is_vhdl is True or (
+            ss.top_is_vhdl is None and self.design.rtl.sources[-1].type is SourceType.Vhdl
+        ):
+            # generics were already handled by GHDL and the synthesized design is no longer parametric
+            self.design.rtl.parameters = {}
+
         script_path = self.copy_from_template(
             "yosys_fpga_synth.tcl" if ss.fpga else "yosys_synth.tcl",
             lstrip_blocks=True,
@@ -277,7 +288,7 @@ def run(self) -> None:
             clock_period_ps=clock_period_ps,
             abc_script_file=abc_script_file,
         )
-        log.info("Yosys script: %s", self.run_path.relative_to(Path.cwd()) / script_path)
+        log.info("Yosys script: %s", script_path.absolute())
         args = ["-c", script_path]
         if ss.log_file:
             args.extend(["-L", ss.log_file])

src/tool.py

@@ -10,7 +10,7 @@
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 from .dataclass import Field, XedaBaseModel, validator
-from .utils import cached_property
+from .utils import cached_property, try_convert
 from .flow import Flow
 
 log = logging.getLogger(__name__)
@@ -80,10 +80,11 @@ class Docker(XedaBaseModel):
     def cpuinfo(self) -> Optional[List[List[str]]]:
         try:
             ret = self.run_dockerized("cat", "/proc/cpuinfo", stdout=True)
-        except:
-            return None
-        assert ret is not None
-        return [x.split("\n") for x in re.split(r"\n\s*\n", ret, re.MULTILINE)]
+        except:  # noqa
+            ret = None
+        if ret is not None:
+            return [x.split("\n") for x in re.split(r"\n\s*\n", ret, re.MULTILINE)]
+        return None
 
     @cached_property
     def nproc(self) -> int:
@@ -259,25 +260,24 @@ def fake_cpu_info(file=".xeda_cpuinfo", ncores=4):
                 f.write(f"{k}{ws}: {v}")
 
 
-def _run_processes(commands: List[List[str]], cwd: OptionalPath = None) -> None:
-    """Run a list commands to completion. Throws if any of them did not execute and exit normally"""
-    # if args:
-    #     args = [str(a) for a in args]
-    # if env:
-    #     env = {str(k): str(v) for k, v in env.items()}
-    processes = []
+def _run_processes(commands: List[List[Any]], cwd: OptionalPath = None, env=None) -> None:
+    """Run a list commands to completion. Raises an exception if any of them did not execute and exit normally"""
+    for args in commands:
+        args = [str(a) for a in args]
+    if env:
+        env = {str(k): str(v) for k, v in env.items() if v is not None}
+    processes: List[subprocess.Popen] = []
     for cmd in commands:
         log.info("Running `%s`", " ".join(cmd))
         with subprocess.Popen(
             cmd,
             cwd=cwd,
             shell=False,
-            #   stdout=subprocess.PIPE if stdout else None,
             bufsize=1,
             universal_newlines=True,
             encoding="utf-8",
             errors="replace",
-            #   env=env
+            env=env,
         ) as proc:
             assert isinstance(proc.args, list)
             log.info("Started %s[%d]", str(proc.args[0]), proc.pid)
@@ -288,7 +288,7 @@ def _run_processes(commands: List[List[str]], cwd: OptionalPath = None) -> None:
 
     for p in processes:
         if p.returncode != 0:
-            raise Exception(f"Process exited with return code {p.returncode}")
+            raise NonZeroExitCode(p.args, p.returncode)
 
 
 class Tool(XedaBaseModel):
@@ -361,13 +361,10 @@ def version(self) -> Tuple[str, ...]:
     @cached_property
     def nproc(self) -> int:
         if self.dockerized:
-            n = None
             try:
-                ret = self.execute("nproc", stdout=True)
-                if ret:
-                    n = int(ret)
-            except:
-                pass
+                n = try_convert(self.execute("nproc", stdout=True), int)
+            except:  # noqa
+                n = None
             assert self.docker
             return n or self.docker.nproc
         else:
@@ -380,22 +377,16 @@ def _version_is_gte(
         """check if `tool_version` is greater than or equal to `required_version`"""
         log.debug(f"[gte] {tool_version}  ?  {required_version}")
         for tool_part, req_part in zip(tool_version, required_version):
-            try:
-                req_part_val = int(req_part)
-            except ValueError:
-                req_part_val = -1
-            try:
-                tool_part_val = int(tool_part)
-            except ValueError:
+            req_part_val = try_convert(req_part, int, default=-1)
+            assert req_part_val is not None
+            tool_part_val = try_convert(tool_part, int)
+            if tool_part_val is None:
                 match = re.match(r"(\d+)[+.-\._](\w+)", tool_part)
                 if match:
                     tool_part = match.group(1)
-                    try:
-                        tool_part_val = int(tool_part)
-                        assert req_part_val > -1
+                    tool_part_val = try_convert(tool_part, int)
+                    if tool_part_val is not None and req_part_val > -1:
                         return tool_part_val >= req_part_val
-                    except ValueError:
-                        pass
                 tool_part_val = -1
 
             if tool_part_val < req_part_val:  # if equal, continue
@@ -440,20 +431,9 @@ def _run_remote(
 
         sshfs_opts = ["directport=10000", "idmap=user", "exec", "compression=yes"]
 
-        sshfs_cmd = [
-            "sshfs",
-            "-o",
-            ",".join(sshfs_opts),
-            f"localhost:{wd}",
-            remote_sshfs_mount_dir,
-        ]
+        sshfs_cmd = ["sshfs", "-o", ",".join(sshfs_opts), f"localhost:{wd}", remote_sshfs_mount_dir]
 
-        ssh_proc = [
-            "ssh",
-            self.remote.hostname,
-            "-p",
-            str(self.remote.port),
-        ]
+        ssh_proc = ["ssh", self.remote.hostname, "-p", str(self.remote.port)]
 
         sshfs_proc = (
             ssh_proc
@@ -470,14 +450,7 @@ def _run_remote(
             + remote_cmd
         )
 
-        ncat_proc = [
-            "ncat",
-            "-l",
-            "-p",
-            f"{client_nc_port}",
-            "-e",
-            "/usr/libexec/sftp-server",
-        ]
+        ncat_proc = ["ncat", "-l", "-p", f"{client_nc_port}", "-e", "/usr/libexec/sftp-server"]
         _run_processes([sshfs_proc, ncat_proc])
 
     def run(

src/utils.py

@@ -12,18 +12,8 @@
 from functools import cached_property, reduce
 from pathlib import Path
 from types import TracebackType
-from typing import (
-    Any,
-    Dict,
-    Iterable,
-    List,
-    Optional,
-    OrderedDict,
-    Tuple,
-    Type,
-    TypeVar,
-    Union,
-)
+from typing import Any, Dict, Iterable, List, Optional, OrderedDict, Tuple, Type, TypeVar, Union
+import unittest
 from xml.etree import ElementTree
 
 from varname import argname
@@ -51,6 +41,8 @@
     "dump_json",
     "toml_loads",
     "parse_xml",
+    "try_convert_to_primitives",
+    "try_convert",
     # list/container utils
     "unique",
     # str utils
@@ -223,7 +215,18 @@ def dict_merge(
     return rtn_dct
 
 
-def try_convert(
+_T1 = TypeVar("_T1")
+_D1 = TypeVar("_D1")
+
+
+def try_convert(value: Any, typ: Type[_T1], default: Optional[_D1] = None) -> Union[None, _T1, _D1]:
+    try:
+        return typ(value)  # type: ignore
+    except:  # noqa
+        return default
+
+
+def try_convert_to_primitives(
     s: Any, convert_lists: bool = False
 ) -> Union[bool, int, float, str, List[Union[bool, int, float, str, List[Any]]]]:
     if s is None:
@@ -235,7 +238,7 @@ def try_convert(
             return s.strip("\"'")
         if convert_lists and s.startswith("[") and s.endswith("]"):
             s = re.sub(r"\s+", "", s)
-            return try_convert(list(s.strip("][").split(",")))
+            return try_convert_to_primitives(list(s.strip("][").split(",")))
         # Should NOT convert dict, set, etc!
         if re.match(r"^\d+$", s):
             return int(s)
@@ -253,7 +256,7 @@ def try_convert(
     if isinstance(s, (tuple)):
         s = list(s)
     if isinstance(s, (list)):
-        return [try_convert(e) for e in s]
+        return [try_convert_to_primitives(e) for e in s]
     return str(s)
 
 
@@ -435,3 +438,12 @@ def first_value(d: Dict[_K, _V]) -> Optional[_V]:  # pyright: ignore reportInval
 
 def first_key(d: Dict[_K, _V]) -> Optional[_K]:  # pyright: ignore reportInvalidTypeVarUse
     return next(iter(d)) if d else None
+
+
+class UnitTestUtils(unittest.TestCase):
+    def test_try_convert(self):
+        self.assertEqual(try_convert(" ", int), None)
+        self.assertEqual(try_convert("", int), None)
+        self.assertEqual(try_convert("xx", int), None)
+        self.assertEqual(try_convert("1234", int), 1234)
+        self.assertEqual(try_convert("1234.5", float), 1234.5)