From 6d374b0a6391312b81abfc48132e578be5ebadbb Mon Sep 17 00:00:00 2001 From: Richard Lin Date: Sun, 31 May 2026 22:11:54 -0700 Subject: [PATCH 1/5] refactor --- edg/core/Blocks.py | 23 +++++++++++++++++------ edg/core/HierarchyBlock.py | 2 +- edg/parts/microcontroller/Esp32.py | 4 +--- edg/parts/microcontroller/Esp32c3.py | 11 ++++------- edg/parts/microcontroller/Esp32s3.py | 4 +--- edg/parts/microcontroller/Stm32f303.py | 1 - 6 files changed, 24 insertions(+), 21 deletions(-) diff --git a/edg/core/Blocks.py b/edg/core/Blocks.py index 9342586dc..1fead099d 100644 --- a/edg/core/Blocks.py +++ b/edg/core/Blocks.py @@ -289,7 +289,7 @@ def __init__(self) -> None: self._param_docs = IdentityDict[ConstraintExpr, str]() self._ports: SubElementDict[BasePort] = self.manager.new_dict(BasePort) - self._required_ports = IdentitySet[BasePort]() + self._required_ports = IdentityDict[BasePort, Optional[BoolExpr]]() # port -> optional expr for required self._port_docs = IdentityDict[BasePort, str]() self._connects = self.manager.new_dict(Connection, anon_prefix="anon_link") @@ -397,12 +397,18 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: for name, port in self._ports.items(): if port in self._required_ports: if isinstance(port, Port): - port.is_connected()._populate_expr_proto(edgir.add_pair(pb.constraints, f"(reqd){name}"), ref_map) + connected_condition = port.is_connected() elif isinstance(port, Vector): - (port.length() > 0)._populate_expr_proto(edgir.add_pair(pb.constraints, f"(reqd){name}"), ref_map) + connected_condition = port.length() > 0 else: raise ValueError(f"unknown non-optional port type {port}") + precondition = self._required_ports[port] + if precondition is not None: + self.require(precondition.implies(connected_condition), f"(reqd){name}") + else: + self.require(connected_condition, f"(reqd){name}") + self._constraints.finalize() # needed for source locator generation self._add_doc_metadata() @@ -525,7 +531,7 @@ def assign( T = TypeVar("T", bound=BasePort) - def Port(self, tpe: T, *, optional: bool = False, doc: Optional[str] = None) -> T: + def Port(self, tpe: T, *, optional: BoolLike = False, doc: Optional[str] = None) -> T: """Registers a port for this Block""" if self._elaboration_state != BlockElaborationState.init: raise BlockDefinitionError( @@ -539,8 +545,13 @@ def Port(self, tpe: T, *, optional: bool = False, doc: Optional[str] = None) -> elt = tpe._bind(self) self._ports.register(elt) - if not optional: - self._required_ports.add(elt) + if isinstance(optional, bool): + if not optional: + self._required_ports[elt] = None + elif isinstance(optional, BoolExpr): + self._required_ports[elt] = ~optional + else: + raise EdgTypeError(f"optional flag to Port(...)", optional, (bool, BoolExpr)) if doc is not None: self._port_docs[elt] = doc diff --git a/edg/core/HierarchyBlock.py b/edg/core/HierarchyBlock.py index 8c108c47a..a05824976 100644 --- a/edg/core/HierarchyBlock.py +++ b/edg/core/HierarchyBlock.py @@ -672,7 +672,7 @@ def ArgParameter(self, param: CastableType, *, doc: Optional[str] = None) -> Con T = TypeVar("T", bound=BasePort) @override - def Port(self, tpe: T, tags: Iterable[PortTag] = [], *, optional: bool = False, doc: Optional[str] = None) -> T: + def Port(self, tpe: T, tags: Iterable[PortTag] = [], *, optional: BoolLike = False, doc: Optional[str] = None) -> T: """Registers a port for this Block""" if not isinstance(tpe, (Port, Vector)): raise NotImplementedError("Non-Port (eg, Vector) ports not (yet?) supported") diff --git a/edg/parts/microcontroller/Esp32.py b/edg/parts/microcontroller/Esp32.py index f4e9eda35..1a8b03341 100644 --- a/edg/parts/microcontroller/Esp32.py +++ b/edg/parts/microcontroller/Esp32.py @@ -62,7 +62,6 @@ class Esp32_Wroom_32_Device( def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [], **kwargs: Any) -> None: super().__init__(**kwargs) - self._model = self.ArgParameter(_model) self._allowed_pins = self.ArgParameter(_allowed_pins) self.generator_param(self._allowed_pins) @@ -85,8 +84,7 @@ def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [] pullup_capable=True, pulldown_capable=True, ) - self.chip_pu = self.Port(self._dio_model, optional=True) - self.require((~self._model).implies(self.chip_pu.is_connected()), "chip_pu must not be left floating") + self.chip_pu = self.Port(self._dio_model, optional=_model) # section 2.4, table 5: strapping IOs that need a fixed value to boot, TODO currently not allocatable post-boot self.io0 = self.Port(self._dio_model, optional=True) # default pullup (SPI boot), set low to download boot self.io2 = self.Port( diff --git a/edg/parts/microcontroller/Esp32c3.py b/edg/parts/microcontroller/Esp32c3.py index 3e0c9db58..fab2092ba 100644 --- a/edg/parts/microcontroller/Esp32c3.py +++ b/edg/parts/microcontroller/Esp32c3.py @@ -58,7 +58,6 @@ def _system_pinmap(self) -> Mapping[Union[Iterable[str], str], Union[Passive, Ha def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [], **kwargs: Any) -> None: super().__init__(**kwargs) - self._model = self.ArgParameter(_model) self._allowed_pins = self.ArgParameter(_allowed_pins) self.generator_param(self._allowed_pins) @@ -108,19 +107,17 @@ def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [] pullup_capable=True, pulldown_capable=True, ) - self.en = self.Port(DigitalSink.from_bidir(self._dio_model), optional=True) # needs external pullup - self.io2 = self.Port(self._dio_model, optional=True) # needs external pullup; affects IO glitching on boot - self.io8 = self.Port(self._dio_model, optional=True) # needs external pullup, required for download boot + self.en = self.Port(DigitalSink.from_bidir(self._dio_model), optional=_model) # needs external pullup + self.io2 = self.Port(self._dio_model, optional=_model) # needs external pullup; affects IO glitching on boot + self.io8 = self.Port(self._dio_model, optional=_model) # needs external pullup, required for download boot self.io9 = self.Port( self._dio_model, optional=True ) # internally pulled up for SPI boot, connect to GND for download - self.require((~self._model).implies(self.en.is_connected() & self.io2.is_connected() & self.io8.is_connected())) # similarly, the programming UART is fixed and allocated separately self.uart0 = self.Port(UartPort(self._dio_model), optional=True) - self.lna_in = self.Port(Passive(), optional=True) - self.require((~self._model).implies(self.lna_in.is_connected())) + self.lna_in = self.Port(Passive(), optional=_model) @override def generate(self) -> None: diff --git a/edg/parts/microcontroller/Esp32s3.py b/edg/parts/microcontroller/Esp32s3.py index 15ecd520d..0e9edff76 100644 --- a/edg/parts/microcontroller/Esp32s3.py +++ b/edg/parts/microcontroller/Esp32s3.py @@ -65,7 +65,6 @@ class Esp32s3_Wroom_1_Device( def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [], **kwargs: Any) -> None: super().__init__(**kwargs) - self._model = self.ArgParameter(_model) self._allowed_pins = self.ArgParameter(_allowed_pins) self.generator_param(self._allowed_pins) @@ -89,8 +88,7 @@ def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [] pulldown_capable=True, ) - self.chip_pu = self.Port(self._dio_model, optional=True) - self.require((~self._model).implies(self.chip_pu.is_connected()), "chip_pu must not be left floating") + self.chip_pu = self.Port(self._dio_model, optional=_model) self.io0 = self.Port( self._dio_model, optional=True ) # table 2-11, default pullup (SPI boot), set low to download boot diff --git a/edg/parts/microcontroller/Stm32f303.py b/edg/parts/microcontroller/Stm32f303.py index ed03e19c6..7ffad4f5e 100644 --- a/edg/parts/microcontroller/Stm32f303.py +++ b/edg/parts/microcontroller/Stm32f303.py @@ -1,6 +1,5 @@ from typing import * -from deprecated import deprecated from typing_extensions import override from ...circuits import * From 8d042a78cd841740f3c8ce0865e3cf0b5785ff19 Mon Sep 17 00:00:00 2001 From: Richard Lin Date: Sun, 31 May 2026 22:14:23 -0700 Subject: [PATCH 2/5] wip --- edg/parts/microcontroller/Esp32c3.py | 4 ++-- edg/parts/microcontroller/Rp2040.py | 17 ++++------------- 2 files changed, 6 insertions(+), 15 deletions(-) diff --git a/edg/parts/microcontroller/Esp32c3.py b/edg/parts/microcontroller/Esp32c3.py index fab2092ba..46a096af1 100644 --- a/edg/parts/microcontroller/Esp32c3.py +++ b/edg/parts/microcontroller/Esp32c3.py @@ -92,9 +92,9 @@ def __init__(self, _model: BoolLike = False, _allowed_pins: ArrayStringLike = [] # 10ppm requirement from ESP32-C3-WROOM schematic, and in ESP32 hardware design guidelines self.xtal = self.Port( # vdda domain assumed - CrystalDriver(frequency_limits=40 * MHertz(tol=10e-6), voltage_out=self.vdda.link().voltage), optional=True + CrystalDriver(frequency_limits=40 * MHertz(tol=10e-6), voltage_out=self.vdda.link().voltage), + optional=_model, ) - self.require((~self._model).implies(self.xtal.is_connected())) # section 2.4: strapping IOs that need a fixed value to boot, and currently can't be allocated as GPIO # TODO model from different 3.3v domains diff --git a/edg/parts/microcontroller/Rp2040.py b/edg/parts/microcontroller/Rp2040.py index fba96530e..d1076bde0 100644 --- a/edg/parts/microcontroller/Rp2040.py +++ b/edg/parts/microcontroller/Rp2040.py @@ -111,19 +111,10 @@ def __init__(self, *, _model: BoolLike = False, _allowed_pins: ArrayStringLike = pulldown_capable=True, ) - self.qspi = self.Port(SpiController(self._dio_std_model), optional=True) # TODO actually QSPI - self.qspi_cs = self.Port(self._dio_std_model, optional=True) - self.qspi_sd2 = self.Port(self._dio_std_model, optional=True) - self.qspi_sd3 = self.Port(self._dio_std_model, optional=True) - self.require( - (~self._model).implies( - self.qspi.is_connected() - & self.qspi_cs.is_connected() - & self.qspi_sd2.is_connected() - & self.qspi_sd3.is_connected() - ), - "SPI memory required", - ) + self.qspi = self.Port(SpiController(self._dio_std_model), optional=self._model) # TODO actually QSPI + self.qspi_cs = self.Port(self._dio_std_model, optional=self._model) + self.qspi_sd2 = self.Port(self._dio_std_model, optional=self._model) + self.qspi_sd3 = self.Port(self._dio_std_model, optional=self._model) self.xosc = self.Port( CrystalDriver( From 72ee5b5bdaf7b8a8bae9e224c994e95e88e2ffc0 Mon Sep 17 00:00:00 2001 From: Richard Lin Date: Sun, 31 May 2026 22:17:42 -0700 Subject: [PATCH 3/5] Update Blocks.py --- edg/core/Blocks.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/edg/core/Blocks.py b/edg/core/Blocks.py index 1fead099d..1da14db0f 100644 --- a/edg/core/Blocks.py +++ b/edg/core/Blocks.py @@ -405,9 +405,10 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: precondition = self._required_ports[port] if precondition is not None: - self.require(precondition.implies(connected_condition), f"(reqd){name}") - else: - self.require(connected_condition, f"(reqd){name}") + connected_condition = precondition.implies(connected_condition) + + # add required constraints first + connected_condition._populate_expr_proto(edgir.add_pair(pb.constraints, f"(reqd){name}"), ref_map) self._constraints.finalize() # needed for source locator generation From f1400bdbacea58742f5bb5808356847fc8852fc1 Mon Sep 17 00:00:00 2001 From: Richard Lin Date: Sun, 31 May 2026 23:28:57 -0700 Subject: [PATCH 4/5] add unit test --- edg/core/Blocks.py | 3 ++- edg/core/test_block.py | 47 ++++++++++++++++++++++++++++-------------- 2 files changed, 33 insertions(+), 17 deletions(-) diff --git a/edg/core/Blocks.py b/edg/core/Blocks.py index 1da14db0f..a2147b12b 100644 --- a/edg/core/Blocks.py +++ b/edg/core/Blocks.py @@ -289,7 +289,7 @@ def __init__(self) -> None: self._param_docs = IdentityDict[ConstraintExpr, str]() self._ports: SubElementDict[BasePort] = self.manager.new_dict(BasePort) - self._required_ports = IdentityDict[BasePort, Optional[BoolExpr]]() # port -> optional expr for required + self._required_ports = IdentityDict[BasePort, Optional[BoolExpr]]() # port -> expr for not-required self._port_docs = IdentityDict[BasePort, str]() self._connects = self.manager.new_dict(Connection, anon_prefix="anon_link") @@ -405,6 +405,7 @@ def _populate_def_proto_block_base(self, pb: edgir.BlockLikeTypes) -> None: precondition = self._required_ports[port] if precondition is not None: + self._check_constraint(precondition) connected_condition = precondition.implies(connected_condition) # add required constraints first diff --git a/edg/core/test_block.py b/edg/core/test_block.py index 8e7c54fb9..6aa99b799 100644 --- a/edg/core/test_block.py +++ b/edg/core/test_block.py @@ -14,6 +14,7 @@ def __init__(self) -> None: self.base_float = self.Parameter(FloatExpr()) self.base_port = self.Port(TestPortBase()) # required to test required constraint self.base_port_constr = self.Port(TestPortBase(self.base_float), optional=True) + self.base_port_optional = self.Port(TestPortBase(), optional=self.base_port.is_connected()) @abstract_block_default(lambda: TestBlock) @@ -56,23 +57,36 @@ def test_param_def(self) -> None: self.assertTrue(self.pb.params[0].value.HasField("floating")) def test_port_def(self) -> None: - self.assertEqual(len(self.pb.ports), 2) + self.assertEqual(len(self.pb.ports), 3) self.assertEqual(self.pb.ports[0].name, "base_port") self.assertEqual(self.pb.ports[0].value.lib_elem.target.name, "edg.core.test_elaboration_common.TestPortBase") self.assertEqual(self.pb.ports[1].name, "base_port_constr") self.assertEqual(self.pb.ports[1].value.lib_elem.target.name, "edg.core.test_elaboration_common.TestPortBase") + self.assertEqual(self.pb.ports[2].name, "base_port_optional") + self.assertEqual(self.pb.ports[2].value.lib_elem.target.name, "edg.core.test_elaboration_common.TestPortBase") - def test_connected_constraint(self) -> None: + def test_required_constraint(self) -> None: expected_constr = edgir.ValueExpr() expected_constr.ref.steps.add().name = "base_port" expected_constr.ref.steps.add().reserved_param = edgir.IS_CONNECTED self.assertEqual(self.pb.constraints[0].name, "(reqd)base_port") self.assertEqual(self.pb.constraints[0].value, expected_constr) + def test_optional_required_constraint(self) -> None: + expected_constr = edgir.ValueExpr() + expected_constr.binary.op = edgir.BinaryExpr.Op.IMPLIES + expected_constr.binary.lhs.unary.op = edgir.UnaryExpr.Op.NOT + expected_constr.binary.lhs.unary.val.ref.steps.add().name = "base_port" + expected_constr.binary.lhs.unary.val.ref.steps.add().reserved_param = edgir.IS_CONNECTED + expected_constr.binary.rhs.ref.steps.add().name = "base_port_optional" + expected_constr.binary.rhs.ref.steps.add().reserved_param = edgir.IS_CONNECTED + self.assertEqual(self.pb.constraints[1].name, "(reqd)base_port_optional") + self.assertEqual(self.pb.constraints[1].value, expected_constr) + def test_port_init(self) -> None: - self.assertEqual(self.pb.constraints[1].name, "(init)base_port_constr.float_param") + self.assertEqual(self.pb.constraints[2].name, "(init)base_port_constr.float_param") self.assertEqual( - self.pb.constraints[1].value, edgir.AssignRef(["base_port_constr", "float_param"], ["base_float"]) + self.pb.constraints[2].value, edgir.AssignRef(["base_port_constr", "float_param"], ["base_float"]) ) @@ -114,10 +128,11 @@ def test_superclass(self) -> None: self.assertEqual(self.pb.ports[1].value.lib_elem.target.name, "edg.core.test_elaboration_common.TestPortBase") def test_port_def(self) -> None: - self.assertEqual(len(self.pb.ports), 3) - self.assertEqual(self.pb.ports[2].name, "port_lit") - self.assertEqual(self.pb.ports[2].value.lib_elem.target.name, "edg.core.test_elaboration_common.TestPortBase") + self.assertEqual(len(self.pb.ports), 4) + self.assertEqual(self.pb.ports[3].name, "port_lit") + self.assertEqual(self.pb.ports[3].value.lib_elem.target.name, "edg.core.test_elaboration_common.TestPortBase") self.assertEqual(self.pb.constraints[0].name, "(reqd)base_port") + self.assertEqual(self.pb.constraints[1].name, "(reqd)base_port_optional") def test_param_def(self) -> None: self.assertEqual(len(self.pb.params), 4) @@ -129,17 +144,17 @@ def test_param_def(self) -> None: self.assertTrue(self.pb.params[3].value.HasField("array")) def test_superclass_init(self) -> None: - self.assertEqual(self.pb.constraints[1].name, "(init)base_port_constr.float_param") + self.assertEqual(self.pb.constraints[2].name, "(init)base_port_constr.float_param") self.assertEqual( - self.pb.constraints[1].value, edgir.AssignRef(["base_port_constr", "float_param"], ["base_float"]) + self.pb.constraints[2].value, edgir.AssignRef(["base_port_constr", "float_param"], ["base_float"]) ) def test_port_init(self) -> None: - self.assertEqual(self.pb.constraints[2].name, "(init)port_lit.float_param") + self.assertEqual(self.pb.constraints[3].name, "(init)port_lit.float_param") def test_param_init(self) -> None: - self.assertEqual(self.pb.constraints[3].name, "(init)range_init") - self.assertEqual(self.pb.constraints[3].value, edgir.AssignLit(["range_init"], Range(-4.2, -1.3))) + self.assertEqual(self.pb.constraints[4].name, "(init)range_init") + self.assertEqual(self.pb.constraints[4].value, edgir.AssignLit(["range_init"], Range(-4.2, -1.3))) expected_assign = edgir.ValueExpr() expected_assign.assign.dst.CopyFrom(edgir.LocalPathList(["array_init"])) @@ -147,14 +162,14 @@ def test_param_init(self) -> None: expected_array.vals.add().CopyFrom(edgir.lit_to_expr(False)) expected_array.vals.add().CopyFrom(edgir.lit_to_expr(True)) expected_array.vals.add().CopyFrom(edgir.lit_to_expr(False)) - self.assertEqual(self.pb.constraints[4].name, "(init)array_init") - self.assertEqual(self.pb.constraints[4].value, expected_assign) + self.assertEqual(self.pb.constraints[5].name, "(init)array_init") + self.assertEqual(self.pb.constraints[5].value, expected_assign) expected_assign = edgir.ValueExpr() expected_assign.assign.dst.CopyFrom(edgir.LocalPathList(["array_empty"])) expected_assign.assign.src.array.SetInParent() - self.assertEqual(self.pb.constraints[5].name, "(init)array_empty") - self.assertEqual(self.pb.constraints[5].value, expected_assign) + self.assertEqual(self.pb.constraints[6].name, "(init)array_empty") + self.assertEqual(self.pb.constraints[6].value, expected_assign) def test_docs(self) -> None: self.assertEqual(self.pb.meta.members.node["_docs"].members.node[""].text_leaf, "Test docstring") From 736ac65174b9af5fae9df4cb552e57e19afd707c Mon Sep 17 00:00:00 2001 From: Richard Lin Date: Sun, 31 May 2026 23:40:20 -0700 Subject: [PATCH 5/5] Update Blocks.py --- edg/core/Blocks.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/edg/core/Blocks.py b/edg/core/Blocks.py index a2147b12b..ab306b20a 100644 --- a/edg/core/Blocks.py +++ b/edg/core/Blocks.py @@ -289,7 +289,7 @@ def __init__(self) -> None: self._param_docs = IdentityDict[ConstraintExpr, str]() self._ports: SubElementDict[BasePort] = self.manager.new_dict(BasePort) - self._required_ports = IdentityDict[BasePort, Optional[BoolExpr]]() # port -> expr for not-required + self._required_ports = IdentityDict[BasePort, Optional[BoolExpr]]() # port -> precondition for required self._port_docs = IdentityDict[BasePort, str]() self._connects = self.manager.new_dict(Connection, anon_prefix="anon_link")