Coverage for addmo_examples/executables/exe_data_tuning_fixed.py: 59%
87 statements
« prev ^ index » next coverage.py v7.4.4, created at 2025-08-31 13:05 +0000
1# # Data Tuning (Fixed) Tutorial
3# Importing of necessary libraries and walk through of internal data tuning, created with default config
4import os
5import json
6import pandas as pd
7from addmo.util.definitions import results_dir_data_tuning
8from addmo.util.load_save_utils import root_dir
9from addmo.util.experiment_logger import LocalLogger
10from addmo.util.experiment_logger import WandbLogger
11from addmo.util.experiment_logger import ExperimentLogger
12from addmo.s2_data_tuning.config.data_tuning_config import DataTuningFixedConfig
13from addmo.s2_data_tuning.data_tuner_fixed import DataTunerByConfig
14from addmo.util.load_save import load_data
15from addmo.util.plotting_utils import save_pdf
16from addmo.util.data_handling import split_target_features
17from addmo.s5_insights.model_plots.time_series import plot_timeseries_combined
18from addmo.util.load_save import load_config_from_json
20def _exe_data_tuning(config, user_input='y'):
21 """
22 Execute the system_data tuning process in a fixed manner.
23 """
25 # Configure the logger
26 LocalLogger.active = True
27 if LocalLogger.active:
28 LocalLogger.directory = results_dir_data_tuning(config,user_input)
30 WandbLogger.project = "addmo-tests_data_tuning_fixed"
31 WandbLogger.active = False
32 if WandbLogger.active:
33 WandbLogger.directory = results_dir_data_tuning(config,user_input)
35 # Initialize logging
36 ExperimentLogger.start_experiment(config=config)
38 # Create the system_data tuner
39 tuner = DataTunerByConfig(config=config)
41 # Load the system_data
42 xy_raw = load_data(config.abs_path_to_data)
44 # Split the system_data
45 x, y = split_target_features(config.name_of_target, xy_raw)
47 # Tune the system_data
48 tuned_x = tuner.tune_fixed(xy_raw)
50 # Merge target and features
51 xy_tuned = tuned_x.join(y)
53 # Drop NaNs
54 xy_tuned = xy_tuned.dropna()
56 # Log the tuned system_data
57 file_name = 'tuned_xy_fixed'
58 ExperimentLogger.log_artifact(xy_tuned, file_name, art_type='system_data')
60 # Finish logging
61 ExperimentLogger.finish_experiment()
63 # Return file paths for plotting data
64 saved_data_path = os.path.join(LocalLogger.directory, file_name + '.csv')
65 data = pd.read_csv(saved_data_path, delimiter=",", index_col=0, encoding="latin1", header=0)
66 config_path = os.path.join(LocalLogger.directory, "config.json")
67 with open(config_path, 'r') as f:
68 plot_config = json.load(f)
70 # Plot tuned data
72 figures = plot_timeseries_combined(plot_config,data)
73 for fig in figures:
74 fig.show()
75 os.makedirs(LocalLogger.directory, exist_ok=True)
76 for idx, fig in enumerate(figures):
77 suffix = "_2weeks" if idx == 1 else ""
78 plot_path = os.path.join(LocalLogger.directory, f"{file_name}{suffix}")
79 save_pdf(fig, plot_path)
80 # Data tuning completed
81 print("Finished")
83def default_config_exe_data_tuning_fixed(user_input='y'):
84 """Execute the system_data tuning process with default config.
85 Parameters:
86 user_input : str, optional
87 If 'y', the contents of the target results directory will be overwritten.
88 If 'd', the directory contents will be deleted. Default is 'y'."""
89 # Initialize a default config (without loading JSON)
90 config = DataTuningFixedConfig()
91 # Run data tuning execution
92 _exe_data_tuning(config, user_input)
94# Tutorial for executing data tuning on user defined config
96def exe_data_tuning_fixed(user_input='y'):
97 """Execute the system_data tuning process from a config file."""
98 # Path to the config file
99 path_to_config = os.path.join(
100 root_dir(), 'addmo', 's2_data_tuning', 'config', 'data_tuning_config.json'
101 )
102 # Create the config object
103 config = load_config_from_json(path_to_config, DataTuningFixedConfig)
104 # Configure the logger
105 LocalLogger.active = True
106 if LocalLogger.active:
107 LocalLogger.directory = results_dir_data_tuning(config, user_input)
109 WandbLogger.project = "addmo-tests_data_tuning_fixed"
110 WandbLogger.active = False
111 if WandbLogger.active:
112 WandbLogger.directory = results_dir_data_tuning(config, user_input)
114 # Initialize logging
115 ExperimentLogger.start_experiment(config=config)
117 # Create the system_data tuner
118 tuner = DataTunerByConfig(config=config)
120 # Load the system_data
121 xy_raw = load_data(config.abs_path_to_data)
123 # Split the system_data
124 x, y = split_target_features(config.name_of_target, xy_raw)
126 # Tune the system_data
127 tuned_x = tuner.tune_fixed(xy_raw)
129 # Merge target and features
130 xy_tuned = tuned_x.join(y)
132 # Drop NaNs
133 xy_tuned = xy_tuned.dropna()
135 # Log the tuned system_data
136 file_name = 'tuned_xy_fixed'
137 ExperimentLogger.log_artifact(xy_tuned, file_name, art_type='system_data')
139 # Finish logging
140 ExperimentLogger.finish_experiment()
142 # Return file paths for plotting data
143 saved_data_path = os.path.join(LocalLogger.directory, file_name + '.csv')
144 data = pd.read_csv(saved_data_path, delimiter=",", index_col=0, encoding="latin1", header=0)
145 config_path = os.path.join(LocalLogger.directory, "config.json")
146 with open(config_path, 'r') as f:
147 plot_config = json.load(f)
149 # Plot tuned data
151 figures = plot_timeseries_combined(plot_config, data)
152 for fig in figures:
153 fig.show()
154 os.makedirs(LocalLogger.directory, exist_ok=True)
155 for idx, fig in enumerate(figures):
156 suffix = "_2weeks" if idx == 1 else ""
157 plot_path = os.path.join(LocalLogger.directory, f"{file_name}{suffix}")
158 save_pdf(fig, plot_path)
159 # Data tuning completed
160 print("Finished")
162if __name__ == '__main__':
163 # Ask the user to overwrite or delete existing results
164 user_input = input("To overwrite the existing content type in 'data_tuning_experiment_fixed' results directory <y>, for deleting the current contents type <d>: ")
165 # Execute data tuning
166 exe_data_tuning_fixed(user_input)