Coverage for src/instawell/processing/step_07_find_min_temp.py: 93%

28 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-11-07 15:47 -0600

1import logging 

2 

3import pandas as pd 

4 

5from instawell.core.exp_context import ExperimentContext 

6from instawell.core.steps import StepFiles 

7from instawell.utils.logging_util import setup_experiment_logging 

8from instawell.utils.utils import split_unqcon_column 

9 

10# set logging level to INFO 

11logger = logging.getLogger(__name__) 

12 

13 

14def find_min_temperature(ctx: ExperimentContext) -> None: 

15 """Finds the minimum temperature for each unique condition in the derivative data.""" 

16 if ctx.log_to_file: 

17 setup_experiment_logging( 

18 experiment_dir=ctx.experiment_dir, filename="experiment.log", level=ctx.log_level 

19 ) 

20 # Build the path to the derivative data 

21 derivative_data_path = ctx.experiment_dir / StepFiles.DERIVATIVE_DATA 

22 if not derivative_data_path.exists(): 

23 raise FileNotFoundError(f"Derivative data file not found: {derivative_data_path}") 

24 # Load the derivative data 

25 data = pd.read_csv(derivative_data_path) 

26 # ensure the first column is 'Temperature', we know it should be b/c we construct it that way 

27 if data.columns[0] != "Temperature": 

28 raise ValueError("The first column must be 'Temperature'.") 

29 min_temps = {} 

30 for col in data.columns: # Skip the first column (Temperature) 

31 if col in ["Temperature", "index"]: 

32 continue 

33 min_index = data[col].idxmin() 

34 # if min_index is not None: 

35 # BUG? this should probnably always be an int 

36 min_temp = data["Temperature"].iloc[int(min_index)] 

37 min_temps[col] = min_temp 

38 # need to convert the min_temps dictionary to a DataFrame 

39 min_temps_df = pd.DataFrame(list(min_temps.items()), columns=["unqcond", "min_temperature"]) 

40 # Save the min temperatures to a CSV file 

41 min_temps_df = split_unqcon_column(min_temps_df) 

42 min_temps_path = ctx.experiment_dir / StepFiles.MIN_TEMPERATURES_DATA 

43 min_temps_df.to_csv(min_temps_path, index=False) 

44 logging.info(f"Min temperatures saved to {min_temps_path}")