From c205a8c06cd77f5b7c79bef39a1ebcf14aa83b75 Mon Sep 17 00:00:00 2001 From: "Mohamed A. Bamakhrama" Date: Thu, 6 Aug 2026 16:07:07 +0300 Subject: [PATCH] Support multidimensional ports Add support for multidimensional design and IP ports Signed-off-by: Mohamed A. Bamakhrama --- tests/tests_ir/backend/test_yaml.py | 81 +++++++++++++++++++ tests/tests_ir/frontend/test_yaml.py | 56 +++++++++++++ topwrap/backend/yaml/backend.py | 20 ++++- topwrap/backend/yaml/common/ip_core_schema.py | 27 +++++++ topwrap/frontend/yaml/design.py | 36 +++++++-- topwrap/frontend/yaml/design_schema.py | 13 ++- topwrap/frontend/yaml/ip_core.py | 22 +++++ 7 files changed, 244 insertions(+), 11 deletions(-) diff --git a/tests/tests_ir/backend/test_yaml.py b/tests/tests_ir/backend/test_yaml.py index 1761dc88..4c3855eb 100644 --- a/tests/tests_ir/backend/test_yaml.py +++ b/tests/tests_ir/backend/test_yaml.py @@ -41,6 +41,7 @@ Bits, BitStruct, Dimensions, + LogicArray, LogicBitSelect, LogicFieldSelect, LogicSelect, @@ -259,6 +260,49 @@ def test_complex_port(self): }, } + def test_multidimensional_port(self): + ty = Bits( + dimensions=[ + Dimensions(upper=ElaboratableValue(1), lower=ElaboratableValue(0)), + Dimensions(upper=ElaboratableValue(7), lower=ElaboratableValue(0)), + ] + ) + + top = Module( + id=Identifier(name="top"), + ports=[ + Port( + name="foo", + direction=PortDirection.IN, + type=ty, + default_value=ElaboratableValue(4), + ), + ], + ) + + backend = IpCoreDescriptionBackend() + + out = backend.represent(top) + [out] = backend.serialize(out) + tree = yaml.safe_load(out.content) + + assert tree == { + "id": {"name": "top", "library": "libdefault", "vendor": "vendor", "version": "0.1"}, + "signals": { + "in": [ + { + "name": "foo", + "dimensions": [["1", "0"], ["7", "0"]], + "default": "4", + }, + ], + }, + } + + frontend = IPCoreDescriptionFrontend() + mod = frontend.parse_str(out.content) + _compare_modules(top, mod) + def test_parameters(self): mod = Module( id=Identifier(name="top"), @@ -612,6 +656,43 @@ def test_config_output(self): expected_obj: dict[str, str] = {name: h.to_str() for name, h in rep.items()} assert repo_dict == expected_obj + def test_multidimensional_top_level_ports_roundtrip(self): + design_yaml = """ + name: top + external: + ports: + in: + - name: in_arr + dimensions: + - [1, 0] + - [7, 0] + out: + - name: out_vec + dimensions: + - [15, 0] + """ + + front = DesignDescriptionFrontend() + orig_des = front.parse_str(design_yaml) + + back = DesignDescriptionBackend() + out = back.represent(orig_des.parent) + [out] = back.serialize(out) + + new_des = front.parse_str(out.content) + + in_arr = new_des.parent.ports.find_by_name_or_error("in_arr") + out_vec = new_des.parent.ports.find_by_name_or_error("out_vec") + + assert isinstance(in_arr.type, LogicArray) + assert in_arr.type.dimensions == [ + Dimensions(upper=ElaboratableValue(1), lower=ElaboratableValue(0)), + Dimensions(upper=ElaboratableValue(7), lower=ElaboratableValue(0)), + ] + assert isinstance(out_vec.type, LogicArray) + assert out_vec.type.dimensions == [ + Dimensions(upper=ElaboratableValue(15), lower=ElaboratableValue(0)) + ] class TestDesignPositionsBackend: def test_positions(self): diff --git a/tests/tests_ir/frontend/test_yaml.py b/tests/tests_ir/frontend/test_yaml.py index 31fbc41e..d2138b60 100644 --- a/tests/tests_ir/frontend/test_yaml.py +++ b/tests/tests_ir/frontend/test_yaml.py @@ -262,8 +262,64 @@ def test_config_field(self): assert all(type(out.config.repositories[k]) is type(rep[k]) for k in keys) assert all(out.config.repositories[k].to_str() == rep[k].to_str() for k in keys) + def test_multidimensional_top_level_ports(self): + des = """ + name: top + external: + ports: + in: + - name: in_arr + dimensions: + - [1, 0] + - [7, 0] + out: + - name: out_vec + dimensions: + - [15, 0] + """ + + mod = DesignDescriptionFrontend().parse_str(des) + + in_arr = mod.parent.ports.find_by_name_or_error("in_arr") + out_vec = mod.parent.ports.find_by_name_or_error("out_vec") + + assert isinstance(in_arr.type, LogicArray) + assert in_arr.type.dimensions == [ + Dimensions(upper=ElaboratableValue(1), lower=ElaboratableValue(0)), + Dimensions(upper=ElaboratableValue(7), lower=ElaboratableValue(0)), + ] + assert isinstance(out_vec.type, LogicArray) + assert out_vec.type.dimensions == [ + Dimensions(upper=ElaboratableValue(15), lower=ElaboratableValue(0)) + ] + class TestIPCoreDescriptionFrontend: + def test_multidimensional_signal(self): + ip = """ + id: + name: top + vendor: vendor + library: libdefault + signals: + in: + - name: in_arr + dimensions: + - [1, 0] + - [7, 0] + default: 4 + """ + + mod = IPCoreDescriptionFrontend().parse_str(ip) + in_arr = mod.ports.find_by_name_or_error("in_arr") + + assert isinstance(in_arr.type, LogicArray) + assert in_arr.type.dimensions == [ + Dimensions(upper=ElaboratableValue(1), lower=ElaboratableValue(0)), + Dimensions(upper=ElaboratableValue(7), lower=ElaboratableValue(0)), + ] + assert in_arr.default_value == ElaboratableValue("4") + def test_parse_on_mem_yaml(self): ip = Path("examples/ir_examples/interconnect/ips/mem.yaml") mod = IPCoreDescriptionFrontend().parse_file(ip) diff --git a/topwrap/backend/yaml/backend.py b/topwrap/backend/yaml/backend.py index cb2c119f..2d9bccb3 100644 --- a/topwrap/backend/yaml/backend.py +++ b/topwrap/backend/yaml/backend.py @@ -18,6 +18,7 @@ IPCoreInterface, IPCoreIntfPorts, IPCoreParameter, + IPCorePortDefinition, IPCorePorts, IPCoreStruct, IPCoreStructField, @@ -28,6 +29,7 @@ ConnectionsSection, DesignDescription, DesignExternalIntfs, + DesignExternalPortDefinition, DesignExternalPorts, DesignExternalSection, DesignInverterPosition, @@ -149,7 +151,11 @@ def _represent_signal( raise ValueError("Trying to slice a single bit") elif isinstance(type, Bits): if len(type.dimensions) > 1: - raise ValueError("IP core YAML format only supports one-dimensional bit vectors") + return IPCorePortDefinition( + name=name, + dimensions=tuple((d.upper.value, d.lower.value) for d in type.dimensions), + default=default.value if default else None, + ) bound = (type.dimensions[0].upper.value, type.dimensions[0].lower.value) else: @@ -536,11 +542,19 @@ def _represent_external_ports(self, mod: Module) -> DesignExternalPorts: outputs = [] inouts = [] + def represent_port(port: Port): + if isinstance(port.type, LogicArray) and isinstance(port.type.item, Bit): + return DesignExternalPortDefinition( + name=port.name, + dimensions=[(dim.upper.value, dim.lower.value) for dim in port.type.dimensions], + ) + return port.name + for port in mod.non_intf_ports(): if port.direction is PortDirection.IN: - inputs.append(port.name) + inputs.append(represent_port(port)) elif port.direction is PortDirection.OUT: - outputs.append(port.name) + outputs.append(represent_port(port)) elif port.direction is PortDirection.INOUT: # Look for connection that this port is a part of, then from that # find the module port it's connected to. diff --git a/topwrap/backend/yaml/common/ip_core_schema.py b/topwrap/backend/yaml/common/ip_core_schema.py index 49bbb788..a7076acb 100644 --- a/topwrap/backend/yaml/common/ip_core_schema.py +++ b/topwrap/backend/yaml/common/ip_core_schema.py @@ -33,6 +33,7 @@ from topwrap.util import get_config, get_interface_by_id _StrOrInt = Union[str, int] +IPCoreDimension = Tuple[_StrOrInt, _StrOrInt] @marshmallow_dataclass.dataclass(frozen=True) @@ -55,11 +56,26 @@ def _validate(self, self_obj: Dict[str, Any], **kwargs: Any) -> bool: return True +@marshmallow_dataclass.dataclass(frozen=True) +class IPCorePortDefinition(MarshmallowDataclassExtensions): + name: str + dimensions: Tuple[IPCoreDimension, ...] = ext_field(tuple, inline_depth=1) + default: Optional[_StrOrInt] = ext_field(None) + + @marshmallow.validates_schema + def _validate(self, self_obj: Dict[str, Any], **kwargs: Any) -> bool: + if len(self_obj["dimensions"]) == 0: + raise marshmallow.ValidationError("Signal dimensions cannot be empty") + + return True + + Signal = Union[ str, Tuple[str, _StrOrInt, _StrOrInt], Tuple[str, _StrOrInt, _StrOrInt, _StrOrInt, _StrOrInt], IPCoreComplexSignal, + IPCorePortDefinition, ] @@ -89,6 +105,17 @@ def raw(self) -> Signal: @staticmethod def from_sig_and_dir(sig: Signal, dir: LegacyPortDirection) -> "IPCorePort": + if isinstance(sig, IPCorePortDefinition): + upper_bound, lower_bound = sig.dimensions[0] + return IPCorePort( + name=sig.name, + direction=dir, + upper_bound=upper_bound, + lower_bound=lower_bound, + upper_slice=upper_bound, + lower_slice=lower_bound, + ) + if isinstance(sig, IPCoreComplexSignal): name = sig.name if sig.name is not None else str(sig.path) assert name is not None diff --git a/topwrap/frontend/yaml/design.py b/topwrap/frontend/yaml/design.py index 9bfa0af6..a16d3ffd 100644 --- a/topwrap/frontend/yaml/design.py +++ b/topwrap/frontend/yaml/design.py @@ -12,6 +12,7 @@ from topwrap.backend.yaml.common.ip_core_schema import param_to_ir_param from topwrap.frontend.yaml.design_schema import ( DesignDescription, + DesignExternalPortDefinition, DesignIP, DesignNodePosition, DesignPositionDefinition, @@ -36,7 +37,7 @@ ResetPolarity, ) from topwrap.model.design import ClockDomain, Design, ModuleInstance, ResetDomain -from topwrap.model.hdl_types import Bit +from topwrap.model.hdl_types import Bit, Bits, Dimensions from topwrap.model.interconnect import Interconnect from topwrap.model.interface import Interface, InterfaceDefinition, InterfaceMode, InterfaceSignal from topwrap.model.memory_map import MemoryMap as IRMemoryMap @@ -136,15 +137,36 @@ def _parse_components(self, desc: DesignDescription, design: Design, source: Opt parsed, _ = self._parse_hier(source, hdesc, hname) design.add_component(ModuleInstance(name=hname, module=parsed.parent)) - def _parse_ports(self, desc: DesignDescription) -> dict[str, tuple[PortDirection, bool]]: + # Parse external port names and determine if they are multidimensional + def _external_decl_name(self, decl: str | DesignExternalPortDefinition) -> str: + return decl if isinstance(decl, str) else decl.name + + def _external_decl_type(self, decl: DesignExternalPortDefinition): + if len(decl.dimensions) == 0: + return Bit() + return Bits( + dimensions=[ + Dimensions(ElaboratableValue(upper), ElaboratableValue(lower)) + for upper, lower in decl.dimensions + ] + ) + + def _parse_ports(self, desc: DesignDescription, mod: Module) -> dict[str, tuple[PortDirection, bool]]: declared_exts = dict[str, tuple[PortDirection, bool]]() for port, group in ((True, desc.external.ports), (False, desc.external.interfaces)): for dir, decls in ((PortDirection.IN, group.input), (PortDirection.OUT, group.output)): for d in decls: - if d in declared_exts: - logger.warning(f"Skipping duplicated external IO: '{d}'") + name = self._external_decl_name(d) if port else d + if name in declared_exts: + logger.warning(f"Skipping duplicated external IO: '{name}'") continue - declared_exts[d] = (dir, port) + + if port and isinstance(d, DesignExternalPortDefinition): + mod.add_port( + Port(name=name, direction=dir, type=self._external_decl_type(d)) + ) + + declared_exts[name] = (dir, port) return declared_exts def _parse_connections( @@ -163,6 +185,8 @@ def _parse_connections( def _add_ports(self, mod: Module, declared_exts: dict[str, tuple[PortDirection, bool]]): for name, (dir, port) in declared_exts.items(): + if mod.ios.find_by_name(name) is not None: + continue if port: mod.add_port(Port(name=name, direction=dir, type=Bit())) else: @@ -218,7 +242,7 @@ def _parse_hier( # Gather declarations of external ports and interfaces so that # they can be instantiated with the inferred type later on - declared_exts = self._parse_ports(desc) + declared_exts = self._parse_ports(desc, mod) # Parse regular connections between ports, interfaces and externals self._parse_connections(desc, design, declared_exts) diff --git a/topwrap/frontend/yaml/design_schema.py b/topwrap/frontend/yaml/design_schema.py index f7991743..790f9c1c 100644 --- a/topwrap/frontend/yaml/design_schema.py +++ b/topwrap/frontend/yaml/design_schema.py @@ -36,10 +36,19 @@ def path(self): return self.file.to_path() +@marshmallow_dataclass.dataclass(frozen=True) +class DesignExternalPortDefinition(MarshmallowDataclassExtensions): + name: str + dimensions: List[Tuple[Union[str, int], Union[str, int]]] = ext_field(list, inline_depth=1) + + +DesignExternalPort = Union[str, DesignExternalPortDefinition] + + @marshmallow_dataclass.dataclass(frozen=True) class DesignExternalPorts(MarshmallowDataclassExtensions): - input: List[str] = ext_field(list, data_key="in") - output: List[str] = ext_field(list, data_key="out") + input: List[DesignExternalPort] = ext_field(list, data_key="in") + output: List[DesignExternalPort] = ext_field(list, data_key="out") inout: List[Tuple[str, str]] = ext_field(list, inline_depth=1) @cached_property diff --git a/topwrap/frontend/yaml/ip_core.py b/topwrap/frontend/yaml/ip_core.py index f4ef501d..128afff0 100644 --- a/topwrap/frontend/yaml/ip_core.py +++ b/topwrap/frontend/yaml/ip_core.py @@ -12,6 +12,7 @@ IPCoreComplexSignal, IPCoreDescription, IPCoreDescriptionFrontendException, + IPCorePortDefinition, IPCoreStruct, IPCoreType, Signal, @@ -118,6 +119,27 @@ def to_dims(lst: Sequence[Union[str, int]]): lower=ElaboratableValue(lst[1]), ) + if isinstance(signal, IPCorePortDefinition): + type = Bits(dimensions=[to_dims(dim) for dim in signal.dimensions]) + default = ElaboratableValue(signal.default) if signal.default is not None else None + + if default is not None and direction is not PortDirection.IN: + raise IPCoreDescriptionFrontendException( + f"Default value '{default}' assigned to non-input port '{signal.name}'" + ) + + if not (port := mod.ports.find_by_name(signal.name)): + mod.add_port( + port := Port( + name=signal.name, + type=type, + direction=direction, + default_value=default, + ) + ) + + return port + if isinstance(signal, IPCoreComplexSignal): if signal.name is not None: slice = None if signal.slice is None else to_dims(signal.slice)