APOG_public/acsmnode
0
0
Fork 0
mirror of https://gitea.psi.ch/APOG/acsmnode.git synced 2025-06-28 12:30:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of https://gitlab.psi.ch/apog/acsmnode

Browse Source
This commit is contained in:
florez_j
2025-01-30 13:52:13 +01:00
parent 1061163e77 8b25bdb990
commit fa9a20c35c
5 changed files with 339 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
pipelines/params/calibration_factors.yaml
View File

@ -1,103 +1,103 @@
# Define common factors # Define common factors
# Get values from data/<station>/<year>/config_acsm_<year>.r, values used in Tofware analysis. # Get values from data/<station>/<year>/config_acsm_<year>.r, values used in Tofware analysis.
IE: &IE 145.9 IE: &IE 145.9
ABRefWave: &ABRefWave 254000 ABRefWave: &ABRefWave 254000
RIE_SO4: &RIE_SO4 0.63 RIE_SO4: &RIE_SO4 0.63
RIE_NH4: &RIE_NH4 3.495 RIE_NH4: &RIE_NH4 3.495
RIE_Org: &RIE_Org 1.4 RIE_Org: &RIE_Org 1.4
# Get values from data/<station>/<year>/cal.csv # Get values from data/<station>/<year>/cal.csv
IE_correct: &IE_correct 145.9 IE_correct: &IE_correct 145.9
AB_ref_correct: &AB_ref_correct 254000 AB_ref_correct: &AB_ref_correct 254000
RIE_SO4_correct: &RIE_SO4_correct 0.63 RIE_SO4_correct: &RIE_SO4_correct 0.63
RIE_NH4_correct: &RIE_NH4_correct 3.495 RIE_NH4_correct: &RIE_NH4_correct 3.495
RIE_Org_correct: &RIE_Org_correct 1.4 RIE_Org_correct: &RIE_Org_correct 1.4
flow_ref_correct: &flow_ref_correct 1.36 flow_ref_correct: &flow_ref_correct 1.36
# Define mappings for associated variables # Define mappings for associated variables
variables: variables:
# all_dat[, NO3_correct := (NO3_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, NO3_correct := (NO3_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
NO3_11000: NO3_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, SO4_correct := (SO4_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)] # all_dat[, SO4_correct := (SO4_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)]
SO4_11000: SO4_11000:
num: [*IE, *RIE_SO4, *AB_ref_correct] num: [*IE, *RIE_SO4, *AB_ref_correct]
den: [*IE_correct, *RIE_SO4_correct, *ABRefWave] den: [*IE_correct, *RIE_SO4_correct, *ABRefWave]
# all_dat[, NH4_correct := (NH4_11000 * IE * RIE_NH4 * AB_ref_correct) / (IE_correct * RIE_NH4_correct * ABRefWave)] # all_dat[, NH4_correct := (NH4_11000 * IE * RIE_NH4 * AB_ref_correct) / (IE_correct * RIE_NH4_correct * ABRefWave)]
NH4_11000: NH4_11000:
num: [*IE, *RIE_NH4, *AB_ref_correct] num: [*IE, *RIE_NH4, *AB_ref_correct]
den: [*IE_correct, *RIE_NH4_correct, *ABRefWave] den: [*IE_correct, *RIE_NH4_correct, *ABRefWave]
# all_dat[, Org_correct := (Org_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, Org_correct := (Org_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
Org_11000: Org_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, Chl_correct := (Chl_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, Chl_correct := (Chl_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
Chl_11000: Chl_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, Org_44_11000_correct := (Org_44_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, Org_44_11000_correct := (Org_44_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
Org_44_11000: Org_44_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, Org_43_11000_correct := (Org_43_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, Org_43_11000_correct := (Org_43_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
Org_43_11000: Org_43_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, Org_60_11000_correct := (Org_60_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, Org_60_11000_correct := (Org_60_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
Org_60_11000: Org_60_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, NO3_30_11000_correct := (NO3_30_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)] # all_dat[, NO3_30_11000_correct := (NO3_30_11000 * IE * AB_ref_correct) / (IE_correct * ABRefWave)]
NO3_30_11000: NO3_30_11000:
num: [*IE, *AB_ref_correct] num: [*IE, *AB_ref_correct]
den: [*IE_correct, *ABRefWave] den: [*IE_correct, *ABRefWave]
# all_dat[, SO4_98_11000_correct := (SO4_98_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)] # all_dat[, SO4_98_11000_correct := (SO4_98_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)]
SO4_98_11000: SO4_98_11000:
num: [*IE, *RIE_SO4, *AB_ref_correct] num: [*IE, *RIE_SO4, *AB_ref_correct]
den: [*IE_correct, *RIE_SO4_correct, *ABRefWave] den: [*IE_correct, *RIE_SO4_correct, *ABRefWave]
# all_dat[, SO4_81_11000_correct := (SO4_81_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)] # all_dat[, SO4_81_11000_correct := (SO4_81_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)]
SO4_81_11000: SO4_81_11000:
num: [*IE, *RIE_SO4, *AB_ref_correct] num: [*IE, *RIE_SO4, *AB_ref_correct]
den: [*IE_correct, *RIE_SO4_correct, *ABRefWave] den: [*IE_correct, *RIE_SO4_correct, *ABRefWave]
# all_dat[, SO4_82_11000_correct := (SO4_82_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)] # all_dat[, SO4_82_11000_correct := (SO4_82_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)]
SO4_82_11000: SO4_82_11000:
num: [*IE, *RIE_SO4, *AB_ref_correct] num: [*IE, *RIE_SO4, *AB_ref_correct]
den: [*IE_correct, *RIE_SO4_correct, *ABRefWave] den: [*IE_correct, *RIE_SO4_correct, *ABRefWave]
# all_dat[, SO4_62_11000_correct := (SO4_62_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)] # all_dat[, SO4_62_11000_correct := (SO4_62_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)]
SO4_62_11000: SO4_62_11000:
num: [*IE, *RIE_SO4, *AB_ref_correct] num: [*IE, *RIE_SO4, *AB_ref_correct]
den: [*IE_correct, *RIE_SO4_correct, *ABRefWave] den: [*IE_correct, *RIE_SO4_correct, *ABRefWave]
# all_dat[, SO4_48_11000_correct := (SO4_48_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)] # all_dat[, SO4_48_11000_correct := (SO4_48_11000 * IE * RIE_SO4 * AB_ref_correct) / (IE_correct * RIE_SO4_correct * ABRefWave)]
SO4_48_11000: SO4_48_11000:
num: [*IE, *RIE_SO4, *AB_ref_correct] num: [*IE, *RIE_SO4, *AB_ref_correct]
den: [*IE_correct, *RIE_SO4_correct, *ABRefWave] den: [*IE_correct, *RIE_SO4_correct, *ABRefWave]

18
pipelines/params/diagnostic_variable_limits.yaml
View File

@ -1,18 +0,0 @@
# Define limits for diagnostic variables (src: data/<station>/<year>/global_config.r)
VaporizerTemp_C :
lower_lim : {value : 400, description : "heater"}
upper_lim : {value : 610, description : "heater"}
ABsamp :
lower_lim : {value : 20000, description : "not specified yet"}
upper_lim : {value : 500000, description : "not specified yet"}
FlowRate_ccs :
lower_lim : {value : 1.23, description : "not specified yet"}
upper_lim : {value : 1.45, description : "not specified yet"}
FilamentEmission_mA :
lower_lim : {value : 0.65, description : "not specified yet"}
upper_lim : {value : 1.5, description : "not specified yet"}

36
pipelines/params/limits_of_detection.yaml Normal file
View File

@ -0,0 +1,36 @@
# Get values from data/<station>/<year>/config_acsm_2023.r
LOD :
standard_name : "limit_of_detection"
description : "Limit of detection for various variables, at different temporal resolutions"
datetime : "2023-07-06"
variables:
NO3_11000 :
resolution :
40s : 0.21
4m : 0.09
1h : 0.02
SO4_11000:
resolution :
40s : 0.24
4m : 0.10
1h : 0.03
NH4_11000:
resolution :
40s : 0.98
4m : 0.40
1h : 0.1
Org_11000:
resolution :
40s : 0.12
4m : 0.51
1h : 0.13
Chl_11000:
resolution :
40s : 1.26
4m : 0.05
1h : 0.01

22
pipelines/params/operational_variable_ranges.yaml Normal file
View File

@ -0,0 +1,22 @@
# Get values from data/<station>/<year>/global_config.r
# Define operational ranges for diagnostic variables
operational_range:
description : "Defines the value range of a particular variable"
variables:
VaporizerTemp_C :
lower_lim : 400
upper_lim : 610
description : "heater temperature"
ABsamp :
lower_lim : 20000
upper_lim : 500000
FlowRate_ccs :
lower_lim : 1.23
upper_lim : 1.45
FilamentEmission_mA :
lower_lim : 0.65
upper_lim : 1.5

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)