Skip to content

dqm_ml_core.api.gap_processor

Gap processor base class.

This module contains the GapProcessor base class that all domain gap processors must inherit from.

logger = logging.getLogger(__name__) module-attribute

GapProcessor

Bases: Processor

Base class for all domain gap processors.

Gap processors compute distribution shift between two datasets (e.g., MMD, FID, KL divergence). The primary lifecycle methods are select_features (per-batch column selection aware of previous features), compute_batch_metric (batch aggregation), compute (final dataset-level statistics), and compute_delta (pairwise comparison).

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class GapProcessor(Processor):
    """
    Base class for all domain gap processors.

    Gap processors compute distribution shift between two datasets
    (e.g., MMD, FID, KL divergence). The primary lifecycle methods are
    ``select_features`` (per-batch column selection aware of previous features),
    ``compute_batch_metric`` (batch aggregation), ``compute`` (final dataset-level
    statistics), and ``compute_delta`` (pairwise comparison).
    """

    def select_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Extract relevant columns from a batch, resolving patterns against
        both batch columns and previously computed upstream features.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping column names to pyarrow Arrays.
        """
        available = list(prev_features.keys()) + batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        features: dict[str, pa.Array] = {}
        for col in cols:
            if col in prev_features:
                continue
            if col not in batch.schema.names:
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue
            features[col] = batch.column(col)

        return features

    def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Aggregate features into intermediate statistics for the current batch.

        Args:
            features: Dictionary of feature arrays computed on the batch.

        Returns:
            A dictionary of aggregated statistics.
        """
        return {}

    def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
        """
        Perform the final dataset-level aggregation of batch statistics.

        Args:
            batch_metrics: The aggregated intermediate statistics from all batches.

        Returns:
            A dictionary containing the final dataset-level statistics.
        """
        # SonarQube raises a warning because batch_metrics is not used.
        # It is irrelevant because compute is implemented in child classes which use batch_metrics.
        return {}

    def compute_delta(self, source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
        """
        Compare metrics between two different dataselections.

        Args:
            source: Final metrics from the source dataselection.
            target: Final metrics from the target dataselection.

        Returns:
            A dictionary containing distance or difference scores.
        """
        return {}

compute(batch_metrics: dict[str, pa.Array]) -> dict[str, Any]

Perform the final dataset-level aggregation of batch statistics.

Parameters:

Name Type Description Default
batch_metrics dict[str, Array]

The aggregated intermediate statistics from all batches.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing the final dataset-level statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
    """
    Perform the final dataset-level aggregation of batch statistics.

    Args:
        batch_metrics: The aggregated intermediate statistics from all batches.

    Returns:
        A dictionary containing the final dataset-level statistics.
    """
    # SonarQube raises a warning because batch_metrics is not used.
    # It is irrelevant because compute is implemented in child classes which use batch_metrics.
    return {}

compute_batch_metric(features: dict[str, pa.Array]) -> dict[str, pa.Array]

Aggregate features into intermediate statistics for the current batch.

Parameters:

Name Type Description Default
features dict[str, Array]

Dictionary of feature arrays computed on the batch.

 required

Returns:

Type Description
dict[str, Array]

A dictionary of aggregated statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
59
60
61
62
63
64
65
66
67
68
69
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Aggregate features into intermediate statistics for the current batch.

    Args:
        features: Dictionary of feature arrays computed on the batch.

    Returns:
        A dictionary of aggregated statistics.
    """
    return {}

compute_delta(source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]

Compare metrics between two different dataselections.

Parameters:

Name Type Description Default
source dict[str, Any]

Final metrics from the source dataselection.

 required
target dict[str, Any]

Final metrics from the target dataselection.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing distance or difference scores.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
85
86
87
88
89
90
91
92
93
94
95
96
def compute_delta(self, source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
    """
    Compare metrics between two different dataselections.

    Args:
        source: Final metrics from the source dataselection.
        target: Final metrics from the target dataselection.

    Returns:
        A dictionary containing distance or difference scores.
    """
    return {}

select_features(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Extract relevant columns from a batch, resolving patterns against both batch columns and previously computed upstream features.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping column names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def select_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Extract relevant columns from a batch, resolving patterns against
    both batch columns and previously computed upstream features.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping column names to pyarrow Arrays.
    """
    available = list(prev_features.keys()) + batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    features: dict[str, pa.Array] = {}
    for col in cols:
        if col in prev_features:
            continue
        if col not in batch.schema.names:
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue
        features[col] = batch.column(col)

    return features