Custom connectors

Connectors ensure data transfer from a source port to a sink port. By default, they are of type cosapp.ports.Connector, which transmits shallow copies of source variables to sink variables, and performs unit conversion when necessary.

It is also possible for users to define their own connector classes, as long as they derive from base class cosapp.base.BaseConnector, and implement abstract method transfer. Two cases are possible:

1. Specific port connectors, for peer-to-peer connections.

2. Generic connectors, capable of handling any source/sink pair, including ports of different types.

This tutorial briefly illustrates these two situations.

Warning: data transfer is a fundamental feature of multi-system simulations, so we strongly discourage the use of custom general connectors, unless strictly necessary. In particular, custom connectors should never be used as a substitute for an additional system in a composite model.

Peer-to-peer connectors

Specific port connectors are implemented as an inner class Connector, defined inside the targetted port class. When a specific port connector class exists, all peer-to-peer connections (including those created by a pulling) will automatically use it.

In the example below, we define port class XyzPort, with peer-to-peer connector XyzPort.Connector:

from cosapp.base import Port, System, BaseConnector
from cosapp.utils import set_log, LogLevel
import numpy


class XyzPort(Port):
    """Port with a specific peer-to-peer connector"""
    def setup(self):
        self.add_variable('x', 1.0)
        self.add_variable('y', 2.0)
        self.add_variable('z', 3.0)

    class Connector(BaseConnector):
        """Custom connector for `XyzPort` objects
        """
        def __init__(self, name: str, sink: Port, source: Port, *args, **kwargs):
            super().__init__(name, sink, source, mapping=dict(zip('xyz', 'xzy')))

        def transfer(self) -> None:
            source = self.source
            self.sink.set_values(
                x = source.x,
                y = source.z,
                z = -source.y,
            )


class XyzSystem(System):
    def setup(self):
        self.add_input(XyzPort, 'p_in')
        self.add_output(XyzPort, 'p_out')

    def compute(self):
        p_in = self.p_in
        self.p_out.set_values(
            x = p_in.x * 2,
            y = p_in.y - p_in.x,
            z = p_in.z - p_in.x,
        )


class XyzToVector(System):
    def setup(self):
        self.add_input(XyzPort, 'p_in')
        self.add_outward('v', numpy.zeros(3))

    def compute(self):
        p = self.p_in
        self.v = numpy.array([p.x, p.y, p.z])


top = System('top')
s1 = top.add_child(XyzSystem('s1'))
s2 = top.add_child(XyzToVector('s2'))

set_log()
top.connect(s1.p_out, s2.p_in)  # automatically uses `XyzPort.Connector`

s1.p_in.set_values(x=0.1, y=1.0, z=2.0)
top.run_once()

print()
for var in ("s1.p_in", "s1.p_out", "s2.p_in", "s2.v"):
    print(f"top.{var}:\t{top[var]!r}")

's1.p_out' and 's2.p_in' connected by port-specific connector `XyzPort.Connector`.

top.s1.p_in:    XyzPort: {'x': 0.1, 'y': 1.0, 'z': 2.0}
top.s1.p_out:   XyzPort: {'x': 0.2, 'y': 0.9, 'z': 1.9}
top.s2.p_in:    XyzPort: {'x': 0.2, 'y': 1.9, 'z': -0.9}
top.s2.v:       array([ 0.2,  1.9, -0.9])

General connectors

General connectors are supposed to handle data transfer between any source/sink port pair. We show below two examples of such connectors;

• PlainConnector, transfering data using a simple assignment syntax;

• DeepCopyConnector, assigning deep copies of source variables to sink variables. More information on shallow and deep copy may be found here.

from cosapp.base import BaseConnector
import copy


class PlainConnector(BaseConnector):
    """Plain connector performing simple variable assignments.
    See `BaseConnector` for base class details.
    """
    def transfer(self) -> None:
        source, sink = self.source, self.sink

        for target, origin in self.mapping.items():
            # Implement: sink.target = source.origin
            setattr(sink, target, getattr(source, origin))


class DeepCopyConnector(BaseConnector):
    """Deep copy connector.
    """
    def transfer(self) -> None:
        source, sink = self.source, self.sink

        for target, origin in self.mapping.items():
            # Implement: sink.target = deepcopy(source.origin)
            value = getattr(source, origin)
            setattr(sink, target, copy.deepcopy(value))

For most data structures, plain assignment in Python usually binds the two sides of the equal sign by simple reference, with no copy. This, in fact, is the reason why the default connector class cosapp.ports.Connector perform copies, to avoid undesired side effects, such as trivially nil residues during loop resolution, for instance.

In the next example, we use PlainConnector in a port connection involving numpy arrays.

from cosapp.base import Port, System
import numpy


class FramePort(Port):
    def setup(self):
        self.add_variable('x', numpy.zeros(3), unit='m', desc="Centre-of-mass position")
        self.add_variable('w', numpy.zeros(3), unit='1/s', desc="Angular velocity")


class Displacement(System):
    def setup(self):
        self.add_input(FramePort, 'fr_in')
        self.add_output(FramePort, 'fr_out')
        self.add_inward('delta', numpy.zeros(3), unit='m', desc="Displacement vector")

    def compute(self):
        self.fr_out.x = self.fr_in.x + self.delta
        self.fr_out.w = self.fr_in.w


top = System('top')
d1 = top.add_child(Displacement('d1'), pulling='fr_in')
d2 = top.add_child(Displacement('d2'), pulling='fr_out')

top.connect(d1.fr_out, d2.fr_in, cls=PlainConnector)  # specify connector

top.fr_in.x = numpy.zeros(3)
top.d1.delta = numpy.r_[0.0, 0.1, 0.2]
top.d2.delta = numpy.r_[1.0, 0.5, 0.0]

# Before system execution, connected arrays are different objects:
assert d1.fr_out.x is not d2.fr_in.x

print(
    "Before execution:",
    f"{d1.fr_out.x is d2.fr_in.x = }",
    sep="\n",
)

# After system execution, arrays are identical objects,
# owing to simple assignment in `PlainConnector`:
top.run_once()

print(
    "\nAfter execution:",
    f"{d1.fr_out.x is d2.fr_in.x = }",
    f"{top.fr_out.x = }",
    sep="\n",
)

Before execution:
d1.fr_out.x is d2.fr_in.x = False

After execution:
d1.fr_out.x is d2.fr_in.x = True
top.fr_out.x = array([1. , 0.6, 0.2])

Try it out with DeepCopyConnector (or nothing) instead of PlainConnector, and see what happens!

Notes

• This stunt was performed by trained professionals, do not try this at home! In this very simple example, the absence of copy may lead to a slight improvement in performance, but can have very bad, hard-to-debug side effects.

• DeepCopyConnector is safe, and readily available in module cosapp.ports.connectors. It may be useful when one wants to ensure a complete, recursive copy of complex composite data structures, but is slower than default class Connector. Importantly, it does not perform unit conversion.

• Module cosapp.ports.connectors also contains a shallow copy connector, named CopyConnector. Unlike Connector, though, it does not handle unit conversion.

• If a port contains a type-specific connector, peer-to-peer connections use the dedicated connector by default (see Peer-to-peer connectors). However, this behaviour can be overriden by specifying optional argument cls in System.connect.

Direct value connectors PlainConnector, CopyConnector and DeepCopyConnector (from module cosapp.ports.connectors) can be useful to define peer-to-peer connectors. For instance:

from cosapp.base import Port
from cosapp.ports.connectors import CopyConnector
import numpy


class FramePort(Port):

    def setup(self):
        self.add_variable('x', numpy.zeros(3), unit='m', desc="Centre-of-mass position")
        self.add_variable('w', numpy.zeros(3), unit='1/s', desc="Angular velocity")

    Connector = CopyConnector  # port-specific peer-to-peer connector