Source code for parq_tools.utils.profile_compare_utils

import copy
import math
from typing import Any, Iterable, Optional, Sequence

import pandas as pd


DEFAULT_COMPARISON_METRICS: tuple[str, ...] = (
    "mean",
    "std",
    "min",
    "max",
    "n_missing",
    "p_missing",
    "n_distinct",
)


def _to_python_scalar(value: Any) -> Any:
    if hasattr(value, "item"):
        try:
            return value.item()
        except (TypeError, ValueError):
            return value
    return value


[docs] def normalize_for_export(value: Any) -> Any: """Convert non-JSON-serializable numpy/pandas values into Python scalars.""" if isinstance(value, dict): return {str(k): normalize_for_export(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [normalize_for_export(v) for v in value] value = _to_python_scalar(value) if isinstance(value, float): if math.isnan(value): return None if math.isinf(value): return "inf" if value > 0 else "-inf" return value
[docs] def is_numeric(value: Any) -> bool: value = _to_python_scalar(value) return isinstance(value, (int, float)) and not isinstance(value, bool)
[docs] def within_tolerance(a: Any, b: Any, abs_tol: float, rel_tol: float) -> bool: a = _to_python_scalar(a) b = _to_python_scalar(b) if is_numeric(a) and is_numeric(b): if isinstance(a, float) and math.isnan(a) and isinstance(b, float) and math.isnan(b): return True if isinstance(a, float) and math.isnan(a): return False if isinstance(b, float) and math.isnan(b): return False diff = abs(float(a) - float(b)) scale = max(abs(float(a)), abs(float(b))) return diff <= max(abs_tol, rel_tol * scale) return a == b
[docs] def build_column_summary( values: Sequence[Any], abs_tol: float, rel_tol: float, ) -> dict[str, Any]: if not values: return {"within_tolerance": True, "deltas_from_first": []} baseline = values[0] deltas: list[Optional[dict[str, Any]]] = [None] all_within = True for value in values[1:]: if is_numeric(baseline) and is_numeric(value): baseline_float = float(baseline) value_float = float(value) abs_delta = abs(value_float - baseline_float) rel_delta = abs_delta / abs(baseline_float) if baseline_float != 0 else (0.0 if abs_delta == 0 else float("inf")) deltas.append({"abs": abs_delta, "rel": rel_delta}) else: deltas.append(None) if not within_tolerance(baseline, value, abs_tol=abs_tol, rel_tol=rel_tol): all_within = False return { "within_tolerance": all_within, "deltas_from_first": deltas, }
[docs] def build_profile_comparison_summary( descriptions: Sequence[Any], labels: Sequence[str], abs_tol: float = 0.0, rel_tol: float = 0.0, metrics: Optional[Iterable[str]] = None, ) -> dict[str, Any]: metric_names = list(metrics) if metrics is not None else list(DEFAULT_COMPARISON_METRICS) all_columns: list[str] = [] for desc in descriptions: for column in desc.variables.keys(): if column not in all_columns: all_columns.append(column) columns: dict[str, Any] = {} overview = { "equal": 0, "different": 0, "missing_in_some": 0, "type_mismatch": 0, "total_columns": len(all_columns), } for column in all_columns: variable_entries = [desc.variables.get(column) for desc in descriptions] present = [entry is not None for entry in variable_entries] types = [entry.get("type") if entry else None for entry in variable_entries] reasons: list[str] = [] status = "equal" if not all(present): status = "missing_in_some" reasons.append("missing_in_some_datasets") elif len({str(dtype) for dtype in types}) > 1: status = "type_mismatch" reasons.append("type_mismatch") metric_payload: dict[str, Any] = {} if status == "equal": metric_differences: list[str] = [] for metric_name in metric_names: values = [entry.get(metric_name) if entry else None for entry in variable_entries] comparison = build_column_summary(values, abs_tol=abs_tol, rel_tol=rel_tol) metric_payload[metric_name] = { "values": values, **comparison, } if not comparison["within_tolerance"]: metric_differences.append(metric_name) if metric_differences: status = "different" reasons.append(f"metric_differences:{','.join(metric_differences)}") if status in overview: overview[status] += 1 columns[column] = { "present": present, "types": types, "status": status, "metrics": metric_payload, "reasons": reasons, } return normalize_for_export( { "labels": list(labels), "overview": overview, "columns": columns, "tolerance": {"abs_tol": abs_tol, "rel_tol": rel_tol}, "metrics_compared": metric_names, } )
[docs] def get_changed_columns_from_summary(summary: dict[str, Any]) -> list[str]: changed = [] for column, payload in summary["columns"].items(): if payload.get("status") != "equal": changed.append(column) return changed
def _slice_correlation_value(value: Any, keep_columns: set[str]) -> Any: if isinstance(value, pd.DataFrame): shared_columns = [col for col in value.columns if col in keep_columns] shared_index = [idx for idx in value.index if idx in keep_columns] if not shared_columns or not shared_index: return pd.DataFrame() return value.loc[shared_index, shared_columns] return value def _recompute_table_stats(description: Any) -> None: variables = description.variables table = description.table if not variables: table["n_var"] = 0 table["memory_size"] = 0 table["record_size"] = 0 table["n_cells_missing"] = 0 table["n_vars_with_missing"] = 0 table["n_vars_all_missing"] = 0 table["p_cells_missing"] = 0 table["types"] = {} return n = table.get("n", 0) memory_size = 0 n_cells_missing = 0 n_vars_with_missing = 0 n_vars_all_missing = 0 type_counts: dict[str, int] = {} for variable in variables.values(): var_memory = variable.get("memory_size", 0) var_missing = variable.get("n_missing", 0) var_n = variable.get("n", n) var_type = str(variable.get("type", "Unsupported")) memory_size += var_memory if isinstance(var_memory, (int, float)) else 0 n_cells_missing += var_missing if isinstance(var_missing, (int, float)) else 0 if isinstance(var_missing, (int, float)) and var_missing > 0: n_vars_with_missing += 1 if isinstance(var_missing, (int, float)) and isinstance(var_n, (int, float)) and var_missing == var_n: n_vars_all_missing += 1 type_counts[var_type] = type_counts.get(var_type, 0) + 1 table["n_var"] = len(variables) table["memory_size"] = memory_size table["record_size"] = memory_size / n if n else 0 table["n_cells_missing"] = n_cells_missing table["n_vars_with_missing"] = n_vars_with_missing table["n_vars_all_missing"] = n_vars_all_missing denom = len(variables) * n table["p_cells_missing"] = n_cells_missing / denom if denom else 0 table["types"] = type_counts
[docs] def prune_description_to_columns(description: Any, keep_columns: Sequence[str]) -> Any: """Prune a BaseDescription-like object to the requested columns.""" keep = set(keep_columns) pruned = copy.deepcopy(description) pruned.variables = {k: v for k, v in pruned.variables.items() if k in keep} pruned.scatter = { k1: {k2: v2 for k2, v2 in row.items() if k2 in keep} for k1, row in pruned.scatter.items() if k1 in keep } pruned.correlations = { corr_name: _slice_correlation_value(corr_value, keep) for corr_name, corr_value in pruned.correlations.items() } # Missing diagrams can couple to the original full variable set; reset for safe compare rendering. pruned.missing = {} pruned.alerts = [ alert for alert in pruned.alerts if getattr(alert, "column_name", None) in keep or getattr(alert, "column_name", None) is None ] _recompute_table_stats(pruned) return pruned