hermes.base.nodes¶

The module is responsible for wrapping common functionality of Consumer Nodes. It defines a unified interface to the ZeroMQ message exchange layer for all concrete Consumers.

consumer_interface - base interface for Consumer function.

consumer - abstract Consumer with common data ingestion logic to subclass.

consumer_interface ¶

Classes:

Name	Description
`ConsumerInterface`	Interface for the Consumer Node component.

ConsumerInterface ¶

Bases: NodeInterface

Interface for the Consumer Node component.

_is_done `abstractmethod` `property` ¶

_is_done: bool

Read-only property identifying if the Node completed operation.

Returns:

Name	Type	Description
`bool`	`bool`	Whether the Node completed its function.

_activate_data_poller `abstractmethod` ¶

_activate_data_poller() -> None

Start listening for new data from other Nodes.

_activate_kill_poller `abstractmethod` ¶

_activate_kill_poller() -> None

Start listening for the KILL signal on the special PUB/SUB socket that coordinates program termination.

_deactivate_kill_poller `abstractmethod` ¶

_deactivate_kill_poller() -> None

Stop listening for the KILL signal.

_get_sync_socket `abstractmethod` ¶

_get_sync_socket() -> SyncSocket

Get reference to the socket used for synchronization of the Node to its master Broker.

Returns:

Type	Description
`SyncSocket`	zmq.SyncSocket: ZeroMQ socket of the Node connected to the local master Broker.

_initialize `abstractmethod` ¶

_initialize() -> None

Node-specific initialization procedure.

Pre-run setup of the backend specific to the Node implementaiton. Generic setup should be run first.

_log_source_tag `abstractmethod` `classmethod` ¶

_log_source_tag() -> str

Read-only property uniquely identifying the Node.

Returns:

Name	Type	Description
`str`	`str`	Unique key identifying the Node in the data exchange.

_on_poll `abstractmethod` ¶

_on_poll(poll_res: tuple[list[SyncSocket], list[int]]) -> None

Callback to perform some logic everytime some data transactions are received by the Poller.

Generic entry-point for all types of Nodes, based on their active Poller settings. NOTE: if Node in JoinState, kill socket is no longer in the Poller and only higher-level logic is triggered.

Parameters:

Name	Type	Description	Default
`poll_res` ¶	`tuple[list[SyncSocket], list[int]]`	Reference to the complete captured result of listening by the Poller.	required

_on_sync_complete `abstractmethod` ¶

_on_sync_complete() -> None

Callback to perform some logic after synchronization of Nodes is completed and indicated by the Broker.

_poll `abstractmethod` ¶

_poll() -> tuple[list[SyncSocket], list[int]]

Block for new ZeroMQ data to collect at the Poller.

Listens for events when new data is received from or when new data can be written to sockets, based on the active Poller settings of the Node implementation.

Returns:

Type	Description
`tuple[list[SyncSocket], list[int]]`	tuple[list[zmq.SyncSocket], list[int]]: Result of listening on the sockets registered by the Poller.

_send_kill_to_broker `abstractmethod` ¶

_send_kill_to_broker() -> None

Send a slave KILL signal to the local Broker in case program termination by the slave Node is recorded.

_set_state `abstractmethod` ¶

_set_state(state) -> None

User-defined logic for FSM state transition.

Parameters:

Name	Type	Description	Default
`state` ¶	`Any`	New state to transition to.	required

_trigger_stop `abstractmethod` ¶

_trigger_stop() -> None

Trigger to the Node's internal procedures and background threads to gracefully wrap-up.

Producer: stops sampling data, continue sending already captured until none is left, with last message labeled 'END'. Consumer: continues listening to data until each of subscribed Producers sent the last message. Pipeline: continues listening to data to produce results until each data sources sent the last message, and then labels the last message with 'END'.

consumer ¶

Classes:

Name	Description
`Consumer`	An abstract class to interface with a particular data consumer.

Consumer ¶

Bases: ConsumerInterface, Node

An abstract class to interface with a particular data consumer.

Subscribes to the modalities specified in and parametrized by stream_in_specs.

Methods:

Name	Description
`__call__`	Node objects are callable to start the FSM as entry-point.
`__init__`	Constructor of the Consumer parent class.

call ¶

__call__()

Node objects are callable to start the FSM as entry-point.

init ¶

__init__(
    host_ip: str,
    stream_in_specs: list[dict],
    logging_spec: dict,
    port_sub: str = PORT_FRONTEND,
    port_sync: str = PORT_SYNC_HOST,
    port_killsig: str = PORT_KILL,
    log_history_filepath: str | None = None,
) -> None

Parameters:

Name	Type	Description	Default
`host_ip` ¶	`str`	IP address of the local master Broker.	required
`stream_in_specs` ¶	`list[dict]`	List of mappings of user-configured incoming modalities.	required
`logging_spec` ¶	`dict`	Mapping of Storage object parameters to user-defined configuration values.	required
`port_sub` ¶	`str`	Local port to subscribe to for incoming relayed data from the local master Broker. Defaults to PORT_FRONTEND.	`PORT_FRONTEND`
`port_sync` ¶	`str`	Local port to listen to for local master Broker's startup coordination. Defaults to PORT_SYNC_HOST.	`PORT_SYNC_HOST`
`port_killsig` ¶	`str`	Local port to listen to for local master Broker's termination signal. Defaults to PORT_KILL.	`PORT_KILL`
`log_history_filepath` ¶	`str \| None`	File path to the system log file. Defaults to None.	`None`

_log_source_tag `abstractmethod` `classmethod` ¶

_log_source_tag() -> str

Read-only property uniquely identifying the Node.

Returns:

Name	Type	Description
`str`	`str`	Unique key identifying the Node in the data exchange.

_poll_data_packets ¶

_poll_data_packets() -> None

Receive data packets in a steady state.

Gets called every time one of the requestes modalities produced new data. In normal operation mode, all messages are 2-part.

_poll_ending_data_packets ¶

_poll_ending_data_packets() -> None

Receive data packets from producers and monitor for end-of-stream signal.

When system triggered a safe exit, Pipeline gets a mix of normal 2-part messages and 3-part 'END' message from each Producer that safely exited. It's more efficient to dynamically switch the callback instead of checking every message.

Processes packets on each modality until all data sources sent the 'END' packet. If triggered to stop and no more available data, sends empty 'END' packet and joins.

hermes.base.nodes¶

consumer_interface ¶

ConsumerInterface ¶

_is_done `abstractmethod` `property` ¶

_activate_data_poller `abstractmethod` ¶

_activate_kill_poller `abstractmethod` ¶

_deactivate_kill_poller `abstractmethod` ¶

_get_sync_socket `abstractmethod` ¶

_initialize `abstractmethod` ¶

_log_source_tag `abstractmethod` `classmethod` ¶

_on_poll `abstractmethod` ¶

`poll_res` ¶

_on_sync_complete `abstractmethod` ¶

_poll `abstractmethod` ¶

_send_kill_to_broker `abstractmethod` ¶

_set_state `abstractmethod` ¶

`state` ¶

_trigger_stop `abstractmethod` ¶

consumer ¶

Consumer ¶

call ¶

init ¶

`host_ip` ¶

`stream_in_specs` ¶

`logging_spec` ¶

`port_sub` ¶

`port_sync` ¶

`port_killsig` ¶

`log_history_filepath` ¶

_log_source_tag `abstractmethod` `classmethod` ¶

_poll_data_packets ¶

_poll_ending_data_packets ¶

hermes.base.nodes¶

consumer_interface ¶

ConsumerInterface ¶

_is_done abstractmethod property ¶

_activate_data_poller abstractmethod ¶

_activate_kill_poller abstractmethod ¶

_deactivate_kill_poller abstractmethod ¶

_get_sync_socket abstractmethod ¶

_initialize abstractmethod ¶

_log_source_tag abstractmethod classmethod ¶

_on_poll abstractmethod ¶

poll_res ¶

_on_sync_complete abstractmethod ¶

_poll abstractmethod ¶

_send_kill_to_broker abstractmethod ¶

_set_state abstractmethod ¶

state ¶

_trigger_stop abstractmethod ¶

consumer ¶

Consumer ¶

__call__ ¶

__init__ ¶

host_ip ¶

stream_in_specs ¶

logging_spec ¶

port_sub ¶

port_sync ¶

port_killsig ¶

log_history_filepath ¶

_log_source_tag abstractmethod classmethod ¶

_poll_data_packets ¶

_poll_ending_data_packets ¶

_is_done `abstractmethod` `property` ¶

_activate_data_poller `abstractmethod` ¶

_activate_kill_poller `abstractmethod` ¶

_deactivate_kill_poller `abstractmethod` ¶

_get_sync_socket `abstractmethod` ¶

_initialize `abstractmethod` ¶

_log_source_tag `abstractmethod` `classmethod` ¶

_on_poll `abstractmethod` ¶

`poll_res` ¶

_on_sync_complete `abstractmethod` ¶

_poll `abstractmethod` ¶

_send_kill_to_broker `abstractmethod` ¶

_set_state `abstractmethod` ¶

`state` ¶

_trigger_stop `abstractmethod` ¶

call ¶

init ¶

`host_ip` ¶

`stream_in_specs` ¶

`logging_spec` ¶

`port_sub` ¶

`port_sync` ¶

`port_killsig` ¶

`log_history_filepath` ¶

_log_source_tag `abstractmethod` `classmethod` ¶