declearn.data_info.aggregate_data_info

Combine individual 'data_info' dict into a single one.

Parameters:

Name	Type	Description	Default
clients_data_info	List[Dict[str, Any]]	List of client-wise data specifications.	required
required_fields	Optional[Set[str]]	Optional set of fields to target among provided information. If set, raise if a field is missing from any client, and use only these fields in the returned dict.	None

Returns:

Name	Type	Description
data_info	dict[str, any]	Aggregated data specifications derived from individual ones. Fields are either required_fields or the intersection of the individual dicts' fields. Values are the result of DataInfoField.combine(...) called on individual values, provided a DataInfoField subclass is associated with the field's name.

Raises:

Type	Description
KeyError	If a field in required_fields is missing from at least one dict included in clients_data_info.
ValueError	If any value of a shared field is invalid, or if values from a field are incompatible for combination.

Warns:

Type	Description
UserWarning	If one of the return fields has no corresponding registered DataInfoField specification class. In that case, return the list of individual values in lieu of aggregated value.

Notes

See declearn.data_info.register_data_info_field for details on how to register a DataInfoField subclass. See the latter (also part of declearn.data_info) for the field specification API.

Source code in declearn/data_info/_base.py

def aggregate_data_info(
    clients_data_info: List[Dict[str, Any]],
    required_fields: Optional[Set[str]] = None,
) -> Dict[str, Any]:
    """Combine individual 'data_info' dict into a single one.

    Parameters
    ----------
    clients_data_info: list[dict[str, any]]
        List of client-wise data specifications.
    required_fields: set[str] or None, default=None
        Optional set of fields to target among provided information.
        If set, raise if a field is missing from any client, and use
        only these fields in the returned dict.

    Returns
    -------
    data_info: dict[str, any]
        Aggregated data specifications derived from individual ones.
        Fields are either `required_fields` or the intersection of
        the individual dicts' fields.
        Values are the result of `DataInfoField.combine(...)` called
        on individual values, provided a `DataInfoField` subclass is
        associated with the field's name.

    Raises
    ------
    KeyError
        If a field in `required_fields` is missing from at least one
        dict included in `clients_data_info`.
    ValueError
        If any value of a shared field is invalid, or if values from
        a field are incompatible for combination.

    Warns
    -----
    UserWarning
        If one of the return fields has no corresponding registered
        `DataInfoField` specification class.
        In that case, return the list of individual values in lieu
        of aggregated value.

    Notes
    -----
    See `declearn.data_info.register_data_info_field` for details on
    how to register a `DataInfoField` subclass. See the latter (also
    part of `declearn.data_info`) for the field specification API.
    """
    # Select shared fields across clients, or required ones.
    fields = set.intersection(*[set(info) for info in clients_data_info])
    if required_fields is not None:
        missing = set(required_fields).difference(fields)
        if missing:
            raise KeyError(
                f"Missing required fields in at least one dict: {missing}"
            )
        fields = required_fields
    # Gather and spec-based-aggregate individual values.
    data_info = {}  # type: Dict[str, Any]
    for field in fields:
        values = [info[field] for info in clients_data_info]
        spec = DATA_INFO_FIELDS.get(field)
        if spec is None:
            warnings.warn(
                f"Unspecified 'data_info' field '{field}': "
                "returning list of individual values."
            )
            data_info[field] = values
        else:
            data_info[field] = spec.combine(*values)
    # Return aggregated information.
    return data_info