APOG_public/acsmnode
0
0
Fork 0
mirror of https://gitea.psi.ch/APOG/acsmnode.git synced 2025-06-29 04:40:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

Save changes to calibration factors and apply_calibration_factors.py after git pull operation

Browse Source
This commit is contained in:
florez_j
2025-01-30 13:33:44 +01:00
parent 1f119991f2
commit 95f13d05b2
1 changed files with 178 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

356
pipelines/steps/apply_calibration_factors.py
View File

@ -1,179 +1,179 @@
import sys, os import sys, os
try: try:
thisFilePath = os.path.abspath(__file__) thisFilePath = os.path.abspath(__file__)
print(thisFilePath) print(thisFilePath)
except NameError: except NameError:
print("[Notice] The __file__ attribute is unavailable in this environment (e.g., Jupyter or IDLE).") print("[Notice] The __file__ attribute is unavailable in this environment (e.g., Jupyter or IDLE).")
print("When using a terminal, make sure the working directory is set to the script's location to prevent path issues (for the DIMA submodule)") print("When using a terminal, make sure the working directory is set to the script's location to prevent path issues (for the DIMA submodule)")
#print("Otherwise, path to submodule DIMA may not be resolved properly.") #print("Otherwise, path to submodule DIMA may not be resolved properly.")
thisFilePath = os.getcwd() # Use current directory or specify a default thisFilePath = os.getcwd() # Use current directory or specify a default
dimaPath = os.path.normpath(os.path.join(thisFilePath, "..", "..",'..')) # Move up to project root dimaPath = os.path.normpath(os.path.join(thisFilePath, "..", "..",'..')) # Move up to project root
projectPath = os.path.normpath(os.path.join(dimaPath,'..')) projectPath = os.path.normpath(os.path.join(dimaPath,'..'))
print(dimaPath) print(dimaPath)
import numpy as np import numpy as np
import pandas as pd import pandas as pd
from math import prod # To replace multiplyall from math import prod # To replace multiplyall
import argparse import argparse
import yaml import yaml
# Set up project root directory # Set up project root directory
#root_dir = os.path.abspath(os.curdir) #root_dir = os.path.abspath(os.curdir)
#sys.path.append(root_dir) #sys.path.append(root_dir)
sys.path.append(dimaPath) sys.path.append(dimaPath)
import dima.src.hdf5_ops as dataOps import dima.src.hdf5_ops as dataOps
def apply_calibration_factors(data_table, calibration_factors): def apply_calibration_factors(data_table, calibration_factors):
""" """
Calibrates the species data in the given data table using a calibration factor. Calibrates the species data in the given data table using a calibration factor.
Parameters: Parameters:
data_table (pd.DataFrame): The input data table with variables to calibrate. data_table (pd.DataFrame): The input data table with variables to calibrate.
calibration_factor (dict): Dictionary containing 'standard' calibration factors calibration_factor (dict): Dictionary containing 'standard' calibration factors
with 'num' and 'den' values as dictionaries of multipliers. with 'num' and 'den' values as dictionaries of multipliers.
Returns: Returns:
pd.DataFrame: A new data table with calibrated variables. pd.DataFrame: A new data table with calibrated variables.
""" """
# Make a copy of the input table to avoid modifying the original # Make a copy of the input table to avoid modifying the original
new_data_table = data_table.copy() new_data_table = data_table.copy()
# Initialize a dictionary to rename variables # Initialize a dictionary to rename variables
variable_rename_dict = {} variable_rename_dict = {}
# Loop through the column names in the data table # Loop through the column names in the data table
for variable_name in new_data_table.select_dtypes(include=["number"]).columns: for variable_name in new_data_table.select_dtypes(include=["number"]).columns:
if variable_name in calibration_factors['variables'].keys(): # use standard calibration factor if variable_name in calibration_factors['variables'].keys(): # use standard calibration factor
#print(variable_name) #print(variable_name)
# Extract numerator and denominator values # Extract numerator and denominator values
numerator = prod(calibration_factors['variables'][variable_name]['num']) numerator = prod(calibration_factors['variables'][variable_name]['num'])
denominator = prod(calibration_factors['variables'][variable_name]['den']) denominator = prod(calibration_factors['variables'][variable_name]['den'])
# Apply calibration to each variable # Apply calibration to each variable
new_data_table[variable_name] = new_data_table[variable_name].mul((numerator / denominator)) new_data_table[variable_name] = new_data_table[variable_name].mul((numerator / denominator))
# Add renaming entry # Add renaming entry
variable_rename_dict[variable_name] = f"{variable_name}_correct" variable_rename_dict[variable_name] = f"{variable_name}_correct"
else: # use specifies dependent calibration factor else: # use specifies dependent calibration factor
print(f'There is no calibration factors for variable {variable_name}. The variable will remain the same.') print(f'There is no calibration factors for variable {variable_name}. The variable will remain the same.')
# Rename the columns in the new data table # Rename the columns in the new data table
new_data_table.rename(columns=variable_rename_dict, inplace=True) new_data_table.rename(columns=variable_rename_dict, inplace=True)
return new_data_table return new_data_table
def record_data_lineage(path_to_output_file, metadata): def record_data_lineage(path_to_output_file, metadata):
path_to_output_dir, output_file = os.path.split(path_to_output_file) path_to_output_dir, output_file = os.path.split(path_to_output_file)
path_to_metadata_file = '/'.join([path_to_output_dir,'data_lineage_metadata.json']) path_to_metadata_file = '/'.join([path_to_output_dir,'data_lineage_metadata.json'])
# Ensure the file exists # Ensure the file exists
if not os.path.exists(path_to_metadata_file): if not os.path.exists(path_to_metadata_file):
with open(path_to_metadata_file, 'w') as f: with open(path_to_metadata_file, 'w') as f:
json.dump({}, f) # Initialize empty JSON json.dump({}, f) # Initialize empty JSON
# Read the existing JSON # Read the existing JSON
with open(path_to_metadata_file, 'r') as metadata_file: with open(path_to_metadata_file, 'r') as metadata_file:
try: try:
json_dict = json.load(metadata_file) json_dict = json.load(metadata_file)
except json.JSONDecodeError: except json.JSONDecodeError:
json_dict = {} # Start fresh if file is invalid json_dict = {} # Start fresh if file is invalid
# Compute relative output file path and update the JSON object # Compute relative output file path and update the JSON object
relpath_to_output_file = os.path.relpath(path_to_output_file, start=projectPath).replace(os.sep, '/') relpath_to_output_file = os.path.relpath(path_to_output_file, start=projectPath).replace(os.sep, '/')
json_dict[relpath_to_output_file] = metadata json_dict[relpath_to_output_file] = metadata
# Write updated JSON back to the file # Write updated JSON back to the file
with open(path_to_metadata_file, 'w') as metadata_file: with open(path_to_metadata_file, 'w') as metadata_file:
json.dump(json_dict, metadata_file, indent=4) json.dump(json_dict, metadata_file, indent=4)
print(f"Metadata for calibrated data saved to {path_to_metadata_file}") print(f"Metadata for calibrated data saved to {path_to_metadata_file}")
return 0 return 0
if __name__ == '__main__': if __name__ == '__main__':
# Set up argument parsing # Set up argument parsing
parser = argparse.ArgumentParser(description="Calibrate species data using calibration factors.") parser = argparse.ArgumentParser(description="Calibrate species data using calibration factors.")
parser.add_argument('data_file', type=str, help="Path to the input HDF5 file containing the data table.") parser.add_argument('data_file', type=str, help="Path to the input HDF5 file containing the data table.")
parser.add_argument('dataset_name', type=str, help ='Relative path to data_table (i.e., dataset name) in HDF5 file') parser.add_argument('dataset_name', type=str, help ='Relative path to data_table (i.e., dataset name) in HDF5 file')
parser.add_argument('calibration_file', type=str, help="Path to the input YAML file containing calibration factors.") parser.add_argument('calibration_file', type=str, help="Path to the input YAML file containing calibration factors.")
parser.add_argument('output_file', type=str, help="Path to save the output calibrated data as a CSV file.") parser.add_argument('output_file', type=str, help="Path to save the output calibrated data as a CSV file.")
args = parser.parse_args() args = parser.parse_args()
# Load input data and calibration factors # Load input data and calibration factors
try: try:
#data_table = pd.read_json(args.data_file) #data_table = pd.read_json(args.data_file)
print(args.data_file) print(args.data_file)
dataManager = dataOps.HDF5DataOpsManager(args.data_file) dataManager = dataOps.HDF5DataOpsManager(args.data_file)
dataManager.load_file_obj() dataManager.load_file_obj()
dataset_name = '/'+args.dataset_name dataset_name = '/'+args.dataset_name
data_table = dataManager.extract_dataset_as_dataframe('/'+args.dataset_name) data_table = dataManager.extract_dataset_as_dataframe('/'+args.dataset_name)
dataManager.extract_and_load_dataset_metadata() dataManager.extract_and_load_dataset_metadata()
dataset_metadata_df = dataManager.dataset_metadata_df.copy() dataset_metadata_df = dataManager.dataset_metadata_df.copy()
print(dataset_metadata_df.head()) print(dataset_metadata_df.head())
dataset_name_idx = dataset_metadata_df.index[(dataset_metadata_df['dataset_name']==args.dataset_name).to_numpy()] dataset_name_idx = dataset_metadata_df.index[(dataset_metadata_df['dataset_name']==args.dataset_name).to_numpy()]
data_table_metadata = dataset_metadata_df.loc[dataset_name_idx,:] data_table_metadata = dataset_metadata_df.loc[dataset_name_idx,:]
parent_instrument = data_table_metadata.loc[dataset_name_idx,'parent_instrument'].values[0] parent_instrument = data_table_metadata.loc[dataset_name_idx,'parent_instrument'].values[0]
parent_file = data_table_metadata.loc[dataset_name_idx,'parent_file'].values[0] parent_file = data_table_metadata.loc[dataset_name_idx,'parent_file'].values[0]
dataManager.unload_file_obj() dataManager.unload_file_obj()
print(args.calibration_file) print(args.calibration_file)
with open(args.calibration_file, 'r') as stream: with open(args.calibration_file, 'r') as stream:
calibration_factors = yaml.load(stream, Loader=yaml.FullLoader) calibration_factors = yaml.load(stream, Loader=yaml.FullLoader)
except Exception as e: except Exception as e:
print(f"Error loading input files: {e}") print(f"Error loading input files: {e}")
exit(1) exit(1)
path_to_output_dir, ext = os.path.splitext(args.data_file) path_to_output_dir, ext = os.path.splitext(args.data_file)
print('Path to output directory :', path_to_output_dir) print('Path to output directory :', path_to_output_dir)
# Perform calibration # Perform calibration
try: try:
processingScriptRelPath = os.path.relpath(thisFilePath,start=projectPath) processingScriptRelPath = os.path.relpath(thisFilePath,start=projectPath)
print(processingScriptRelPath) print(processingScriptRelPath)
metadata = {'actris_level' : 1, 'processing_script': processingScriptRelPath.replace(os.sep,'/')} metadata = {'actris_level' : 1, 'processing_script': processingScriptRelPath.replace(os.sep,'/')}
path_to_output_file, ext = os.path.splitext('/'.join([path_to_output_dir,parent_instrument,parent_file])) path_to_output_file, ext = os.path.splitext('/'.join([path_to_output_dir,parent_instrument,parent_file]))
path_to_calibrated_file = ''.join([path_to_output_file, '_calibrated.csv']) path_to_calibrated_file = ''.join([path_to_output_file, '_calibrated.csv'])
#path_tail, path_head = os.path.split(path_to_calibrated_file) #path_tail, path_head = os.path.split(path_to_calibrated_file)
#path_to_metadata_file = '/'.join([path_tail, 'data_lineage_metadata.json']) #path_to_metadata_file = '/'.join([path_tail, 'data_lineage_metadata.json'])
print('Path to output file :', path_to_calibrated_file) print('Path to output file :', path_to_calibrated_file)
import dima.utils.g5505_utils as utils import dima.utils.g5505_utils as utils
import json import json
calibrated_table = apply_calibration_factors(data_table, calibration_factors) calibrated_table = apply_calibration_factors(data_table, calibration_factors)
metadata['processing_date'] = utils.created_at() metadata['processing_date'] = utils.created_at()
calibrated_table.to_csv(path_to_calibrated_file, index=False) calibrated_table.to_csv(path_to_calibrated_file, index=False)
status = record_data_lineage(path_to_calibrated_file, metadata) status = record_data_lineage(path_to_calibrated_file, metadata)
print(f"Calibrated data saved to {path_to_calibrated_file}") print(f"Calibrated data saved to {path_to_calibrated_file}")
except Exception as e: except Exception as e:
print(f"Error during calibration: {e}") print(f"Error during calibration: {e}")
exit(1) exit(1)