Coverage for aixcalibuha/utils/__init__.py: 33%

1""" 

2Package containing utility functions used in different packages. 

3Contains a statistics analyzer and a visualizer. 

4""" 

5import os 

6from typing import Union, List 

7from pathlib import Path 

8from aixcalibuha import CalibrationClass 

9from ebcpy import TimeSeriesData 

10import pandas as pd 

11 

12 

13def convert_mat_to_suffix(mat_result_file, variable_names, suffix_files, parquet_engine='pyarrow', compression='snappy'): 

14 """ 

15 Postprocess the mat result files. 

16 

17 :param str mat_result_file: The path to the MATLAB result file. 

18 :param List[str] variable_names: The names of the variables to extract. 

19 :param str suffix_files: The suffix for the output files. 

20 :param str parquet_engine: The engine to use for saving parquet files. 

21 """ 

22 df = TimeSeriesData(mat_result_file, variable_names=variable_names).to_df() 

23 df_path = Path(mat_result_file).with_suffix("." + suffix_files) 

24 if suffix_files == "csv": 

25 df.to_csv(df_path) 

26 elif suffix_files == "parquet": 

27 df_for_disk = df.copy() 

28 for col in df_for_disk.columns: 

29 if isinstance(df_for_disk[col].dtype, pd.SparseDtype): 

30 df_for_disk[col] = df_for_disk[col].sparse.to_dense() 

31 df.to_parquet( 

32 df_path, 

33 engine=parquet_engine, 

34 compression=compression, 

35 index=True 

36 ) 

37 else: 

38 raise ValueError(f"Unsupported file suffix: {suffix_files}. " 

39 "Supported suffixes are 'csv' and 'parquet'.") 

40 os.remove(mat_result_file) 

41 return df_path 

42 

43 

44def empty_postprocessing(mat_result, **_kwargs): 

45 return mat_result 

46 

47 

48def validate_cal_class_input( 

49 calibration_classes: Union[CalibrationClass, List[CalibrationClass]] 

50) -> List[CalibrationClass]: 

51 """Check if given list contains only CalibrationClass objects or is one 

52 and return a list in both cases. Else raise an error""" 

53 if isinstance(calibration_classes, list): 

54 for cal_class in calibration_classes: 

55 if not isinstance(cal_class, CalibrationClass): 

56 raise TypeError(f"calibration_classes is of type {type(cal_class).__name__} " 

57 f"but should be CalibrationClass") 

58 elif isinstance(calibration_classes, CalibrationClass): 

59 calibration_classes = [calibration_classes] 

60 else: 

61 raise TypeError(f"calibration_classes is of type {type(calibration_classes).__name__} " 

62 f"but should be CalibrationClass or list") 

63 return calibration_classes 

64 

65 

66class MaxIterationsReached(Exception): 

67 """ 

68 Exception raised for when the calibration 

69 ends because the maximum number of 

70 allowed iterations is reached. 

71 """ 

72 

73class MaxTimeReached(Exception): 

74 """ 

75 Exception raised for when the calibration 

76 ends because the maximum calibration time is reached. 

77 """