cosapp.ports.connectors

Classes connecting Port of foreign System to transfer variable values.

Functions

MakeDirectConnector(classname[, transform])

Connector factory using a simple transfer function, with no unit conversion.

Classes

BaseConnector(name, sink, source[, mapping])	This class connect two ports without enforcing that all port variables are connected.
Connector(name, sink, source[, mapping])	Shallow copy connector.

Exceptions

ConnectorError

Raised if a connector cannot be created between two Port.

class cosapp.ports.connectors.BaseConnector(name: str, sink: BasePort, source: BasePort, mapping: str | list[str] | dict[str, str] | None = None)

Bases: ABC

This class connect two ports without enforcing that all port variables are connected.

The link is oriented from the source to the sink.

Parameters:

• name (str) – Name of the connector

• source (BasePort) – Port from which originate the variables

• mapping (str or list[str] or dict[str, str]) – (list of) common name(s) or mapping name dictionary

• sink (BasePort) – Port to which the variables are transferred

contextual_name(context: System | None = None, with_mapping=True) → str

Contextual name of the connector, of the kind system[source -> sink].

static format_mapping(mapping: str | Collection[str] | dict[str, str], /) → dict[str, str]

Returns suitable name mapping for connectors, from different kinds of argument mapping.

Parameters:

• mapping [str | list[str] | dict[str, str] | None]:

  Name mapping, given as either a string (single variable), a collection of strings, or a full name mapping, as a dictionary.

Returns:

dict[str, str]: name mapping suitable for connectors.

info() → tuple[str, str] | tuple[str, str, dict[str, str]]

Returns connector information in a tuple.

If the name mapping is complete, with identical names, it is omitted, and the output tuple is formatted as: - (target_name, source_name)

Otherwise, output is formatted as: - (target_name, source_name, name_mapping)

Returns:

tuple representing connector

Return type:

tuple

is_mirror() → bool

Returns True if connector is an identical, one-to-one mapping between two ports of the same kind; False otherwise.

property mapping: dict[str, str]

Variable name mapping between the sink (key) and the source (value).

Type:

dict[str, str]

property name: str

name of the connector.

Type:

str

port_names() → tuple[str, str]

Returns source and sink contextual names as a str tuple.

Returns:

(source_name, sink_name) tuple

Return type:

tuple

preserves_names() → bool

Returns True if connector mapping preserves variable names, False otherwise.

pretty_mapping() → str

Pretty formatting of the variable name mapping applied by the connector.

remove_variables(names: Iterable[str]) → None

Remove the provided variables from this connection.

The provided names should be sink names.

Parameters:

names (Iterable[str]) – Collection of sink variable names to be removed.

property sink: BasePort

Port to which values are transferred.

Type:

BasePort

sink_variable(source_variable: str) → str

Returns the name of the sink variable associated to source_variable

sink_variables() → Iterator[str]

property source: BasePort

Port from which transferred values originate.

Type:

BasePort

source_variable(sink_variable: str) → str

Returns the name of the source variable associated to sink_variable

source_variables() → Iterator[str]

to_dict() → dict[str, tuple[str, str] | tuple[str, str, dict[str, str]]]

Converts connector into a single-key dictionary. The key is the connector name; associated value is the tuple returned by method info().

Returns:

Dictionary {name: info_tuple} representing connector

Return type:

dict

abstract transfer() → None

Transfer values from source to sink.

update_mapping(mapping: dict[str, str]) → None

Extend current mapping with additional dictionary.

Parameters:

mapping (dict[str, str]) – Variable name mapping extending current mapping.

class cosapp.ports.connectors.Connector(name: str, sink: BasePort, source: BasePort, mapping: str | list[str] | dict[str, str] | None = None)

Bases: BaseConnector

Shallow copy connector. See BaseConnector for base class details.

remove_variables(names: Iterable[str]) → None

Remove the provided variables from this connection.

The provided names should be sink names.

Parameters:

names (Iterable[str]) – Collection of sink variable names to be removed.

property sink: BasePort

Port to which values are transferred.

Type:

BasePort

property source: BasePort

Port from which transferred values originate.

Type:

BasePort

transfer() → None

Transfer values from source to sink.

update_mapping(mapping: dict[str, str]) → None

Extend current mapping with additional dictionary.

Parameters:

mapping (dict[str, str]) – Variable name mapping extending current mapping.

update_unit_conversion(name: str | None = None) → None

Update the physical unit conversion on the connector.

If name is not None, update the conversion only for the connexion towards that variable.

Parameters:

name (str, optional) – Name of the variable for which unit conversion needs an update; default None (i.e. all conversions will be updated).

Raises:

UnitError – If unit conversion from source to sink is not possible

exception cosapp.ports.connectors.ConnectorError

Bases: Exception

Raised if a connector cannot be created between two Port.

cosapp.ports.connectors.MakeDirectConnector(classname: str, transform: Callable | None = None, **kwargs) → type[BaseConnector]

Connector factory using a simple transfer function, with no unit conversion.