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[source]#

Bases: ReprMixin, ABC

Abstract base class of all loggers.

abstract close() → None[source]#

Close the logger.

epoch_end(epoch: int) → None[source]#

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[source]#

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][source]#

Extra keys for __repr__() and __str__().

abstract property filename: str#

Name of the log file.

abstract flush() → None[source]#

Flush the message buffer.

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

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[source]#

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[source]#

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[source]#

Reset the logger.

static set_log_dir(log_dir: Path | str | None = None) → Path[source]#

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)[source]#

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[source]#

Close the logger.

property filename: str#

Name of the log file.

flush() → None[source]#

Flush the message buffer.

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

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[source]#

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[source]#

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[source]#

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)[source]#

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[source]#

Close the logger.

property filename: str#

Name of the log file.

flush() → None[source]#

Flush the message buffer.

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

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[source]#

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[source]#

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[source]#

Reset the logger.

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

static strftime(time: datetime) → str[source]#
static strptime(time: str) → datetime[source]#
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)[source]#

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[source]#

Close the logger.

epoch_end(epoch: int) → None[source]#

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[source]#

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][source]#

Extra keys for __repr__() and __str__().

flush() → None[source]#

Flush the message buffer.

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

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[source]#

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[source]#

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[source]#

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)[source]#

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[source]#

Close the logger.

epoch_end(epoch: int) → None[source]#

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[source]#

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[source]#

Flush the message buffer.

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

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[source]#

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[source]#

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[source]#

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[source]#

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[source]#

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[source]#

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[source]#

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[source]#
fl_sim.utils.misc.find_longest_common_substring(strings: Sequence[str], min_len: int | None = None, ignore: str | None = None) → str[source]#

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[source]#

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[source]#

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

fl_sim.utils.misc.is_notebook() → bool[source]#

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

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

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[source]#

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[source]#

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

Parameters:

seed (int) – Random seed.

Return type:

None