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: cosapp.ports.port.BasePort, source: cosapp.ports.port.BasePort, mapping: Optional[Union[str, List[str], Dict[str, str]]] = None)[source]

Bases: abc.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: Union[str, List[str], Dict[str, str]]) → Dict[str, str][source]

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

  • 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.

Dict[str, str]: suitable mapping for connectors.

info() → Union[Tuple[str, str], Tuple[str, str, Dict[str, str]]][source]

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

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

property mapping

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

Type

Dict[str, str]

property name

name of the connector.

Type

str

port_names() → Tuple[str, str][source]

Returns source and sink contextual names as a str tuple.

Returns

(source_name, sink_name) tuple

Return type

tuple

preserves_names() → bool[source]

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

remove_variables(names: Iterable[str]) → None[source]

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

Port to which values are transferred.

Type

BasePort

sink_variable(source_variable: str) → str[source]

Returns the name of the sink variable associated to source_variable

sink_variables() → Iterator[str][source]
property source

Port from which transferred values originate.

Type

BasePort

source_variable(sink_variable: str) → str[source]

Returns the name of the source variable associated to sink_variable

source_variables() → Iterator[str][source]
to_dict() → Dict[str, Union[Tuple[str, str], Tuple[str, str, Dict[str, str]]]][source]

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

Transfer values from source to sink.

update_mapping(mapping: Dict[str, str]) → None[source]

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: cosapp.ports.port.BasePort, source: cosapp.ports.port.BasePort, mapping: Optional[Union[str, List[str], Dict[str, str]]] = None)[source]

Bases: cosapp.ports.connectors.BaseConnector

Shallow copy connector. See BaseConnector for base class details.

remove_variables(names: Iterable[str]) → None[source]

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

Port to which values are transferred.

Type

BasePort

property source

Port from which transferred values originate.

Type

BasePort

transfer() → None[source]

Transfer values from source to sink.

update_mapping(mapping: Dict[str, str]) → None[source]

Extend current mapping with additional dictionary.

Parameters

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

update_unit_conversion(name: Optional[str] = None) → None[source]

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

Bases: Exception

Raised if a connector cannot be created between two Port.

message

Error message

Type

str

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

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