<commit_before>"""Initialise the DAGs for all the pipelines required to process different datasets containing MRI images"""

# Please keep keywords airflow and DAG in this file, otherwise the safe mode in DagBag may skip this file

import logging

from airflow import configuration
from common_steps import default_config

from preprocessing_pipelines.mri_notify_failed_processing import mri_notify_failed_processing_dag
from preprocessing_pipelines.mri_notify_skipped_processing import mri_notify_skipped_processing_dag
from preprocessing_pipelines.mri_notify_successful_processing import mri_notify_successful_processing_dag
from preprocessing_pipelines.continuously_pre_process_incoming import continuously_preprocess_incoming_dag
from preprocessing_pipelines.daily_pre_process_incoming import daily_preprocess_incoming_dag
from preprocessing_pipelines.flat_pre_process_incoming import flat_preprocess_incoming_dag
from preprocessing_pipelines.pre_process_images import pre_process_images_dag
from etl_pipelines.daily_ehr_incoming import daily_ehr_incoming_dag
from etl_pipelines.flat_ehr_incoming import flat_ehr_incoming_dag
from etl_pipelines.ehr_to_i2b2 import ehr_to_i2b2_dag
from reorganisation_pipelines.flat_reorganise import flat_reorganisation_dag
from reorganisation_pipelines.reorganise import reorganise_dag


def register_dag(dag):
    dag_id = dag.dag_id
    var_name = "%s_dag" % dag_id
    globals()[var_name] = dag
    logging.info("Add DAG %s", dag_id)
    return dag_id


def register_reorganisation_dags(dataset, dataset_section, email_errors_to):
    reorganisation_section = dataset_section + ':reorganisation'
    default_config(reorganisation_section, 'INPUT_FOLDER_DEPTH', '1')

    reorganisation_input_folder = configuration.get(reorganisation_section, 'INPUT_FOLDER')
    depth = int(configuration.get(reorganisation_section, 'INPUT_FOLDER_DEPTH'))
    max_active_runs = int(configuration.get(reorganisation_section, 'MAX_ACTIVE_RUNS'))
    reorganisation_pipelines = configuration.get(reorganisation_section, 'PIPELINES').split(',')

    if reorganisation_pipelines and len(reorganisation_pipelines) > 0:
        reorganisation_dag_id = register_dag(reorganise_dag(dataset=dataset,
                                                            section=reorganisation_section,<commit_msg>bugfix<commit_after>"""Initialise the DAGs for all the pipelines required to process different datasets containing MRI images"""

# Please keep keywords airflow and DAG in this file, otherwise the safe mode in DagBag may skip this file

import logging

from airflow import configuration
from common_steps import default_config

from preprocessing_pipelines.mri_notify_failed_processing import mri_notify_failed_processing_dag
from preprocessing_pipelines.mri_notify_skipped_processing import mri_notify_skipped_processing_dag
from preprocessing_pipelines.mri_notify_successful_processing import mri_notify_successful_processing_dag
from preprocessing_pipelines.continuously_pre_process_incoming import continuously_preprocess_incoming_dag
from preprocessing_pipelines.daily_pre_process_incoming import daily_preprocess_incoming_dag
from preprocessing_pipelines.flat_pre_process_incoming import flat_preprocess_incoming_dag
from preprocessing_pipelines.pre_process_images import pre_process_images_dag
from etl_pipelines.daily_ehr_incoming import daily_ehr_incoming_dag
from etl_pipelines.flat_ehr_incoming import flat_ehr_incoming_dag
from etl_pipelines.ehr_to_i2b2 import ehr_to_i2b2_dag
from reorganisation_pipelines.flat_reorganise import flat_reorganisation_dag
from reorganisation_pipelines.reorganise import reorganise_dag


def register_dag(dag):
    dag_id = dag.dag_id
    var_name = "%s_dag" % dag_id
    globals()[var_name] = dag
    logging.info("Add DAG %s", dag_id)
    return dag_id


def register_reorganisation_dags(dataset, dataset_section, email_errors_to):
    reorganisation_section = dataset_section + ':reorganisation'
    default_config(reorganisation_section, 'INPUT_FOLDER_DEPTH', '1')

    reorganisation_input_folder = configuration.get(reorganisation_section, 'INPUT_FOLDER')
    depth = int(configuration.get(reorganisation_section, 'INPUT_FOLDER_DEPTH'))
    max_active_runs = int(configuration.get(reorganisation_section, 'MAX_ACTIVE_RUNS'))
    reorganisation_pipelines = configuration.get(reorganisation_section, 'PIPELINES').split(',')

    if reorganisation_pipelines and len(reorganisation_pipelines) > 0 and reorganisation_pipelines[0] != '':
        reorganisation_dag_id = register_dag(reorganise_dag(dataset=dataset,
                                                            section=reorganisation_section,
{'fraction_non_alphanumeric': 0.049282194128990785, 'fraction_numerical': 0.002785515320334262, 'mean_word_length': 6.871838111298482, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5069634', 'n_tokens_mistral': 1421, 'n_tokens_neox': 1375, 'n_words': 275}