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(message)

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

static format_mapping(mapping: str | List[str] | Dict[str, str]) → Dict[str, str]

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

Parameters:

• mapping, Union[str, List[str], Dict[str, str]]:

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

Returns:

Dict[str, str]: suitable mapping 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.

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(message: str)

Bases: Exception

Raised if a connector cannot be created between two Port.

message

Error message

Type:

str

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

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