fedbiomed.researcher.secagg

@property
def context(self) -> Union[dict, None]:
    """Getter for secagg context element content

    Returns:
        secagg context element, or `None` if it doesn't exist
    """
    return self._context

def delete(
        self,
        timeout: float = 0
) -> bool:
    """Delete secagg context element on defined parties.

    Args:
        timeout: maximum duration for the deletion phase. Defaults to `environ['TIMEOUT']` if unset
            or equals 0.

    Returns:
        True if secagg context element could be deleted for all parties, False if at least
            one of the parties could not delete context element.

    Raises:
        FedbiomedSecaggError: bad argument type
    """
    if isinstance(timeout, int):
        timeout = float(timeout)  # accept int (and bool...)
    try:
        self._v.validate(timeout, float)
    except ValidatorError as e:
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `timeout`: {e}'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    self._status = False
    self._context = None
    msg = {
        'researcher_id': self._researcher_id,
        'secagg_id': self._secagg_id,
        'element': self._element.value,
        'job_id': self._job_id,
        'command': 'secagg-delete',
    }
    return self._secagg_round(msg, 'secagg-delete', False, self._delete_payload, timeout)

@property
def job_id(self) -> Union[str, None]:
    """Getter for secagg context element job_id

    Returns:
        secagg context element job_ib (or None if no job_id is attached to the element)
    """
    return self._job_id

@staticmethod
def load_state(
        state: Dict[str, Any]
) -> 'SecaggContext':

    """
    Method for loading secagg state from breakpoint state

    Args:
        state: The state that will be loaded
    """

    # Get class
    cls = getattr(importlib.import_module(state["module"]), state["class"])

    # Validate job id
    spec = get_method_spec(cls)
    if 'job_id' in spec:
        secagg = cls(**state["arguments"])
    else:
        state["arguments"].pop('job_id')
        secagg = cls(**state["arguments"])

    for key, value in state["attributes"].items():
        setattr(secagg, key, value)

    return secagg

@property
def parties(self) -> str:
    """Getter for secagg parties

    Returns:
        Parties that participates secure aggregation
    """
    return self._parties

def save_state(self) -> Dict[str, Any]:
    """Method for saving secagg state for saving breakpoints

    Returns:
        The state of the secagg
    """
    # `_v` and `_requests` dont need to be savec (properly initiated in constructor)
    state = {
        "class": type(self).__name__,
        "module": self.__module__,
        "arguments": {
            "secagg_id": self._secagg_id,
            "parties": self._parties,
            "job_id": self._job_id,

        },
        "attributes": {
            "_status": self._status,
            "_context": self._context,
            "_researcher_id": self._researcher_id,
        }
    }
    return state

@property
def secagg_id(self) -> str:
    """Getter for secagg context element ID 

    Returns:
        secagg context element unique ID
    """
    return self._secagg_id

def set_job_id(self, job_id: Union[str, None]) -> None:
    """Setter for secagg context element job_id

    Args:
        job_id: ID of the job to which this secagg context element is attached.

    Raises:
        FedbiomedSecaggError: bad argument type or value
    """

    if not isinstance(job_id, (str, type(None))):
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `job_id` must be a str or None if the ' \
                  f'context is set for biprime.'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    self._job_id = job_id

def setup(
        self,
        timeout: float = 0
) -> bool:
    """Setup secagg context element on defined parties.

    Args:
        timeout: maximum time waiting for answers from other nodes, after completing the setup locally.
            It does not include the time for the local setup payload.
            Defaults to `environ['TIMEOUT']` if unset or equals 0.

    Returns:
        True if secagg context element could be setup for all parties, False if at least
            one of the parties could not setup context element.

    Raises:
        FedbiomedSecaggError: bad argument type
    """
    if isinstance(timeout, int):
        timeout = float(timeout)  # accept int (and bool...)
    try:
        self._v.validate(timeout, float)
    except ValidatorError as e:
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `timeout`: {e}'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    msg = {
        'researcher_id': self._researcher_id,
        'secagg_id': self._secagg_id,
        'element': self._element.value,
        'job_id': self._job_id,
        'parties': self._parties,
        'command': 'secagg',
    }

    return self._secagg_round(msg, 'secagg', True, self._payload, timeout)

@property
def status(self) -> bool:
    """Getter for secagg context element status

    Returns:
        `True` if secagg context element exists, `False` otherwise
    """
    return self._status

def activate(self, status) -> bool:
    """Set activate status of secure aggregation

    Returns:
        Status of secure aggregation True if it is activated
    """

    if not isinstance(status, bool):
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: The argument `status` for activation should be True or False, "
            f"but got {type(status)} "
        )

    self._active = status

    return self._active

@property
def active(self) -> bool:
    """Gets secagg activation status

    Returns:
        bool, True if secagg is activated
    """
    return self._active

def aggregate(
        self,
        round_: int,
        total_sample_size: int,
        model_params: Dict[str, List[int]],
        encryption_factors: Union[Dict[str, List[int]], None] = None,
) -> List[float]:
    """Aggregates given model parameters

    Args:
        round_: current training round number
        total_sample_size: sum of number of samples used by all nodes
        model_params: model parameters from the participating nodes
        encryption_factors: encryption factors from the participating nodes

    Returns:
        Aggregated parameters

    Raises:
        FedbiomedSecureAggregationError: secure aggregation context not properly configured
        FedbiomedSecureAggregationError: secure aggregation computation error
    """

    if self._biprime is None or self._servkey is None:
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: Can not aggregate parameters, one of Biprime or Servkey context is"
            f"not configured. Please setup secure aggregation before the aggregation.")

    if not self._biprime.status or not self._servkey.status:
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: Can not aggregate parameters, one of Biprime or Servkey context is"
            f"not set properly")

    biprime = self._biprime.context["context"]["biprime"]
    key = self._servkey.context["context"]["server_key"]

    num_nodes = len(model_params)

    aggregate = functools.partial(self._secagg_crypter.aggregate,
                                  current_round=round_,
                                  num_nodes=num_nodes,
                                  key=key,
                                  total_sample_size=total_sample_size,
                                  biprime=biprime,
                                  clipping_range=self.clipping_range)

    # Validate secure aggregation
    if self._secagg_random is not None:

        if encryption_factors is None:
            raise FedbiomedSecureAggregationError(
                f"{ErrorNumbers.FB417.value}: Secure aggregation random validation has been set but the encryption "
                f"factors are not provided. Please provide encrypted `secagg_random` values in different parties. "
                f"Or to not set/get `secagg_random()` before the aggregation.")

        logger.info("Validating secure aggregation results...")
        encryption_factors = [f for k, f in encryption_factors.items()]
        validation: List[float] = aggregate(params=encryption_factors)

        if len(validation) != 1 or not math.isclose(validation[0], self._secagg_random, abs_tol=0.03):
            raise FedbiomedSecureAggregationError(
                f"{ErrorNumbers.FB417.value}: Aggregation is failed due to incorrect decryption."
            )
        logger.info("Validation is completed.")

    elif encryption_factors is not None:
        logger.warning("Encryption factors are provided while secagg random is None. Please make sure secure "
                       "aggregation steps are applied correctly.")

    logger.info("Aggregating encrypted parameters. This process may take some time depending on model size.")
    # Aggregate parameters
    params = [p for _, p in model_params.items()]
    aggregated_params = aggregate(params=params)

    return aggregated_params

@property
def biprime(self) -> Union[None, SecaggBiprimeContext]:
    """Gets biprime object

    Returns:
        Biprime object, None if biprime is not setup
    """
    return self._biprime

@property
def job_id(self) -> Union[str, None]:
    """Gets secagg associated job_id

    Returns:
        str of associated job_id if it exists, or None
    """
    return self._job_id

@classmethod
def load_state(
        cls,
        state: Dict
) -> 'SecureAggregation':
    """Create a `SecureAggregation` object from a saved state

    Args:
        state: saved state to restore in the created object

    Returns:
        The created `SecureAggregation` object
    """

    secagg = cls(**state["arguments"])

    if state["attributes"]["_biprime"] is not None:
        state["attributes"]["_biprime"] = SecaggBiprimeContext. \
            load_state(state=state["attributes"]["_biprime"])

    if state["attributes"]["_servkey"] is not None:
        state["attributes"]["_servkey"] = SecaggServkeyContext. \
            load_state(state=state["attributes"]["_servkey"])

    # Set attributes
    for name, val in state["attributes"].items():
        setattr(secagg, name, val)

    return secagg

@property
def parties(self) -> Union[List[str], None]:
    """Gets secagg parties

    Returns:
        List of secagg parties if it exists, or None
    """
    return self._parties

def save_state(self) -> Dict[str, Any]:
    """Saves state of the secagg

    Returns:
        The secagg state to be saved
    """

    state = {
        "class": type(self).__name__,
        "module": self.__module__,
        "arguments": {
            'active': self._active,
            'timeout': self.timeout,
            'clipping_range': self.clipping_range,
        },
        "attributes": {
            "_biprime": self._biprime.save_state() if self._biprime is not None else None,
            "_servkey": self._servkey.save_state() if self._servkey is not None else None,
            "_job_id": self._job_id,
            "_parties": self._parties
        }
    }

    return state

@property
def servkey(self) -> Union[None, SecaggServkeyContext]:
    """Gets servkey object

    Returns:
        Servkey object, None if servkey is not setup
    """
    return self._servkey

def setup(self,
          parties: List[str],
          job_id: str,
          force: bool = False):
    """Setup secure aggregation instruments.

    Requires setting `parties` and `job_id` if they are not set in previous secagg
    setups. It is possible to execute without any argument if SecureAggregation
    has already `parties` and `job_id` defined. This feature provides researcher
    execute `secagg.setup()` if any connection issue

    Args:
        parties: Parties that participates secure aggregation
        job_id: The id of the job of experiment
        force: Forces secagg setup even context is already existing

    Returns:
        Status of setup

    Raises
        FedbiomedSecureAggregationError: Invalid argument type
    """

    if not isinstance(parties, list):
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: Expected argument `parties` list but got {type(parties)}"
        )

    if not isinstance(job_id, str):
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: Expected argument `job_id` string but got {type(parties)}"
        )

    self._configure_round(parties, job_id)

    if self._biprime is None or self._servkey is None:
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: server key or biprime contexts is not fully configured."
        )

    if not self._biprime.status or force:
        self._biprime.setup(timeout=self.timeout)

    if not self._servkey.status or force:
        self._servkey.setup(timeout=self.timeout)

    return True

def train_arguments(self) -> Dict:
    """Gets train arguments for secagg train request

    Returns:
        Arguments that is going tobe attached to the experiment.
    """
    return {'secagg_servkey_id': self._servkey.secagg_id if self._servkey is not None else None,
            'secagg_biprime_id': self._biprime.secagg_id if self._biprime is not None else None,
            'secagg_random': self._secagg_random,
            'secagg_clipping_range': self.clipping_range}