Add frequency calculations to CAN, SPI, I2C, I2S, UART links

edg/core/Blocks.py

@@ -492,12 +492,12 @@ def check_subexpr(expr: Union[ConstraintExpr, BasePort]) -> None:  # TODO rewrit
         for subexpr in constraint._get_exprs():
             check_subexpr(subexpr)
 
-    def require(self, constraint: BoolLike, name: Optional[str] = None, *, unchecked: bool = False) -> BoolExpr:
+    def require(self, constraint: BoolLike, name: Optional[str] = None, *, _unchecked: bool = False) -> BoolExpr:
         constraint_typed = BoolExpr._to_expr_type(constraint)
         if not isinstance(name, (str, type(None))):
             raise EdgTypeError(f"require(...) name", name, (str, None))
 
-        if not unchecked:  # before we have const prop need to manually set nested params
+        if not _unchecked:  # allow breaking structural / abstraction rules, used in some unit tests
             self._check_constraint(constraint_typed)
 
         self._constraints.register(constraint_typed)

edg/electronics_interfaces/CanPort.py

@@ -16,13 +16,12 @@ def __init__(self) -> None:
         self.transceiver = self.Port(CanTransceiverPort(DigitalBidir.empty()))
         self.passive = self.Port(Vector(CanPassivePort(DigitalBidir.empty())), optional=True)
 
-        # TODO write custom top level digital constraints
-        # TODO model frequency ... somewhere
+        # 0-1 Mbit/s are standard CANbus, 1-8 Mbit/s imply CAN-FD
+        self.bitrate_limit = self.Parameter(RangeExpr())
 
     @override
     def contents(self) -> None:
         super().contents()
-        # TODO future: digital constraints through link inference
 
         self.txd = self.connect(
             self.controller.txd, self.transceiver.txd, self.passive.map_extract(lambda port: port.txd), flatten=True
@@ -31,13 +30,14 @@ def contents(self) -> None:
             self.controller.rxd, self.transceiver.rxd, self.passive.map_extract(lambda port: port.rxd), flatten=True
         )
 
+        self.assign(self.bitrate_limit, self.controller.bitrate_limit.intersect(self.transceiver.bitrate_limit))
+        self.require(self.bitrate_limit != RangeExpr.EMPTY, "no compatible bitrate between devices")
+
 
 class CanControllerPort(Port[CanLogicLink]):
     link_type = CanLogicLink
 
     def __init__(self, model: Optional[DigitalBidir] = None, *, bitrate_limit: RangeLike = RangeExpr.ALL) -> None:
-        # 0-1 Mbit/s are standard CANbus
-        # 1-8 Mbit/s imply CAN-FD
         super().__init__()
         if model is None:  # ideal by default
             model = DigitalBidir()
@@ -49,12 +49,13 @@ def __init__(self, model: Optional[DigitalBidir] = None, *, bitrate_limit: Range
 class CanTransceiverPort(Port[CanLogicLink]):
     link_type = CanLogicLink
 
-    def __init__(self, model: Optional[DigitalBidir] = None) -> None:
+    def __init__(self, model: Optional[DigitalBidir] = None, *, bitrate_limit: RangeLike = RangeExpr.ALL) -> None:
         super().__init__()
         if model is None:  # ideal by default
             model = DigitalBidir()
         self.txd = self.Port(DigitalSink.from_bidir(model))
         self.rxd = self.Port(DigitalSource.from_bidir(model))
+        self.bitrate_limit = self.Parameter(RangeExpr(bitrate_limit))
 
 
 class CanPassivePort(Port[CanLogicLink]):
@@ -75,8 +76,7 @@ def __init__(self) -> None:
         super().__init__()
         self.nodes = self.Port(Vector(CanDiffPort.empty()))  # TODO mark as required
 
-        # TODO write custom top level digital constraints
-        # TODO future: digital constraints through link inference
+        self.bitrate_limit = self.Parameter(RangeExpr())
 
     @override
     def contents(self) -> None:
@@ -85,13 +85,16 @@ def contents(self) -> None:
         self.canh = self.connect(self.nodes.map_extract(lambda node: node.canh), flatten=True)
         self.canl = self.connect(self.nodes.map_extract(lambda node: node.canl), flatten=True)
 
+        self.assign(self.bitrate_limit, self.nodes.intersection(lambda x: x.bitrate_limit))
+        self.require(self.bitrate_limit != RangeExpr.EMPTY, "no compatible bitrate between devices")
+
 
 class CanDiffBridge(PortBridge):
     def __init__(self) -> None:
         super().__init__()
 
         self.outer_port = self.Port(CanDiffPort.empty())
-        self.inner_link = self.Port(CanDiffPort.empty())
+        self.inner_link = self.Port(CanDiffPort())
 
     @override
     def contents(self) -> None:
@@ -105,12 +108,16 @@ def contents(self) -> None:
         self.connect(self.outer_port.canl, self.canl_bridge.outer_port)
         self.connect(self.canl_bridge.inner_link, self.inner_link.canl)
 
+        self.assign(self.outer_port.bitrate_limit, self.inner_link.link().bitrate_limit)
+
 
 class CanDiffPort(Port[CanDiffLink]):
     link_type = CanDiffLink
     bridge_type = CanDiffBridge
 
-    def __init__(self) -> None:
+    def __init__(self, *, bitrate_limit: RangeLike = RangeExpr.ALL) -> None:
         super().__init__()
         self.canh = self.Port(Passive())
         self.canl = self.Port(Passive())
+
+        self.bitrate_limit = self.Parameter(RangeExpr(bitrate_limit))

edg/electronics_interfaces/I2cPort.py

@@ -7,7 +7,7 @@
 
 
 class I2cLink(Link):
-    """I2C connection, using terminology from the auhtoritative NXP specification at
+    """I2C connection, using terminology from the authoritative NXP specification at
     https://www.nxp.com/docs/en/user-guide/UM10204.pdf.
     """
 
@@ -25,7 +25,7 @@ def __init__(self) -> None:
         self.addresses = self.Parameter(
             ArrayIntExpr(self.targets.flatten(lambda x: x.addresses).concat(self.controller.addresses))
         )
-        self.frequency = self.Parameter(RangeExpr())
+        self.frequency_limit = self.Parameter(RangeExpr())
 
         self.has_pull = self.Parameter(BoolExpr(self.pull.any_connected()))
 
@@ -38,9 +38,10 @@ def contents(self) -> None:
 
         self.require(self.addresses.all_unique(), "conflicting addresses on I2C bus")
         self.assign(
-            self.frequency,
+            self.frequency_limit,
             self.controller.frequency_limit.intersect(self.targets.intersection(lambda x: x.frequency_limit)),
         )
+        self.require(self.frequency_limit != RangeExpr.EMPTY, "no compatible frequency between devices")
 
         self.scl = self.connect(
             self.pull.map_extract(lambda device: device.scl),
@@ -82,6 +83,7 @@ def contents(self) -> None:
                 DigitalBidir.empty(),
                 has_pullup=self.inner_link.link().has_pull,
                 addresses=self.inner_link.link().addresses,
+                frequency_limit=self.inner_link.link().frequency_limit,
             )
         )
 
@@ -128,7 +130,13 @@ def __init__(self) -> None:
     def contents(self) -> None:
         super().contents()
 
-        self.outer_port.init_from(I2cTarget(DigitalBidir.empty(), addresses=self.inner_link.link().addresses))
+        self.outer_port.init_from(
+            I2cTarget(
+                DigitalBidir.empty(),
+                addresses=self.inner_link.link().addresses,
+                frequency_limit=self.inner_link.link().frequency_limit,
+            )
+        )
 
         self.scl_bridge = self.Block(DigitalSinkBridge())
         self.connect(self.outer_port.scl, self.scl_bridge.outer_port)

edg/electronics_interfaces/I2sPort.py

@@ -13,39 +13,65 @@ def __init__(self) -> None:
         self.target_receiver = self.Port(I2sTargetReceiver(DigitalSink.empty()))
         # TODO: multiple receivers, target transmitters, eg microphones
 
+        self.sample_rate_limit = self.Parameter(RangeExpr())
+        self.bit_limit = self.Parameter(RangeExpr())
+
     @override
     def contents(self) -> None:
         super().contents()
         self.sck = self.connect(self.controller.sck, self.target_receiver.sck)
         self.ws = self.connect(self.controller.ws, self.target_receiver.ws)
         self.sd = self.connect(self.controller.sd, self.target_receiver.sd)
 
+        self.assign(
+            self.sample_rate_limit, self.controller.sample_rate_limit.intersect(self.target_receiver.sample_rate_limit)
+        )
+        self.require(self.sample_rate_limit != RangeExpr.EMPTY, "no compatible sample rate between devices")
+        self.assign(self.bit_limit, self.controller.bit_limit.intersect(self.target_receiver.bit_limit))
+        self.require(self.bit_limit != RangeExpr.EMPTY, "no compatible sample bits (resolution) between devices")
+
 
 class I2sController(Port[I2sLink]):
     """Controller is both controller (drives SCK and WS lines) and transmitter (SD is output)"""
 
     link_type = I2sLink
 
-    def __init__(self, model: Optional[DigitalBidir] = None, *, bitrate_limit: RangeLike = RangeExpr.ALL) -> None:
+    def __init__(
+        self,
+        model: Optional[DigitalBidir] = None,
+        *,
+        sample_rate_limit: RangeLike = RangeExpr.ALL,
+        bit_limit: RangeLike = RangeExpr.ALL,
+    ) -> None:
         super().__init__()
         if model is None:
             model = DigitalBidir()  # ideal by default
         self.sck = self.Port(DigitalSource.from_bidir(model))
         self.ws = self.Port(DigitalSource.from_bidir(model))
         self.sd = self.Port(model)  # bidirectional
 
-        self.bitrate_limit = self.Parameter(RangeExpr(bitrate_limit))  # bitrate
+        self.sample_rate_limit = self.Parameter(RangeExpr(sample_rate_limit))
+        self.bit_limit = self.Parameter(RangeExpr(bit_limit))
 
 
 class I2sTargetReceiver(Port[I2sLink]):
     """Target means SCK and WS are inputs, receiver means SD is input"""
 
     link_type = I2sLink
 
-    def __init__(self, model: Optional[DigitalSink] = None) -> None:
+    def __init__(
+        self,
+        model: Optional[DigitalSink] = None,
+        *,
+        sample_rate_limit: RangeLike = RangeExpr.ALL,
+        bit_limit: RangeLike = RangeExpr.ALL,
+    ) -> None:
         super().__init__()
         if model is None:
             model = DigitalSink()  # ideal by default
         self.sck = self.Port(model)
         self.ws = self.Port(model)
         self.sd = self.Port(model)
+
+        self.sample_rate_limit = self.Parameter(RangeExpr(sample_rate_limit))
+        self.bit_limit = self.Parameter(RangeExpr(bit_limit))

edg/electronics_interfaces/MergedBlocks.py

@@ -3,7 +3,7 @@
 
 from ..electronics_model import *
 from .VoltagePorts import VoltageSource, VoltageSink, VoltageLink
-from .DigitalPorts import DigitalSource, DigitalSink, DigitalLink
+from .DigitalPorts import DigitalSource, DigitalSink, DigitalLink, DigitalBidir
 from .AnalogPort import AnalogSource, AnalogSink, AnalogLink
 from .SpiPort import SpiController, SpiPeripheral, SpiLink
 
@@ -125,11 +125,13 @@ def generate(self) -> None:
 
         self.ins.defined()
         for in_request in self.get(self.ins.requested()):
-            in_port = self.ins.append_elt(SpiPeripheral.empty(), in_request)
+            in_port = self.ins.append_elt(SpiPeripheral(DigitalBidir.empty()), in_request)
             self.connect(miso_net, in_port.miso)
             self.connect(self.sck_merge.ins.request(in_request), in_port.sck)
             self.connect(self.mosi_merge.ins.request(in_request), in_port.mosi)
 
+        self.assign(self.out.frequency_limit, self.ins.hull(lambda x: x.link().frequency_limit))
+
     def connected_from(self, *ins: Port[SpiLink]) -> "MergedSpiController":
         for in_port in ins:
             builder.block().connect(in_port, self.ins.request())

edg/electronics_interfaces/SpiPort.py

@@ -21,6 +21,8 @@ def __init__(self) -> None:
         self.controller = self.Port(SpiController(DigitalBidir.empty()))
         self.peripherals = self.Port(Vector(SpiPeripheral(DigitalBidir.empty())))
 
+        self.frequency_limit = self.Parameter(RangeExpr())
+
     @override
     def contents(self) -> None:
         super().contents()
@@ -33,6 +35,11 @@ def contents(self) -> None:
         self.mosi = self.connect(
             self.controller.mosi, self.peripherals.map_extract(lambda device: device.mosi), flatten=True
         )
+        self.assign(
+            self.frequency_limit,
+            self.controller.frequency_limit.intersect(self.peripherals.hull(lambda x: x.frequency_limit)),
+        )
+        self.require(self.frequency_limit != RangeExpr.EMPTY, "no compatible frequency between devices")
 
 
 class SpiController(Port[SpiLink]):

edg/electronics_interfaces/UartPort.py

@@ -11,13 +11,18 @@ def __init__(self) -> None:
         self.a = self.Port(UartPort(DigitalBidir.empty()))
         self.b = self.Port(UartPort(DigitalBidir.empty()))
 
+        self.baud_limit = self.Parameter(RangeExpr())
+
     @override
     def contents(self) -> None:
         super().contents()
 
         self.a_tx = self.connect(self.a.tx, self.b.rx)
         self.b_tx = self.connect(self.b.tx, self.a.rx)
 
+        self.assign(self.baud_limit, self.a.baud_limit.intersect(self.b.baud_limit))
+        self.require(self.baud_limit != RangeExpr.EMPTY, "no compatible baud rate between devices")
+
 
 class UartPort(Port[UartLink]):
     link_type = UartLink