Coverage for src/instawell/processing/step_06_calc_derivative.py: 87%

1import logging 

2 

3import numpy as np 

4import pandas as pd 

5 

6from instawell.core.exp_context import ExperimentContext 

7from instawell.core.steps import StepFiles 

8from instawell.utils.logging_util import setup_experiment_logging 

9 

10# set logging level to INFO 

11logger = logging.getLogger(__name__) 

12 

13 

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

15 """Calclulates the derivative of the background subtracted data with respect to Temperature. And Min-Max scales the data afterwards.""" 

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 a path the the bg subtracted data 

21 bg_data_path = ctx.experiment_dir / StepFiles.BG_SUB_DATA 

22 if not bg_data_path.exists(): 

23 raise FileNotFoundError(f"BG subtracted data file not found: {bg_data_path}") 

24 # Load the background subtracted data 

25 data = pd.read_csv(bg_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 

30 # ensure that the min max scaled data is NOT being passed here 

31 # Calculate the derivative with respect to Temperature 

32 data_derivative = data.copy() 

33 for col in data.columns[1:]: # Skip the first column (Temperature) 

34 data_derivative[col] = np.gradient(data[col], data["Temperature"]) 

35 # multiply by -1 to flip the sign 

36 data_derivative[col] *= -1 

37 

38 for col in data_derivative.columns: 

39 if col.startswith("Temperature"): 

40 continue 

41 if data_derivative[col].max() - data_derivative[col].min() == 0: 

42 logger.info( 

43 f"Column {col} has zero range in derivative data; skipping derivative for this column." 

44 ) 

45 continue # Avoid division by zero 

46 data_derivative[col] = (data_derivative[col] - data_derivative[col].min()) / ( 

47 data_derivative[col].max() - data_derivative[col].min() 

48 ) 

49 

50 # Save the derivative data 

51 derivative_data_path = ctx.experiment_dir / StepFiles.DERIVATIVE_DATA 

52 data_derivative.to_csv(derivative_data_path, index=False) 

53 logging.info(f"Derivative data saved to {derivative_data_path}")