libmuscle.port_manager module
- class libmuscle.port_manager.PortManager(index: List[int], declared_ports: Dict[Operator, List[str]] | None)[source]
Bases:
objectManages sending and receiving ports of the current instance.
- connect_ports(peer_info: PeerInfo) None[source]
Connect the ports to their peers.
This is the second stage in the simulation wiring process.
Peers here are instances, and the information about them received from the manager is in peer_info. We are going to create a set of Port objects here, one for each of our ports.
If the user gave us a set of ports (i.e. they were declared in the code) then we’ll create a Port object for each of those, with information about the attached conduit / peer (if any) from peer_info. If the user did not give us any ports (which is legal), then the ports will be created entirely from the information received from the manager. The user then has to use Instance.list_ports() to see what ports they got, and do something with them.
- Parameters:
peer_info – Information about our peers from the manager.
- get_message_counts() Dict[str, List[int]][source]
Get message counts for all ports on the communicator.
- Returns:
A dictionary indexed by port name containing a list of counts, one for each slot of the corresponding port.
- get_port(port_name: str) Port[source]
Returns a Port object describing a port with the given name.
- Parameters:
port_name – Name of the port to retrieve.
- Returns:
A Port object for the port
- list_ports() Dict[Operator, List[str]][source]
Returns a description of the ports this PortManager has.
- Returns:
A dictionary, indexed by Operator, containing lists of port names. Operators with no associated ports are not included.
- port_exists(port_name: str) bool[source]
Returns whether a port with the given name exists.
- Parameters:
port_name – Port name to check.
- Returns:
True iff the port exists