Coverage for packages/dqm-ml-core/src/dqm_ml_core/api/processor.py: 100%

1"""Base processor class. 

2 

3This module contains the Processor base class that all 

4metrics, features, and gap processors must inherit from. 

5""" 

6 

7import logging 

8from typing import Any 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13class Processor: 

14 """ 

15 Base class for all Data Quality metrics, feature extractors, and gap processors. 

16 

17 Provides shared initialization, failure-rate checking, and column resolution. 

18 Lifecycle methods are defined in the appropriate subclass: 

19 :class:`FeaturesProcessor`, :class:`MetricsProcessor`, or :class:`GapProcessor`. 

20 """ 

21 

22 def __init__(self, name: str, config: dict[str, Any] | None): 

23 """ 

24 Initialize the processor. 

25 

26 Args: 

27 name: Unique name of the processor instance. 

28 config: Configuration dictionary (optional). 

29 """ 

30 self.name = name 

31 config = config or {} 

32 self.storage_raw = config.pop("storage", None) 

33 self.config = config 

34 self.errors_config = None 

35 self.compute_device: str = "cpu" 

36 self.compute_seed: int | None = None 

37 self.current_path_prefix: dict[str, str] = {} 

38 self._failure_count = 0 

39 self._total_count = 0 

40 

41 self.input_columns: list[str] = [] 

42 self.exclude_columns: list[str] | None = None 

43 self.columns_config = None 

44 if "columns" in self.config and isinstance(self.config["columns"], dict): 

45 from dqm_ml_core.models.columns import ColumnsConfig 

46 

47 self.columns_config = ColumnsConfig.model_validate(self.config["columns"]) 

48 self.input_columns = self.columns_config.input 

49 self.exclude_columns = self.columns_config.exclude 

50 

51 def _check_failure_rate(self) -> None: 

52 if self.errors_config and self.errors_config.max_failure_rate is not None and self._total_count > 0: 

53 failure_rate = self._failure_count / self._total_count 

54 if failure_rate > self.errors_config.max_failure_rate: 

55 raise RuntimeError( 

56 f"Failure rate {failure_rate:.1%} exceeds max {self.errors_config.max_failure_rate:.1%}" 

57 ) 

58 

59 def needed_columns(self) -> list[str]: 

60 """ 

61 Return the list of raw input columns required for processing. 

62 

63 Returns: 

64 A list of column names. 

65 """ 

66 return self.input_columns 

67 

68 def reset(self) -> None: 

69 """Reset per-selection state between dataselections. 

70 

71 Processors that cache per-selection state (e.g. histogram bin 

72 edges in RepresentativenessProcessor) MUST override this to 

73 clear that state. Called by DatasetJob after each selection. 

74 """