CoSApp examples

Tips & Tricks

Purge all drivers from a system

Often necessary to start off a new, clean simulation workflow.

Drivers are stored in a plain dictionary drivers. Thus, cleaning all drivers from system s is as simple as

s.drivers.clear()

Set targets on output values

See tutorial on method add_target.

Loop through all elements in a composite system tree

Browsing through composite trees is usually tackled with recursive functions. However, writing such functions can sometimes be tricky.

As of version 0.11.7, a recursive iterator makes it very easy to loop through system or driver trees.

system = SomeComplexSystem('system')

for elem in system.tree():
    # do whatever

This functionality might come handy if one wishes to collect data from all sub-systems, say.

from cosapp.base import System

a = System('a')
# First layer
a.add_child(System('aa'))
a.add_child(System('ab'))
# Second layer
a.ab.add_child(System('aba'))
a.ab.add_child(System('abb'))
a.ab.add_child(System('abc'))

print([elem.name for elem in a.tree()])

['aa', 'aba', 'abb', 'abc', 'ab', 'a']

System.tree() accepts optional argument downwards, which determines whether the iterator should yield elements from top to bottom (downwards=True), or from bottom to top (downwards=False - default behaviour).

print([elem.name for elem in a.tree(downwards=True)])

['a', 'aa', 'ab', 'aba', 'abb', 'abc']

Note that the iterator follows the execution order of all elements in the tree:

print([elem.name for elem in a.tree()])

a.exec_order = ['ab', 'aa']
a.ab.exec_order = ['abc', 'aba', 'abb']
print([elem.name for elem in a.tree()])

['aa', 'aba', 'abb', 'abc', 'ab', 'a']
['abc', 'aba', 'abb', 'ab', 'aa', 'a']

Synchronize sub-system inputs

Inputs/inwards of a system specify the necessary information for local execution of the system. Very often, though, inputs of two or more sub-systems must take the same value, in the context of their parent assembly.

For instance, consider a system PlaneWing with inward length. When gathered in a higher-level system Airplane as left_wing and right_wing, it may be desirable to force both wings to have the same length, by construction.

Such input synchronization is natively handled by CoSApp through the pulling mechanism:

class Airplane(System):
    def setup(self):
        self.add_child(PlaneWing('left_wing', pulling={'length': 'wing_length'}))
        self.add_child(PlaneWing('right_wing', pulling={'length': 'wing_length'}))

Explaination:

Pulling an input/inward creates a parent-to-child downward connector. Thus, the first add_child statement will create new inward wing_length on the fly, as well as a connector wing_length -> left_wing.length. When the right wing sub-system is created, a second connector wing_length -> right_wing.length is generated, using existing inward wing_length.

plane = Airplane('plane')

plane.wing_length = 12.3
plane.run_once()  # will propagate `wing_length` to both wings

Notes

• Connectors only transfer information when run_once() or run_drivers() is invoked.

• Any value directly assigned to plane.left_wing.length, say, will be eventually superseded by plane.wing_length at each model execution.

• Renaming pulled inputs/inwards at parent level is not mandatory. Passing a list of attribute names rather than a dictionary as pulling argument will trigger the same mechanisms, with no name changes. In the example above, it is just more explicit to expose inward wing_length rather than just length in the context of the airplane.

Create collections of sub-systems and/or ports in systems

Consider an arbitrary number of resistors in series, modeled as an assembly of subsystems R1, R2, etc.

In this case, the number of resistors n is provided at setup, and the n resistors are created with a loop of the kind:

def setup(self, n=2):
    for i in range(n)
        self.add_child(Resistor(f"R{i}"))

While this is creation loop is unavoidable, it can be interesting to get an attribute storing all resistors as an iterable collection.

One can pull this trick using method add_property, as illustrated below:

from cosapp.base import Port, System

class ElecPort(Port):
    def setup(self):
        self.add_variable("I", 1.0, unit="A", desc="Current")
        self.add_variable("V", 0.0, unit="V", desc="Voltage")

class Resistor(System):
    def setup(self):
        self.add_input(ElecPort, 'elec_in')
        self.add_output(ElecPort, 'elec_out')
        self.add_inward("R", 1e2, unit="ohm", desc="Resistance")

    def compute(self):
        self.elec_out.I = I = self.elec_in.I
        self.elec_out.V = self.elec_in.V - self.R * I

class ResistorSeries(System):
    def setup(self, n=2):
        self.add_property('n', max(int(n), 2))

        # Create a tuple of children, and store it as a property
        self.add_property('resistors', tuple(
            self.add_child(Resistor(f"R{i}"))
            for i in range(self.n)
        ))
        R = self.resistors

        for previous, current in zip(R, R[1:]):
            self.connect(current.elec_in, previous.elec_out)

        # Pull first `elec_in` and last `elec_out`
        self.add_input(ElecPort, 'elec_in')
        self.add_output(ElecPort, 'elec_out')

        self.connect(self.elec_in, R[0].elec_in)
        self.connect(self.elec_out, R[-1].elec_out)

The same trick can be used to create collections of ports, if needed.

Use of property resistors makes the code clearer, and allows one to loop through the child/port collection without resorting to syntaxes of the kind s[f"R{i}"] for i in range(s.n).

In the next cell, for example, we show two identical ways of initializing the resistances:

s = ResistorSeries('s', n=5)

for res in s.resistors:
    res.R = 1.25e3

# instead of:
for i in range(s.n):
    s[f"R{i}"].R = 1.25e3

Note in particular that the first expression, apart from being much simpler, is also more robust to name changes in ResistorSeries, as we do not need to know the naming convention of sub-systems (R1, R2, etc.), and of their total number n.

System properties are immutable by nature, so resistors cannot be redefined. Furthermore, use of tuple garanties that each individual element of the collection is also immutable, which is exactly what we wish to achieve here.

import logging

# Check that resistors cannot be individually reassigned:
try:
    s.resistors[0] = None

except Exception as error:
    logging.exception(error)

ERROR:root:'tuple' object does not support item assignment
Traceback (most recent call last):
  File "<ipython-input-1-940a2c6173b3>", line 5, in <module>
    s.resistors[0] = None
TypeError: 'tuple' object does not support item assignment

Write alternative ways to create a system

Consider a system class containing a 2D numpy array, which we would like to define from local data or from a file. In the next cell, we show a first, awkward implementation, where all parameters are provided to the setup method, and different courses of action are decided, depending on data type.

from cosapp.base import System
import numpy

class MySystem(System):
    def setup(self, n: int, **options):
        self.add_property('n', int(n))

        self.add_inward('x', numpy.zeros((self.n, 3)))

        # Get data from array
        try:
            x = options['x']
        except KeyError:
            pass
        else:
            if numpy.shape(x) == self.x.shape:
                self.x = numpy.copy(x)
        # Get data from file
        try:
            filename = options['filename']
        except KeyError:
            pass
        else:
            x = numpy.load(filename)
            if x.shape == self.x.shape:
                self.x = x
        # And so on...

s = MySystem('s', n=2, x=[[0, 0.1, 0.2], [-0.5, 0.9, 0.4]])

print(s.x)

[[ 0.   0.1  0.2]
 [-0.5  0.9  0.4]]

The complexity of parsing through options at setup can be simplified using ad-hoc functions referred to as factories, declared as class methods by decorator @classmethod. Note that the first argument of class methods is always cls (as opposed to self for traditional object methods), denoting the class itself. Factories are special class method whose job is to create and return a new class instance (an object) from a specific set of arguments.

Class methods are invoked with syntax ClassName.method_name(...).

In the next cell, we implement factory methods from_data and from_file, each with its own signature:

from cosapp.base import System
import numpy

class MySystem(System):
    def setup(self, n: int):
        self.add_property('n', int(n))
        self.add_inward('x', numpy.zeros((self.n, 3)))

    @classmethod
    def from_data(cls, name: str, data) -> "MySystem":
        n = cls.check_shape(data)
        s = cls(name, n=n)  # newly created system
        s.x = numpy.array(data, dtype=float)
        return s

    @classmethod
    def from_file(cls, name: str, filename) -> "MySystem":
        data = numpy.load(filename)
        return cls.from_data(name, data)

    @staticmethod
    def check_shape(data) -> int:
        """Checks that `data` is a (N x 3) array-like object.
        If so, returns integer N. If not, raises `ValueError`.
        """
        shape = numpy.shape(data)
        ok = len(shape) == 2 and shape[1] == 3
        if not ok:
            raise ValueError("data must be a (N x 3) array-like object")
        return shape[0]

s1 = MySystem('s1', n=4)  # usual way
print(f"s1.x =\n{s1.x}")

# Create a system from existing data with `from_data`
s2 = MySystem.from_data('s2', [[0, 0.1, 0.2], [-0.5, 0.9, 0.4]])
print(f"\ns2.x =\n{s2.x}")

s1.x =
[[0. 0. 0.]
 [0. 0. 0.]
 [0. 0. 0.]
 [0. 0. 0.]]

s2.x =
[[ 0.   0.1  0.2]
 [-0.5  0.9  0.4]]

import logging

try:
    s3 = MySystem.from_file('s3', 'data.pickle')

except Exception as error:
    logging.error(
        f" {type(error).__name__}: {error}"
        f"\nDoes not work for s3 since the file does not exist, but you get the idea!"
    )
else:
    print(f"\ns3.x =\n{s3.x}")

ERROR:root: FileNotFoundError: [Errno 2] No such file or directory: 'data.pickle'
Does not work for s3 since the file does not exist, but you get the idea!