cosapp.systems.system

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

Classes

ConversionType(value)

Enumeration of System conversion type.

ExecutionOrdering(value)

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)[source]

Bases: enum.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)[source]

Bases: enum.Enum

Enumeration of System algorithm to define component execution order.

MANUAL = 'manual'
class cosapp.systems.system.System(name: str, **kwargs)[source]

Bases: cosapp.core.module.Module, cosapp.core.time.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: cosapp.patterns.visitor.Visitor) → None[source]

Specifies course of action when visited by visitor

property active_surrogate

True if surrogate model is activated, False otherwise.

Type

bool

add_child(child: AnySystem, execution_index: Optional[int] = None, pulling: Optional[Union[str, List[str], Dict[str, str]]] = None, desc: str = '') → AnySystem[source]

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

  • child [System] (-) – 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.

  • pulling [str or list[str] or dict[str (-) – 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)

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

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

Returns

Return type

child

add_design_method(name: str) → cosapp.core.numerics.basics.MathematicalProblem[source]

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[source]

Add a driver to this system.

Parameters

driver (Driver) – Driver to add

Returns

The added driver

Return type

Driver

add_equation(equation: Union[str, Iterable[Union[dict, str, Tuple[str, str]]]], name: Optional[str] = None, reference: Union[numbers.Number, numpy.ndarray, str] = 1) → cosapp.core.numerics.basics.MathematicalProblem[source]

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: Optional[Union[str, cosapp.multimode.event.Event, cosapp.multimode.event.EventState, cosapp.multimode.zeroCrossing.ZeroCrossing]] = None, final: bool = False) → cosapp.multimode.event.Event[source]

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

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

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

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

  • trigger [Union[str (-) – 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.

  • final [bool (-) – 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: Optional[Dict[str, Any]] = None, desc: str = '') → AnyPort[source]

Add an input Port to the System.

This function cannot be called outside System.setup.

Parameters

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

  • name [str] (-) – Port name

  • desc [str (-) – Port description

  • optional] – Port description

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

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

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

Returns

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: Union[str, Dict[str, Any]], value: Any = 1, unit: str = '', dtype: Optional[Union[Any, Tuple[Any, ...]]] = None, valid_range: Optional[Tuple[Any, Any]] = None, invalid_comment: str = '', limits: Optional[Tuple[Any, Any]] = None, out_of_limits_comment: str = '', desc: str = '', distribution: Optional[cosapp.utils.distributions.distribution.Distribution] = None, scope: cosapp.ports.enum.Scope = <Scope.PRIVATE: 1>) → None[source]

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: Optional[Union[Any, Tuple[Any, ...]]] = None, desc: str = '', scope: cosapp.ports.enum.Scope = <Scope.PRIVATE: 1>) → None[source]

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: Optional[Dict[str, Any]] = None, desc: str = '') → AnyPort[source]

Add an output Port to the System.

This function cannot be called outside System.setup.

Parameters

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

  • name [str] (-) – Port name

  • desc [str (-) – Port description

  • optional] – Port description

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

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

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

Returns

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: Union[str, Dict[str, Any]], value: Any = 1, unit: str = '', dtype: Optional[Union[Any, Tuple[Any, ...]]] = None, valid_range: Optional[Tuple[Any, Any]] = None, invalid_comment: str = '', limits: Optional[Tuple[Any, Any]] = None, out_of_limits_comment: str = '', desc: str = '', scope: cosapp.ports.enum.Scope = <Scope.PUBLIC: -1>) → None[source]

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: Optional[Any] = None, unit: str = '', dtype: Optional[Union[Any, Tuple[Any, ...]]] = None, desc: str = '', init: Optional[Any] = None, scope: cosapp.ports.enum.Scope = <Scope.PRIVATE: 1>) → None[source]

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[source]

Create new read-only property name, set to value

add_rate(name: str, source: Any, initial_value: Optional[Union[numbers.Number, numpy.ndarray, str]] = None, desc: Optional[str] = None) → None[source]

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: Union[str, Iterable[Union[dict, str, Tuple[str, str]]]], reference: Union[numbers.Number, numpy.ndarray, str] = 1, weak=False) → cosapp.core.numerics.basics.MathematicalProblem[source]

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: Optional[str] = None, max_time_step: Union[numbers.Number, str] = inf, max_abs_step: Union[numbers.Number, str] = inf) → None[source]

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: Union[str, Iterable[Union[dict, str]]], max_abs_step: numbers.Number = inf, max_rel_step: numbers.Number = inf, lower_bound: numbers.Number = - inf, upper_bound: numbers.Number = inf) → cosapp.core.numerics.basics.MathematicalProblem[source]

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[cosapp.ports.connectors.Connector][source]

Iterator yielding all connectors within system.

all_events() → Iterator[cosapp.multimode.event.Event][source]

Recursive iterator on all events in complete system tree.

any_active_driver() → bool[source]
append_name2variable(additional_mapping: Iterable[Tuple[str, cosapp.core.variableref.VariableReference]]) → None[source]

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() → cosapp.core.numerics.basics.MathematicalProblem[source]

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

Returns

The assembled mathematical problem.

Return type

MathematicalProblem

call_clean_run(skip_driver: bool = False) → None[source]

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[source]

Execute setup_run recursively on all modules.

Parameters

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

check(name: Optional[str] = None) → Union[Dict[str, cosapp.ports.enum.Validity], cosapp.ports.enum.Validity][source]

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[source]

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

Connectors between sub-systems, referenced by system names.

Type

Dict[str, List[Connector]]

close_loops()[source]

Close loops opened to allow system resolution.

connect(object1: Union[cosapp.ports.port.BasePort, cosapp.systems.system.System], object2: Union[cosapp.ports.port.BasePort, cosapp.systems.system.System], mapping: Optional[Union[str, List[str], Dict[str, str]]] = None, cls: Optional[Type[cosapp.ports.connectors.BaseConnector]] = None, **kwargs) → None[source]

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, cosapp.ports.connectors.Connector][source]

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

convert_to(*args, **kwargs) → None[source]

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

Raises

TypeError – If the current System cannot be converted.

design(method: str) → cosapp.core.numerics.basics.MathematicalProblem[source]

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
drivers
dump_surrogate(filename: str) → None[source]

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

events() → Iterator[cosapp.multimode.event.Event][source]

Iterator on all events locally defined on system.

export_structure() → Dict[source]

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() → cosapp.core.numerics.basics.MathematicalProblem[source]

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

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

Type

bool

incoming_connectors() → Iterator[cosapp.ports.connectors.Connector][source]

Iterator yielding all connectors targetting system.

property input_mapping

free input mapping

Type

Dict[str, VariableReference]

inputs
is_clean(direction: Optional[cosapp.ports.enum.PortType] = None) → bool[source]

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[source]

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

is_output_var(name: str) → bool[source]

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

is_running() → bool[source]

Is this System in execution?

Returns

In execution status

Return type

bool

is_standalone() → bool[source]

Is this System able to solve itself?

Returns

Ability to solve the system or not.

Return type

bool

classmethod load(filepath: Union[str, pathlib.Path, _io.StringIO], name: Optional[str] = None) → cosapp.systems.system.System[source]

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) → cosapp.systems.system.System[source]

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[source]

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: 0>) → bool[source]

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: Union[pandas.core.frame.DataFrame, Dict[str, List[float]]], model=<class 'cosapp.utils.surrogate_models.kriging.FloatKrigingSurrogate'>, activate=True, data_out: Optional[Union[pandas.core.frame.DataFrame, Dict[str, Any]]] = None, postsynch: Union[str, List[str]] = '*', *args, **kwargs) → cosapp.systems.systemSurrogate.SystemSurrogate[source]

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
new_problem(name='problem') → cosapp.core.numerics.basics.MathematicalProblem[source]

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()[source]

Open closed loops in children relations.

outputs
pop_child(name: str) → cosapp.systems.system.System[source]

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[source]

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[cosapp.ports.port.BasePort][source]

Iterator on all system ports (both inputs and outputs)

property problem

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

AttributeError unless system is being modified by a driver.

Type

MathematicalProblem

property properties

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]

run_children_drivers() → None[source]

Solve the children System in the execution order.

run_drivers() → None[source]

Run the drivers defined on this System.

run_once() → None[source]

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[source]

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: cosapp.ports.enum.PortType) → None[source]

Set to clean ports of a certain direction.

Parameters

direction (PortType) – Direction to set

set_dirty(direction: cosapp.ports.enum.PortType) → None[source]

Set to dirty ports of a certain direction.

Parameters

direction (PortType) – Direction to set

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

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[source]

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][source]

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[source]

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[source]

Public API to export system to a dictionary

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

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[source]

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()[source]
property transients

Returns a dictionary containing all transient unknowns in current system tree

transition() → None[source]

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()[source]
property unknowns

Get the unknowns for the current System.

Type

Dict[str, Unknown]

update()[source]

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.