API

anonymization_manager.config.AnonymizationConfig

Bases: BaseModel

Configuration object for the anonymization workflow.

Attributes:

Name	Type	Description
data	str	Path to the input dataset. Supported formats include CSV, Excel, JSON, and SQLite (.db) files.
identifiers	list[str]	List of direct identifiers (e.g., name, SSN, phone number).
quasi_identifiers	list[str]	List of quasi-identifying attributes requiring generalization (e.g., age, zipcode, occupation).
sensitive_attributes	list[str]	Attributes considered sensitive (e.g., disease, salary) If not empty, either l-diversity or t-closeness must be specified.
insensitive_attributes	list[str]	Attributes that are neither identifiers nor sensitive and are carried through unchanged.
hierarchies	dict[str, str]	Mapping from quasi-identifiers to CSV hierarchy files.
k	int	k value for k-anonymity. Must be positive integer.
l	int	l value for l-diversity. Must be positive integer.
t	float	t value for t-closeness. Must be a float in [0,1].
suppression_limit	float	Maximum percentage of suppressed rows allowed (0-100%). Must be a float in [0,1].
backend	str	Anonymization backend to use, either 'arx' or 'anjana'. Defaults to 'arx'.
quality_metric	dict[Any]	A dictionary holding the information related to the quality metric. For more information, check the documentation.
attribute_weights	dict[str, float]	A set assigning weight "importance" to each attribute.

Source code in src/anonymization_manager/config.py

class AnonymizationConfig(BaseModel):
    """
    Configuration object for the anonymization workflow.

    Attributes:
        data (str):
            Path to the input dataset. Supported formats include CSV, Excel,
            JSON, and SQLite (.db) files.

        identifiers (list[str], optional):
            List of direct identifiers (e.g., name, SSN, phone number).

        quasi_identifiers (list[str], optional):
            List of quasi-identifying attributes requiring generalization
            (e.g., age, zipcode, occupation).

        sensitive_attributes (list[str], optional):
            Attributes considered sensitive (e.g., disease, salary)
            If not empty, either l-diversity or t-closeness must be specified.

        insensitive_attributes (list[str], optional):
            Attributes that are neither identifiers nor sensitive and are carried through unchanged.

        hierarchies (dict[str, str]):
            Mapping from quasi-identifiers to CSV hierarchy files.

        k (int, optional):
            k value for k-anonymity.
            Must be positive integer.

        l (int, optional):
            l value for l-diversity.
            Must be positive integer.

        t (float, optional):
            t value for t-closeness.
            Must be a float in [0,1].

        suppression_limit (float, optional):
            Maximum percentage of suppressed rows allowed (0-100%). Must be a float in [0,1].

        backend (str, optional):
            Anonymization backend to use, either 'arx' or 'anjana'.
            Defaults to 'arx'.

        quality_metric (dict[Any], optional):
            A dictionary holding the information related to the quality metric. For
            more information, check the documentation.
        attribute_weights (dict[str, float], optional):
            A set assigning weight "importance" to each attribute.
    """
    data: str
    identifiers: Optional[List[str]] = Field(default_factory=list)
    quasi_identifiers: Optional[List[str]] = Field(default_factory=list)
    sensitive_attributes: Optional[List[str]] = Field(default_factory=list)
    insensitive_attributes: Optional[List[str]] = Field(default_factory=list)
    hierarchies: Optional[Dict[str, str]] = Field(default_factory=dict)
    k: Optional[int] = Field(None, gt=0, description="k must be an integer > 0!")
    l: Optional[int] = Field(None, gt=0, description="l must be an integer > 0!")
    t: Optional[float] = Field(None, ge=0.0, le=1.0, description="t must be a float in [0,1]!")
    quality_metric: Optional[MetricConfig] = Field(None)
    suppression_limit: Optional[float] = Field(None, ge=0.0, le=1.0)
    backend: Optional[BackendType] = "arx"
    attribute_weights: Optional[Dict[str, Annotated[float, Field(ge=0)]]] = None

    @classmethod
    def from_json(cls, json_path: str):
        """
        Constructs an AnonymizationConfig from a JSON file.

        Args:
            json_path (str): Path to the JSON configuration file.

        Returns:
            AnonymizationConfig: The constructed and validated configuration object.
        """
        with open(json_path, "r") as file:
            config_json = json.load(file)

        attributes = {
            key: config_json[key]
            for key in cls.__annotations__
            if key in config_json
        }
        return cls(**attributes)

    @model_validator(mode="after")
    def _validate_attributes(self) -> "AnonymizationConfig":
        """
        Validates all the attribute lists.

        Checks:
            - Attribute names are unique across identifiers, quasi-identifiers,
            sensitive attributes, and insensitive attributes

        Raises:
            ValueError: If attribute names overlap across categories.
        """
        attr_list = {
            "identifiers": self.identifiers,
            "quasi_identifiers": self.quasi_identifiers,
            "sensitive_attributes": self.sensitive_attributes,
            "insensitive_attributes": self.insensitive_attributes,
        }
        # --- Checks that the attribute names do not overlap.
        all_attrs = sum(attr_list.values(), [])
        if len(all_attrs) != len(set(all_attrs)):
            raise ValueError(
                f"Attribute names must be unique across all types!"
            )

        return self

    @field_validator("data")
    @classmethod
    def _validate_dataset(cls, path: str) -> str:
        """
        Validates the dataset path.

        Checks:
            - Dataset file exists at the given path

        Raises:
            FileNotFoundError: If the file does not exist at the given path.
        """
        # --- Checks that the dataset file exists.
        if not os.path.exists(path):
            raise FileNotFoundError(
                f"The dataset could not be located at {path!r}!"
            )
        return path

    @model_validator(mode="after")
    def _validate_hierarchies(self) -> "AnonymizationConfig":
        """
        Validates the hierarchies provided for the quasi-identifiers.

        Checks:
            - Each quasi-identifier exists in `quasi_identifiers`
            - Each hierarchy file exists at the specified path

        Raises:
            ValueError: If a key is not a quasi-identifier.
            FileNotFoundError: If any hierarchy file cannot be located at the given path.
        """
        # --- Checks if the hierarchies are valid ---
        for qid, hierarchy_path in self.hierarchies.items():
            # --- Checks that the quasi-identifier exists ---
            if qid not in self.quasi_identifiers:
                raise ValueError(
                    f"Cannot create hierarchy for {qid!r}, since it is not a quasi-identifier!"
                )

            # --- Checks that the hierarchy path exists.
            if not os.path.exists(hierarchy_path):
                raise FileNotFoundError(
                    f"Cannot create hierarchy for {qid!r}, the path {hierarchy_path!r} could not be located!"
                )

        return self

    @model_validator(mode="after")
    def _validate_sensitive_attributes_and_l_t(self) -> "AnonymizationConfig":
        """
        Validates the relationship between sensitive attribute and l-diversity/t-closeness.

        Checks:
            - If `sensitive_attributes` is non-empty, `l` or `t` must be specified.
            - If `l` or `t` is specified, `sensitive_attributes` must not be empty.

        Raises:
            ValueError: If sensitive attributes exist without `l` or `t`, or if
            `l`/`t` is specified without sensitive attributes.
        """
        has_sensitive = bool(self.sensitive_attributes)
        has_l_t = self.l is not None or self.t is not None

        if has_sensitive and not has_l_t:
            raise ValueError(
                "`l` or `t` must be specified when anonymizing with sensitive attributes!"
            )

        if not has_sensitive and has_l_t:
            raise ValueError(
                "If `l` or `t` is specified, `sensitive_attributes` must not be empty!"
            )

        return self

    @model_validator(mode="after")
    def _validate_privacy_model_presence(self) -> "AnonymizationConfig":
        """
        Validates that at least one privacy model is specified.

        Checks:
            - At least one of `k`, `l`, or `t` is provided.

        Raises:
            ValueError: If `k`, `l`, and `t` are all set to None.
        """
        if self.k is None and self.l is None and self.t is None:
            raise ValueError(
                "At least one of `k`, `l` or `t` must be specified!"
            )

        return self

    @model_validator(mode="after")
    def _validate_backend_compatibility(self) -> "AnonymizationConfig":
        """
            Validates that ARX-only parameters are not used with the Anjana backend.

            Checks that `quality_metric` and `attribute_weights` are only used when
            the ARX backend is selected.

            Raises:
                ValueError: If anjana is used with the `quality_metric` or `attribute_weights`.
        """
        if self.backend == "anjana":
            if self.quality_metric is not None:
                raise ValueError(
                    "Anjana does not support `quality_metric` as a parameter!"
                )
            if self.attribute_weights is not None:
                raise ValueError(
                    "Anjana does not support `attribute_weights` as a parameter!"
                )

        return self

from_json(json_path) classmethod

Constructs an AnonymizationConfig from a JSON file.

Parameters:

Name	Type	Description	Default
json_path	str	Path to the JSON configuration file.	required

Returns:

Name	Type	Description
AnonymizationConfig		The constructed and validated configuration object.

Source code in src/anonymization_manager/config.py

@classmethod
def from_json(cls, json_path: str):
    """
    Constructs an AnonymizationConfig from a JSON file.

    Args:
        json_path (str): Path to the JSON configuration file.

    Returns:
        AnonymizationConfig: The constructed and validated configuration object.
    """
    with open(json_path, "r") as file:
        config_json = json.load(file)

    attributes = {
        key: config_json[key]
        for key in cls.__annotations__
        if key in config_json
    }
    return cls(**attributes)

anonymization_manager.core.AnonymizationManager

Entry point for the anonymization workflow.

Directs execution to the appropriate backend adapter, wraps the result, and returns it to the caller.

Source code in src/anonymization_manager/core.py

class AnonymizationManager:
    """
    Entry point for the anonymization workflow.

    Directs execution to the appropriate backend adapter, wraps
    the result, and returns it to the caller.
    """

    def anonymize(config: AnonymizationConfig) -> AnonymizedData:
        """
        Anonymizes the dataset using the anonymization config.

        Args: 
            config (AnonymizationConfig):
                The configuration the anonymization manager must respect.

        Returns:
            AnonymizedData: A unified wrapper around the backend-specific result object.

        Raises:
            Exception: If the underlying anonymization engine fails.
        """
        if config.backend == None or config.backend == "arx":
            return AnonymizedData(ARXAnonymizer.anonymize(config))
        elif config.backend == "anjana":
            return AnonymizedData(AnjanaAnonymizer.anonymize(config))
        else:
            logger.warning(f"Unsupported backend: {config.backend}, using ARX")
            return AnonymizedData(ARXAnonymizer.anonymize(config))

anonymize(config)

Anonymizes the dataset using the anonymization config.

Parameters:

Name	Type	Description	Default
config	AnonymizationConfig	The configuration the anonymization manager must respect.	required

Returns:

Name	Type	Description
AnonymizedData	AnonymizedData	A unified wrapper around the backend-specific result object.

Raises:

Type	Description
Exception	If the underlying anonymization engine fails.

Source code in src/anonymization_manager/core.py

def anonymize(config: AnonymizationConfig) -> AnonymizedData:
    """
    Anonymizes the dataset using the anonymization config.

    Args: 
        config (AnonymizationConfig):
            The configuration the anonymization manager must respect.

    Returns:
        AnonymizedData: A unified wrapper around the backend-specific result object.

    Raises:
        Exception: If the underlying anonymization engine fails.
    """
    if config.backend == None or config.backend == "arx":
        return AnonymizedData(ARXAnonymizer.anonymize(config))
    elif config.backend == "anjana":
        return AnonymizedData(AnjanaAnonymizer.anonymize(config))
    else:
        logger.warning(f"Unsupported backend: {config.backend}, using ARX")
        return AnonymizedData(ARXAnonymizer.anonymize(config))

anonymization_manager.adapters.arx.ARXResult

Wrapper for the ARX Java Result object.

Provides Pythonic access to anonymization results, equivalence class statistics, and various quality metrics.

Source code in src/anonymization_manager/adapters/arx/arx.py

class ARXResult:
    """
    Wrapper for the ARX Java Result object.

    Provides Pythonic access to anonymization results, equivalence class
    statistics, and various quality metrics.
    """

    def __init__(self, java_arx_result) -> None:
        """
        Initializes the ARXResult wrapper.

        Args:
            java_arx_result (jpype._jclass.org.deidentifier.arx.ARXResult):
                The Java Arx result object.
        """
        self.arx_result = java_arx_result

    @staticmethod
    def _data_handle_to_dataframe(data_handle: JClass) -> pd.DataFrame:
        """
        Converts a Java ARX DataHandle object to a pandas DataFrame.

        Args:
            data_handle (jpype._jclass.org.deidentifier.arx.DataHandle):
                The ARX DataHandle Object.

        Returns:
            pd.DataFrame: The dataset as a pandas DataFrame.
        """
        column_names = [
            str(data_handle.getAttributeName(i))
            for i in range(data_handle.getNumColumns())
        ]

        data = []

        for i in range(data_handle.getNumRows()):
            row = [
                str(data_handle.getValue(i, j))
                for j in range(data_handle.getNumColumns())
            ]
            data.append(row)

        df = pd.DataFrame(data, columns=column_names)
        return df

    def get_anonymized_data_as_dataframe(self) -> pd.DataFrame:
        """
        Returns the anonymized dataset as a pandas DataFrame.

        Returns:
            pd.Dataframe: Anonymized data.
        """
        data_handle = self.arx_result.getOutput()
        return ARXResult._data_handle_to_dataframe(data_handle)

    def get_raw_data_as_dataframe(self) -> pd.DataFrame:
        """
        Returns the original (raw) dataset as a pandas DataFrame.

        Returns:
            pd.DataFrame: Original data.
        """
        data_handle = self.arx_result.getInput()
        return ARXResult._data_handle_to_dataframe(data_handle)

    def get_transformations(self) -> dict[str, int]:
        """
        Gets the generalization levels applied to quasi-identifiers.

        Returns:
            dict[str, int]: Mapping of quasi-identifier names to their generalization level.
        """
        output_data = self.arx_result.getOutput()
        quasi_identifiers = (
            output_data.getDefinition()
            .getQuasiIdentifyingAttributes()
            .toArray()
        )
        transformations = {
            str(quasi_identifier): int(self.arx_result.getOutput().getGeneralization(
                quasi_identifier
            ))
            for quasi_identifier in quasi_identifiers
        }
        return transformations

    def get_anonymization_time(self) -> int:
        """
        Returns the wall-clock time taken for anonymization.

        Returns:
            int: Time in milliseconds.
        """
        return int(self.arx_result.getTime())

    def store_as_csv(self, output_path: str) -> None:
        """
        Stores the anonymized dataset as CSV file.

        Args:
            output_path (str): File path to save the CSV.
        """
        output = self.arx_result.getOutput()
        output.save(output_path, ",")

    def get_average_equivalence_class_size(self) -> float:
        """
        Returns the average size of equivalence classes.

        Returns:
            float: Average equivalence class size.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getEquivalenceClassStatistics()
            .getAverageEquivalenceClassSize()
        )

    def get_number_of_suppressed_records(self) -> int:
        """
        Returns the number of suppressed (removed) records.

        Returns:
            int: Number of suppressed records.
        """
        return int(
            self.arx_result.getOutput()
            .getStatistics()
            .getEquivalenceClassStatistics()
            .getNumberOfSuppressedRecords()
        )

    def get_max_equivalence_class_size(self) -> int:
        """
        Returns the maximum equivalence class size.

        Returns:
            int: Maximum equivalence class size.
        """
        return int(
            self.arx_result.getOutput()
            .getStatistics()
            .getEquivalenceClassStatistics()
            .getMaximalEquivalenceClassSize()
        )

    def get_min_equivalence_class_size(self) -> int:
        """
        Returns the minimum equivalence class size.

        Returns:
            int: Minimum equivalence class size.
        """
        return int(
            self.arx_result.getOutput()
            .getStatistics()
            .getEquivalenceClassStatistics()
            .getMinimalEquivalenceClassSize()
        )

    def get_number_of_equivalence_classes(self) -> int:
        """
        Returns the number of equivalence classes.

        Returns:
            int: Number of equivalence classes.
        """
        return int(
            self.arx_result.getOutput()
            .getStatistics()
            .getEquivalenceClassStatistics()
            .getNumberOfEquivalenceClasses()
        )

    def get_discernability_metric(self) -> float:
        """
        Returns the discernability metric, a measure of information loss.

        Returns:
            float: discernability metric value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getDiscernibility()
            .getValue()
        )

    def get_average_class_size_metric(self) -> float:
        """
        Returns the average class size metric.

        Note:
            This metric is different from the average equivalence class size.

        Returns:
            float: Average class size metric value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getAverageClassSize()
            .getValue()
        )

    def get_granularity_metric(self, attribute: str) -> float:
        """
        Returns the granularity metric for a specific attribute.

        Args:
            attribute (str): The attribute name.

        Returns:
            float: Granularity metric value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getGranularity()
            .getValue(attribute)
        )

    def get_ssesst_metric(self) -> float:
        """
        Returns the SSESST metric value.

        Returns:
            float: SSESST metric value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getSSESST()
            .getValue()
        )

    def get_record_level_squared_error_metric(self) -> float:
        """
        Returns the record-level squared error metric.

        Returns:
            float: Record-level squared error value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getRecordLevelSquaredError()
            .getValue()
        )

    def get_attribute_level_squared_error_metric(
        self, attribute: str
    ) -> float:
        """
        Returns the attribute level squared metric for a specific attribute.

        Args:
            attribute (str): The attribute name.

        Returns:
            float: Attribute-level squared error value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getAttributeLevelSquaredError()
            .getValue(attribute)
        )

    def get_non_uniform_entropy_metric(self, attribute: str) -> float:
        """
        Returns the non-uniform entropy metric for a specific attribute.

        Args:
            attribute (str): The attribute name.

        Returns:
            float: Non-uniform entropy value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getNonUniformEntropy()
            .getValue(attribute)
        )

    def get_generalization_intensity_metric(self, attribute: str) -> float:
        """
        Returns the generalization intensity metric for a specific attribute.

        Args:
            attribute (str): The attribute name.

        Returns:
            float: Generalization intensity value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getGeneralizationIntensity()
            .getValue(attribute)
        )

    def get_ambiguity_metric(self) -> float:
        """
        Returns the ambiguity metric.

        Returns:
            float: Ambiguity metric value.
        """
        return float(
            self.arx_result.getOutput()
            .getStatistics()
            .getQualityStatistics()
            .getAmbiguity()
            .getValue()
        )

__init__(java_arx_result)

Initializes the ARXResult wrapper.

Parameters:

Name	Type	Description	Default
java_arx_result	ARXResult	The Java Arx result object.	required

Source code in src/anonymization_manager/adapters/arx/arx.py

def __init__(self, java_arx_result) -> None:
    """
    Initializes the ARXResult wrapper.

    Args:
        java_arx_result (jpype._jclass.org.deidentifier.arx.ARXResult):
            The Java Arx result object.
    """
    self.arx_result = java_arx_result

get_ambiguity_metric()

Returns the ambiguity metric.

Returns:

Name	Type	Description
float	float	Ambiguity metric value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_ambiguity_metric(self) -> float:
    """
    Returns the ambiguity metric.

    Returns:
        float: Ambiguity metric value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getAmbiguity()
        .getValue()
    )

get_anonymization_time()

Returns the wall-clock time taken for anonymization.

Returns:

Name	Type	Description
int	int	Time in milliseconds.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_anonymization_time(self) -> int:
    """
    Returns the wall-clock time taken for anonymization.

    Returns:
        int: Time in milliseconds.
    """
    return int(self.arx_result.getTime())

get_anonymized_data_as_dataframe()

Returns the anonymized dataset as a pandas DataFrame.

Returns:

Type	Description
DataFrame	pd.Dataframe: Anonymized data.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_anonymized_data_as_dataframe(self) -> pd.DataFrame:
    """
    Returns the anonymized dataset as a pandas DataFrame.

    Returns:
        pd.Dataframe: Anonymized data.
    """
    data_handle = self.arx_result.getOutput()
    return ARXResult._data_handle_to_dataframe(data_handle)

get_attribute_level_squared_error_metric(attribute)

Returns the attribute level squared metric for a specific attribute.

Parameters:

Name	Type	Description	Default
attribute	str	The attribute name.	required

Returns:

Name	Type	Description
float	float	Attribute-level squared error value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_attribute_level_squared_error_metric(
    self, attribute: str
) -> float:
    """
    Returns the attribute level squared metric for a specific attribute.

    Args:
        attribute (str): The attribute name.

    Returns:
        float: Attribute-level squared error value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getAttributeLevelSquaredError()
        .getValue(attribute)
    )

get_average_class_size_metric()

Returns the average class size metric.

Note

This metric is different from the average equivalence class size.

Returns:

Name	Type	Description
float	float	Average class size metric value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_average_class_size_metric(self) -> float:
    """
    Returns the average class size metric.

    Note:
        This metric is different from the average equivalence class size.

    Returns:
        float: Average class size metric value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getAverageClassSize()
        .getValue()
    )

get_average_equivalence_class_size()

Returns the average size of equivalence classes.

Returns:

Name	Type	Description
float	float	Average equivalence class size.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_average_equivalence_class_size(self) -> float:
    """
    Returns the average size of equivalence classes.

    Returns:
        float: Average equivalence class size.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getEquivalenceClassStatistics()
        .getAverageEquivalenceClassSize()
    )

get_discernability_metric()

Returns the discernability metric, a measure of information loss.

Returns:

Name	Type	Description
float	float	discernability metric value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_discernability_metric(self) -> float:
    """
    Returns the discernability metric, a measure of information loss.

    Returns:
        float: discernability metric value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getDiscernibility()
        .getValue()
    )

get_generalization_intensity_metric(attribute)

Returns the generalization intensity metric for a specific attribute.

Parameters:

Name	Type	Description	Default
attribute	str	The attribute name.	required

Returns:

Name	Type	Description
float	float	Generalization intensity value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_generalization_intensity_metric(self, attribute: str) -> float:
    """
    Returns the generalization intensity metric for a specific attribute.

    Args:
        attribute (str): The attribute name.

    Returns:
        float: Generalization intensity value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getGeneralizationIntensity()
        .getValue(attribute)
    )

get_granularity_metric(attribute)

Returns the granularity metric for a specific attribute.

Parameters:

Name	Type	Description	Default
attribute	str	The attribute name.	required

Returns:

Name	Type	Description
float	float	Granularity metric value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_granularity_metric(self, attribute: str) -> float:
    """
    Returns the granularity metric for a specific attribute.

    Args:
        attribute (str): The attribute name.

    Returns:
        float: Granularity metric value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getGranularity()
        .getValue(attribute)
    )

get_max_equivalence_class_size()

Returns the maximum equivalence class size.

Returns:

Name	Type	Description
int	int	Maximum equivalence class size.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_max_equivalence_class_size(self) -> int:
    """
    Returns the maximum equivalence class size.

    Returns:
        int: Maximum equivalence class size.
    """
    return int(
        self.arx_result.getOutput()
        .getStatistics()
        .getEquivalenceClassStatistics()
        .getMaximalEquivalenceClassSize()
    )

get_min_equivalence_class_size()

Returns the minimum equivalence class size.

Returns:

Name	Type	Description
int	int	Minimum equivalence class size.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_min_equivalence_class_size(self) -> int:
    """
    Returns the minimum equivalence class size.

    Returns:
        int: Minimum equivalence class size.
    """
    return int(
        self.arx_result.getOutput()
        .getStatistics()
        .getEquivalenceClassStatistics()
        .getMinimalEquivalenceClassSize()
    )

get_non_uniform_entropy_metric(attribute)

Returns the non-uniform entropy metric for a specific attribute.

Parameters:

Name	Type	Description	Default
attribute	str	The attribute name.	required

Returns:

Name	Type	Description
float	float	Non-uniform entropy value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_non_uniform_entropy_metric(self, attribute: str) -> float:
    """
    Returns the non-uniform entropy metric for a specific attribute.

    Args:
        attribute (str): The attribute name.

    Returns:
        float: Non-uniform entropy value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getNonUniformEntropy()
        .getValue(attribute)
    )

get_number_of_equivalence_classes()

Returns the number of equivalence classes.

Returns:

Name	Type	Description
int	int	Number of equivalence classes.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_number_of_equivalence_classes(self) -> int:
    """
    Returns the number of equivalence classes.

    Returns:
        int: Number of equivalence classes.
    """
    return int(
        self.arx_result.getOutput()
        .getStatistics()
        .getEquivalenceClassStatistics()
        .getNumberOfEquivalenceClasses()
    )

get_number_of_suppressed_records()

Returns the number of suppressed (removed) records.

Returns:

Name	Type	Description
int	int	Number of suppressed records.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_number_of_suppressed_records(self) -> int:
    """
    Returns the number of suppressed (removed) records.

    Returns:
        int: Number of suppressed records.
    """
    return int(
        self.arx_result.getOutput()
        .getStatistics()
        .getEquivalenceClassStatistics()
        .getNumberOfSuppressedRecords()
    )

get_raw_data_as_dataframe()

Returns the original (raw) dataset as a pandas DataFrame.

Returns:

Type	Description
DataFrame	pd.DataFrame: Original data.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_raw_data_as_dataframe(self) -> pd.DataFrame:
    """
    Returns the original (raw) dataset as a pandas DataFrame.

    Returns:
        pd.DataFrame: Original data.
    """
    data_handle = self.arx_result.getInput()
    return ARXResult._data_handle_to_dataframe(data_handle)

get_record_level_squared_error_metric()

Returns the record-level squared error metric.

Returns:

Name	Type	Description
float	float	Record-level squared error value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_record_level_squared_error_metric(self) -> float:
    """
    Returns the record-level squared error metric.

    Returns:
        float: Record-level squared error value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getRecordLevelSquaredError()
        .getValue()
    )

get_ssesst_metric()

Returns the SSESST metric value.

Returns:

Name	Type	Description
float	float	SSESST metric value.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_ssesst_metric(self) -> float:
    """
    Returns the SSESST metric value.

    Returns:
        float: SSESST metric value.
    """
    return float(
        self.arx_result.getOutput()
        .getStatistics()
        .getQualityStatistics()
        .getSSESST()
        .getValue()
    )

get_transformations()

Gets the generalization levels applied to quasi-identifiers.

Returns:

Type	Description
dict[str, int]	dict[str, int]: Mapping of quasi-identifier names to their generalization level.

Source code in src/anonymization_manager/adapters/arx/arx.py

def get_transformations(self) -> dict[str, int]:
    """
    Gets the generalization levels applied to quasi-identifiers.

    Returns:
        dict[str, int]: Mapping of quasi-identifier names to their generalization level.
    """
    output_data = self.arx_result.getOutput()
    quasi_identifiers = (
        output_data.getDefinition()
        .getQuasiIdentifyingAttributes()
        .toArray()
    )
    transformations = {
        str(quasi_identifier): int(self.arx_result.getOutput().getGeneralization(
            quasi_identifier
        ))
        for quasi_identifier in quasi_identifiers
    }
    return transformations

store_as_csv(output_path)

Stores the anonymized dataset as CSV file.

Parameters:

Name	Type	Description	Default
output_path	str	File path to save the CSV.	required

Source code in src/anonymization_manager/adapters/arx/arx.py

def store_as_csv(self, output_path: str) -> None:
    """
    Stores the anonymized dataset as CSV file.

    Args:
        output_path (str): File path to save the CSV.
    """
    output = self.arx_result.getOutput()
    output.save(output_path, ",")

anonymization_manager.adapters.anjana.AnjanaResult

Wrapper class for Anjana's anonymized results.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

class AnjanaResult:
    """
    Wrapper class for Anjana's anonymized results.
    """

    def __init__(
        self,
        result: pd.DataFrame,
        raw: pd.DataFrame,
        config: AnonymizationConfig,
        time: int,
    ):
        self.result = result
        self.raw = raw
        self.config = config
        self.time = time
        self.quasi_identifiers = config.quasi_identifiers

    def get_anonymized_data_as_dataframe(self) -> pd.DataFrame:
        """
        Returns the anonymized data from ARX as a pandas dataframe.
        """
        return self.result

    def get_raw_data_as_dataframe(self) -> pd.DataFrame:
        """
        Returns the original dataset as a dataframe.
        """
        return self.raw

    def get_transformations(self) -> dict[str, int]:
        """
        Returns the transformations applied to each quasi-identifier.
        """
        hierarchies = {
            key: dict(pd.read_csv(path, header=None))
            for key, path in self.config.hierarchies.items()
        }

        qi: list[str] = self.quasi_identifiers
        transformations: list[int] = utils.get_transformation(
            self.result, qi, hierarchies
        )

        return dict(zip(qi, transformations))

    def store_as_csv(self, output_path: str) -> None:
        """
        Stores the anonymized dataset as .csv file.
        """
        self.result.to_csv(output_path)

    def get_anonymization_time(self) -> int:
        """
        Returns the time it took to anonymize the dataset (Wall Clock).
        """
        return self.time

    # HACK?
    def get_average_equivalence_class_size(self) -> float:
        """
        Returns the average equivalence class size.
        """
        eq_classes = self.result.groupby(list(self.quasi_identifiers)).size()
        return float(eq_classes.mean())

    def get_number_of_suppressed_records(self) -> int:
        """
        Returns the number of suppressed records, i.e. removed from the dataset.
        """
        # Suppressed = original rows not in anonymized (by index)
        return len(self.raw) - len(self.result)

    # TODO
    def get_max_equivalence_class_size(self) -> int:
        """
        Returns the maximum size of an equivalence class present in the anonymized dataset.
        """
        ...

    # TODO
    def get_min_equivalence_class_size(self) -> int:
        """
        Returns the minimum size of an equivalence class present in the anonymized dataset.
        """
        ...

    # TODO
    def get_number_of_equivalence_classes(self) -> int:
        """
        Returns the number of equivalence classes present in the anonymized dataset.
        """
        ...

    # TODO
    def get_discernability_metric(self) -> float:
        """
        Returns the discernibility metric for the anonymized dataset.
        """
        ...
        eq_classes = self.result.groupby(list(self.quasi_identifiers)).size()
        return np.sum(eq_classes**2)

    # TODO
    def get_average_class_size_metric(self) -> float:
        """
        Returns the average class metric, not to be confused with the other similarly named method.
        """
        ...

    # TODO
    def get_granularity_metric(self, attribute: str) -> float:
        """
        Returns the granularity metric for the specific attribute.
        """
        ...

    # TODO
    def get_ssesst_metric(self) -> float:
        """
        Returns the ssesst metric for the anonymized dataset.
        """
        ...

    # TODO
    def get_record_level_squared_error_metric(self) -> float:
        """
        Returns the record level squared metric for the anonymized dataset.
        """
        ...

    # TODO
    def get_attribute_level_squared_error_metric(
        self, attribute: str
    ) -> float:
        """
        Returns the attribute level squared metric for the anonymized dataset.
        """
        ...

    # TODO
    def get_non_uniform_entropy_metric(self, attribute: str) -> float:
        """
        Returns the non uniform entropy metric for the specific attribute in the anonymized dataset.
        """
        ...

    # TODO
    def get_generalization_intensity_metric(self, attribute: str) -> float:
        """
        Returns the generalization intensity metric for the specific attribute in the anonymized dataset.
        """
        ...

    # TODO
    def get_ambiguity_metric(self) -> float:
        """
        Returns the ambiguity metric for the anonymized dataset.
        """
        ...

get_ambiguity_metric()

Returns the ambiguity metric for the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_ambiguity_metric(self) -> float:
    """
    Returns the ambiguity metric for the anonymized dataset.
    """
    ...

get_anonymization_time()

Returns the time it took to anonymize the dataset (Wall Clock).

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_anonymization_time(self) -> int:
    """
    Returns the time it took to anonymize the dataset (Wall Clock).
    """
    return self.time

get_anonymized_data_as_dataframe()

Returns the anonymized data from ARX as a pandas dataframe.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_anonymized_data_as_dataframe(self) -> pd.DataFrame:
    """
    Returns the anonymized data from ARX as a pandas dataframe.
    """
    return self.result

get_attribute_level_squared_error_metric(attribute)

Returns the attribute level squared metric for the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_attribute_level_squared_error_metric(
    self, attribute: str
) -> float:
    """
    Returns the attribute level squared metric for the anonymized dataset.
    """
    ...

get_average_class_size_metric()

Returns the average class metric, not to be confused with the other similarly named method.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_average_class_size_metric(self) -> float:
    """
    Returns the average class metric, not to be confused with the other similarly named method.
    """
    ...

get_average_equivalence_class_size()

Returns the average equivalence class size.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_average_equivalence_class_size(self) -> float:
    """
    Returns the average equivalence class size.
    """
    eq_classes = self.result.groupby(list(self.quasi_identifiers)).size()
    return float(eq_classes.mean())

get_discernability_metric()

Returns the discernibility metric for the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_discernability_metric(self) -> float:
    """
    Returns the discernibility metric for the anonymized dataset.
    """
    ...
    eq_classes = self.result.groupby(list(self.quasi_identifiers)).size()
    return np.sum(eq_classes**2)

get_generalization_intensity_metric(attribute)

Returns the generalization intensity metric for the specific attribute in the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_generalization_intensity_metric(self, attribute: str) -> float:
    """
    Returns the generalization intensity metric for the specific attribute in the anonymized dataset.
    """
    ...

get_granularity_metric(attribute)

Returns the granularity metric for the specific attribute.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_granularity_metric(self, attribute: str) -> float:
    """
    Returns the granularity metric for the specific attribute.
    """
    ...

get_max_equivalence_class_size()

Returns the maximum size of an equivalence class present in the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_max_equivalence_class_size(self) -> int:
    """
    Returns the maximum size of an equivalence class present in the anonymized dataset.
    """
    ...

get_min_equivalence_class_size()

Returns the minimum size of an equivalence class present in the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_min_equivalence_class_size(self) -> int:
    """
    Returns the minimum size of an equivalence class present in the anonymized dataset.
    """
    ...

get_non_uniform_entropy_metric(attribute)

Returns the non uniform entropy metric for the specific attribute in the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_non_uniform_entropy_metric(self, attribute: str) -> float:
    """
    Returns the non uniform entropy metric for the specific attribute in the anonymized dataset.
    """
    ...

get_number_of_equivalence_classes()

Returns the number of equivalence classes present in the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_number_of_equivalence_classes(self) -> int:
    """
    Returns the number of equivalence classes present in the anonymized dataset.
    """
    ...

get_number_of_suppressed_records()

Returns the number of suppressed records, i.e. removed from the dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_number_of_suppressed_records(self) -> int:
    """
    Returns the number of suppressed records, i.e. removed from the dataset.
    """
    # Suppressed = original rows not in anonymized (by index)
    return len(self.raw) - len(self.result)

get_raw_data_as_dataframe()

Returns the original dataset as a dataframe.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_raw_data_as_dataframe(self) -> pd.DataFrame:
    """
    Returns the original dataset as a dataframe.
    """
    return self.raw

get_record_level_squared_error_metric()

Returns the record level squared metric for the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_record_level_squared_error_metric(self) -> float:
    """
    Returns the record level squared metric for the anonymized dataset.
    """
    ...

get_ssesst_metric()

Returns the ssesst metric for the anonymized dataset.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_ssesst_metric(self) -> float:
    """
    Returns the ssesst metric for the anonymized dataset.
    """
    ...

get_transformations()

Returns the transformations applied to each quasi-identifier.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def get_transformations(self) -> dict[str, int]:
    """
    Returns the transformations applied to each quasi-identifier.
    """
    hierarchies = {
        key: dict(pd.read_csv(path, header=None))
        for key, path in self.config.hierarchies.items()
    }

    qi: list[str] = self.quasi_identifiers
    transformations: list[int] = utils.get_transformation(
        self.result, qi, hierarchies
    )

    return dict(zip(qi, transformations))

store_as_csv(output_path)

Stores the anonymized dataset as .csv file.

Source code in src/anonymization_manager/adapters/anjana/anjana.py

def store_as_csv(self, output_path: str) -> None:
    """
    Stores the anonymized dataset as .csv file.
    """
    self.result.to_csv(output_path)