CoSApp examples

Replacing a sub-system dynamically

Utility function cosapp.utils.swap_system allows one to replace on the fly an existing sub-system by another System instance.

from cosapp.utils import swap_system

help(swap_system)

Help on function swap_system in module cosapp.utils.swap_system:

swap_system(old_system: 'System', new_system: 'System', init_values=True)
    Replace `old_system` by `new_system`.

    Parameters
    ----------
    - old_system [System]: System to replace
    - new_system [System]: Replacement system
    - init_values [bool, optional]: If `True` (default), original
        system values are copied into the replacement system.

    Returns
    -------
    old_system [System]: the original system, devoid of parent.

Example

from cosapp.base import System
from cosapp.drivers import NonLinearSolver


class NominalComponent(System):
    def setup(self):
        self.add_inward('a', 1.0)
        self.add_inward('x', 1.0)
        self.add_outward('y', 0.0)

    def compute(self) -> None:
        self.y = self.a * self.x**2 - 1


class DegradedComponent(System):
    def setup(self):
        self.add_inward('a', 1.0)
        self.add_inward('x', 1.0)
        self.add_outward('y', 0.0)

    def compute(self) -> None:
        self.y = self.x - self.a


class CompositeSystem(System):

    def setup(self):
        a = self.add_child(NominalComponent('a'), pulling='x')
        b = self.add_child(NominalComponent('b'), pulling='y')

        self.connect(a, b, {'y': 'x'})  # a.y -> b.x

from cosapp.utils import swap_system

head = CompositeSystem('head')

solver = head.add_driver(NonLinearSolver('solver'))
solver.add_unknown('x', max_abs_step=0.25).add_equation('y == 0')

head.run_drivers()

print(
    "Original config:\n",
    solver.problem,
    sep="\n",
)

Original config:

Unknowns [1]
  x = 1.414213562373095
Equations [1]
  y == 0 := -8.881784197001252e-16

Next, we swap head.a with a newly created system of type DegradedComponent, and retrieve the original sub-system as original_a. After the replacement, original_a is a parentless, stand-alone system.

original_a = swap_system(head.a, DegradedComponent('a'))

# Checks
print(
    f"{head.a.parent = }",
    f"{original_a.parent = }",
    f"{type(head.a) = }",
    sep="\n",
)

head.a.parent = head - CompositeSystem
original_a.parent = None
type(head.a) = <class '__main__.DegradedComponent'>

In the process, existing connectors within the parent system are maintained:

head.connectors()

{'b.outwards -> outwards': Connector(head.outwards <- b.outwards, ['y']),
 'a.outwards -> b.inwards': Connector(b.inwards <- a.outwards, {'x': 'y'}),
 'inwards -> a.inwards': Connector(a.inwards <- head.inwards, ['x'])}

If we rerun the model, we can see that the mathematical problem is maintained. However, the obtained solution differs from the previous one, since the overall behaviour of system head has changed.

# Re-run;
head.run_drivers()

print(
    "Modified config:\n",
    solver.problem,
    sep="\n",
)

Modified config:

Unknowns [1]
  x = 2.0
Equations [1]
  y == 0 := 0.0

We can revert to the original configuration, by re-swapping current head.a with previously stored object original_a:

# Revert to original sub-system
swap_system(head.a, original_a)

head.run_drivers()

print(
    "Recovered config:\n",
    solver.problem,
    sep="\n",
)

Recovered config:

Unknowns [1]
  x = 1.4142135623730954
Equations [1]
  y == 0 := 1.7763568394002505e-15

Function swap_system can be useful in the context of an event-driven transition, for instance (see tutorial on discrete events):

from cosapp.base import System
from cosapp.utils import swap_system


class ThreasholdSystem(System):

    def setup(self):
        a = self.add_child(NominalComponent('a'), pulling='x')
        b = self.add_child(NominalComponent('b'), pulling='y')

        self.connect(a, b, {'y': 'x'})

        self.add_inward('y_max', 3.14)
        self.add_event('failure', trigger="y > y_max")
        self.add_event('recovery', trigger="y < y_max")

    def transition(self):
        if self.failure.present:
            swap_system(self.a, DegradedComponent('a'))

        if self.recovery.present:
            swap_system(self.a, NominalComponent('a'))

Ground rules

To avoid undesired side-effects, the substitute system (second argument) must not be part of an existing system tree (i.e. its parent should be None):

head1 = CompositeSystem('head1')
head2 = CompositeSystem('head2')

def print_exception(error: Exception):
    print(f"{type(error).__name__}: {error!s}")

try:
    swap_system(head1.a, head2.b)

except Exception as error:
    print_exception(error)

ValueError: System 'head2.b' already belongs to a system tree.

Oppositely, the swapped system (first argument) must be the child of a higher-level system:

try:
    swap_system(head1, CompositeSystem('new'))

except Exception as error:
    print_exception(error)

ValueError: Cannot replace top system 'head1'.