p2pfl.management.logger.logger module

P2PFL Logger.

Note

Not all is typed because the python logger is not typed (yep, is a TODO…).

class p2pfl.management.logger.logger.ColoredFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

Formatter that adds color to the log messages.

format(record)

Format the log record with color.

Parameters:

record – The log record.

exception p2pfl.management.logger.logger.NodeNotRegistered

Bases: Exception

Exception raised when a node is not registered.

class p2pfl.management.logger.logger.P2PFLogger(nodes=None, disable_locks=False)

Bases: object

Class that manages the node logging (not a singleton).

Parameters:

• p2pfl_web_services – The P2PFL Web Services to log and monitor the nodes remotely.

• nodes (dict[str, dict[str, Any]] | None)

• disable_locks (bool)

add_handler(handler)

Add a handler to the logger.

Parameters:

handler (Handler) – The logger handler.

Return type:

None

cleanup()

Cleanup the logger.

Return type:

None

connect(**kwargs)

Establish connection/setup for the logger.

This method should be overridden by loggers that require connection setup. By default, it does nothing.

Parameters:

**kwargs (Any) – Connection parameters specific to each logger type.

Return type:

None

critical(node, message)

Log a critical message.

Parameters:

• node (str) – The node name.

• message (str) – The message to log.

Return type:

None

debug(node, message)

Log a debug message.

Parameters:

• node (str) – The node name.

• message (str) – The message to log.

Return type:

None

error(node, message)

Log an error message.

Parameters:

• node (str) – The node name.

• message (str) – The message to log.

Return type:

None

experiment_finished(node)

Notify the experiment end.

Parameters:

node (str) – The node address.

Return type:

None

experiment_started(node, experiment)

Notify the experiment start.

Parameters:

• node (str) – The node address.

• experiment (Experiment) – The experiment.

Return type:

None

experiment_updated(node, experiment)

Notify the round end.

Parameters:

• node (str) – The node address.

• experiment (Experiment) – The experiment to update.

Return type:

None

finish()

Finish any logging activities, like closing a W&B run.

This method is a placeholder and is meant to be implemented by a decorator. By default, it does nothing.

Return type:

None

get_global_logs()

Get the logs.

Parameters:

• node – The node name.

• exp – The experiment name.

Return type:

dict[str, dict[str, dict[str, list[tuple[int, float]]]]]

Returns:

The logs.

get_level()

Get the logger level.

Return type:

int

Returns

The logger level.

get_level_name(lvl)

Get the logger level name.

Parameters:

lvl (int) – The logger level.

Return type:

str

Returns:

The logger level name.

get_local_logs()

Get the logs.

Parameters:

• node – The node name.

• exp – The experiment name.

Return type:

dict[str, dict[int, dict[str, dict[str, list[tuple[int, float]]]]]]

Returns:

The logs.

get_messages(direction='all', node=None, cmd=None, round_num=None, limit=None)

Get communication messages with optional filtering.

Parameters:

• direction (str) – Filter by message direction (“all”, “sent”, or “received”).

• node (str | None) – Filter by node address (optional).

• cmd (str | None) – Filter by command type (optional).

• round_num (int | None) – Filter by round number (optional).

• limit (int | None) – Limit the number of messages returned per node (optional).

Return type:

list[dict[str, Any]]

Returns:

A flat list of message dictionaries. Each message includes a ‘direction’ field indicating whether it was ‘sent’ or ‘received’.

get_nodes()

Get the registered nodes.

Return type:

dict[str, dict[Any, Any]]

Returns:

The registered nodes.

get_system_metrics()

Get the system metrics.

Return type:

dict[datetime, dict[str, float]]

Returns:

The system metrics.

info(node, message)

Log an info message.

Parameters:

• node (str) – The node name.

• message (str) – The message to log.

Return type:

None

log(level, node, message)

Log a message.

Parameters:

• level (int) – The log level.

• node (str) – The node name.

• message (str) – The message to log.

Return type:

None

log_communication(node, direction, cmd, source_dest, package_type, package_size, round_num=None, additional_info=None)

Log a communication event.

Parameters:

• node (str) – The node address.

• direction (str) – Direction of communication (“sent” or “received”).

• cmd (str) – The command or message type.

• source_dest (str) – Source (if receiving) or destination (if sending) node.

• package_type (str) – Type of package (“message” or “weights”).

• package_size (int) – Size of the package in bytes (if available).

• round_num (int | None) – The federated learning round number (if applicable).

• additional_info (dict[str, Any] | None) – Additional information as a dictionary.

Return type:

None

log_metric(addr, metric, value, step=None, round=None)

Log a metric.

Parameters:

• addr (str) – The node name.

• metric (str) – The metric to log.

• value (float) – The value.

• step (int | None) – The step.

• round (int | None) – The round.

Return type:

None

register_node(node)

Register a node.

Parameters:

node (str) – The node address.

Return type:

None

reset()

Reset the logger state between experiments.

This clears all stored metrics, messages, and system logs while keeping the logger configuration and handlers intact.

Return type:

None

set_level(level)

Set the logger level.

Parameters:

level (int | str) – The logger level.

Return type:

None

unregister_node(node)

Unregister a node.

Parameters:

node (str) – The node address.

Return type:

None

warning(node, message)

Log a warning message.

Parameters:

• node (str) – The node name.

• message (str) – The message to log.

Return type:

None