Coverage for packages/dqm-ml-images/src/dqm_ml_images/visual_features.py: 86%
195 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-21 08:27 +0000
1"""Visual feature extraction processor for image quality assessment.
3This module contains the VisualFeaturesProcessor class that extracts
4visual quality features from images including luminosity, contrast,
5blur, and entropy.
6"""
8import io
9import logging
10from pathlib import Path
11from typing import Any
13from dqm_ml_core import FeaturesProcessor
14from dqm_ml_core.models.columns import ColumnsConfig
15from dqm_ml_core.models.processors import _LUMINOSITY_STANDARDS, ImageFeaturesProcessorConfig
16from dqm_ml_core.utils.image_loading import ImageLoadingMixin
17import numpy as np
18from PIL import Image
19import pyarrow as pa
20from scipy import signal
22# COMPATIBILITY : from typing import Any, override # When support of 3.10 and 3.11 will be removed
23from typing_extensions import override
25logger = logging.getLogger(__name__)
28class VisualFeaturesProcessor(ImageLoadingMixin, FeaturesProcessor):
29 """
30 Computes basic image quality features per sample.
32 Features:
33 - Luminosity: Mean intensity of the image. By default, it is the
34 average gray level mapped to the [0, 1] range.
35 - Contrast: RMS contrast, calculated as the standard deviation of
36 the gray level intensities, mapped to the [0, 1] range.
37 - Blur: Measured as the variance of the Laplacian of the image. A
38 higher value indicates more edges and higher sharpness.
39 - Entropy: Shannon entropy of the image's grayscale histogram.
40 Measures the information content or complexity.
42 This processor operates purely at the feature extraction level
43 (per-sample).
44 """
46 DEFAULT_OUTPUTS: dict[str, str] = {
47 "luminosity": "luminosity",
48 "contrast": "contrast",
49 "blur": "blur",
50 "entropy": "entropy",
51 }
53 output_features: dict[str, str]
55 def __init__(self, name: str = "image_features", config: dict[str, Any] | None = None) -> None:
56 """
57 Initialize the visual features processor.
59 Args:
60 name: Unique name of the processor instance.
61 config: Configuration dictionary containing:
62 - input_columns: List containing the image column name.
63 - output_features: Mapping of feature names to column names.
64 - grayscale: Whether to convert images to grayscale.
65 - normalize: Whether to normalize pixel values to [0, 1].
66 - entropy_bins: Number of bins for entropy calculation.
67 - clip_percentiles: Tuple of (low, high) percentiles.
68 - laplacian_kernel: Laplacian kernel size ('3x3' or '5x5').
69 - path_prefix: Base directory for resolving relative paths
70 (set via dataloader ``mode`` config instead).
71 """
72 super().__init__(name, config)
74 self.columns_config: ColumnsConfig | None = None
75 raw_columns = self.config.get("columns")
76 if isinstance(raw_columns, dict):
77 self.columns_config = ColumnsConfig.model_validate(raw_columns)
79 cfg = ImageFeaturesProcessorConfig.model_validate({**self.config, "name": self.name})
81 self._configure_storage(self.config)
82 self._configure_columns(cfg)
83 self._configure_params(cfg)
84 self._validate_output_features()
86 def _configure_storage(self, cfg: dict[str, Any]) -> None:
87 """Configure S3 filesystem support from config.
89 Args:
90 cfg: Configuration dictionary.
91 """
92 self.s3_fs = None
93 storage_cfg = self.storage_raw
94 if not storage_cfg: 94 ↛ 97line 94 didn't jump to line 97 because the condition on line 94 was always true
95 return
97 from dqm_ml_core.models.global_ import StorageConfig
98 from dqm_ml_job.utils import get_s3_filesystem
100 storage_config = StorageConfig.model_validate(storage_cfg)
101 if storage_config.type == "s3":
102 self.s3_fs = get_s3_filesystem(storage_config)
104 def _configure_columns(self, cfg: ImageFeaturesProcessorConfig) -> None:
105 """Configure input and output columns.
107 Args:
108 cfg: Configuration dictionary.
109 """
110 if not hasattr(self, "input_columns") or "input" not in (self.config.get("columns") or {}):
111 self.input_columns = ["image_bytes"]
113 if not hasattr(self, "output_features") or not self.output_features: 113 ↛ exitline 113 didn't return from function '_configure_columns' because the condition on line 113 was always true
114 self.output_features = self.DEFAULT_OUTPUTS.copy()
116 def _configure_params(self, cfg: ImageFeaturesProcessorConfig) -> None:
117 """Configure processing parameters.
119 Args:
120 cfg: Configuration dictionary.
121 """
122 self.grayscale: bool = cfg.grayscale
123 self.normalize: bool = cfg.normalize
124 self.entropy_bins: int = cfg.histogram.bins if cfg.histogram else 256
125 self.clip_percentiles = cfg.clip_percentiles
126 self.laplacian_kernel: str = cfg.laplacian_kernel
127 raw = cfg.luminosity_weights
128 if isinstance(raw, str): 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 self._luminosity_weights = _LUMINOSITY_STANDARDS[raw]
130 else:
131 self._luminosity_weights = raw if raw is not None else _LUMINOSITY_STANDARDS["bt709"]
133 def _validate_output_features(self) -> None:
134 """Validate and populate default output features configuration.
136 Ensures output_features is a dict and fills in missing feature keys
137 with defaults.
139 Raises:
140 ValueError: If output_features is not a dictionary.
141 """
142 if not isinstance(self.output_features, dict): 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 raise ValueError(f"[{self.name}] 'output_features' must be a dict of metric->column_name")
144 for k in ("luminosity", "contrast", "blur", "entropy"):
145 if k not in self.output_features: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 self.output_features[k] = self.DEFAULT_OUTPUTS[k]
148 def _apply_clip_and_normalize(self, gray: np.ndarray) -> np.ndarray:
149 """Apply percentile clipping and optional normalization to a grayscale array.
151 Args:
152 gray: Input grayscale array.
154 Returns:
155 Clipped and optionally normalized array.
156 """
157 if self.clip_percentiles is None:
158 return gray
159 p_lo, p_hi = self.clip_percentiles
160 lo = np.percentile(gray, p_lo)
161 hi = np.percentile(gray, p_hi)
162 if hi <= lo: 162 ↛ 164line 162 didn't jump to line 164 because the condition on line 162 was always true
163 return gray
164 gray = np.clip(gray, lo, hi)
165 if self.normalize:
166 gray = (gray - lo) / max(1e-12, (hi - lo))
167 return gray
169 def _handle_image_error(self, exc: Exception, idx: int) -> None:
170 """Check error config and either raise or record the failure.
172 Increments failure counters and checks failure rate threshold.
174 rate against
175 configured maximum.
177 Args:
178 exc: The exception that occurred.
179 idx: Sample index for logging.
180 """
181 self._check_image_fail_fast(exc, "on_transform_error", "on_unsupported_format")
182 self._failure_count += 1
183 self._total_count += 1
184 self._check_failure_rate()
185 logger.exception(f"[{self.name}] failed to process sample {idx}: {exc}")
187 def _process_single_image(self, idx: int, v: Any, column: str | None = None) -> np.ndarray | None:
188 """Convert a raw image value to a grayscale numpy array, applying clipping.
190 Args:
191 idx: Position index for error logging.
192 v: Raw image value (bytes, path, PIL Image, or ndarray).
193 column: The input column name (used to resolve path prefix).
195 Returns:
196 Grayscale numpy array or None if processing fails.
197 """
198 try:
199 gray = self._to_gray_np(v, column=column)
200 return self._apply_clip_and_normalize(gray)
201 except Exception as e:
202 self._handle_image_error(e, idx)
203 return None
205 @staticmethod
206 def _compute_scalar_feature(
207 gray_images: list[np.ndarray | None],
208 func: Any,
209 normalize: bool,
210 ) -> pa.Array:
211 """Compute a per-image scalar feature using a given function.
213 Args:
214 gray_images: List of grayscale image arrays (or None for failed images).
215 func: Callable that takes a grayscale array and returns a scalar.
216 normalize: Whether pixel values are normalized to [0, 1].
218 Returns:
219 PyArrow array of feature values.
220 """
221 values = []
222 for gray in gray_images:
223 if gray is not None:
224 values.append(float(func(gray if normalize else gray / 255.0)))
225 else:
226 values.append(float("nan"))
227 return pa.array(values, type=pa.float32())
229 def _output_column_name(self, col: str, feature_key: str) -> str:
230 """Generate output column name for a feature with rename/prefix/suffix.
232 Args:
233 col: Input column name.
234 feature_key: Feature key (luminosity, contrast, blur, entropy).
236 Returns:
237 Fully qualified output column name with rename, prefix, and suffix applied.
238 """
239 name = self.output_features.get(feature_key, feature_key)
240 if self.columns_config and self.columns_config.rename: 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true
241 for r in self.columns_config.rename:
242 if r.from_ == feature_key:
243 name = r.to
244 break
245 return super()._resolve_output_name(col, name)
247 @override
248 def generated_features(self) -> list[str]:
249 """Return list of feature column names this processor will generate.
251 Combines each input column with each feature key (luminosity, contrast,
252 blur, entropy) applying column modifiers (prefix, suffix, rename).
254 Returns:
255 List of output feature column names.
256 """
257 if not self.input_columns:
258 return []
259 return [
260 self._output_column_name(col, fk)
261 for col in self.input_columns
262 for fk in ("luminosity", "contrast", "blur", "entropy")
263 ]
265 @override
266 def compute_features(
267 self,
268 batch: pa.RecordBatch,
269 prev_features: dict[str, pa.Array] | None = None,
270 ) -> dict[str, pa.Array]:
271 """Compute per-sample image features for all configured input columns.
273 Args:
274 batch: Input batch of data containing image columns.
275 prev_features: Previously computed features (not used in this processor).
277 Returns:
278 Dictionary mapping feature names to their computed values.
279 """
280 if not self.input_columns:
281 logger.warning(f"[{self.name}] no input_columns configured")
282 return {}
284 result: dict[str, pa.Array] = {}
285 for image_column in self.input_columns:
286 if image_column not in batch.schema.names:
287 logger.warning(f"[{self.name}] column '{image_column}' not found in batch")
288 continue
290 values = batch.column(image_column).to_pylist()
291 gray_images = [self._process_single_image(idx, v, column=image_column) for idx, v in enumerate(values)]
293 for fk in ("luminosity", "contrast", "blur", "entropy"):
294 func = {"luminosity": np.mean, "contrast": np.std}.get(fk)
295 if fk == "blur":
296 arr = self._compute_scalar_feature(gray_images, self._variance_of_laplacian, True)
297 elif fk == "entropy":
298 arr = self._compute_scalar_feature(gray_images, self._entropy, True)
299 else:
300 arr = self._compute_scalar_feature(gray_images, func, self.normalize)
301 result[self._output_column_name(image_column, fk)] = arr
303 return result
305 @override
306 def reset(self) -> None:
307 """Reset processor state for new processing run.
309 Resets failure counters and total count for fresh processing.
310 """
311 self._failure_count = 0
312 self._total_count = 0
314 # --- helpers --------------------------------------------------------------
316 def _to_gray_np(self, image_data: Any, column: str | None = None) -> np.ndarray:
317 """Convert various input types to a 2D grayscale numpy array.
319 If `self.normalize` is True, returns float32 in [0,1].
320 Otherwise returns uint8 [0,255].
322 Args:
323 image_data: Input data (PIL Image, bytes, string path, or numpy array).
324 column: The input column name (used to resolve path prefix).
326 Returns:
327 2D grayscale array in grayscale.
328 """
329 if isinstance(image_data, Image.Image): 329 ↛ 330line 329 didn't jump to line 330 because the condition on line 329 was never true
330 return self._pil_to_gray(image_data)
331 if isinstance(image_data, (bytes, bytearray)):
332 return self._pil_to_gray(Image.open(io.BytesIO(image_data)))
333 if isinstance(image_data, str):
334 image = self._load_image_from_path(image_data, column=column)
335 if image is None:
336 raise ValueError(f"Failed to load image from path: {image_data}")
337 return self._pil_to_gray(image)
338 if isinstance(image_data, np.ndarray): 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 return self._ndarray_to_gray(image_data)
340 raise ValueError(f"Unsupported type for image input: {type(image_data)}")
342 def _load_image_from_path(self, path: str, column: str | None = None) -> Image.Image | None:
343 """Load a PIL Image from a string path (S3 or local).
345 Resolves relative paths by looking up the prefix for the configured
346 input column from ``self.current_path_prefix`` (set by the job for
347 each selection).
349 Args:
350 path: File path or relative path.
351 column: The input column name (used to resolve the path prefix).
353 Returns:
354 PIL Image object.
356 Raises:
357 ValueError: If the file does not exist and ``fail_fast`` is configured.
358 """
359 img = super()._open_image_from_path(path, column)
360 if img is not None:
361 return img
363 # File not found — check error configuration
364 if (
365 self.errors_config
366 and self.errors_config.tabular
367 and self.errors_config.tabular.on_file_not_found == "fail_fast"
368 ):
369 prefix = self._current_image_prefix(column)
370 img_path = Path(prefix) / path if prefix else Path(path)
371 raise ValueError(f"Path does not exist: {img_path}")
372 return None
374 def _pil_to_gray(self, img: Image.Image) -> np.ndarray:
375 """Convert a PIL Image to a 2D grayscale numpy array.
377 Args:
378 img: PIL Image to convert.
380 Returns:
381 2D grayscale array.
382 """
383 if self.grayscale and img.mode != "L":
384 img = img.convert("L")
385 elif not self.grayscale and img.mode not in ("RGB", "L"): 385 ↛ 386line 385 didn't jump to line 386 because the condition on line 385 was never true
386 img = img.convert("RGB")
388 gray_np = np.array(img)
389 if gray_np.ndim == 3:
390 r, g, b = self._luminosity_weights
391 gray_np = r * gray_np[..., 0] + g * gray_np[..., 1] + b * gray_np[..., 2]
393 return self._to_float01(gray_np) if self.normalize else gray_np.astype(np.uint8)
395 def _ndarray_to_gray(self, arr: np.ndarray) -> np.ndarray:
396 """Convert a numpy image array to 2D grayscale.
398 Args:
399 arr: Input array (2D gray, 3D RGB/RGBA).
401 Returns:
402 2D grayscale array.
404 Raises:
405 ValueError: If the shape is unsupported.
406 """
407 if arr.ndim == 2:
408 gray = arr
409 elif arr.ndim == 3 and arr.shape[2] in (3, 4):
410 rgb = arr[..., :3].astype(np.float32)
411 r, g, b = self._luminosity_weights
412 gray = r * rgb[..., 0] + g * rgb[..., 1] + b * rgb[..., 2]
413 else:
414 raise ValueError(f"Unsupported ndarray shape {arr.shape}")
416 return self._to_float01(gray) if self.normalize else gray.astype(np.uint8)
418 @staticmethod
419 def _to_float01(arr: np.ndarray) -> np.ndarray:
420 """Normalize array to [0, 1] range using min-max scaling.
422 Args:
423 arr: Input numpy array.
425 Returns:
426 Normalized array with float32 values in [0, 1].
427 """
428 arr = arr.astype(np.float32)
429 vmin, vmax = float(arr.min()), float(arr.max())
430 arr = (arr - vmin) / (vmax - vmin) if vmax > vmin else np.zeros_like(arr, dtype=np.float32)
431 return arr
433 def _variance_of_laplacian(self, gray: np.ndarray) -> float:
434 """Variance of Laplacian as a blur metric.
436 Args:
437 gray: Grayscale image array.
439 Returns:
440 Variance of Laplacian (higher values indicate
441 more edges/sharpness).
442 """
443 gray = gray.astype(np.float32)
444 if self.laplacian_kernel == "5x5":
445 kernel = np.array(
446 [
447 [0, 0, -1, 0, 0],
448 [0, -1, -2, -1, 0],
449 [-1, -2, 16, -2, -1],
450 [0, -1, -2, -1, 0],
451 [0, 0, -1, 0, 0],
452 ],
453 dtype=np.float32,
454 )
455 else:
456 kernel = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32)
458 # Use scipy for optimized convolution
459 lap = signal.convolve2d(gray, kernel, mode="same")
460 return float(np.var(lap))
462 def _entropy(self, gray: np.ndarray) -> float:
463 """Shannon entropy of the gray histogram (natural log).
465 Args:
466 gray: Grayscale image array.
468 Returns:
469 Shannon entropy value. Returns NaN if histogram sum is zero.
470 """
471 if self.normalize:
472 # histogram on [0,1]
473 hist, _ = np.histogram(gray, bins=self.entropy_bins, range=(0.0, 1.0))
474 else:
475 # uint8 range
476 hist, _ = np.histogram(gray, bins=min(256, self.entropy_bins), range=(0, 255))
477 prob = hist.astype(np.float64)
478 total = prob.sum()
479 if total <= 0: 479 ↛ 480line 479 didn't jump to line 480 because the condition on line 479 was never true
480 return float("nan")
481 prob /= total
482 # avoid log(0)
483 prob = prob[prob > 0]
484 return float(-(prob * np.log(prob)).sum())