Source code for fedn.utils.helpers.plugins.numpyhelper


import numpy as np

from fedn.utils.helpers.helperbase import HelperBase


[docs] class Helper(HelperBase): """ FEDn helper class for models weights/parameters that can be transformed to numpy ndarrays. """ def __init__(self): """ Initialize helper. """ super().__init__() self.name = "numpyhelper"
[docs] def increment_average(self, m1, m2, n, N): """ Update a weighted incremental average of model weights. :param m1: Current parameters. :type model: list of numpy ndarray :param m2: next parameters. :type model_next: list of numpy ndarray :param n: Number of examples used for updating m2. :type n: int :param N: Total number of examples (accumulated). :type N: int :return: Updated incremental weighted average. :rtype: list of numpy ndarray """ return [np.add(x, n*(y-x)/N) for x, y in zip(m1, m2)]
[docs] def add(self, m1, m2, a=1.0, b=1.0): """ m1*a + m2*b :param model: Current model weights. :type model: list of ndarrays :param model_next: New model weights. :type model_next: list of ndarrays :return: Incremental weighted average of model weights. :rtype: list of ndarrays """ return [x*a+y*b for x, y in zip(m1, m2)]
[docs] def subtract(self, m1, m2, a=1.0, b=1.0): """ m1*a - m2*b. :param m1: Current model weights. :type m1: list of ndarrays :param m2: New model weights. :type m2: list of ndarrays :return: m1*a-m2*b :rtype: list of ndarrays """ return self.add(m1, m2, a, -b)
[docs] def divide(self, m1, m2): """ Subtract weights. :param m1: Current model weights. :type m1: list of ndarrays :param m2: New model weights. :type m2: list of ndarrays :return: m1/m2. :rtype: list of ndarrays """ return [np.divide(x, y) for x, y in zip(m1, m2)]
[docs] def multiply(self, m1, m2): """ Multiply m1 by m2. :param m1: Current model weights. :type m1: list of ndarrays :param m2: New model weights. :type m2: list of ndarrays :return: m1.*m2 :rtype: list of ndarrays """ return [np.multiply(x, y) for (x, y) in zip(m1, m2)]
[docs] def sqrt(self, m1): """ Sqrt of m1, element-wise. :param m1: Current model weights. :type model: list of ndarrays :param model_next: New model weights. :type model_next: list of ndarrays :return: sqrt(m1) :rtype: list of ndarrays """ return [np.sqrt(x) for x in m1]
[docs] def power(self, m1, a): """ m1 raised to the power of m2. :param m1: Current model weights. :type m1: list of ndarrays :param m2: New model weights. :type a: float :return: m1.^m2 :rtype: list of ndarrays """ return [np.power(x, a) for x in m1]
[docs] def norm(self, m): """ Return the norm (L1) of model weights. :param m: Current model weights. :type m: list of ndarrays :return: norm of m :rtype: float """ n = 0.0 for x in m: n += np.linalg.norm(x, 1) return n
[docs] def ones(self, m1, a): """ Return a list of numpy arrays of the same shape as m1, filled with ones. :param m1: Current model weights. :type m1: list of ndarrays :param a: Scalar value. :type a: float :return: list of numpy arrays of the same shape as m1, filled with ones. :rtype: list of ndarrays """ res = [] for x in m1: res.append(np.ones(np.shape(x))*a) return res
[docs] def save(self, weights, path=None): """ Serialize weights to file. The serialized model must be a single binary object. :param weights: List of weights in numpy format. :param path: Path to file. :return: Path to file. """ if not path: path = self.get_tmp_path() weights_dict = {} for i, w in enumerate(weights): weights_dict[str(i)] = w np.savez_compressed(path, **weights_dict) return path
[docs] def load(self, fh): """ Load weights from file or filelike. :param fh: file path, filehandle, filelike. :return: List of weights in numpy format. """ a = np.load(fh) weights = [] for i in range(len(a.files)): weights.append(a[str(i)]) return weights