Skip to content

dqm_ml_core.api.features_processor

Features processor base class.

This module contains the FeaturesProcessor base class that all feature extraction processors must inherit from.

FeaturesProcessor

Bases: Processor

Base class for all feature extraction processors.

Feature processors transform raw data into per-sample features (e.g., image luminosity, embeddings). The primary lifecycle method is compute_features, which produces a dict of feature arrays from each batch of raw data.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
13
14
15
16
17
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
class FeaturesProcessor(Processor):
    """
    Base class for all feature extraction processors.

    Feature processors transform raw data into per-sample features
    (e.g., image luminosity, embeddings). The primary lifecycle method
    is ``compute_features``, which produces a dict of feature arrays
    from each batch of raw data.
    """

    def _resolve_output_name(self, col: str, feature_key: str) -> str:
        """Build the output column name with prefix and suffix applied.

        Args:
            col: The input column name.
            feature_key: The feature key to append after ``col_``.

        Returns:
            The fully qualified output column name.
        """
        from dqm_ml_core.models.columns import ColumnsConfig

        cfg: ColumnsConfig | None = getattr(self, "columns_config", None)
        base = f"{col}_{feature_key}"
        p = cfg.prefix if cfg else ""
        s = cfg.suffix if cfg else ""
        return f"{p}{base}{s}"

    def _check_image_fail_fast(self, exc: Exception, *error_attrs: str) -> None:
        if not (self.errors_config and self.errors_config.images):
            return
        image_errors = self.errors_config.images
        for attr in error_attrs:
            if getattr(image_errors, attr, None) == "fail_fast":
                raise exc

    def generated_features(self) -> list[str]:
        """
        Return the list of columns generated by this processor during feature extraction.

        Returns:
            A list of feature names.
        """
        outputs = getattr(self, "output_features", {})
        return list(outputs.values())

    def compute_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Transform a raw data batch into features.

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

        Returns:
            A dictionary mapping feature names to pyarrow Arrays.
        """
        features = {}

        available = batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        for col in cols:
            if col in prev_features:
                continue
            features[col] = batch.column(col)

        return features

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

Transform a raw data batch into 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 feature names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
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
def compute_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Transform a raw data batch into features.

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

    Returns:
        A dictionary mapping feature names to pyarrow Arrays.
    """
    features = {}

    available = batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    for col in cols:
        if col in prev_features:
            continue
        features[col] = batch.column(col)

    return features

generated_features() -> list[str]

Return the list of columns generated by this processor during feature extraction.

Returns:

Type Description
list[str]

A list of feature names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
49
50
51
52
53
54
55
56
57
def generated_features(self) -> list[str]:
    """
    Return the list of columns generated by this processor during feature extraction.

    Returns:
        A list of feature names.
    """
    outputs = getattr(self, "output_features", {})
    return list(outputs.values())