Ports

What is a Port?!

A Port is a collection of variables that helps transfer data from one System to another. Without such an object, connections between systems should be done variable-by-variable, ouch!

Types

IN or OUT

Port instances inside systems are directional objects; they can be either IN or OUT. This definition ensures compatibility at connection. However, the definition of a Port class only contains the list and description of its variables, irrespective of direction.

Available ports and compatibility

CoSApp allow user-defined ports, stored in libraries or created in specific projects. They define a frozen collection of variables that can be shared between multiple systems. They are compatible by construction (see the Port connection section of this tutorial).

Create a port

Import CoSApp core package

# import CoSApp base classes
from cosapp.base import System, Port

Define a new port

class DemoPort(Port):

    def setup(self):
        self.add_variable('a', 1.0)
        self.add_variable('b', 2.0)
        self.add_variable('c', 3.0)

Use it in a system

We now create a new System using DemoPort:

class DemoSystem(System):

    def setup(self):
        self.add_input(DemoPort, 'p_in')
        self.add_output(DemoPort, 'p_out')

    def compute(self):
        self.p_out.a = self.p_in.c
        self.p_out.b = self.p_in.b
        self.p_out.c = self.p_in.a

s = DemoSystem(name='s')
# Set `s.p_in` variables one by one...
s.p_in.a = 0.2
s.p_in.b = 0.5
# ... or use multi-variable setter `set_values`:
s.p_in.set_values(a=0.2, c=1.5)

In the example above, p_in and p_out are two instances of DemoPort. All instances of DemoSystem, such as s, will possess attributes p_in and p_out. Use of methods add_input and add_output provides a clear interface to create ports with the desired direction.

Run the system to confirm the expected behaviour

s.run_once()
s.p_in

s.p_in: DemoPort

a: 0.2
b: 0.5
c: 1.5

s.p_out

s.p_out: DemoPort

a: 1.5
b: 0.5
c: 0.2

Set information on the variables

All Port variables may be given optional information:

• unit: Physical unit of the variable, given by a string. Units are not enforced inside a System. This means that system developers are responsible for unit conversions (if required) in method compute. However, CoSApp will take care of unit conversion during data transfer via a connector (see Port connection).

• desc: Short description of the variable.

• dtype: If you need to force certain data type(s) on a variable, a tuple of acceptable types can be provided through this keyword. If that information is not supplied, dtype is inferred from the variable value; e.g. a number (integer or floating point) will be typed as Number.

class AdvancedDemoPort(Port):

    def setup(self):
        self.add_variable('a', 3e2, unit='degK', dtype=float, desc='Temperature')
        self.add_variable('b', 0.1, unit='MPa', dtype=(int, float), desc='Pressure')
        self.add_variable('c', 1.0, unit='kg/s', desc='Mass flowrate')


class AdvancedDemoSystem(System):

    def setup(self):
        self.add_input(AdvancedDemoPort, 'p_in')
        self.add_output(AdvancedDemoPort, 'p_out')

    def compute(self):
        self.p_out.a = self.p_in.c
        self.p_out.b = self.p_in.b
        self.p_out.c = self.p_in.a


sa = AdvancedDemoSystem('sa')
print('Input port:')
sa.p_in

Input port:

sa.p_in: AdvancedDemoPort

a: 300 degK	Temperature
b: 0.1 MPa	Pressure
c: 1 kg/s	Mass flowrate

Port connection

Introduction

Port connection creates a dedicated object (of type Connector) managing one-way data transfers between two Port instances. Connectors are handled at system levels, via method connect, and can be accessed through attribute connectors as a dictionnary.

h = System('head')
h.add_child(DemoSystem('demo1'))
h.add_child(DemoSystem('demo2'))

h.connect(h.demo2.p_in, h.demo1.p_out)

If you need to connect a subset of the Port variables, or connect inwards or outwards, the connect() method of System can be called in setup with additional arguments, providing variable mapping from the first port to the other.

class MonitorSystem(System):

    def setup(self):
        self.add_inward('a')
        self.add_inward('b')
        self.add_inward('x')
        self.add_outward('result')

    def compute(self):
        self.result = self.a

h.add_child(MonitorSystem('monitor'))
# Connect `h.monitor.a` to `h.demo1.p_out.a`
h.connect(h.monitor.inwards, h.demo1.p_out, 'a')

h.demo1.p_in.set_values(a=50., b=0., c=25.)
h.run_once()

print(
    f"{h.monitor.a = }",
    f"{h.demo1.p_out = }",
    sep="\n",
)

h.monitor.a = 25.0
h.demo1.p_out = DemoPort: {'a': 25.0, 'b': 0.0, 'c': 50.0}

In the example above, variable a from port inwards has been connected to variable a from port demo1.p_out. This is the simplest mapping, when the variable name is identical in poth ports.

Other options are possible:

• A list variable names, if all names exist in both ports:

h.connect(h.monitor.inwards, h.demo1.p_out, ['a', 'b'])

• An explicit name mapping, if names differ:

h.connect(h.monitor.inwards, h.demo1.p_out, {'a': 'a', 'x': 'c'})

• Connecting two systems, with a port and/or variable mapping:

h.connect(h.monitor, h.demo1, {'x': 'p_out.c'})

In the last example, h.x will be connected to h.demo1.p_out.c. Note that h.x and h.inwards.x refer to the same variable.

print(f"{h.demo2.p_in = }")

h.demo2.p_in = DemoPort: {'a': 25.0, 'b': 0.0, 'c': 50.0}

print(f"{h.monitor.outwards = }")

h.monitor.outwards = ExtensiblePort: {'result': 25.0}

Example with unit conversions

from cosapp.base import Port, Scope, System


class SiFlowPort(Port):
    """Flow port in SI units"""
    def setup(self):
        self.add_variable('q', 1.0, unit='kg/s', desc='Mass flowrate')
        self.add_variable('p', 1e5, unit='Pa', dtype=(int, float), desc='Pressure')
        self.add_variable('T', 3e2, unit='degK', dtype=float, desc='Temperature', scope=Scope.PROTECTED)


class UsFlowPort(Port):
    """Flow port in US units"""
    def setup(self):
        self.add_variable('q', 1.0, unit='lb/min', desc='Mass flowrate')
        self.add_variable('p', 14.5, unit='psi', dtype=(int, float), desc='Pressure')
        self.add_variable('T', 32.0, unit='degF', dtype=float, desc='Temperature')


class SiUnitSystem(System):

    def setup(self):
        self.add_input(SiFlowPort, 'fl_in')
        self.add_output(SiFlowPort, 'fl_out')

    def compute(self):
        # Simply assign `in` values to `out` port
        self.fl_out.set_from(self.fl_in)


class UsUnitSystem(System):

    def setup(self):
        self.add_input(UsFlowPort, 'fl_in')
        self.add_output(UsFlowPort, 'fl_out')

    def compute(self):
        # Simply assign `in` values to `out` port
        self.fl_out.set_from(self.fl_in)


class TemperatureCheck(System):

    def setup(self):
        self.add_inward('T', 0.0, unit='degC')
        self.add_inward('Tmax', 60.0, unit='degC')
        self.add_outward('too_hot', False)

    def compute(self):
        self.too_hot = (self.T > self.Tmax)


top = System('top')
si = top.add_child(SiUnitSystem('si'))
us = top.add_child(UsUnitSystem('us'))
checker = top.add_child(TemperatureCheck('checker'))

top.connect(si.fl_out, us.fl_in)
top.connect(us, checker, {'fl_out.T': 'T'})

# Set entry conditions
top.si.fl_in.set_values(q=2.5, p=1e5, T=293.15)
top.run_once()

top.si.fl_out

top.si.fl_out: SiFlowPort

q: 2.5 kg/s	Mass flowrate
p: 1e+05 Pa	Pressure
T 🔒 : 293.15 degK	Temperature

top.us.fl_in

top.us.fl_in: UsFlowPort

q: 330.69 lb/min	Mass flowrate
p: 14.504 psi	Pressure
T: 68 degF	Temperature

print(f"{top.checker.too_hot = }")

top.checker.inwards

top.checker.too_hot = False

top.checker.inwards: ExtensiblePort

T 🔒🔒 : 20 degC
Tmax 🔒🔒 : 60 degC

Unit compatibility is checked at connection:

import logging

try:
    top.connect(us, checker, {'fl_out.p': 'Tmax'})

except Exception as error:
    logging.error(error)

ERROR:root:Unit degC is not compatible with psi.

Useful functions

Method set_values allows one to set multiple variables of a port at once, using keyword arguments:

port.set_values(x=1.2, y=0.5, v=numpy.zeros(3))

# Rather than
port.x = 1.2
port.y = 0.5
port.v = numpy.zeros(3)

Method set_from assigns port values from another port. If the source and destination ports are of different types, it selects common variable names between them. This behaviour can be overriden by specifying optional argument check_names=False.

By default, values are simply assigned. If another operation is required, one can specify a transfer function applied before the assignments, via optional argument transfer. For example, a shallow copy can be enforced with:

import copy

# Copy variables of `source` port into `destination` port
destination.set_from(source, transfer=copy.copy)

Method items() returns an iterator yieding (varname, value) tuples, akin to dict.items():

for varname, value in port.items():
    print(f"{varname} = {value}")

# Send port variables into a dictionary:
data = dict(port.items())

Congrats! Now you know the basics of Port in CoSApp.

Next you will discover how to solve mathematical problems using Drivers.