sdap_ingest_manager/collections_ingester/collection_ingestion.py
def collection_row_callback(collection,
                            collection_config_template,
                            granule_file_list_root_path,
                            dataset_configuration_root_path,
                            history_root_path,
                            deconstruct_nfs=False,
                            **pods_run_kwargs
                            ):
    """ Create the configuration and launch the ingestion
        for the given collection row
    """
    dataset_id = collection['id']
    netcdf_variable = collection['variable']
    netcdf_file_pattern = collection['path']

    if 'forward_processing' in collection.keys():
        forward_processing = collection['forward_processing']
    else:
        forward_processing = False

    granule_list_file_path = os.path.join(granule_file_list_root_path,
                                          f'{dataset_id}-granules.lst')
    dataset_ingestion_history_manager = sdap_ingest_manager.history_manager\
        .DatasetIngestionHistoryFile(history_root_path, dataset_id, lambda x: str(os.path.getmtime(x)))


    time_range = {}
    for time_boundary in {"from", "to"}:
        if time_boundary in collection.keys() and collection[time_boundary]:
            # add prefix "from" because is a reserved name which can not be used as function argument
            time_range[f'date_{time_boundary}'] = datetime.fromisoformat(collection[time_boundary])
            logger.info(f"time criteria {time_boundary} is {time_range[f'date_{time_boundary}']}")

    create_granule_list(netcdf_file_pattern,
                        dataset_ingestion_history_manager,
                        granule_list_file_path,
                        deconstruct_nfs=deconstruct_nfs,
                        **time_range,
                        forward_processing=forward_processing)

    dataset_configuration_file_path = os.path.join(dataset_configuration_root_path,
                                                   f'{dataset_id}-config.yml')

    create_dataset_config(dataset_id,
                          netcdf_variable,
                          collection_config_template,
                          dataset_configuration_file_path)
    cwd = os.getcwd()
    # group must match the following regex (([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?
    # and when suffixed with uid must not exceed 64 characters
    prog = re.compile(GROUP_PATTERN)
    group = prog.match(dataset_id[:19])
    if group is None:
        group_name = GROUP_DEFAULT_NAME
    else:
        group_name = group.group(0)
    pods_run_kwargs['file_list_path'] = granule_list_file_path
    pods_run_kwargs['job_config'] = dataset_configuration_file_path
    pods_run_kwargs['job_group'] = group_name
    pods_run_kwargs['ningester_version'] = '1.1.0'
    pods_run_kwargs['delete_successful'] = True
    pods_run_kwargs['hist_manager'] = dataset_ingestion_history_manager

    def param_to_str_arg(k, v):
        k_with_dash = k.replace('_', '-')
        str_k = f'--{k_with_dash}'
        if type(v) == bool:
            if v:
                return [str_k]
            else:
                return []
        elif isinstance(v, list):
            return [str_k, ','.join(v)]
        else:
            return [str_k, str(v)]

    pod_launch_options = [param_to_str_arg(k, v) for (k,v) in pods_run_kwargs.items()]
    flat_pod_launch_options = [item for option in pod_launch_options for item in option]
    pod_launch_cmd = ['run_granules'] + flat_pod_launch_options
    logger.info("launch pod with command:\n%s", " ".join(pod_launch_cmd))
    sdap_ingest_manager.granule_ingester.create_and_run_jobs(**pods_run_kwargs)
 Create the configuration and launch the ingestion
        for the given collection row
    
Create the configuration and launch the ingestion
for the given collection row
Create the configuration and launch the ingestion for the given collection row
def collection_row_callback(collection,
                            collection_config_template,
                            granule_file_list_root_path,
                            dataset_configuration_root_path,
                            history_root_path,
                            deconstruct_nfs=False,
                            **pods_run_kwargs
                            ):
    dataset_id = collection['id']
    netcdf_variable = collection['variable']
    netcdf_file_pattern = collection['path']
    if 'forward_processing' in collection.keys():
        forward_processing = collection['forward_processing']
    else:
        forward_processing = False
    granule_list_file_path = os.path.join(granule_file_list_root_path,
                                          f'{dataset_id}-granules.lst')
    dataset_ingestion_history_manager = sdap_ingest_manager.history_manager\
        .DatasetIngestionHistoryFile(history_root_path, dataset_id, lambda x: str(os.path.getmtime(x)))
    time_range = {}
    for time_boundary in {"from", "to"}:
        if time_boundary in collection.keys() and collection[time_boundary]:
            time_range[f'date_{time_boundary}'] = datetime.fromisoformat(collection[time_boundary])
            logger.info(f"time criteria {time_boundary} is {time_range[f'date_{time_boundary}']}")
    create_granule_list(netcdf_file_pattern,
                        dataset_ingestion_history_manager,
                        granule_list_file_path,
                        deconstruct_nfs=deconstruct_nfs,
                        **time_range,
                        forward_processing=forward_processing)
    dataset_configuration_file_path = os.path.join(dataset_configuration_root_path,
                                                   f'{dataset_id}-config.yml')
    create_dataset_config(dataset_id,
                          netcdf_variable,
                          collection_config_template,
                          dataset_configuration_file_path)
    cwd = os.getcwd()
    prog = re.compile(GROUP_PATTERN)
    group = prog.match(dataset_id[:19])
    if group is None:
        group_name = GROUP_DEFAULT_NAME
    else:
        group_name = group.group(0)
    pods_run_kwargs['file_list_path'] = granule_list_file_path
    pods_run_kwargs['job_config'] = dataset_configuration_file_path
    pods_run_kwargs['job_group'] = group_name
    pods_run_kwargs['ningester_version'] = '1.1.0'
    pods_run_kwargs['delete_successful'] = True
    pods_run_kwargs['hist_manager'] = dataset_ingestion_history_manager
    def param_to_str_arg(k, v):
        k_with_dash = k.replace('_', '-')
        str_k = f'--{k_with_dash}'
        if type(v) == bool:
            if v:
                return [str_k]
            else:
                return []
        elif isinstance(v, list):
            return [str_k, ','.join(v)]
        else:
            return [str_k, str(v)]
    pod_launch_options = [param_to_str_arg(k, v) for (k,v) in pods_run_kwargs.items()]
    flat_pod_launch_options = [item for option in pod_launch_options for item in option]
    pod_launch_cmd = ['run_granules'] + flat_pod_launch_options
    logger.info("launch pod with command:\n%s", " ".join(pod_launch_cmd))
    sdap_ingest_manager.granule_ingester.create_and_run_jobs(**pods_run_kwargs)
def collection_row_callback ( collection , collection_config_template , granule_file_list_root_path , dataset_configuration_root_path , history_root_path , deconstruct_nfs = False , ** pods_run_kwargs ) : dataset_id = collection [ 'id' ] netcdf_variable = collection [ 'variable' ] netcdf_file_pattern = collection [ 'path' ] if 'forward_processing' in collection . keys ( ) : forward_processing = collection [ 'forward_processing' ] else : forward_processing = False granule_list_file_path = os . path . join ( granule_file_list_root_path , f'{dataset_id}-granules.lst' ) dataset_ingestion_history_manager = sdap_ingest_manager . history_manager . DatasetIngestionHistoryFile ( history_root_path , dataset_id , lambda x : str ( os . path . getmtime ( x ) ) ) time_range = { } for time_boundary in { "from" , "to" } : if time_boundary in collection . keys ( ) and collection [ time_boundary ] : time_range [ f'date_{time_boundary}' ] = datetime . fromisoformat ( collection [ time_boundary ] ) logger . info ( f"time criteria {time_boundary} is {time_range[f'date_{time_boundary}']}" ) create_granule_list ( netcdf_file_pattern , dataset_ingestion_history_manager , granule_list_file_path , deconstruct_nfs = deconstruct_nfs , ** time_range , forward_processing = forward_processing ) dataset_configuration_file_path = os . path . join ( dataset_configuration_root_path , f'{dataset_id}-config.yml' ) create_dataset_config ( dataset_id , netcdf_variable , collection_config_template , dataset_configuration_file_path ) cwd = os . getcwd ( ) prog = re . compile ( GROUP_PATTERN ) group = prog . match ( dataset_id [ : 19 ] ) if group is None : group_name = GROUP_DEFAULT_NAME else : group_name = group . group ( 0 ) pods_run_kwargs [ 'file_list_path' ] = granule_list_file_path pods_run_kwargs [ 'job_config' ] = dataset_configuration_file_path pods_run_kwargs [ 'job_group' ] = group_name pods_run_kwargs [ 'ningester_version' ] = '1.1.0' pods_run_kwargs [ 'delete_successful' ] = True pods_run_kwargs [ 'hist_manager' ] = dataset_ingestion_history_manager def param_to_str_arg ( k , v ) : k_with_dash = k . replace ( '_' , '-' ) str_k = f'--{k_with_dash}' if type ( v ) == bool : if v : return [ str_k ] else : return [ ] elif isinstance ( v , list ) : return [ str_k , ',' . join ( v ) ] else : return [ str_k , str ( v ) ] pod_launch_options = [ param_to_str_arg ( k , v ) for ( k , v ) in pods_run_kwargs . items ( ) ] flat_pod_launch_options = [ item for option in pod_launch_options for item in option ] pod_launch_cmd = [ 'run_granules' ] + flat_pod_launch_options logger . info ( "launch pod with command:\n%s" , " " . join ( pod_launch_cmd ) ) sdap_ingest_manager . granule_ingester . create_and_run_jobs ( ** pods_run_kwargs )
Create the configuration and launch the ingestion
for the given collection row
Create the configuration and launch the ingestion for the given collection row
""" Create the configuration and launch the ingestion
        for the given collection row
    """ # add prefix "from" because is a reserved name which can not be used as function argument # group must match the following regex (([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])? # and when suffixed with uid must not exceed 64 characters
collection
collection_config_template
granule_file_list_root_path
dataset_configuration_root_path
history_root_path
deconstruct_nfs