fl_sim.utils

This module contains various utilities for the fl-sim package.

fl_sim.utils.loggers

This module contains various loggers.

class fl_sim.utils.loggers.BaseLogger

Bases: ReprMixin, ABC

Abstract base class of all loggers.

abstract close() → None

Close the logger.

epoch_end(epoch: int) → None

Actions to be performed at the end of each epoch.

Parameters:

epoch (int) – The number of the current epoch.

Return type:

None

epoch_start(epoch: int) → None

Actions to be performed at the start of each epoch.

Parameters:

epoch (int) – The number of the current epoch.

Return type:

None

extra_repr_keys() → List[str]

Extra keys for __repr__() and __str__().

abstract property filename: str

Name of the log file.

abstract flush() → None

Flush the message buffer.

abstract classmethod from_config(config: Dict[str, Any]) → Any

Create a logger instance from a configuration.

property log_dir: str

Directory to save the log file.

abstract log_message(msg: str, level: int = 20) → None

Log a message.

Parameters:

• msg (str) – The message to be logged.

• level (int, optional) – The level of the message, can be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL

Return type:

None

abstract log_metrics(client_id: int | None, metrics: Dict[str, Real | Tensor], step: int | None = None, epoch: int | None = None, part: str = 'val') → None

Log metrics.

Parameters:

• client_id (int) – Index of the client, None for the server.

• metrics (dict) – The metrics to be logged.

• step (int, optional) – The current number of (global) steps of training.

• epoch (int, optional) – The current epoch number of training.

• part (str, default "val") – The part of the training data the metrics computed from, can be "train" or "val" or "test", etc.

Return type:

None

abstract reset() → None

Reset the logger.

static set_log_dir(log_dir: Path | str | None = None) → Path

Set the log directory.

Parameters:

log_dir (str or pathlib.Path) – The log directory.

Return type:

None

class fl_sim.utils.loggers.CSVLogger(algorithm: str, dataset: str, model: str, log_dir: Path | str | None = None, log_suffix: str | None = None, verbose: int = 1)

Bases: BaseLogger

Logger that logs to a CSV file.

Parameters:

• algorithm (str) – Used to form the prefix of the log file.

• dataset (str) – Used to form the prefix of the log file.

• model (str) – Used to form the prefix of the log file.

• log_dir (str or pathlib.Path, optional) – Directory to save the log file

• log_suffix (str, optional) – Suffix of the log file.

• verbose (int, default 1) – The verbosity level. Not used in this logger, but is kept for compatibility with other loggers.

close() → None

Close the logger.

property filename: str

Name of the log file.

flush() → None

Flush the message buffer.

classmethod from_config(config: Dict[str, Any]) → CSVLogger

Create a CSVLogger instance from a configuration.

Parameters:

config (dict) –

Configuration for the logger. The following keys are used:

• "algorithm": str, name of the algorithm.

• "dataset": str, name of the dataset.

• "model": str, name of the model.

• "log_dir": str or pathlib.Path, optional, directory to save the log file.

• "log_suffix": str, optional, suffix of the log file.

Returns:

A CSVLogger instance.

Return type:

CSVLogger

log_message(msg: str, level: int = 20) → None

Log a message.

Parameters:

• msg (str) – The message to be logged.

• level (int, optional) – The level of the message, can be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL

Return type:

None

log_metrics(client_id: int | None, metrics: Dict[str, Real | Tensor], step: int | None = None, epoch: int | None = None, part: str = 'val') → None

Log metrics.

Parameters:

• client_id (int) – Index of the client, None for the server.

• metrics (dict) – The metrics to be logged.

• step (int, optional) – The current number of (global) steps of training.

• epoch (int, optional) – The current epoch number of training.

• part (str, default "val") – The part of the training data the metrics computed from, can be "train" or "val" or "test", etc.

Return type:

None

reset() → None

Reset the logger.

Close the current logger and create a new one, with new log file name.

class fl_sim.utils.loggers.JsonLogger(algorithm: str, dataset: str, model: str, fmt: str = 'json', log_dir: Path | str | None = None, log_suffix: str | None = None, verbose: int = 1)

Bases: BaseLogger

Logger that logs to a JSON file, or a yaml file.

The structure is as follows for example:

{
    "train": {
        "client0": [
            {
                "epoch": 1,
                "step": 1,
                "time": "2020-01-01 00:00:00",
                "loss": 0.1,
                "acc": 0.2,
                "top3_acc": 0.3,
                "top5_acc": 0.4,
                "num_samples": 100
            }
        ]
    },
    "val": {
        "client0": [
            {
                "epoch": 1,
                "step": 1,
                "time": "2020-01-01 00:00:00",
                "loss": 0.1,
                "acc": 0.2,
                "top3_acc": 0.3,
                "top5_acc": 0.4,
                "num_samples": 100
            }
        ]
    }
}

Parameters:

• algorithm (str) – Used to form the prefix of the log file.

• dataset (str) – Used to form the prefix of the log file.

• model (str) – Used to form the prefix of the log file.

• fmt ({"json", "yaml"}, optional) – Format of the log file.

• log_dir (str or pathlib.Path, optional) – Directory to save the log file

• log_suffix (str, optional) – Suffix of the log file.

• verbose (int, default 1) – The verbosity level. Not used in this logger, but is kept for compatibility with other loggers.

close() → None

Close the logger.

property filename: str

Name of the log file.

flush() → None

Flush the message buffer.

classmethod from_config(config: Dict[str, Any]) → JsonLogger

Create a JsonLogger instance from a configuration.

Parameters:

config (dict) –

Configuration for the logger. The following keys are used:

• "algorithm": str, name of the algorithm.

• "dataset": str, name of the dataset.

• "model": str, name of the model.

• "fmt": {“json”, “yaml”}, optional, format of the log file, default: "json".

• "log_dir": str or pathlib.Path, optional, directory to save the log file.

• "log_suffix": str, optional, suffix of the log file.

Returns:

A JsonLogger instance.

Return type:

JsonLogger

log_message(msg: str, level: int = 20) → None

Log a message.

Parameters:

• msg (str) – The message to be logged.

• level (int, optional) – The level of the message, can be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL

Return type:

None

log_metrics(client_id: int | None, metrics: Dict[str, Real | Tensor], step: int | None = None, epoch: int | None = None, part: str = 'val') → None

Log metrics.

Parameters:

• client_id (int) – Index of the client, None for the server.

• metrics (dict) – The metrics to be logged.

• step (int, optional) – The current number of (global) steps of training.

• epoch (int, optional) – The current epoch number of training.

• part (str, default "val") – The part of the training data the metrics computed from, can be "train" or "val" or "test", etc.

Return type:

None

reset() → None

Reset the logger.

Close the current logger and create a new one, with new log file name.

static strftime(time: datetime) → str

static strptime(time: str) → datetime

class fl_sim.utils.loggers.LoggerManager(algorithm: str, dataset: str, model: str, log_dir: Path | str | None = None, log_suffix: str | None = None, verbose: int = 1)

Bases: ReprMixin

Manager for loggers.

Parameters:

• algorithm (str) – Used to form the prefix of the log file.

• dataset (str) – Used to form the prefix of the log file.

• model (str) – Used to form the prefix of the log file.

• log_dir (str or pathlib.Path, optional) – Directory to save the log file

• log_suffix (str, optional) – Suffix of the log file.

• verbose (int, default 1) – The verbosity level.

close() → None

Close the logger.

epoch_end(epoch: int) → None

Actions to be performed at the end of each epoch.

Parameters:

epoch (int) – The number of the current epoch.

Return type:

None

epoch_start(epoch: int) → None

Actions to be performed at the start of each epoch.

Parameters:

epoch (int) – The number of the current epoch.

Return type:

None

extra_repr_keys() → List[str]

Extra keys for __repr__() and __str__().

flush() → None

Flush the message buffer.

classmethod from_config(config: Dict[str, Any]) → LoggerManager

Create a LoggerManager instance from a configuration.

Parameters:

config (dict) –

Configuration of the logger manager. The following keys are used:

• "algorithm": str, algorithm name.

• "dataset": str, dataset name.

• "model": str, model name.

• "log_dir": str or pathlib.Path, optional, directory to save the log files.

• "log_suffix": str, optional, suffix of the log files.

• "txt_logger": bool, optional, whether to add a TxtLogger instance.

• "csv_logger": bool, optional, whether to add a CSVLogger instance.

• "json_logger": bool, optional, whether to add a JsonLogger instance.

• "fmt": {“json”, “yaml”}, optional, format of the json log file, default: "json", valid when "json_logger" is True.

• "verbose": int, optional, verbosity level of the logger manager.

Returns:

A LoggerManager instance.

Return type:

LoggerManager

property log_dir: str

Directory to save the log files.

log_message(msg: str, level: int = 20) → None

Log a message.

Parameters:

• msg (str) – The message to be logged.

• level (int, optional) – The level of the message, can be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL

Return type:

None

log_metrics(client_id: int | None, metrics: Dict[str, Real | Tensor], step: int | None = None, epoch: int | None = None, part: str = 'val') → None

Log a message.

Parameters:

• msg (str) – The message to be logged.

• level (int, optional) – The level of the message, can be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL

Return type:

None

property log_suffix: str

Suffix of the log files.

property loggers: List[BaseLogger]

The list of loggers.

reset() → None

Reset the logger.

class fl_sim.utils.loggers.TxtLogger(algorithm: str, dataset: str, model: str, log_dir: Path | str | None = None, log_suffix: str | None = None, verbose: int = 1)

Bases: BaseLogger

Logger that logs to a text file.

Parameters:

• algorithm (str) – Used to form the prefix of the log file.

• dataset (str) – Used to form the prefix of the log file.

• model (str) – Used to form the prefix of the log file.

• log_dir (str or pathlib.Path, optional) – Directory to save the log file. If None, use the default log directory. If not absolute, use DEFAULT_LOG_DIR/log_dir.

• log_suffix (str, optional) – Suffix of the log file.

• verbose (int, default 1) – The verbosity level.

close() → None

Close the logger.

epoch_end(epoch: int) → None

Actions to be performed at the end of each epoch.

Parameters:

epoch (int) – The number of the current epoch.

Return type:

None

epoch_start(epoch: int) → None

Actions to be performed at the start of each epoch.

Parameters:

epoch (int) – The number of the current epoch.

Return type:

None

property filename: str

Name of the log file.

flush() → None

Flush the message buffer.

classmethod from_config(config: Dict[str, Any]) → TxtLogger

Create a TxtLogger instance from a configuration.

Parameters:

config (dict) –

Configuration for the logger. The following keys are used:

• "algorithm": str, name of the algorithm.

• "dataset": str, name of the dataset.

• "model": str, name of the model.

• "log_dir": str or pathlib.Path, optional, directory to save the log file.

• "log_suffix": str, optional, suffix of the log file.

Returns:

A TxtLogger instance.

Return type:

TxtLogger

log_message(msg: str, level: int = 20) → None

Log a message.

Parameters:

• msg (str) – The message to be logged.

• level (int, optional) – The level of the message, can be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL

Return type:

None

log_metrics(client_id: int | None, metrics: Dict[str, Real | Tensor], step: int | None = None, epoch: int | None = None, part: str = 'val') → None

Log metrics.

Parameters:

• client_id (int) – Index of the client, None for the server.

• metrics (dict) – The metrics to be logged.

• step (int, optional) – The current number of (global) steps of training.

• epoch (int, optional) – The current epoch number of training.

• part (str, default "val") – The part of the training data the metrics computed from, can be "train" or "val" or "test", etc.

Return type:

None

property long_sep: str

Long separator for logging messages.

reset() → None

Reset the logger.

Close the current logger and create a new one, with new log file name.

property short_sep: str

Short separator for logging messages.

fl_sim.utils.imports

This module contains utilities for dynamic imports.

fl_sim.utils.imports.load_module_from_file(file_path: str | Path) → module

Load a module from a file.

Parameters:

file_path (str or pathlib.Path) – The path of the file.

Returns:

The loaded module.

Return type:

types.ModuleType

fl_sim.utils.misc

This module provides miscellaneous utilities.

fl_sim.utils.misc.add_kwargs(func: callable, **kwargs: Any) → callable

Add keyword arguments to a function.

This function is used to add keyword arguments to a function in order to make it compatible with other functions。

Parameters:

• func (callable) – The function to be decorated.

• kwargs (dict) – The keyword arguments to be added.

Returns:

The decorated function, with the keyword arguments added.

Return type:

callable

fl_sim.utils.misc.clear_logs(pattern: str = '*', directory: Path | str | None = None) → None

Clear given log files in given directory.

Parameters:

• pattern (str, optional) – Pattern of log files to be cleared, by default “*” The searching will be executed by pathlib.Path.rglob().

• directory (str or Path, optional) – Directory to be searched, by default the default log directory LOG_DIR will be used. If the given directory is not absolute, it will be joined with LOG_DIR.

fl_sim.utils.misc.default_dict_to_dict(d: defaultdict | dict | list | tuple) → dict | list

Convert default dict to dict.

Parameters:

d (Union[defaultdict, dict, list, tuple]) – Input default dict, or dict, list, tuple containing default dicts.

Returns:

Converted dict.

Return type:

Union[dict, list]

fl_sim.utils.misc.experiment_indicator(name: str) → Callable

fl_sim.utils.misc.find_longest_common_substring(strings: Sequence[str], min_len: int | None = None, ignore: str | None = None) → str

Find the longest common substring of a list of strings.

Parameters:

• strings (sequence of str) – The list of strings.

• min_len (int, optional) – The minimum length of the common substring.

• ignore (str, optional) – The substring to be ignored.

Returns:

The longest common substring.

Return type:

str

fl_sim.utils.misc.get_scheduler(scheduler_name: str, optimizer: Optimizer, config: dict | None) → _LRScheduler

Get learning rate scheduler.

Parameters:

• scheduler_name (str) – Name of the scheduler.

• optimizer (torch.optim.Optimizer) – Optimizer.

• config (dict) – Configuration of the scheduler.

Returns:

Learning rate scheduler.

Return type:

torch.optim.lr_scheduler._LRScheduler

fl_sim.utils.misc.get_scheduler_info(scheduler_name: str) → dict

Get information of the scheduler, including the required and optional configs.

fl_sim.utils.misc.is_notebook() → bool

Check if the current environment is a notebook (Jupyter or Colab).

Implementation adapted from [1].

Parameters:

None –

Returns:

Whether the code is running in a notebook

Return type:

bool

References

[1]

https://stackoverflow.com/questions/15411967/how-can-i-check-if-code-is-executed-in-the-ipython-notebook

fl_sim.utils.misc.make_serializable(x: ndarray | generic | dict | list | tuple) → list | dict | Number

Make an object serializable.

This function is used to convert all numpy arrays to list in an object, and also convert numpy data types to python data types in the object, so that it can be serialized by json.

Parameters:

x (Union[numpy.ndarray, numpy.generic, dict, list, tuple]) – Input data, which can be numpy array (or numpy data type), or dict, list, tuple containing numpy arrays (or numpy data type).

Returns:

Converted data.

Return type:

Union[list, dict, numbers.Number]

Examples

>>> import numpy as np
>>> from fl_sim.utils.misc import make_serializable
>>> x = np.array([1, 2, 3])
>>> make_serializable(x)
[1, 2, 3]
>>> x = {"a": np.array([1, 2, 3]), "b": np.array([4, 5, 6])}
>>> make_serializable(x)
{'a': [1, 2, 3], 'b': [4, 5, 6]}
>>> x = [np.array([1, 2, 3]), np.array([4, 5, 6])]
>>> make_serializable(x)
[[1, 2, 3], [4, 5, 6]]
>>> x = (np.array([1, 2, 3]), np.array([4, 5, 6]).mean())
>>> obj = make_serializable(x)
>>> obj
[[1, 2, 3], 5.0]
>>> type(obj[1]), type(x[1])
(float, numpy.float64)

fl_sim.utils.misc.ordered_dict_to_dict(d: OrderedDict | dict | list | tuple) → dict | list

Convert ordered dict to dict.

Parameters:

d (Union[OrderedDict, dict, list, tuple]) – Input ordered dict, or dict, list, tuple containing ordered dicts.

Returns:

Converted dict.

Return type:

Union[dict, list]

fl_sim.utils.misc.set_seed(seed: int) → None

Set random seed for numpy and pytorch, as well as disable cudnn to ensure reproducibility.

Parameters:

seed (int) – Random seed.

Return type:

None