Coverage for packages/dqm-ml-core/src/dqm_ml_core/utils/processor_runner.py: 96%
90 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-21 08:27 +0000
1"""Processor runner utility for executing metrics on DataFrames.
3This module contains the ProcessorRunner class that provides a high-level
4API for running feature and metric processors directly on Pandas DataFrames.
5"""
7from __future__ import annotations
9import logging
10from typing import Any
12import numpy as np
13from pandas import DataFrame
14import pyarrow as pa
16from dqm_ml_core.api.gap_processor import GapProcessor
17from dqm_ml_core.api.metrics_processor import MetricsProcessor
18from dqm_ml_core.api.processor import Processor
20logger = logging.getLogger(__name__)
23class ProcessorRunner:
24 """
25 Orchestrator for executing metric processors on in-memory Pandas DataFrames.
27 This class provides a high-level API for users who want to compute metrics
28 directly on DataFrames without using the full YAML-driven pipeline.
29 """
31 def __init__(self, config: dict[str, Any] | None = None) -> None:
32 """
33 Initialize the runner.
35 Args:
36 config: Optional configuration for metric default behaviors.
37 """
38 self.config = config or {}
40 @staticmethod
41 def _df_to_record_batch(df: DataFrame) -> pa.RecordBatch:
42 """Convert a pandas DataFrame to a pyarrow RecordBatch.
44 Handles conversion of object columns containing uniform-size numpy
45 arrays (e.g. embeddings) to FixedSizeListArray, which is required
46 by processors like DomainGapProcessor.
47 """
48 arrays: list[pa.Array] = []
49 for col_name in df.columns:
50 series = df[col_name]
51 if series.dtype == object and len(series) > 0:
52 first = series.iloc[0]
53 if isinstance(first, np.ndarray): 53 ↛ 60line 53 didn't jump to line 60 because the condition on line 53 was always true
54 dim = first.shape[0]
55 if all(isinstance(v, np.ndarray) and v.shape == (dim,) for v in series): 55 ↛ 60line 55 didn't jump to line 60 because the condition on line 55 was always true
56 flat = np.vstack(series.values).astype(np.float64)
57 arr = pa.FixedSizeListArray.from_arrays(pa.array(flat.flatten().tolist()), dim)
58 arrays.append(arr)
59 continue
60 arrays.append(pa.Array.from_pandas(series))
61 table = pa.table(dict(zip(df.columns, arrays, strict=True)))
62 return table.to_batches()[0]
64 def _compute_batch_level(self, df: DataFrame, processors: list[Processor]) -> tuple[dict[str, Any], dict[str, Any]]:
65 """Compute features and batch-level metrics from non-Gap processors.
67 Returns:
68 Tuple of (batch_features, batch_metrics) merged dicts.
69 """
70 from dqm_ml_core.api.features_processor import FeaturesProcessor
72 batch = self._df_to_record_batch(df)
73 batch_features: dict[str, Any] = {}
74 batch_metrics: dict[str, Any] = {}
76 for metric in processors:
77 logger.debug(f"Processing metric {metric.__class__.__name__}")
78 if isinstance(metric, FeaturesProcessor):
79 batch_features |= metric.compute_features(batch, prev_features=batch_features)
80 elif isinstance(metric, MetricsProcessor): 80 ↛ 85line 80 didn't jump to line 85 because the condition on line 80 was always true
81 # select_columns returns intermediate columns for metric computation,
82 # not final features - use only for compute_batch_metric
83 intermediate = metric.select_columns(batch, prev_features=batch_features)
85 if isinstance(metric, MetricsProcessor):
86 batch_metrics |= metric.compute_batch_metric(intermediate)
88 return batch_features, batch_metrics
90 def _compute_dataset_level(self, batch_metrics: dict[str, Any], processors: list[Processor]) -> dict[str, Any]:
91 """Compute dataset-level metrics from MetricsProcessor instances."""
92 dataset_metrics: dict[str, Any] = {}
93 for metric in processors:
94 if not isinstance(metric, MetricsProcessor):
95 continue
96 logger.debug(f"Computing final score for {metric.__class__.__name__}")
97 dataset_metrics |= metric.compute(batch_metrics=batch_metrics)
98 return dataset_metrics
100 def run(self, df: DataFrame, processors: list[Processor]) -> dict[str, Any]:
101 """
102 Execute the provided processors on a DataFrame.
104 Args:
105 df: The input Pandas DataFrame.
106 processors: List of initialized Processor instances.
108 Returns:
109 A dictionary containing the aggregated dataset-level metrics
110 and/or per-sample features. Features from FeaturesProcessor
111 instances are included alongside metrics from MetricsProcessor instances.
112 """
113 if df.empty or not processors:
114 logger.warning("Empty DataFrame or no processors provided to ProcessorRunner")
115 return {}
117 active = [p for p in processors if not isinstance(p, GapProcessor)]
118 batch_features, batch_metrics = self._compute_batch_level(df, active)
119 dataset_metrics = self._compute_dataset_level(batch_metrics, active)
121 # Merge features (from FeaturesProcessor) and metrics (from MetricsProcessor)
122 result = {}
123 result.update(batch_features) # Per-sample features
124 result.update(dataset_metrics) # Aggregated metrics
125 return result
127 def _compute_features(self, df: DataFrame, features: list[Processor]) -> DataFrame:
128 """Run FeaturesProcessors on a DataFrame and return a new DataFrame with computed columns.
130 Args:
131 df: The input DataFrame.
132 features: List of FeaturesProcessor instances.
134 Returns:
135 A new DataFrame with the original columns plus any new feature columns.
136 """
137 result = self.run(df, features)
138 if not result: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 return df
140 out = df.copy()
141 for col, values in result.items():
142 out[col] = values
143 return out
145 def run_gap(
146 self,
147 source_df: DataFrame,
148 target_df: DataFrame,
149 processor: GapProcessor,
150 features: list[Processor] | None = None,
151 source_selection_name: str = "source",
152 target_selection_name: str = "target",
153 ) -> dict[str, Any]:
154 """
155 Execute a GapProcessor on two DataFrames and compute the domain gap.
157 This method handles the two-dataset execution pattern required by GapProcessor:
158 1. Optionally compute features (e.g. embeddings) on both DataFrames
159 2. Process source DataFrame to compute source statistics
160 3. Process target DataFrame to compute target statistics
161 4. Compute the domain gap delta between source and target
163 Args:
164 source_df: The source dataset DataFrame.
165 target_df: The target dataset DataFrame.
166 processor: An initialized GapProcessor instance.
167 features: Optional list of FeaturesProcessor instances to run on both
168 DataFrames before computing the gap. For example, pass an
169 ImageEmbeddingProcessor to compute embeddings from raw images.
170 source_selection_name: Name for the source selection (default: "source").
171 target_selection_name: Name for the target selection (default: "target").
173 Returns:
174 A dictionary containing:
175 - {metric_name}: The domain gap metric value
176 - "selection_source": Name of the source selection
177 - "selection_target": Name of the target selection
178 """
179 if source_df.empty or target_df.empty:
180 logger.warning("Empty DataFrame provided to run_gap")
181 return {}
183 # Compute features if provided (e.g. embeddings from raw images)
184 if features:
185 source_df = self._compute_features(source_df, features)
186 target_df = self._compute_features(target_df, features)
188 # Process source dataset
189 source_batch = self._df_to_record_batch(source_df)
190 source_features = processor.select_features(source_batch, prev_features={})
191 source_batch_metrics = processor.compute_batch_metric(source_features)
192 source_stats = processor.compute(source_batch_metrics)
194 # Process target dataset
195 target_batch = self._df_to_record_batch(target_df)
196 target_features = processor.select_features(target_batch, prev_features={})
197 target_batch_metrics = processor.compute_batch_metric(target_features)
198 target_stats = processor.compute(target_batch_metrics)
200 # Compute domain gap delta
201 result = processor.compute_delta(source_stats, target_stats)
203 # Add selection metadata
204 result["selection_source"] = pa.array([source_selection_name])
205 result["selection_target"] = pa.array([target_selection_name])
207 return result