diff --git a/b_asic/architecture.py b/b_asic/architecture.py
index 2638cb6220e118919c3827fbc506d760a82b58bc..559cb48564468e0bbc4d5ebd3aa4d0bf14867b0f 100644
--- a/b_asic/architecture.py
+++ b/b_asic/architecture.py
@@ -34,7 +34,7 @@ from b_asic.codegen.vhdl.common import is_valid_vhdl_identifier
from b_asic.operation import Operation
from b_asic.port import InputPort, OutputPort
from b_asic.process import MemoryProcess, MemoryVariable, OperatorProcess, Process
-from b_asic.resources import ProcessCollection
+from b_asic.resources import ProcessCollection, _sanitize_port_option
def _interconnect_dict() -> int:
@@ -435,6 +435,8 @@ class Memory(Resource):
Number of read ports for memory.
write_ports : int, optional
Number of write ports for memory.
+ total_ports : int, optional
+ Total number of read and write ports for memory.
assign : bool, default False
Perform assignment when creating the Memory (using the default properties).
"""
@@ -448,6 +450,7 @@ class Memory(Resource):
entity_name: Optional[str] = None,
read_ports: Optional[int] = None,
write_ports: Optional[int] = None,
+ total_ports: Optional[int] = None,
assign: bool = False,
):
super().__init__(process_collection=process_collection, entity_name=entity_name)
@@ -462,6 +465,10 @@ class Memory(Resource):
raise ValueError(
f"memory_type must be 'RAM' or 'register', not {memory_type!r}"
)
+ if read_ports is not None or write_ports is not None or total_ports is not None:
+ read_ports, write_ports, total_ports = _sanitize_port_option(
+ read_ports, write_ports, total_ports
+ )
read_ports_bound = self._collection.read_ports_bound()
if read_ports is None:
self._output_count = read_ports_bound
@@ -476,6 +483,11 @@ class Memory(Resource):
if write_ports < write_ports_bound:
raise ValueError(f"At least {write_ports_bound} write ports required")
self._input_count = write_ports
+
+ total_ports_bound = self._collection.total_ports_bound()
+ if total_ports is not None and total_ports < total_ports_bound:
+ raise ValueError(f"At least {total_ports_bound} total ports required")
+
self._memory_type = memory_type
if assign:
self.assign()
diff --git a/test/test_architecture.py b/test/test_architecture.py
index ef02ad0c1320072ba7bb1e0080faa8fcbabca7c8..690b8c2720a7b998ff517d3bc31e2ec77b94d63d 100644
--- a/test/test_architecture.py
+++ b/test/test_architecture.py
@@ -1,3 +1,4 @@
+import re
from itertools import chain
from typing import List
@@ -43,26 +44,6 @@ def test_add_remove_process_from_resource(schedule_direct_form_iir_lp_filter: Sc
memory.add_process(PlainMemoryVariable(0, 0, {0: 2}, "PlainMV"))
-def test_extract_processing_elements(schedule_direct_form_iir_lp_filter: Schedule):
- # Extract operations from schedule
- operations = schedule_direct_form_iir_lp_filter.get_operations()
-
- # Split into new process collections on overlapping execution time
- adders = operations.get_by_type_name(Addition.type_name()).split_on_execution_time()
- const_mults = operations.get_by_type_name(
- ConstantMultiplication.type_name()
- ).split_on_execution_time()
-
- # List of ProcessingElements
- processing_elements: List[ProcessingElement] = []
- for adder_collection in adders:
- processing_elements.append(ProcessingElement(adder_collection))
- for const_mult_collection in const_mults:
- processing_elements.append(ProcessingElement(const_mult_collection))
-
- assert len(processing_elements) == len(adders) + len(const_mults)
-
-
def test_memory_exceptions(schedule_direct_form_iir_lp_filter: Schedule):
mvs = schedule_direct_form_iir_lp_filter.get_memory_variables()
operations = schedule_direct_form_iir_lp_filter.get_operations()
@@ -82,6 +63,11 @@ def test_architecture(schedule_direct_form_iir_lp_filter: Schedule):
mvs = schedule_direct_form_iir_lp_filter.get_memory_variables()
operations = schedule_direct_form_iir_lp_filter.get_operations()
+ with pytest.raises(
+ TypeError, match="Different Operation types in ProcessCollection"
+ ):
+ ProcessingElement(operations)
+
# Split operations further into chunks
adders = operations.get_by_type_name(Addition.type_name()).split_on_execution_time()
assert len(adders) == 1
@@ -96,7 +82,10 @@ def test_architecture(schedule_direct_form_iir_lp_filter: Schedule):
# Create necessary processing elements
adder = ProcessingElement(adders[0], entity_name="adder")
- multiplier = ProcessingElement(const_mults[0], entity_name="multiplier")
+ multiplier = ProcessingElement(const_mults[0])
+ assert multiplier.entity_name == "Undefined entity name"
+ multiplier.set_entity_name("multiplier")
+ assert multiplier.entity_name == "multiplier"
input_pe = ProcessingElement(inputs[0], entity_name="input")
output_pe = ProcessingElement(outputs[0], entity_name="output")
processing_elements: List[ProcessingElement] = [
@@ -140,6 +129,7 @@ def test_architecture(schedule_direct_form_iir_lp_filter: Schedule):
with pytest.raises(ValueError, match='32 is not a valid VHDL identifier'):
adder.set_entity_name("32")
assert adder.entity_name == "adder"
+ assert repr(adder) == "adder"
# Create architecture from
architecture = Architecture(
@@ -227,3 +217,47 @@ def test_move_process(schedule_direct_form_iir_lp_filter: Schedule):
architecture.move_process('cmul4.0', memories[0], processing_elements[0])
with pytest.raises(KeyError, match="invalid_name not in"):
architecture.move_process('invalid_name', memories[0], processing_elements[1])
+
+
+def test_resource_errors(precedence_sfg_delays):
+ precedence_sfg_delays.set_latency_of_type(Addition.type_name(), 1)
+ precedence_sfg_delays.set_latency_of_type(ConstantMultiplication.type_name(), 3)
+ precedence_sfg_delays.set_execution_time_of_type(Addition.type_name(), 1)
+ precedence_sfg_delays.set_execution_time_of_type(
+ ConstantMultiplication.type_name(), 1
+ )
+
+ schedule = Schedule(precedence_sfg_delays)
+ operations = schedule.get_operations()
+ additions = operations.get_by_type_name(Addition.type_name())
+ with pytest.raises(
+ ValueError, match='Cannot map ProcessCollection to single ProcessingElement'
+ ):
+ ProcessingElement(additions)
+
+ mv = schedule.get_memory_variables()
+ with pytest.raises(
+ ValueError,
+ match=(
+ "If total_ports is unset, both read_ports and write_ports must be provided."
+ ),
+ ):
+ Memory(mv, read_ports=1)
+ with pytest.raises(
+ ValueError, match=re.escape("Total ports (2) less then read ports (6)")
+ ):
+ Memory(mv, read_ports=6, total_ports=2)
+ with pytest.raises(
+ ValueError, match=re.escape("Total ports (6) less then write ports (7)")
+ ):
+ Memory(mv, read_ports=6, write_ports=7, total_ports=6)
+ with pytest.raises(ValueError, match="At least 6 read ports required"):
+ Memory(mv, read_ports=1, write_ports=1)
+ with pytest.raises(ValueError, match="At least 5 write ports required"):
+ Memory(mv, read_ports=6, write_ports=1)
+ with pytest.raises(ValueError, match="At least 9 total ports required"):
+ Memory(mv, read_ports=6, write_ports=5, total_ports=6)
+ with pytest.raises(
+ ValueError, match="memory_type must be 'RAM' or 'register', not 'foo'"
+ ):
+ Memory(mv, read_ports=6, write_ports=5, total_ports=6, memory_type="foo")