modalities.logging_broker package

Subpackages

• modalities.logging_broker.subscriber_impl package
  • Submodules
  • modalities.logging_broker.subscriber_impl.progress_subscriber module
  • modalities.logging_broker.subscriber_impl.results_subscriber module
  • modalities.logging_broker.subscriber_impl.subscriber_factory module
  • Module contents

Submodules

modalities.logging_broker.message_broker module

class modalities.logging_broker.message_broker.MessageBroker

Bases: MessageBrokerIF

The MessageBroker sends notifications to its subscribers.

add_subscriber(subscription, subscriber)

Adds a single subscriber.

Parameters:

• subscription (MessageTypes)

• subscriber (MessageSubscriberIF)

distribute_message(message)

Distributes message to all subscribers.

Parameters:

message (Message)

class modalities.logging_broker.message_broker.MessageBrokerIF

Bases: ABC

Interface for message broker objects.

abstract add_subscriber(subscription, subscriber)

Parameters:

• subscription (MessageTypes)

• subscriber (MessageSubscriberIF)

abstract distribute_message(message)

Parameters:

message (Message)

modalities.logging_broker.messages module

class modalities.logging_broker.messages.ExperimentStatus(value)

Bases: Enum

An enumeration.

EVALUATION = 'EVALUATION'

TRAIN = 'TRAIN'

class modalities.logging_broker.messages.Message(message_type, payload, global_rank=0, local_rank=0)

Bases: Generic[T]

An object representing a message.

Parameters:

• message_type (MessageTypes)

• payload (T)

• global_rank (int)

• local_rank (int)

global_rank: int = 0

local_rank: int = 0

message_type: MessageTypes

payload: TypeVar(T)

class modalities.logging_broker.messages.MessageTypes(value)

Bases: Enum

An enumeration.

BATCH_PROGRESS_UPDATE = 'PROGRESS_UPDATE'

ERROR_MESSAGE = 'ERROR_MESSAGE'

EVALUATION_RESULT = 'EVALUATION_RESULT'

HIGH_LEVEL_PROGRESS_UPDATE = 'HIGH_LEVEL_PROGRESS_UPDATE'

class modalities.logging_broker.messages.ProgressUpdate(num_steps_done, experiment_status, dataloader_tag)

Bases: object

Object holding the state of the current batch / step computation progress.

Parameters:

• num_steps_done (int)

• experiment_status (ExperimentStatus)

• dataloader_tag (str)

dataloader_tag: str

experiment_status: ExperimentStatus

num_steps_done: int

modalities.logging_broker.publisher module

class modalities.logging_broker.publisher.MessagePublisher(message_broker, global_rank, local_rank)

Bases: MessagePublisherIF[T]

The MessagePublisher sends messages through a message broker.

Parameters:

• message_broker (MessageBroker)

• global_rank (int)

• local_rank (int)

publish_message(payload, message_type)

Publish a message through the message broker.

Parameters:

• payload (T)

• message_type (MessageTypes)

class modalities.logging_broker.publisher.MessagePublisherIF

Bases: ABC, Generic[T]

abstract publish_message(payload, message_type)

Parameters:

• payload (T)

• message_type (MessageTypes)

modalities.logging_broker.subscriber module

class modalities.logging_broker.subscriber.MessageSubscriberIF

Bases: ABC, Generic[T]

Interface for message subscribers.

abstract consume_dict(mesasge_dict)

Parameters:

mesasge_dict (dict[str, Any])

abstract consume_message(message)

Parameters:

message (Message[T])

Module contents