5505-public/dima
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Moved src/data_integration_lib.py -> pipelines/data_integration.py

Browse Source
This commit is contained in:
florez_j
2024-09-17 15:32:23 +02:00
parent 2dd033bcb3
commit 07401c895f
1 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
pipelines/data_integration.py Normal file
View File

@ -0,0 +1,213 @@
import sys
import os
root_dir = os.path.abspath(os.curdir)
sys.path.append(root_dir)
import src.hdf5_lib as hdf5_lib
import utils.g5505_utils as utils
import yaml
import logging
from datetime import datetime
# Importing chain class from itertools
from itertools import chain
from instruments.readers import filereader_registry
allowed_file_extensions = filereader_registry.file_extensions
from datetime import datetime
def load_config_and_setup_logging(yaml_config_file_path, log_dir):
"""Load YAML configuration file, set up logging, and validate required keys and datetime_steps."""
# Define required keys
required_keys = [
'experiment', 'contact', 'input_file_directory', 'output_file_directory',
'instrument_datafolder', 'project', 'actris_level'
]
# Supported integration modes
supported_integration_modes = ['collection', 'single_experiment']
# Set up logging
date = utils.created_at()
utils.setup_logging(log_dir, f"integrate_data_sources_{date}.log")
# Load YAML configuration file
with open(yaml_config_file_path, 'r') as stream:
try:
config_dict = yaml.load(stream, Loader=yaml.FullLoader)
except yaml.YAMLError as exc:
logging.error("Error loading YAML file: %s", exc)
raise ValueError(f"Failed to load YAML file: {exc}")
# Check if required keys are present
missing_keys = [key for key in required_keys if key not in config_dict]
if missing_keys:
raise KeyError(f"Missing required keys in YAML configuration: {missing_keys}")
# Validate integration_mode
integration_mode = config_dict.get('integration_mode', 'collection') # Default to 'collection'
if integration_mode not in supported_integration_modes:
raise ValueError(f"Unsupported integration_mode '{integration_mode}'. Supported modes are {supported_integration_modes}")
# Get dataset start and end dates (optional)
dataset_startdate = config_dict.get('dataset_startdate')
dataset_enddate = config_dict.get('dataset_enddate')
# Validate datetime_steps format if it exists
if 'datetime_steps' in config_dict:
datetime_steps = config_dict['datetime_steps']
expected_format = '%Y-%m-%d %H-%M-%S'
# Check if datetime_steps is a list or a falsy value
if datetime_steps and not isinstance(datetime_steps, list):
raise TypeError(f"datetime_steps should be a list of strings or a falsy value (None, empty), but got {type(datetime_steps)}")
for step_idx, step in enumerate(datetime_steps):
try:
# Attempt to parse the datetime to ensure correct format
config_dict['datetime_steps'][step_idx] = datetime.strptime(step, expected_format)
except ValueError:
raise ValueError(f"Invalid datetime format for '{step}'. Expected format: {expected_format}")
# Set default dates from datetime_steps if not provided
if not dataset_startdate:
dataset_startdate = min(config_dict['datetime_steps']).strftime('%Y-%m-%d') # Earliest datetime step
if not dataset_enddate:
dataset_enddate = max(config_dict['datetime_steps']).strftime('%Y-%m-%d') # Latest datetime step
else:
# If datetime_steps is not present, set the integration mode to 'collection'
logging.info("datetime_steps missing, setting integration_mode to 'collection'.")
config_dict['integration_mode'] = 'collection'
# Set sensible defaults using the current date
current_date = datetime.now().strftime('%Y-%m-%d')
if not dataset_startdate:
dataset_startdate = current_date
if not dataset_enddate:
dataset_enddate = current_date
config_dict['expected_datetime_format'] = '%Y-%m-%d %H-%M-%S'
# Add dataset dates to config for use later
config_dict['dataset_startdate'] = dataset_startdate
config_dict['dataset_enddate'] = dataset_enddate
return config_dict
def generate_datetime_dict(datetime_steps):
""" Generate the datetime augment dictionary from datetime steps. """
datetime_augment_dict = {}
for datetime_step in datetime_steps:
#tmp = datetime.strptime(datetime_step, '%Y-%m-%d %H-%M-%S')
datetime_augment_dict[datetime_step] = [
datetime_step.strftime('%Y-%m-%d'), datetime_step.strftime('%Y_%m_%d'), datetime_step.strftime('%Y.%m.%d'), datetime_step.strftime('%Y%m%d')
]
return datetime_augment_dict
def integrate_data_sources(yaml_config_file_path, log_dir='logs/'):
""" Integrates data sources specified by the input configuration file into HDF5 files.
Parameters:
yaml_config_file_path (str): Path to the YAML configuration file.
log_dir (str): Directory to save the log file.
Returns:
list: List of Paths to the created HDF5 file(s).
"""
def output_filename(name, date, initials):
return f"{name}_{date}_{initials}.h5"
config_dict = load_config_and_setup_logging(yaml_config_file_path, log_dir)
exp_campaign_name = config_dict['experiment']
initials = config_dict['contact']
path_to_input_dir = config_dict['input_file_directory']
path_to_output_dir = config_dict['output_file_directory']
select_dir_keywords = config_dict['instrument_datafolder']
root_metadata_dict = {
'project' : config_dict['project'],
'experiment' : config_dict['experiment'],
'contact' : config_dict['contact'],
'actris_level': config_dict['actris_level']
}
# Get dataset start and end dates.
dataset_startdate = config_dict['dataset_startdate']
dataset_enddate = config_dict['dataset_enddate']
# Handle datetime_steps and integration mode
datetime_augment_dict = {}
select_file_keywords = []
if 'datetime_steps' in config_dict and config_dict['datetime_steps']:
# We augment each datetime step with various datetime formats to compensate for filenaming inconsistencies
datetime_augment_dict = generate_datetime_dict(config_dict['datetime_steps'])
# Create flattened datetime list of datetime object file keywords
select_file_keywords = list(chain.from_iterable(datetime_augment_dict.values()))
# Determine mode and process accordingly
output_filename_path = []
if 'single_experiment' in config_dict.get('integration_mode', 'collection') and datetime_augment_dict:
# Single experiment mode
for datetime_step in datetime_augment_dict.keys():
date_str = datetime_step.strftime('%Y-%m-%d')
select_file_keywords = datetime_augment_dict[datetime_step]
filename = output_filename(exp_campaign_name, initials, date_str)
path_to_rawdata_folder = os.path.splitext(os.path.join(path_to_output_dir, filename))[0]
# Step 1: Search through input directory and make a copy in the output directory that complies with naming constraints
path_to_input_dir_filenames_dict = utils.copy_directory_with_contraints(path_to_input_dir,
path_to_rawdata_folder,
select_dir_keywords,
select_file_keywords,
allowed_file_extensions)
# Step 2: Create HDF5 file
path_to_integrated__stepwise_hdf5_file = hdf5_lib.create_hdf5_file_from_filesystem_path(path_to_rawdata_folder,
path_to_input_dir_filenames_dict,
select_dir_keywords,
root_metadata_dict)
output_filename_path.append(path_to_integrated__stepwise_hdf5_file)
else: # collection
# Collection mode or no datetime_steps
date_str = f'{dataset_startdate}_{dataset_enddate}'
filename = output_filename(exp_campaign_name, initials, date_str)
path_to_rawdata_folder = os.path.splitext(os.path.join(path_to_output_dir, filename))[0]
logging.info("Creating copy of specified experimental campaign folder at: %s", path_to_output_dir)
# Step 1: Search through input directory and make a copy in the output directory that complies with naming constraints
path_to_input_dir_filenames_dict = utils.copy_directory_with_contraints(path_to_input_dir,
path_to_rawdata_folder,
select_dir_keywords,
select_file_keywords,
allowed_file_extensions)
# Step 2: Create HDF5 file
logging.info("Creating HDF5 file at: %s", path_to_output_dir)
path_to_integrated_hdf5_file = hdf5_lib.create_hdf5_file_from_filesystem_path(path_to_rawdata_folder,
path_to_input_dir_filenames_dict,
select_dir_keywords,
root_metadata_dict)
output_filename_path.append(path_to_integrated_hdf5_file )
return output_filename_path