Coverage for kataglyphis_webdavclient/dummy.py: 90%

1# Copyright (c) 2024 Jonas Heinle 

2 

3"""Dummy ML preprocessing helpers used by tests and demos.""" 

4 

5import numpy as np 

6from loguru import logger 

7 

8 

9class SimpleMLPreprocessor: 

10 """Simple preprocessing pipeline for synthetic tabular data.""" 

11 

12 def __init__(self, n_samples: int) -> None: 

13 """Initialize the preprocessor state. 

14 

15 Args: 

16 n_samples: Number of synthetic rows to generate. 

17 """ 

18 self.n_samples = n_samples 

19 self.features = np.array([]) 

20 self.labels = np.array([]) 

21 self.normalized = np.array([]) 

22 self.stats = {} 

23 

24 def generate_synthetic_data(self) -> tuple[np.ndarray, np.ndarray]: 

25 """Generate synthetic feature vectors and binary labels.""" 

26 logger.debug( 

27 f"Generating {self.n_samples} samples of synthetic features and labels..." 

28 ) 

29 rng = np.random.default_rng() 

30 self.features = rng.normal(loc=5.0, scale=2.0, size=(self.n_samples, 3)) 

31 self.labels = (self.features.sum(axis=1) > 15).astype(int) 

32 logger.info(f"First 5 feature vectors: {self.features[:5]}") 

33 logger.info(f"First 5 labels: {self.labels[:5]}") 

34 return self.features, self.labels 

35 

36 def normalize_features(self) -> np.ndarray: 

37 """Normalize features by mean and standard deviation per column.""" 

38 if self.features.size == 0: 

39 logger.warning("No features to normalize.") 

40 return np.array([]) 

41 

42 mean = self.features.mean(axis=0) 

43 std = self.features.std(axis=0) 

44 self.normalized = (self.features - mean) / std 

45 

46 self.stats = {"mean": mean.tolist(), "std": std.tolist()} 

47 logger.debug(f"Feature normalization stats: {self.stats}") 

48 return self.normalized 

49 

50 def apply_joke_labeling(self) -> np.ndarray: 

51 """Map numeric labels to playful text labels.""" 

52 if self.labels.size == 0: 

53 logger.warning("No labels to convert into jokes.") 

54 return np.array([]) 

55 

56 jokes = np.where(self.labels == 1, "Definitely ML", "Possibly Not") 

57 logger.info(f"First 5 joke labels: {jokes[:5]}") 

58 return jokes 

59 

60 def run_pipeline(self) -> dict[str, object]: 

61 """Execute the full preprocessing pipeline and return all outputs.""" 

62 logger.info( 

63 f"Running ML preprocessing pipeline for {self.n_samples} samples..." 

64 ) 

65 self.generate_synthetic_data() 

66 self.normalize_features() 

67 jokes = self.apply_joke_labeling() 

68 

69 result = { 

70 "features": self.features, 

71 "labels": self.labels, 

72 "normalized": self.normalized, 

73 **self.stats, 

74 "joke_labels": jokes, 

75 } 

76 logger.success("ML pipeline complete!") 

77 return result