cosapp.systems.system

Basic classes handling model resolution, system connections and conversion between level of modelings.

Classes

ConversionType(value[, names, module, ...])	Enumeration of System conversion type.
ExecutionOrdering(value[, names, module, ...])	Enumeration of System algorithm to define component execution order.
System(name, **kwargs)	A class to describe generic properties and functions of a component that can be single or made of child System.

class cosapp.systems.system.ConversionType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of System conversion type.

best_fidelity_to_cost = 'best_fidelity_to_cost_ratio'

high_cost = 'highest_cost'

high_fidelity = 'highest_fidelity'

low_cost = 'lowest_cost'

low_fidelity = 'lowest_fidelity'

manual = 'manual'

class cosapp.systems.system.ExecutionOrdering(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of System algorithm to define component execution order.

MANUAL = 'manual'

class cosapp.systems.system.System(name: str, **kwargs)

Bases: Module, TimeObserver

A class to describe generic properties and functions of a component that can be single or made of child System.

Parameters:

name (str) – Name of the System

name

System name

Type:

str

inputs

Dictionary of BasePort containing the values needed to compute the System

Type:

collections.OrderedDict of BasePort

outputs

Dictionary of BasePort containing the values computed by the System

Type:

Dict[BasePort]

residues

Dictionary of residues generated by the System

Type:

collections.OrderedDict of :obj:float

children

Child System of this System

Type:

Dict[System]

parent

Parent System of this System; None if there is no parent.

Type:

System

exec_order

Execution order in which sub-systems are computed.

Type:

MappingView[str]

properties

Dictionary of immutable parameters of the system.

Type:

Dict[str, Any]

design_methods

System pre-defined design methods

Type:

Dict[Equations]

name2variable

Variable name mapping to its value by reference

Type:

Dict[str, VariableReference]

_locked

if True, add_input, add_output, add_inward and add_outward are desactivated. This is the default behavior outside the setup function.

Type:

bool

_active

if False, the System will not execute its compute method

Type:

bool

_is_tree_clean

Reflects the status of the inputs and outputs. clean status means the group of ports was not updated since last computation of the System

Type:

bool

_compute_calls

Store if the System was computed at last call or not (due to inhibition of clean status)

Type:

bool

Examples

To create your own System, you should inherit from this class and define your own setup and compute methods.

>>> import numpy as np
>>>
>>> class RealPort(Port):
>>>
>>>     def setup(self):
>>>         self.add_variable('x',0.)
>>>
>>> class InvertedParabola(System):
>>>
>>>     def setup(self):
>>>         self.add_input(RealPort, 'iterative')
>>>         self.add_input(RealPort, 'target_y')
>>>         self.add_output(RealPort, 'max_root')
>>>         self.add_inward({'a': 0.01,
>>>                        'b': 2.,
>>>                        'c': 4.})
>>>         self.add_outward('roots', None)
>>>         self.add_equation("a * iterative.x**2 + b * iterative.x + c == target_y.x", "y")
>>>
>>>     def compute(self):
>>>
>>>         discriminant = self.b**2 - 4. * self.a * self.c
>>>         if discriminant >= 0.:
>>>             self.roots = ((-self.b + np.sqrt(discriminant)) / (2. * self.a),
>>>                           (-self.b - np.sqrt(discriminant)) / (2. * self.a))
>>>             self.max_roots.x = max(self.roots)
>>>         else:
>>>             self.roots = None
>>>             self.max_roots.x = np.nan

COMMON_PORTS: List[str] = ['INWARDS', 'OUTWARDS', 'MODEVARS_IN', 'MODEVARS_OUT']

INWARDS: ClassVar[str] = 'inwards'

MODEVARS_IN: ClassVar[str] = 'modevars_in'

MODEVARS_OUT: ClassVar[str] = 'modevars_out'

OUTWARDS: ClassVar[str] = 'outwards'

accept(visitor: Visitor) → None

Specifies course of action when visited by visitor

property active_surrogate: bool

True if surrogate model is activated, False otherwise.

Type:

bool

add_child(child: AnySystem, execution_index: int | None = None, pulling: str | Collection[str] | Dict[str, str] | None = None, desc: str = '') → AnySystem

Add a child System to the current System.

When adding a child System, it is possible to specified its position in the execution order.

Child ports or individual inwards and outwards can also be pulled at the parent level by providing either the name of the port/inward/outward or a list of them or the name mapping of the child element (dictionary keys) to the parent element (dictionary values). If the argument is not a dictionary, the name in the parent system will be the same as in the child.

Parameters:

• [System] (- child) – System to add to the current System

• [int (execution_index) – Index of the execution order list at which the System should be inserted; default latest.

• optional] – Index of the execution order list at which the System should be inserted; default latest.

• dict[str (- pulling [str or list[str] or) – Map of child ports to pulled ports at the parent system level; default None (no pulling)

• str] – Map of child ports to pulled ports at the parent system level; default None (no pulling)

• optional] – Map of child ports to pulled ports at the parent system level; default None (no pulling)

• [str (- desc) – Sub-system description in the context of its parent.

• optional] – Sub-system description in the context of its parent.

Return type:

child

add_design_method(name: str) → MathematicalProblem

Add a design method to the System

A design method is a set of free variables and equations that defines a way to design the System It is a easy way to pre-define a design of this System for users

The returned mathematical system is empty. It should be populated with the needed variables and equations (see Examples).

Parameters:

name (str) – The name of the design method

Returns:

The newly created mathematical problem.

Return type:

MathematicalProblem

Examples

>>> system1.add_design_method("method1")         >>>        .add_unknown([
>>>             dict(name="x", max_rel_step=0.1),
>>>             "y"
>>>         ])         >>>        .add_equation([
>>>             "u == 0",
>>>             "v == 800"
>>>         ])

add_driver(driver: AnyDriver) → AnyDriver

Add a driver to this system.

Parameters:

driver (Driver) – Driver to add

Returns:

The added driver

Return type:

Driver

add_equation(equation: str | Iterable[dict | str | Tuple[str, str]], name: str | None = None, reference: Number | ndarray | str = 1) → MathematicalProblem

Add off-design equation.

Equations may be added one by one, or provided by a list of dictionaries to add multiple equations at once. The dictionary keys are the arguments of this method.

Parameters:

• equation (-) – Equation or list of equations to add

• name (-) – Name of the equation; default None => ‘lhs == rhs’

• reference (-) – Reference value(s) used to normalize the equation; default is 1. If value is “norm”, actual reference value is estimated from order of magnitude.

Returns:

The modified mathematical problem

Return type:

MathematicalProblem

add_event(name: str, desc: str = '', trigger: str | Event | EventState | ZeroCrossing | None = None, final: bool = False) → Event

Add an event to system.

An event occurrence can either be determined by the system itself, or triggered by an event from another system.

This function cannot be called outside setup().

Parameters:

• [str] (- name) – Name of the event.

• [str (- desc) – Event description; defaults to ‘’.

• optional] – Event description; defaults to ‘’.

• [Union[str (- trigger) – String, primary or derived event defining the event trigger; defaults to None.

• Event – String, primary or derived event defining the event trigger; defaults to None.

• EventState – String, primary or derived event defining the event trigger; defaults to None.

• ZeroCrossing] – String, primary or derived event defining the event trigger; defaults to None.

• optional] – String, primary or derived event defining the event trigger; defaults to None.

• [bool (- final) – Defines whether or not event is final; defaults to False.

• optional] – Defines whether or not event is final; defaults to False.

Returns:

The newly created event.

Return type:

• event [Event]

Examples

>>> import numpy as np
>>>
>>> class PointMassDynamics(System):
>>>     # Free fall of a point mass, with friction
>>>     def setup(self):
>>>         self.add_inward('mass', 1.2, desc='Mass')
>>>         self.add_inward('cf', 0.1, desc='Friction coefficient')
>>>         self.add_inward('g', np.r_[0, 0, -9.81], unit='m/s**2', desc='External acceleration field')
>>>
>>>         self.add_outward('a', np.zeros(3), unit='m/s**2', desc='Acceleration')
>>>         self.add_transient('v', der='a')
>>>         self.add_transient('x', der='v')
>>>
>>>     def compute(self):
>>>         self.a = self.g - (self.cf / self.mass * np.linalg.norm(self.v)) * self.v
>>>
>>> class BouncingPointMass(System):
>>>     def setup(self):
>>>         self.add_child(PointMassDynamics('dyn'), pulling=['x', 'v', 'a', 'mass', 'cf', 'g'])
>>>         self.add_event('rebound', trigger="x[2] <= 0")
>>>
>>>     def transition(self):
>>>         if self.rebound.present:
>>>             self.v[2] *= -1

add_input(port_class: Type[AnyPort], name: str, variables: Dict[str, Any] | None = None, desc: str = '') → AnyPort

Add an input Port to the System.

This function cannot be called outside System.setup.

Parameters:

• [type[Port]] (- port_class) – Class of the Port to create

• [str] (- name) – Port name

• [str (- desc) – Port description

• optional] – Port description

• [dict[str (- variables) – Dictionary of initial values (default: None)

• Any] – Dictionary of initial values (default: None)

• optional] – Dictionary of initial values (default: None)

Return type:

The created port

Examples

>>> class MyModule(System):
>>>     def setup(self):
>>>         self.add_input(MyPort, 'p_in1')
>>>         self.add_input(MyPort, 'p_in2', {'x': 1.5})

add_inward(definition: str | Dict[str, Any], value: Any = 1, unit: str = '', dtype: Any | Tuple[Any, ...] | None = None, valid_range: Tuple[Any, Any] | None = None, invalid_comment: str = '', limits: Tuple[Any, Any] | None = None, out_of_limits_comment: str = '', desc: str = '', distribution: Distribution | None = None, scope: Scope = Scope.PRIVATE) → None

Add a inward variable to the System.

A inward variable is calculated by the System. But its value is not mandatory in any variables fluxes between this System and another one.

An unique inward variable can be defined by providing directly all arguments. And multiple inward variables can be defined by passing a dict of pair (str, Any) with an entry for each variable.

This function cannot be called outside setup().

Parameters:

• definition (str or Dict[str, Any]) – Name of the unique variable or a dictionary for multiple variables at once

• value (Any, optional) – Value of the variable if definition is a str; default 1

• unit (str, optional) – Variable unit; default empty string (i.e. dimensionless)

• dtype (type or iterable of types, optional) – Variable type; default None (i.e. type of initial value)

• valid_range (Tuple[Any, Any], optional) – Validity range of the variable; default None (i.e. all values are valid)

• invalid_comment (str, optional) – Comment to show in case the value is not valid; default ‘’

• limits (Tuple[Any, Any], optional) – Limits over which the use of the model is wrong; default valid_range

• out_of_limits_comment (str, optional) – Comment to show in case the value is not valid; default ‘’

• desc (str, optional) – Variable description; default ‘’

• distribution (Distribution, optional) – Probability distribution of the variable; default None (no distribution)

• scope (Scope {PRIVATE, PROTECTED, PUBLIC}, optional) – Variable visibility; default PRIVATE

Examples

To add one inward variable, arguments must be directly specified.

>>> system.add_inward('data', 2.)

To add multiple inward variables, a dictionary with one key per data should be provided.

>>> system.add_inward({
>>>     'data1': 42.,
>>>     'data2': False
>>> })

add_inward_modevar(name: str, value: Any = False, unit: str = '', dtype: Any | Tuple[Any, ...] | None = None, desc: str = '', scope: Scope = Scope.PRIVATE) → None

Add an inward mode variable to the System.

A mode variable is a piecewise constant variable in each continuous time phase. Like any input, inward mode variables should not be modified during system transitions.

This function cannot be called outside setup().

Parameters:

• name (str) – Name of the variable

• value (Any, optional) – Value of the variable; default 1

• unit (str, optional) – Variable unit; default empty string (i.e. dimensionless)

• dtype (type) – Variable type; default None (i.e. type of initial value)

• desc (str, optional) – Variable description; default ‘’

• scope (Scope {PRIVATE, PROTECTED, PUBLIC}, optional) – Variable visibility; default PRIVATE

Examples

>>> class MultimodeSystem(System):
>>>     def setup(self):
>>>         self.add_inward_modevar('composite', True)
>>>         self.reconfig()
>>>
>>>     def reconfig(self):
>>>         # Define one or two sub-systems, depending on mode
>>>         for name in list(self.children):
>>>             self.pop_child(name)
>>>         if self.composite:
>>>             a = self.add_child(Foo('a'))
>>>             b = self.add_child(Bar('b'))
>>>             self.connect(a, b, {'x', 'y'})
>>>         else:
>>>             self.add_child(Bogus('c'))
>>>
>>>     def transition(self):
>>>         self.reconfig()

add_output(port_class: Type[AnyPort], name: str, variables: Dict[str, Any] | None = None, desc: str = '') → AnyPort

Add an output Port to the System.

This function cannot be called outside System.setup.

Parameters:

• [type[Port]] (- port_class) – Class of the Port to create

• [str] (- name) – Port name

• [str (- desc) – Port description

• optional] – Port description

• [dict[str (- variables) – Dictionary of initial values (default: None)

• Any] – Dictionary of initial values (default: None)

• optional] – Dictionary of initial values (default: None)

Return type:

The created port

Examples

>>> class MyModule(System):
>>>     def setup(self):
>>>         self.add_output(MyPort, 'p_out1')
>>>         self.add_output(MyPort, 'p_out2', {'y': 1.5})

add_outward(definition: str | Dict[str, Any], value: Any = 1, unit: str = '', dtype: Any | Tuple[Any, ...] | None = None, valid_range: Tuple[Any, Any] | None = None, invalid_comment: str = '', limits: Tuple[Any, Any] | None = None, out_of_limits_comment: str = '', desc: str = '', scope: Scope = Scope.PUBLIC) → None

Add a outward variable to the System.

A outward variable is calculated by the System. But its value is not mandatory in any variables fluxes between this System and another one.

An unique outward variable can be defined by providing directly all arguments. And multiple outward variables can be defined by passing a dict of pair (str, Any) with an entry for each variable.

This function cannot be called outside setup().

Parameters:

• definition (str or Dict[str, Any]) – Name of the unique variable or a dictionary for multiple variables at once

• value (Any, optional) – Value of the variable if definition is a str; default 1

• unit (str, optional) – Variable unit; default empty string (i.e. dimensionless)

• dtype (type or iterable of types, optional) – Variable type; default None (i.e. type of initial value)

• valid_range (Tuple[Any, Any], optional) – Validity range of the variable; default None (i.e. all values are valid)

• invalid_comment (str, optional) – Comment to show in case the value is not valid; default ‘’

• limits (Tuple[Any, Any], optional) – Limits over which the use of the model is wrong; default valid_range

• out_of_limits_comment (str, optional) – Comment to show in case the value is not valid; default ‘’

• desc (str, optional) – Variable description; default ‘’

• scope (Scope {PRIVATE, PROTECTED, PUBLIC}, optional) – Variable visibility; default PUBLIC

Examples

To add an unique variable, arguments must be directly specified.

>>> system.add_outward('info', 2.)

To add multiple variables, a dictionary with one key per outward variable should be provided.

>>> system.add_inward({
>>>     'info1': 42.,
>>>     'info2': False
>>> })

add_outward_modevar(name: str, value: Any | None = None, unit: str = '', dtype: Any | Tuple[Any, ...] | None = None, desc: str = '', init: Any | None = None, scope: Scope = Scope.PRIVATE) → None

Add an outward mode variable to the System.

A mode variable is a piecewise constant variable in each continuous time phase. It may only be modified in the transition method of the System, and never inside method compute. Notwithstanding, its value may be used in compute. The modification of an output mode variable is often associated with the occurence of an event in the system.

This function cannot be called outside setup().

Parameters:

• name (str) – Name of the variable

• value (Any, optional) – Value of the variable; default 1

• unit (str, optional) – Variable unit; default empty string (i.e. dimensionless)

• dtype (type) – Variable type; default None (i.e. type of initial value)

• desc (str, optional) – Variable description; default ‘’

• init (Any, optional) – Value imposed at the beginning of time simulations. If unspecified (default), the variable remains untouched.

• scope (Scope {PRIVATE, PROTECTED, PUBLIC}, optional) – Variable visibility; default PRIVATE

Examples

>>> class ThresholdSystem(System):
>>>     def setup(self):
>>>         self.add_inward('threshold', 1.0)
>>>         self.add_inward('x', 0.0)
>>>         self.add_outward('y', 0.0)
>>>         self.add_event('activates', trigger='x >= threshold')
>>>         self.add_outward_modevar('activated', init='x > threshold')
>>>
>>>     def compute(self):
>>>         self.y = self.x if self.activated else 0.0
>>>
>>>     def transition(self):
>>>         if self.activates.present:
>>>             self.activated = True

add_property(name: str, value: Any) → None

Create new read-only property name, set to value

add_rate(name: str, source: Any, initial_value: Number | ndarray | str = None, desc: str = None) → None

Add a variable monitoring the rate-of-change of a given quantity (referred to as source) as the system evolves. Rates are defined by their source, and are updated by the time driver computing the time evolution of the system.

This function cannot be called outside setup().

Parameters:

• name (str) – Name of the transient variable

• source (Any) – Expression of the quantity of interest whose rate will be computed. The type and unit of the rate are deduced from source.

• desc (str, optional) – Variable description; default None

• initial_value (Number/array, or evaluable expression (str), optional.) – Initial value of the rate (otherwise unknown, as rates are only updated from the first time step on). If specified, it must be consistent with the type of source (scalar vs. array).

Examples

>>> system.add_inward('U', 0.1, desc='Input voltage')
>>> system.add_rate('dUdt', source='U')  # dUdt defined as dU/dt

add_target(name: str | Iterable[dict | str | Tuple[str, str]], reference: Number | ndarray | str = 1, weak=False) → MathematicalProblem

Add deferred off-design equation on a targetted quantity.

Target equations may be added one by one, or provided by a list of dictionaries to add multiple equations at once. The dictionary keys are the arguments of this method.

Parameters:

• name (-) – Name of the equation; default None => ‘lhs == rhs’

• reference (-) – Reference value(s) used to normalize the equation; default is 1. If value is “norm”, actual reference value is estimated from order of magnitude.

• weak (-) – If True, the target is disregarded if the corresponding variable is connected; default is False.

Returns:

The modified mathematical problem

Return type:

MathematicalProblem

add_transient(name: str, der: str, desc: str = None, max_time_step: Number | str = inf, max_abs_step: Number | str = inf) → None

Declare a transient variable, defined implicitly by the expression of its time derivative, and add a time-dependent unknown to the mathematical problem. Transients can be used withinn the system, but their time evolution is computed outside the system, by a time driver.

If name already exists, it must refer to an inward. If name does not exist, a new inward name is created on the fly.

This function cannot be called outside setup().

Parameters:

• name (str) – Name of the transient variable

• der (str) – Expression of the time derivative, used for integration during time driver execution. The type and unit of the transient variable are deduced from der.

• desc (str, optional) – Variable description; default None

• max_time_step (Number or evaluable expression (str), optional) – Maximum time step admissible for the integration of the variable (for stability, typically).

• max_abs_step (Number or evaluable expression compatible with transient variable, optional) – Maximum variable step admissible over one time step (for stability, typically).

Examples

>>> system.add_inward('v', numpy.zeros(3), desc='Velocity')
>>> system.add_inward('a', numpy.zeros(3), desc='Acceleration')
>>> system.add_inward('area', 1.0)
>>> system.add_outward('flowrate', desc='Volumetric flowrate')
>>> system.add_transient('x', der='v')  # x defined by dx/dt = v
>>> system.add_transient('v', der = 'a',
        max_time_step = '0.1 / max(norm(a), 1e-9)',
    )
>>> system.add_transient('h',
        der = 'flowrate / area',
        max_abs_step = 0.1,
    )

add_unknown(name: str | Iterable[dict | str], max_abs_step: Number = inf, max_rel_step: Number = inf, lower_bound: Number = -inf, upper_bound: Number = inf) → MathematicalProblem

Add unknown variables.

You can set variable one by one or provide a list of dictionary to set multiple variable at once. The dictionary key are the arguments of this method.

Parameters:

• name (-) – Name of the variable or list of variable to add

• max_rel_step (-) – Maximal relative step by which the variable can be modified by the numerical solver; default numpy.inf

• max_abs_step (-) – Maximal absolute step by which the variable can be modified by the numerical solver; default numpy.inf

• lower_bound (-) – Lower bound on which the solver solution is saturated; default -numpy.inf

• upper_bound (-) – Upper bound on which the solver solution is saturated; default numpy.inf

Returns:

The modified mathematical problem

Return type:

MathematicalProblem

all_connectors() → Iterator[Connector]

Iterator yielding all connectors within system.

all_events() → Iterator[Event]

Recursive iterator on all events in complete system tree.

any_active_driver() → bool

append_name2variable(additional_mapping: Iterable[Tuple[str, VariableReference]]) → None

Append the Iterable of (str, VariableReference) Tuple to the lookup variables mapping.

The additional mapping is also transfer upward to the parent System.

Parameters:

additional_mapping (Iterable[Tuple[str, VariableReference]]) – Additional list of (str, VariableReference) tuple to append

assembled_problem() → MathematicalProblem

Returns the consolidated mathematical problem, assembled from entire system tree, and accounting for cyclic dependency loops.

Returns:

The assembled mathematical problem.

Return type:

MathematicalProblem

assembled_time_problem() → TimeProblem

Returns the consolidated transient problem, assembled from the entire system tree.

Returns:

The assembled time problem.

Return type:

TimeProblem

call_clean_run(skip_driver: bool = False) → None

Execute clean_run recursively on all modules.

Parameters:

skip_drivers (bool) – Skip calling cosapp.drivers.driver.Driver.clean_run()

call_setup_run(skip_driver: bool = False) → None

Execute setup_run recursively on all modules.

Parameters:

skip_drivers (bool) – Skip calling cosapp.drivers.driver.Driver.setup_run()

check(name: str | None = None) → Dict[str, Validity] | Validity

Get variable value validity for a given variable, port or system or all of them.

If name is not provided, returns a dictionary with the validity of all System variables; i.e. all variables in input and output ports including inwards and outwards of this system and all its children. Else only, the validity for the given variable will be returned.

Parameters:

name (str, optional) – Variable name

Returns:

(Dictionary of) the variable(s) value validity

Return type:

Dict[str, Validity] or Validity

classmethod check_config_dict(params: dict) → None

Check if the provided dictionary respects the convention of a model file.

Parameters:

params (dict) – Dictionary to be tested

Raises:

jsonschema.exceptions.ValidationError – If the provided dictionary does not conform to the JSON schema.

property child_connectors: Dict[str, List[Connector]]

Connectors between sub-systems, referenced by system names.

Type:

Dict[str, List[Connector]]

close_loops()

Close loops opened to allow system resolution.

connect(object1: BasePort | System, object2: BasePort | System, mapping: str | List[str] | Dict[str, str] | None = None, cls: Type[BaseConnector] | None = None, **kwargs) → None

Connect two systems or two ports.

This method connects object1 to object2. If no mapping is provided, connection will be made between variables based on their names. If a name mapping is provided as a list, the name should be present in both objects. If the mapping is specified as a dictionary, keys are expected to belong to object1, and values to object2.

If both objects are systems, mapping is mandatory (full system connections are forbidden). In this case, each (key, value) pair of the mapping leads to a port-to-port connection.

Connections are oriented (i.e. the direction of value transfer is fixed). The decision tree is as following:

• If one port is of type input and the other one of type output, the connection flows from the output to the input.

• Else if one port (portA) belong to the parent System of the other port (portB), the connection flows from portA to portB if portA is an input. And it flows from portB to portA if portA is an output.

• Else if both are inputs, the port is pulled on the parent and connected to the two children.

Connectors can only connect systems at the same level or a system with its parent. In all cases the connectors will be stored by the parent hosting the two connected systems.

Parameters:

• object1 (Union[BasePort, System]) – First end-point of connector.

• object2 (Union[BasePort, System]) – Second end-point of connector.

• mapping (str or List[str] or Dict[str, str], optional) – (List of) common name(s) or mapping name dictionary; default None (i.e. no mapping).

• cls (Type[BaseConnector], optional) – Connector type. When specified, cls must be a specialization of BaseConnector. If cls is None (default), the actual connector type is either Connector, or a port-specific connector, if any. Port bound connectors are automatically selected when both ports are of same type, and when port class contains inner class Connector, derived from BaseConnector.

• **kwargs – Additional keyword arguments forwarded to cls.

Examples

Here is an example, in which we assume a port p_in has inputs of a system s1 that come from some output port p_out of another system s2. In the last example, connect is called with two child systems, rather than ports. The container system being called top:

>>> class DemoPort(Port):
>>>     def setup(self):
>>>         self.add_variable('a', 1.0)
>>>         self.add_variable('b', 2.0)
>>>         self.add_variable('c', 3.0)
>>>
>>> class DummySystem(Port):
>>>     def setup(self):
>>>         self.add_inward('x', 1.0)
>>>         self.add_outward('y', 0.0)
>>>         self.add_input(DemoPort, 'p_in')
>>>         self.add_output(DemoPort, 'p_out')
>>>
>>> top = System('top')
>>> top.add(DummySystem('s1'))
>>> top.add(DummySystem('s2'))
>>>
>>> top.connect(top.s1.p_in, top.s2.p_out, 'a')
>>> top.s1.p_in.a = 4
>>> top.run_once()
>>> assert top.s1.p_in.a == top.s2.p_out.a
>>>
>>> top.connect(top.s1.p_in, top.s2.p_out, {'b': 'c'})
>>> top.run_once()
>>> assert top.s1.p_in.b == top.s2.p_out.c
>>>
>>> top.connect(top.s1, top.s2, {'x': 'y', 'p_in.c': 'p_out.b'})
>>> top.run_once()
>>> assert top.s1.x == top.s2.y
>>> assert top.s1.p_in.c == top.s2.p_out.b

connectors() → Dict[str, Connector]

Constructs a dictionary of all connectors within system, referenced by connector name.

convert_to(*args, **kwargs) → None

Convert system into another System. Note: not implemented for class System.

Raises:

TypeError – If the current System cannot be converted.

design(method: str) → MathematicalProblem

Returns the chosen group of design equations.

Parameters:

method (str) – Name of the group of equations to extract

Returns:

Mathematical system to solve for the chosen method

Return type:

MathematicalProblem

design_methods: Dict[str, MathematicalProblem]

drivers: Dict[str, Driver]

dump_surrogate(filename: str) → None

Dumps system surrogate model (if any) into a binary file.

events() → Iterator[Event]

Iterator on all events locally defined on system.

export_structure() → Dict

Export current system structure to a dictionary, which contains the definition of all ports, the structure of sub-system and the connections between ports.

get_unsolved_problem() → MathematicalProblem

Returns the consolidated mathematical problem, assembled from entire system tree, and accounting for cyclic dependency loops.

Deprecated; use assembled_problem instead.

Returns:

The assembled mathematical problem.

Return type:

MathematicalProblem

property has_surrogate: bool

True if system has a surrogate model (even if inactive), False otherwise.

Type:

bool

incoming_connectors() → Iterator[Connector]

Iterator yielding all connectors targetting system.

property input_mapping: Dict[str, VariableReference]

free input mapping

Type:

Dict[str, VariableReference]

inputs: Dict[str, BasePort]

is_clean(direction: PortType | None = None) → bool

Are the System ports with the given direction clean? If no direction is specified, checks if both directions are clean.

Parameters:

direction (PortType, optional) – Direction of interest

Returns:

Clean status

Return type:

bool

is_input_var(name: str) → bool

Returns True if name is the name of an input variable, False otherwise

is_output_var(name: str) → bool

Returns True if name is the name of an output variable, False otherwise

is_running() → bool

Is this System in execution?

Returns:

In execution status

Return type:

bool

is_standalone() → bool

Is this System able to solve itself?

Returns:

Ability to solve the system or not.

Return type:

bool

classmethod load(filepath: str | Path | StringIO, name: str | None = None) → System

Load configuration from file-like object.

Parameters:

• filepath (str or Path or file-like) – Filepath or file-like object (i.e. has a .read() method)

• name (str, optional) – Name of the newly created system. If unspecified (default), uses the name contained in filepath.

Returns:

The loaded system.

Return type:

System

classmethod load_from_dict(name: str, parameters: dict) → System

Instantiate a System from its name and its parameters.

The construction of the System is done recursively from the lower level. Therefore this function is also called recursively returning to the upper level, the lower System and its connections to the upper System via a dictionary. If no connections are to forwarded the second returned argument is None.

Parameters:

• name (str) – Identifier of the system

• parameters (dict) – The dictionary containing the system definition

Returns:

The generated system

Return type:

System

load_surrogate(filename: str, activate=True) → None

Loads a surrogate model from a binary file created by dump_surrogate.

Parameters:

• filename (-) – Destination file name.

• activate (-) – Boolean determining whether or not surrogate model should be activated once loaded. Default is True.

log_debug_message(handler: HandlerWithContextFilters, record: logging.LogRecord, format: LogFormat = LogFormat.RAW) → bool

Callback method on the system to log more detailed information.

This method will be called by the log handler when log_context() is active if the logging level is lower or equals to VERBOSE_LEVEL. It allows the object to send additional log message to help debugging a simulation.

Note

logger.log method cannot be used here. Use handler.handle(record)

Parameters:

• handler (HandlerWithContextFilters) – Log handler on which additional message should be published.

• record (logging.LogRecord) – Log record

• format (LogFormat) – Format of the message

Returns:

Should the provided record be logged?

Return type:

bool

make_surrogate(data_in: ~pandas.core.frame.DataFrame | ~typing.Dict[str, ~typing.List[float]], model=<class 'cosapp.utils.surrogate_models.kriging.FloatKrigingSurrogate'>, activate=True, data_out: ~pandas.core.frame.DataFrame | ~typing.Dict[str, ~typing.Any] | None = None, postsynch: str | ~typing.List[str] = '*', *args, **kwargs) → SystemSurrogate

Creates a surrogate model superseding the normal behaviour of compute(). The surrogate model is trained from datasets data_in and data_out, given as pandas.DataFrame objects, or dictionnaries interpretable as so.

If no output data are provided (default), they are computed as the response of system to design-of-experiment data_in.

Parameters:

• data_in (-) – Design of experiments for variables in dataframe columns or dict keys. Column/key names must match variable contextual names, as in ‘wing.geom.length’, e.g.

• model (-) – Model class, implementing methods train(x, y) and predict(x), where x and y are 1D arrays. Default is cosapp.utils.surrogate_models.FloatKrigingSurrogate.

• activate (-) – Boolean determining whether or not surrogate model should be activated once created. Default is True.

• data_out (-) – Y-dataset used for surrogate model training. Default is None (dataset is computed).

• postsynch (-) – List of output variable names to synchronize after each surrogate model execution. Default is ‘*’, meaning all outputs are post-synchronized.

• *args (-) –

  Additional parameters passed to model constructor.

• **kwargs (Any) – Additional parameters passed to model constructor.

• Returns

• --------

• SystemSurrogate – System surrogate attached to the system.

name2variable: Dict[str, VariableReference]

new_problem(name='problem') → MathematicalProblem

Create a new, empty MathematicalProblem in the context of system.

Parameters:

name (str) – Name of the mathematical problem. Defaults to ‘problem’.

Returns:

The newly created mathematical problem.

Return type:

MathematicalProblem

open_loops()

Open closed loops in children relations.

outputs: Dict[str, BasePort]

pop_child(name: str) → System

Remove the subsystem called name.

Parameters:

name (str) – Name of the subsystem to be removed

Returns:

The removed subsystem

Return type:

System

Raises:

AttributeError –

pop_name2variable(keys: Iterable[str]) → None

Remove the given keys from the name mapping dictionary.

The keys will be remove from the local mapping. Then the keys list will be sent upward to the parent System for deletion.

Parameters:

keys (Iterable[str]) – Keys to remove

ports() → Iterator[BasePort]

Iterator on all system ports (both inputs and outputs)

property problem: MathematicalProblem

locally defined off-design mathematical problem, without considering sub-system tree.

Raises:

AttributeError unless system is being modified by a driver.

type:

MathematicalProblem

property properties: Dict[str, Any]

list of read-only properties and associated values

Type:

Dict[str, Any]

property rates

Returns a dictionary containing all time derivatives (rates) in current system tree

property residues

Get the residues for the current System.

Type:

Dict[str, Residue]

retrieve_incoming_data() → None

Transfer data from all incoming connectors

run_children_drivers() → None

Solve the children System in the execution order.

run_drivers() → None

Run the drivers defined on this System.

run_once() → None

Run the system once.

Execute the model of this System and its children in the execution order.

Notes

The drivers are not executed when calling this method; only the physical model.

save(fp, indent=2, sort_keys=True) → None

Serialize the System as a JSON formatted stream to fp.

Parameters:

• fp (str or Path or file-like) – A .write()-supporting file-like object or the filename

• indent (int, optional) – Indentation in the file (default: 2)

• sort_keys (bool, optional) – Sort the keys in alphabetic order (default: False)

set_clean(direction: PortType) → None

Set to clean ports of a certain direction.

Parameters:

direction (PortType) – Direction to set

set_dirty(direction: PortType) → None

Set to dirty ports of a certain direction.

Parameters:

direction (PortType) – Direction to set

classmethod set_master(name: str, type_checking: bool = True) → bool

Set the master System

Parameters:

• name (str) – Name of the System calling this context manager

• type_checking (bool, optional) – Whether to activate the type checking in the ports or not (default: True)

Returns:

Is the callee the master System?

Return type:

bool

setup(**kwargs) → None

System port and/or child System are defined in this function.

This function allows to populate a customized System class. The helper functions for the user are:

• add_input : add an input port

• add_output : add an output port

• add_inward : add a inward variable

• add_outward : add a outward variable

• add_child : add a child System

See also

add_input, add_output, add_inward, add_outward, add_child

Examples

Here is an example of System subclassing:

>>> class AdvancedDuct(System):
>>>
>>>     def setup(self):
>>>         self.add_input(FlowPort, 'in')
>>>         self.add_output(FlowPort, 'out')
>>>         self.add_inward({'heat_source': 1.0,
>>>                          'pressure_loss': 0.01})
>>>         self.add_outward('wall_temperature', 400.)
>>>
>>>         self.add_child(AnotherSystem('system2'))

static str_der(order: int) → Callable[[str], str]

Derivate name factory.

Parameters:

order (int) – Derivative order

Returns:

Function generating the derivate name from the variable name

Return type:

Callable[[int, ], str]

tags: ClassVar[FrozenSet[str]] = frozenset({})

to_d3(show=True, size=435) → IPython.display.IFrame

Returns the hierarchical representation of this system in HTML format.

Returns:

IFrame to the HTML formatted representation

Return type:

IPython.display.IFrame

to_dict() → Dict

Public API to export system to a dictionary

to_html(filename: str, embeddable=False) → None

Save the System as HTML using vis.JS library.

Parameters:

• filename (str) – Filename to write to

• embeddable (bool, optional) – Is the HTML to be embedded in an existing page? Default: False

to_json(indent=2, sort_keys=True) → str

Return a string in JSON format representing the System.

Parameters:

• indent (int, optional) – Indentation of the JSON string (default: 2)

• sort_keys (bool, optional) – Sort keys in alphabetic order (default: True)

Returns:

String in JSON format

Return type:

str

touch()

property transients

Returns a dictionary containing all transient unknowns in current system tree

transition() → None

Method describing system transition upon the occurrence of events (if any). Does not do anything by default, but should be implemented for systems with events.

Examples

>>> import numpy as np
>>>
>>> class PointMassDynamics(System):
>>>     # Free fall of a point mass, with friction
>>>     def setup(self):
>>>         self.add_inward('mass', 1.2, desc='Mass')
>>>         self.add_inward('cf', 0.1, desc='Friction coefficient')
>>>         self.add_inward('g', np.r_[0, 0, -9.81], desc='External acceleration field')
>>>         self.add_outward('a', np.zeros(3), desc='Acceleration')
>>>         self.add_transient('v', der='a')
>>>         self.add_transient('x', der='v')
>>>
>>>     def compute(self):
>>>         v_norm = np.linalg.norm(self.v)
>>>         self.a = self.g - (self.cf / self.mass * v_norm) * self.v
>>>
>>> class BouncingPointMass(PointMassDynamics):
>>>     def setup(self):
>>>         super().setup()
>>>         self.add_event('rebound', trigger="x[2] <= 0")
>>>
>>>     def transition(self):
>>>         if self.rebound.present:
>>>             self.v[2] *= -1

tree_transition() → None

Invoke transition in entire system tree

property unknowns

Get the unknowns for the current System.

Type:

Dict[str, Unknown]

update()

Perform complex tasks due to data change.

Some parameters may be a file used to initialize a complex object. Updating that parameter may imply the modification of that complex object. This should be done in this method.

Examples

>>> class TableModel(System):
>>>
>>>     def setup(self):
>>>         # Add a inwards to the file containing the table values
>>>         self.add_inward('table_file', '{myProject}/ressources/my_table.csv')
>>>         # Set a object able to interpolate the table
>>>         self.add_outward('table', Table(self.table_file))
>>>
>>>     def update(self):
>>>         # Update the table object as the source file may have been updated.
>>>         self.table = Table(self.table_file)

Notes

This method is called systematically after the setup() call. Otherwise the user is responsible to explicitly call it when needed.