modalities.logging_broker.subscriber_impl package

Submodules

modalities.logging_broker.subscriber_impl.progress_subscriber module

class modalities.logging_broker.subscriber_impl.progress_subscriber.DummyProgressSubscriber

Bases: MessageSubscriberIF[ProgressUpdate]

consume_dict(message_dict)

Parameters:

message_dict (dict[str, Any])

consume_message(message)

Parameters:

message (Message[ProgressUpdate])

class modalities.logging_broker.subscriber_impl.progress_subscriber.RichProgressSubscriber(train_split_num_steps, eval_splits_num_steps)

Bases: MessageSubscriberIF[ProgressUpdate]

A subscriber object for the RichProgress observable.

Parameters:

• train_split_num_steps (dict[str, tuple[int, int]])

• eval_splits_num_steps (dict[str, int])

consume_dict(message_dict)

Parameters:

message_dict (dict[str, Any])

consume_message(message)

Consumes a message from a message broker.

Parameters:

message (Message[ProgressUpdate])

classmethod register_live_display(live_display)

Only one instance of rich.live.Live can run at the same time. Therefore we use a singleton approach to have only one active,

by storing the active reference as class-field here.

Parameters:

live_display (Live)

modalities.logging_broker.subscriber_impl.results_subscriber module

class modalities.logging_broker.subscriber_impl.results_subscriber.DummyResultSubscriber

Bases: MessageSubscriberIF[EvaluationResultBatch]

consume_dict(message_dict)

Parameters:

message_dict (dict[str, Any])

consume_message(message)

Consumes a message from a message broker.

Parameters:

message (Message[EvaluationResultBatch])

class modalities.logging_broker.subscriber_impl.results_subscriber.EvaluationResultToDiscSubscriber(output_file_path)

Bases: MessageSubscriberIF[EvaluationResultBatch]

A subscriber that writes EvaluationResultBatch messages to a JSONL file.

Parameters:

output_file_path (Path)

consume_dict(message_dict)

Optional: log config data if needed (here: no-op).

Parameters:

message_dict (dict[str, Any])

consume_message(message)

Writes the evaluation result to the JSONL file if rank 0.

Parameters:

message (Message[EvaluationResultBatch])

class modalities.logging_broker.subscriber_impl.results_subscriber.RichResultSubscriber(num_ranks)

Bases: MessageSubscriberIF[EvaluationResultBatch]

Parameters:

num_ranks (int)

consume_dict(message_dict)

Parameters:

message_dict (dict[str, Any])

consume_message(message)

Consumes a message from a message broker.

Parameters:

message (Message[EvaluationResultBatch])

class modalities.logging_broker.subscriber_impl.results_subscriber.WandBEvaluationResultSubscriber(project, experiment_id, mode, logging_directory, config_file_path)

Bases: MessageSubscriberIF[EvaluationResultBatch]

A subscriber object for the WandBEvaluationResult observable.

Parameters:

• project (str)

• experiment_id (str)

• mode (WandbMode)

• logging_directory (Path)

• config_file_path (Path)

consume_dict(message_dict)

Parameters:

message_dict (dict[str, Any])

consume_message(message)

Consumes a message from a message broker.

Parameters:

message (Message[EvaluationResultBatch])

modalities.logging_broker.subscriber_impl.subscriber_factory module

class modalities.logging_broker.subscriber_impl.subscriber_factory.ProgressSubscriberFactory

Bases: object

static get_dummy_progress_subscriber()

Return type:

DummyProgressSubscriber

static get_rich_progress_subscriber(eval_dataloaders, train_dataloader_tag, num_seen_steps, num_target_steps, global_rank)

Return type:

RichProgressSubscriber

Parameters:

• eval_dataloaders (list[LLMDataLoader])

• train_dataloader_tag (str)

• num_seen_steps (int)

• num_target_steps (int)

• global_rank (int)

class modalities.logging_broker.subscriber_impl.subscriber_factory.ResultsSubscriberFactory

Bases: object

static get_dummy_result_subscriber()

Return type:

DummyResultSubscriber

static get_evaluation_result_to_disc_subscriber(output_file_path)

Return type:

EvaluationResultToDiscSubscriber

Parameters:

output_file_path (Path)

static get_rich_result_subscriber(num_ranks, global_rank)

Return type:

RichResultSubscriber

Parameters:

• num_ranks (int)

• global_rank (int)

static get_wandb_result_subscriber(global_rank, project, experiment_id, mode, config_file_path, directory=None)

Return type:

WandBEvaluationResultSubscriber

Parameters:

• global_rank (int)

• project (str)

• experiment_id (str)

• mode (WandbMode)

• config_file_path (Path)

• directory (Path | None)

Module contents