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

基类：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.

参数:

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

返回类型:

None

epoch_start(epoch: int) → None

Actions to be performed at the start of each epoch.

参数:

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

返回类型:

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.

参数:

• 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

返回类型:

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.

参数:

• 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.

返回类型:

None

abstract reset() → None

Reset the logger.

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

Set the log directory.

参数:

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

返回类型:

None

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

基类：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
            }
        ]
    }
}

参数:

• 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.

参数:

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.

返回:

A JsonLogger instance.

返回类型:

JsonLogger

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

Log a message.

参数:

• 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

返回类型:

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.

参数:

• 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.

返回类型:

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: str | Path | None = None, log_suffix: str | None = None, verbose: int = 1, silent: bool = False)

基类：ReprMixin

Manager for loggers.

参数:

• 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.

参数:

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

返回类型:

None

epoch_start(epoch: int) → None

Actions to be performed at the start of each epoch.

参数:

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

返回类型:

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.

参数:

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.

• "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,

• "silent": bool, optional, whether to suppress the output to the console.

返回:

A LoggerManager instance.

返回类型:

LoggerManager

property log_dir: str

Directory to save the log files.

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

Log a message.

参数:

• 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

返回类型:

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.

参数:

• 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

返回类型:

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: str | Path | None = None, log_suffix: str | None = None, verbose: int = 1, silent: bool = False)

基类：BaseLogger

Logger that logs to a text file.

参数:

• 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.

参数:

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

返回类型:

None

epoch_start(epoch: int) → None

Actions to be performed at the start of each epoch.

参数:

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

返回类型:

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.

参数:

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.

返回:

A TxtLogger instance.

返回类型:

TxtLogger

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

Log a message.

参数:

• 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

返回类型:

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.

参数:

• 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.

返回类型:

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) → ModuleType

Load a module from a file.

参数:

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

返回:

The loaded module.

返回类型:

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。

参数:

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

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

返回:

The decorated function, with the keyword arguments added.

返回类型:

callable

fl_sim.utils.misc.aggregate_results_from_json_log(d: dict | str | Path, part: str = 'val', metric: str = 'acc') → ndarray

Aggregate the federated results from json log.

参数:

• d (dict or str or pathlib.Path) -- The dict of the json/yaml log, or the path to the json/yaml log file.

• part (str, default "train") -- The part of the log to aggregate.

• metric (str, default "acc") -- The metric to aggregate.

返回:

The aggregated results (curve).

返回类型:

numpy.ndarray

备注

The parameter d should be a dict similar to the following structure:

{
    "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
            }
        ]
    }
}

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

Clear given log files in given directory.

参数:

• 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.compute_clustering_metrics(y_true: ndarray, y_pred: ndarray) → Dict[str, float]

Compute clustering metrics, including ARI, NMI, and clustering accuracy.

参数:

• y_true (np.ndarray) -- True cluster labels of the data points.

• y_pred (np.ndarray) -- Predicted cluster labels of the data points.

返回:

metrics -- A dictionary containing the computed clustering metrics.

返回类型:

Dict[str, float]

fl_sim.utils.misc.compute_sparsity(model_or_tensor: Module | Tensor) → float

Compute the sparsity of a model or tensor.

参数:

model_or_tensor (torch.nn.Module or torch.Tensor) -- A model or tensor.

返回:

The sparsity of the model or tensor.

返回类型:

float

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

Convert default dict to dict.

参数:

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

返回:

Converted dict.

返回类型:

Union[dict, list]

fl_sim.utils.misc.execute_cmd(cmd: str | List[str], raise_error: bool = True) → Tuple[int, List[str]]

Execute shell command using Popen.

参数:

• cmd (str or list of str) -- Shell command to be executed, or a list of .sh files to be executed.

• raise_error (bool, default True) -- If True, error will be raised when occured.

返回:

• exitcode (int) -- Exit code returned by Popen.

• output_msg (list of str) -- Outputs from stdout of Popen.

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.

参数:

• 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.

返回:

The longest common substring.

返回类型:

str

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

Get learning rate scheduler.

参数:

• scheduler_name (str) -- Name of the scheduler.

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

• config (dict) -- Configuration of the scheduler.

返回:

Learning rate scheduler.

返回类型:

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].

参数:

None

返回:

Whether the code is running in a notebook

返回类型:

bool

引用

[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.

参数:

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).

返回:

Converted data.

返回类型:

Union[list, dict, numbers.Number]

示例

>>> 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.

参数:

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

返回:

Converted dict.

返回类型:

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.

参数:

seed (int) -- Random seed.

返回类型:

None