rebust capture of stdout/stderr

examples/mixed_language/blink/blinky.xeda.toml

@@ -0,0 +1,17 @@
+sources = ["blink.sv"]
+top = "blink"
+clock.port = "clk"
+
+[tb]
+sources = ["blink_tb.cpp"]
+
+[flows.cxxrtl]
+cxxrtl.filename = "blink.cpp"
+
+[flows.yosys_fpga]
+fpga.vendor = "xilinx"
+fpga.family = "xc7"
+
+[flows.vivado_synth]
+fpga.part = "xc7a100tftg256-2L"
+clock.period = 5.0

src/xeda/cli.py

@@ -43,6 +43,7 @@
 from .flow_runner.dse import Dse
 from .tool import ExecutableNotFound, NonZeroExitCode
 from .utils import removeprefix, settings_to_dict
+from .design import DesignValidationError
 
 log = logging.getLogger(__name__)
 
@@ -328,12 +329,12 @@ def run(
  )
  if options.debug:
  raise e
- sys.exit(2)
+ sys.exit(1)
  except NonZeroExitCode as e:
  log.critical("Flow %s failed: NonZeroExitCode %s", flow, " ".join(str(a) for a in e.args))
  if options.debug:
  raise e
- sys.exit(3)
+ sys.exit(1)
  except ExecutableNotFound as e:
  log.critical(
  "Executable '%s' was not found! (tool:%s, flow:%s, PATH:%s)",
@@ -342,17 +343,22 @@ def run(
  flow,
  e.path,
  )
- sys.exit(4)
+ sys.exit(1)
  except FlowSettingsError as e:
  log.critical("%s", e)
  if options.debug:
  raise e
- sys.exit(5)
+ sys.exit(1)
  except FlowException as e: # any flow exception
  log.critical("%s", e)
  if options.debug:
  raise e
- sys.exit(6)
+ sys.exit(1)
+ except DesignValidationError as e: # any flow exception
+ log.critical("%s", e)
+ if options.debug:
+ raise e
+ sys.exit(1)
 
 
 @cli.command(context_settings=CONTEXT_SETTINGS, short_help="List available flows.")

src/xeda/design.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 from glob import glob
-
 import hashlib
 import inspect
 import json
@@ -12,6 +11,7 @@
 from functools import cached_property
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Type, TypeVar, Union
+import yaml
 from urllib.parse import parse_qs, urlparse
 
 
@@ -31,6 +31,7 @@
  settings_to_dict,
  toml_load,
  removesuffix,
+ NonZeroExitCode,
 )
 
 log = logging.getLogger(__name__)
@@ -324,17 +325,68 @@ def _name_validate(cls, value, values) -> str:
 
 
 class Generator(XedaBaseModel):
- cwd: Union[None, str] = None
+ cwd: Optional[str] = None
  executable: Optional[str] = None
+ class_: Optional[str] = Field(None, alias="class")
  args: Union[str, List[str]] = []
- shell: bool = False
  check: bool = True
  env: Optional[Dict[str, str]] = None
  # sweepable parameters used in command
  parameters: dict = {}
  # for xeda to know dependencies, clean previous artifacts, check after generation:
  generated_sources: List[str] = []
 
+ def run(self):
+ assert self.executable
+ self.run_cmd([self.executable, *self.args])
+
+ def run_cmd(self, cmd, check=None, stdout=None, stderr=None):
+ log.info("Running command: '%s'", " ".join(cmd))
+ p = subprocess.run(
+ cmd,
+ cwd=self.cwd,
+ check=False if check is None else self.check,
+ stdout=stdout,
+ stderr=stderr,
+ env=self.env,
+ )
+ if self.check and p.returncode:
+ raise NonZeroExitCode(cmd, p.returncode)
+ return p
+
+ @property
+ def name(self) -> str:
+ return str(self.__class__.__qualname__ or "generator")
+
+
+class ChiselGenerator(Generator):
+ main: Optional[str] = None
+ project: Optional[str] = None
+ # executable = "bloop"
+
+ def run(self):
+ self.check = True
+ if self.project is None:
+ p = self.run_cmd(["bloop", "projects"], stdout=subprocess.PIPE)
+ projects_str = p.stdout.decode()
+ projects = re.split(r"\s+", projects_str)
+ if projects:
+ log.info(f"Found projects: {', '.join(projects)}")
+ self.project = projects[0]
+ else:
+ log.error("No projects found!")
+ raise ValueError("No projects found!")
+ assert self.project, "project not set!"
+ cmd = ["bloop", "run", self.project]
+ if self.main:
+ cmd += ["--main", self.main]
+ if self.args:
+ if isinstance(self.args, str):
+ self.args = self.args.split()
+ cmd.append("--")
+ cmd += self.args
+ self.run_cmd(cmd)
+
 
 class RtlSettings(DVSettings):
  """design.rtl"""
@@ -360,26 +412,29 @@ class RtlSettings(DVSettings):
  clock: Optional[Clock] = None # DEPRECATED # TODO remove
  clock_port: Optional[str] = None # TODO remove?
 
- @root_validator(pre=False)
+ @root_validator(pre=True)
  def rtl_settings_validate(cls, values): # pylint: disable=no-self-argument
  """copy equivalent clock fields (backward compatibility)"""
  clock = values.get("clock")
  clock_port = values.get("clock_port")
  clocks = values.get("clocks")
 
+ if clocks is None:
+ clocks = {}
  if not clock:
  if clock_port:
  clock = Clock(port=clock_port)
  elif len(clocks) == 1:
  clock = list(clocks.values())[0]
  if clock:
+ if isinstance(clock, dict):
+ clock = Clock(**clock)
  if not clock_port:
  clock_port = clock.port
  if not clocks:
  clocks = {"main_clock": clock}
  values["clock"] = clock
- if clocks:
- values["clocks"] = clocks
+ values["clocks"] = clocks
  if clock_port:
  values["clock_port"] = clock_port
  return values
@@ -682,15 +737,17 @@ def process_compatibility(cls, data: Dict[str, Any]) -> Dict[str, Any]:
  clock = data.pop("clock", None)
  if clock:
  clocks = list(clock) if isinstance(clock, (list, tuple)) else [clock]
- if (
- clocks
- and isinstance(clocks, (list, tuple))
- and all(isinstance(c, str) for c in clocks)
- ):
- clocks = {c: {"port": c} for c in clocks}
+ if clocks and isinstance(clocks, (list, tuple)):
+ if all(isinstance(c, str) for c in clocks):
+ clocks = {c: {"port": c} for c in clocks}
+ elif all(isinstance(c, dict) for c in clocks):
+ clocks = {
+ c.get("name", c.get("port")): {"port": c.get("port"), "name": c.get("name")}
+ for c in clocks
+ }
  data["rtl"] = {
- "generator": data.pop("generator", None),
  "sources": data.pop("sources", []),
+ "generator": data.pop("generator", None),
  "parameters": data.pop("parameters", []),
  "defines": data.pop("defines", []),
  "top": data.pop("top", None),
@@ -705,23 +762,27 @@ def process_generation(cls, data: Dict[str, Any]):
  design_root = Path.cwd()
  else:
  design_root = Path(design_root)
- generator = data.get("rtl", {}).get("generator", None)
+ generator = data.get("rtl", {}).pop("generator", None)
  if generator:
  with WorkingDirectory(design_root):
- log.info("Running generator: %s", generator)
  if isinstance(generator, str):
+ log.info("Running generator: %s", generator)
  os.system(generator) # nosec S605
  elif isinstance(generator, (dict, Generator)):
  if isinstance(generator, (dict)):
- generator = Generator(**generator)
- subprocess.run(
- generator.args,
- executable=generator.executable,
- cwd=generator.cwd,
- shell=generator.shell, # nosec S602
- check=generator.check,
- env=generator.env,
- )
+ clazz = generator.get("class")
+ if clazz:
+ assert isinstance(clazz, str)
+ if clazz.lower() == "chisel":
+ generator = ChiselGenerator(**generator)
+ else:
+ raise Exception(f"unkown generator class: {clazz}")
+ else:
+ generator = Generator(**generator)
+ if generator.cwd is None:
+ generator.cwd = str(design_root)
+ log.info("Running generator: %s", generator.name)
+ generator.run()
  else:
  args = generator
  # gen_script = Path(args[0])
@@ -859,6 +920,10 @@ def from_file(
  with open(design_file, "r") as f:
  design_dict = json.load(f)
  design_dict = expand_hierarchy(design_dict)
+ elif design_file.suffix in {".yaml", ".yml"}:
+ with open(design_file, "r") as f:
+ design_dict = yaml.safe_load(f)
+ design_dict = expand_hierarchy(design_dict)
  else:
  raise ValueError(f"File extension `{design_file.suffix}` is not supported.")
  design_dict = hierarchical_merge(design_dict, overrides)

src/xeda/flows/yosys/yosys_fpga.py

@@ -1,6 +1,6 @@
 import logging
 from pathlib import Path
-from typing import List, Literal, Optional, Union
+from typing import Iterable, List, Literal, Optional, Union
 
 from ...dataclass import Field
 from ...flow import FpgaSynthFlow
@@ -138,26 +138,30 @@ def parse_reports(self) -> bool:
  **num_cells_by_type,
  }
  self.results["_utilization"] = design_util
- else:
- if self.settings.fpga:
  if self.settings.fpga.vendor == "xilinx":
- self.parse_report_regex(
- self.artifacts.utilization_report,
- r"=+\s*design hierarchy\s*=+",
- r"DSP48(E\d+)?\s*(?P<DSP48>\d+)",
- r"FDRE\s*(?P<_FDRE>\d+)",
- r"FDSE\s*(?P<_FDSE>\d+)",
- r"number of LCs:\s*(?P<Estimated_LCs>\d+)",
- sequential=True,
- required=False,
- )
- self.results["FFs"] = int(self.results.get("_FDRE", 0)) + int(
- self.results.get("_FDSE", 0)
- )
- if self.settings.fpga.family == "ecp5":
- self.parse_report_regex(
- self.artifacts.utilization_report,
- r"TRELLIS_FF\s+(?P<FFs>\d+)",
- r"LUT4\s+(?P<LUT4>\d+)",
+ self.results["LUT"] = sum_all_resources(
+ design_util, [f"LUT{i}" for i in range(2, 7)]
  )
+ ram32m = sum_all_resources(design_util, ["RAM32M"])
+ if ram32m:
+ self.results["LUT"] += ram32m
+ self.results["LUT:RAM"] = ram32m
+ self.results["FF"] = sum_all_resources(design_util, ["FDRE", "FDSE"])
+ carry4 = sum_all_resources(design_util, ["CARRY4"])
+ if carry4:
+ self.results["CARRY"] = carry4
+ brams = sum_all_resources(design_util, ["RAMB36"])
+ brams_half = sum_all_resources(design_util, ["RAMB18"])
+ brams += brams_half / 2
+ if brams:
+ self.results["BRAM"] = brams
+ dsps = sum_all_resources(design_util, ["DSP48E"])
+ if dsps:
+ self.results["DSP"] = dsps
+
+ # if self.settings.fpga:
  return True
+
+
+def sum_all_resources(design_util: dict, lst: Iterable) -> int:
+ return sum(int(design_util.get(t, 0)) for t in lst)