Conditional optional ports

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 -> precondition for required
         self._port_docs = IdentityDict[BasePort, str]()
 
         self._connects = self.manager.new_dict(Connection, anon_prefix="anon_link")
@@ -397,12 +397,20 @@ 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._check_constraint(precondition)
+                    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
 
         self._add_doc_metadata()
@@ -525,7 +533,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 +547,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

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")

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,32 +144,32 @@ 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"]))
         expected_array = expected_assign.assign.src.array
         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")

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(

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)
 
@@ -93,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
@@ -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:

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

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(

edg/parts/microcontroller/Stm32f303.py

@@ -1,6 +1,5 @@
 from typing import *
 
-from deprecated import deprecated
 from typing_extensions import override
 
 from ...circuits import *