Systems

Simple Systems

A simple System does not contain sub-systems.

Import CoSApp core package

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

Note: Base classes System and Port may also be imported from modules cosapp.systems and cosapp.ports, respectively.

Create a simple system

class DemoPort(Port):
    def setup(self):
        self.add_variable('x')

class Multiply(System):

    def setup(self):
        """Defines system structure"""
        self.add_input(DemoPort, 'p_in')    # define a new input port
        self.add_output(DemoPort, 'p_out')  # define a new output port
        # Solitary variables can also be added,
        # as either `inward` or `outward` variables:
        self.add_inward('K1', 1.0)
        self.add_outward('delta_x', 0.0)

    def compute(self): # `compute` method defines what the system does
        self.p_out.x = self.p_in.x * self.K1
        self.delta_x = self.p_out.x - self.p_in.x

s = Multiply(name='s')

Configuration file `/home/docs/.cosapp.d/cosapp_config.json` cannot be opened; fall back to default.

Run the system to confirm the expected behaviour

s.p_in.x = 10.
s.K1 = 5.
s.run_once()

s.p_out

s.p_out: DemoPort

x: 50

More information on ports (inputs and outputs) can be found in a dedicated Port tutorial.

inwards and outwards ports

Inwards and outwards are orphan variables, which are not declared within a dedicated Port class. Instead, they are declared directly in System.setup, with add_inward and add_outward (for input and output variables, respectively). These orphan variables are dynamically added to special ports of System, called inwards and outwards.

Thus, every system has inwards and outwards ports of different sizes and contents. All other ports possess a fixed number of variables, each with a predefined type.

Typically:

• An inward is an input parameter needed by the system to compute its output. For example, the pressure losses coefficient of a duct is an inward, used to compute the output flow port from its input flow port.

• An outward is an output variable deduced from inputs, such as an intermediate variable of a computation, which is exposed to other systems. For instance, the difference between the input and output pressures in a duct system is a local variable, which may be of interest to neighbouring systems. Another example is the table object read from a filename (the filename being usually an inward).

An inward x (resp. outward) defined in system s can be accessed as either s.x or s.inwards.x (resp. s.outwards.x).

All variables in CoSApp accept additional information:

• unit: Physical unit of the variable, given by a string. CoSApp will take care of unit conversion between connected systems. However, units are not enforced inside a System. Therefore, module developers must ensure that all variables are consistently converted in method compute.

• 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 (int or float) will be typed as Number.

Methods add_input and add_output also accept optional argument desc, if a contextual description is desired.

class MultiplyAdvanced(System):

    def setup(self):
        # Inwput and output ports accept optional description `desc`
        self.add_input(DemoPort, 'p_in', desc='Port containing values to be multiplied')
        # Inward and outward variables accept optional `dtype` and `unit`
        self.add_inward('K1', 1, dtype=int, desc='Multiplication coefficient')
        self.add_outward('delta_x',
            value = 0.0,
            unit = 'Pa',
            dtype = (int, float),
            desc = "Difference between `DemoPort` output and input"
        )
        self.add_output(DemoPort, 'p_out')

    def compute(self):
        self.p_out.x = self.p_in.x * self.K1
        self.delta_x = self.p_out.x - self.p_in.x


advanced = MultiplyAdvanced('advanced')

print(
    f"{advanced.inwards  = }",
    f"{advanced.outwards = }",
    sep="\n"
)

advanced.inwards  = ExtensiblePort: {'K1': 1}
advanced.outwards = ExtensiblePort: {'delta_x': 0.0}

Composite Systems

Composite systems contain sub-systems, referred to as children.

Sub-systems are added to a composite system using method add_child. Like ports and variables, sub-systems can be briefly described in the context of their parent system, with optional argument desc.

Connections between child systems are declared at parent level, with method connect. The typical syntax is parent.connect(child1.portA, child2.portB).

Port connections are described in detail in the Port tutorial.

class MultiplyComplexSystem(System):

    def setup(self):
        # Sub-systems
        self.add_child(Multiply('mult1'), desc='Primary sub-system')
        self.add_child(Multiply('mult2'))

        # Connectors
        self.connect(self.mult1.p_out, self.mult2.p_in)  # connect ports of sub-systems

head = MultiplyComplexSystem('head')

Run the system to confirm the expected behaviour

head.mult1.p_in.x = 10.
head.mult1.K1 = 5.
head.mult2.K1 = 5.
head.run_once()

print(f"{head.mult1.p_in.x = }")
print(f"{head.mult2.p_out.x = }")

head.mult1.p_in.x = 10.0
head.mult2.p_out.x = 250.0

Connect systems

As illustrated in the previous example, ports of two sibling systems can be connected by System.connect. Alternatively, systems can be connected directly, with a port or a variable mapping specifying which parts of the two systems should be connected:

class CompositeSystem(System):

    def setup(self):
        a = self.add_child(Multiply('a'))
        b = self.add_child(Multiply('b'))

        self.connect(a, b, {'p_out.x': 'K1'})  # connect `b.K1` to `a.p_out.x`

head = CompositeSystem(name='head')
head.a.p_in.x = 1.2
head.a.K1 = 5.
head.b.p_in.x = 0.5
head.b.K1 = 2.  # will be overwritten
head.run_once()

print(
    f"{head.a.p_in.x = }",
    f"{head.a.p_out.x = }",
    f"{head.b.p_out.x = }",
    f"{head.b.K1 = }",
    sep="\n",
)

head.a.p_in.x = 1.2
head.a.p_out.x = 6.0
head.b.p_out.x = 3.0
head.b.K1 = 6.0

Note: Full system connections are forbidden, as they can be ambiguous:

import logging
from cosapp.base import ConnectorError

try:
    head.connect(head.a, head.b)  # full system connection

except ConnectorError as error:
    logging.error(error)

ERROR:root:Full system connections are forbidden; please provide port or variable mapping.

Connection between system levels

In above example, we need to know the internal system architecture to access port head.mult1.p_in. Instead, we may want to promote p_in at parent level, as an important system port, and access it as head.p_in.

This can be achieved with parent-child connectors:

class MultiplyComplexSystem1(System):

    def setup(self):
        # inputs / outputs
        self.add_input(DemoPort, 'p_in')
        self.add_output(DemoPort, 'p_out')

        # Children
        self.add_child(Multiply('mult1'))
        self.add_child(Multiply('mult2'))

        # Connections between siblings
        self.connect(self.mult1.p_out, self.mult2.p_in)

        # Parent-child connections
        self.connect(self.p_in, self.mult1.p_in)
        self.connect(self.p_out, self.mult2.p_out)

head = MultiplyComplexSystem1(name='head')

head.p_in.x = 10.
head.mult1.K1 = 1.5
head.mult2.K1 = 5.0
head.run_once()

print(
    f"{head.p_in  = }",
    f"{head.p_out = }",
    f"{head.mult1.K1 * head.mult2.K1 * head.p_in.x = }",
    sep="\n",
)

head.p_in  = DemoPort: {'x': 10.0}
head.p_out = DemoPort: {'x': 75.0}
head.mult1.K1 * head.mult2.K1 * head.p_in.x = 75.0

The exact same thing can be achieved more conveniently with option pulling, in add_child:

class MultiplyComplexSystem2(System):

    def setup(self):
        # Children
        self.add_child(Multiply('mult1'), pulling='p_in')   # expose `p_in` as parent input
        self.add_child(Multiply('mult2'), pulling='p_out')  # expose `p_out` as parent output

        # Connectors
        self.connect(self.mult1.p_out, self.mult2.p_in)

head = MultiplyComplexSystem2(name='head')

head.p_in.x = 10.
head.mult1.K1 = 1.5
head.mult2.K1 = 5.0
head.run_once()

print(
    f"{head.p_in  = }",
    f"{head.p_out = }",
    f"{head.mult1.K1 * head.mult2.K1 * head.p_in.x = }",
    sep="\n",
)

head.p_in  = DemoPort: {'x': 10.0}
head.p_out = DemoPort: {'x': 75.0}
head.mult1.K1 * head.mult2.K1 * head.p_in.x = 75.0

When more that one port need to be pulled up, a list or a tuple of ports may be provided, as in pulling=['portA', 'portB']. It is also possible to change the name at parent level, by providing a name mapping through a dictionary:

class MultiplyComplexWithPulling(System):

    def setup(self):
        # Children
        self.add_child(Multiply('mult1'), pulling={'K1': 'K11', 'p_in': 'p_in'})    # `mult1.K1` mapped as `K11`
        self.add_child(Multiply('mult2'), pulling={'K1': 'K12', 'p_out': 'p_out'})  # `mult2.K1` mapped as `K12`

        # Connectors
        self.connect(self.mult1.p_out, self.mult2.p_in)

head = MultiplyComplexWithPulling(name='head')

head.p_in.x = 10.
head.K11 = 1.5
head.K12 = 5.0
head.run_once()

print(
    f"{head.p_in  = }",
    f"{head.p_out = }",
    f"{head.K11 * head.K12 * head.p_in.x = }",
    sep="\n",
)

head.p_in  = DemoPort: {'x': 10.0}
head.p_out = DemoPort: {'x': 75.0}
head.K11 * head.K12 * head.p_in.x = 75.0

Classical mistakes

Will you find the mistakes in the following systems?

Exec order

class Wrong1(System):

    def setup(self):
        # Children
        self.add_child(Multiply('mult2'), pulling={'K1': 'K12', 'p_out': 'p_out'})
        self.add_child(Multiply('mult1'), pulling={'K1': 'K11', 'p_in': 'p_in'})

        # connectors
        self.connect(self.mult2.p_in, self.mult1.p_out)

head = Wrong1('head')

head.p_in.x = 10.
head.K11 = 5.
head.K12 = 5.
head.run_once()

print(
    f"{head.p_in.x = }",
    f"{head.p_out.x = }, whereas we expected: {head.K11 * head.K12 * head.p_in.x = }!",
    sep="\n",
)

head.p_in.x = 10.0
head.p_out.x = 5.0, whereas we expected: head.K11 * head.K12 * head.p_in.x = 250.0!

Here, the error is caused by a wrong execution order of the sub-systems. By default, sub-systems are computed in their declaration order; in the example above, it means head.mult2 is computed before head.mult1.

The execution order can be displayed with attribute exec_order:

list(head.exec_order)

['mult2', 'mult1']

According to the connection declared in system head, input head.mult2.p_in is mapped to output head.mult1.p_out, which has not been updated yet at the time sub-system mult2 is computed.

Since system head has been executed with run_once, cyclic dependencies are not resolved. Hence the difference between actual and expected results!

In conclusion, the order of execution of children is important. When the system can be resolved without cyclic dependencies, you should make sure that the execution order is consistent with the natural flow of information. In class Wrong1 above, this can be achieved by specifying

self.exec_order = ['mult1', 'mult2']

at the end of method setup.

When complex coupling does involve cyclic dependencies (as is usually the case), a sensible execution order can reduce the number of unknowns necessary to equilibrate the system. Note that in this case, the system can only be balanced through iterative resolution, which in CoSApp is achieved using a Driver. Drivers are presented in a dedicated tutorial.

Execution order can also be set interactively on a specific object. Let’s try again with system head:

head.exec_order = ['mult1', 'mult2']

head.p_in.x = 10.
head.K11 = 5.
head.K12 = 5.
head.run_once()

print(
    f"{head.p_in.x = }",
    f"{head.p_out.x = }",
    f"{head.K11 * head.K12 * head.p_in.x = }",
    sep="\n",
)

head.p_in.x = 10.0
head.p_out.x = 250.0
head.K11 * head.K12 * head.p_in.x = 250.0

Shadowed variables

As was shown earlier, pulling a port/variable creates an attribute at parent level, automatically connected to the child attribute.

However, the connector direction, that is the direction of information flow between the two connected entities, depends on the direction of the pulled port:

• If the pulled port/variable is an input, then the connector will transfer data downwards, i.e. it will copy the parent value down to the connected child value.

• If the pulled port/variable is an output, the flow is reversed, i.e. the connector will transfer the child value up to the parent.

This has a strong implication on pulled input variables, which should always be specified at parent level. If you set the child input directly, its value will be superseded by the parent value before the child system is computed, and the system will not behave as expected:

head = MultiplyComplexWithPulling('head')  # port `p_in` and inward `K1` pulled from `mult1`

# Set pulled inputs at parent level
head.K11 = 2.0
head.K12 = 1.6
head.p_in.x = 10.

# Set pulled inputs at child level
head.mult1.K1 = 1.
head.mult2.K1 = 3.14
head.mult1.p_in.x = 0.

# Execute system once
head.run_once()

print(
    f"{head.p_in  = }",
    f"{head.p_out = }",
    "",
    f"{head.mult1.K1 = }",
    f"{head.mult2.K1 = }",
    f"{head.mult1.p_in.x = }",
    sep="\n",
)

head.p_in  = DemoPort: {'x': 10.0}
head.p_out = DemoPort: {'x': 32.0}

head.mult1.K1 = 2.0
head.mult2.K1 = 1.6
head.mult1.p_in.x = 10.0

Indeed, head.mult1.K1, head.mult2.K1 and head.mult1.p_in.x have been superseded by head.K11, head.K12 and head.p_in.x, respectively.

Next you will learn more about Ports, the interface between systems.