Secagg

def __init__(self, parties: List[str], secagg_id: Union[str, None] = None):
    """Constructor of the class.

    Args:
        parties: list of parties participating to the secagg context element setup, named
            by their unique id (`node_id`, `researcher_id`).
            There must be at least 3 parties, and the first party is this researcher
        secagg_id: optional secagg context element ID to use for this element.
            Default is None, which means a unique element ID will be generated.

    Raises:
        FedbiomedSecaggError: bad argument type or value
    """
    super().__init__(parties, None, secagg_id)

    self._element = SecaggElementTypes.BIPRIME
    self._secagg_manager = _BPrimeManager

def __init__(self, parties: List[str], job_id: Union[str, None], secagg_id: Union[str, None] = None):
    """Constructor of the class.

    Args:
        parties: list of parties participating in the secagg context element setup, named
            by their unique id (`node_id`, `researcher_id`).
            There must be at least 3 parties, and the first party is this researcher
        job_id: ID of the job to which this secagg context element is attached.
            None means the element is not attached to a specific job
        secagg_id: optional secagg context element ID to use for this element.
            Default is None, which means a unique element ID will be generated.

    Raises:
        FedbiomedSecaggError: bad argument type or value
    """
    self._v = Validator()

    self._v.register("nonempty_str_or_none", self._check_secagg_id_type, override=True)
    try:
        self._v.validate(secagg_id, "nonempty_str_or_none")
    except ValidatorError as e:
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `secagg_id` must be a None or non-empty string: {e}'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    try:
        self._v.validate(parties, list)
        for p in parties:
            self._v.validate(p, str)
    except ValidatorError as e:
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `parties` must be a list of strings: {e}'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    if len(parties) < 3:
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `parties` : {parties} : need  ' \
                  'at least 3 parties for secure aggregation'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    if environ['ID'] != parties[0]:
        raise FedbiomedSecaggError(
            f'{ErrorNumbers.FB415.value}: researcher should be the first party.'
        )

    self._secagg_id = secagg_id if secagg_id is not None else 'secagg_' + str(uuid.uuid4())
    self._parties = parties
    self._researcher_id = environ['ID']
    self._requests = Requests()
    self._status = False
    self._context = None
    self._job_id = None

    # set job ID using setter to validate
    self.set_job_id(job_id)

    # one controller per secagg object to prevent any file conflict
    self._MPC = MPCController(
        tmp_dir=environ["TMP_DIR"],
        component_type=ComponentType.RESEARCHER,
        component_id=environ["ID"]
    )

    # to be set in subclasses
    self._secagg_manager = None

def __init__(self, parties: List[str], job_id: str, secagg_id: Union[str, None] = None):
    """Constructor of the class.

    Args:
        parties: list of parties participating in the secagg context element setup, named
            by their unique id (`node_id`, `researcher_id`).
            There must be at least 3 parties, and the first party is this researcher
        job_id: ID of the job to which this secagg context element is attached.
        secagg_id: optional secagg context element ID to use for this element.
            Default is None, which means a unique element ID will be generated.

    Raises:
        FedbiomedSecaggError: bad argument type or value
    """
    super().__init__(parties, job_id, secagg_id)

    if not self._job_id:
        errmess = f'{ErrorNumbers.FB415.value}: bad parameter `job_id` must be non empty string'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    self._element = SecaggElementTypes.SERVER_KEY
    self._secagg_manager = _SKManager

def __init__(
        self,
        active: bool = True,
        clipping_range: Union[None, int] = None,
) -> None:
    """Class constructor

    Assigns default values for attributes

    Args:
        active: True if secure aggregation is activated for the experiment
        clipping_range: Clipping range that will be used for quantization of model
            parameters on the node side. The default will be
            [`VEParameters.CLIPPING_RANGE`][fedbiomed.common.constants.VEParameters].
            The default value will be automatically set on the node side.

    Raises:
        FedbiomedSecureAggregationError: bad argument type
    """

    if not isinstance(active, bool):
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: The argument `active` should be  bool of type, "
            f"but got {type(active)} "
        )

    if clipping_range is not None and \
            (not isinstance(clipping_range, int) or isinstance(clipping_range, bool)):
        raise FedbiomedSecureAggregationError(
            f"{ErrorNumbers.FB417.value}: Clipping range should be None or an integer, "
            f"but got not {type(clipping_range)}"
        )

    self.clipping_range: Optional[int] = clipping_range

    self._active: bool = active
    self._parties: Optional[List[str]] = None
    self._job_id: Optional[str] = None
    self._servkey: Optional[SecaggServkeyContext] = None
    self._biprime: Optional[SecaggBiprimeContext] = None
    self._secagg_random: Optional[float] = None
    self._secagg_crypter: SecaggCrypter = SecaggCrypter()