<commit_before>"""

Reorganisation step: copy files to local folder.

Input data are first copied to a local folder to speed-up processing.

Configuration variables used:

* :reorganisation section
    * INPUT_CONFIG: List of flags defining how incoming imaging data are organised.
    * MIN_FREE_SPACE: minimum percentage of free space available on local disk
* :reorganisation:copy_all_to_local section
    * OUTPUT_FOLDER: destination folder for the local copy

"""


from datetime import timedelta
from textwrap import dedent

from airflow import configuration
from airflow_pipeline.operators import BashPipelineOperator

from common_steps import Step, default_config


def copy_all_to_local_cfg(dag, upstream_step, reorganisation_section, step_section):
    default_config(reorganisation_section, 'INPUT_CONFIG', '')

    dataset_config = configuration.get(reorganisation_section, 'INPUT_CONFIG')
    min_free_space = configuration.getfloat(reorganisation_section, 'MIN_FREE_SPACE')
    output_folder = configuration.get(step_section, 'OUTPUT_FOLDER')

    return copy_all_to_local_step(dag, upstream_step, min_free_space, output_folder, dataset_config)


def copy_all_to_local_step(dag, upstream_step, min_free_space, output_folder, dataset_config):

    copy_all_to_local_cmd = dedent("""
        used="$(df -h /home | grep '/' | grep -Po '[^ ]*(?=%)')"
        if (( 101 - used < {{ params['min_free_space']|float * 100 }} )); then
          echo "Not enough space left, cannot continue"
          exit 1
        fi
        rsync -av $AIRFLOW_INPUT_DIR/ $AIRFLOW_OUTPUT_DIR/
    """)

    copy_all_to_local = BashPipelineOperator(
        task_id='copy_all_to_local',
        bash_command=copy_all_to_local_cmd,
        params={'min_free_space': min_free_space},
        output_folder_callable=lambda **kwargs: output_folder,
        parent_task=upstream_step.task_id,
        priority_weight=upstream_step.priority_weight,
        execution_timeout=timedelta(hours=3),
        dataset_config=dataset_config,
        organised_folder=False,
        dag=dag
    )
<commit_msg>fix copy_all_to_local<commit_after>"""

Reorganisation step: copy files to local folder.

Input data are first copied to a local folder to speed-up processing.

Configuration variables used:

* :reorganisation section
    * INPUT_CONFIG: List of flags defining how incoming imaging data are organised.
    * MIN_FREE_SPACE: minimum percentage of free space available on local disk
* :reorganisation:copy_all_to_local section
    * OUTPUT_FOLDER: destination folder for the local copy

"""


from datetime import timedelta
from textwrap import dedent

from airflow import configuration
from airflow_pipeline.operators import BashPipelineOperator

from common_steps import Step, default_config


def copy_all_to_local_cfg(dag, upstream_step, reorganisation_section, step_section):
    default_config(reorganisation_section, 'INPUT_CONFIG', '')

    dataset_config = configuration.get(reorganisation_section, 'INPUT_CONFIG')
    min_free_space = configuration.getfloat(reorganisation_section, 'MIN_FREE_SPACE')
    output_folder = configuration.get(step_section, 'OUTPUT_FOLDER')

    return copy_all_to_local_step(dag, upstream_step, min_free_space, output_folder, dataset_config)


def copy_all_to_local_step(dag, upstream_step, min_free_space, output_folder, dataset_config):

    copy_all_to_local_cmd = dedent("""
        set -e
        used="$(df -h $AIRFLOW_OUTPUT_DIR/ | grep '/' | grep -Po '[^ ]*(?=%)')"
        is_full=$(echo "101 - used < {{ params['min_free_space']|float * 100 }}" | bc)
        if [ "$is_full" == 1 ]; then
          echo "Not enough space left, cannot continue"
          exit 1
        fi
        rsync -av $AIRFLOW_INPUT_DIR/ $AIRFLOW_OUTPUT_DIR/
    """)

    copy_all_to_local = BashPipelineOperator(
        task_id='copy_all_to_local',
        bash_command=copy_all_to_local_cmd,
        params={'min_free_space': min_free_space},
        output_folder_callable=lambda relative_context_path, **kwargs: output_folder + '/' + relative_context_path,
        parent_task=upstream_step.task_id,
        priority_weight=upstream_step.priority_weight,
        execution_timeout=timedelta(hours=3),
        dataset_config=dataset_config,
        organised_folder=False,
        dag=dag
    )

{'fraction_non_alphanumeric': 0.08414690841469084, 'fraction_numerical': 0.00395165039516504, 'mean_word_length': 4.3189122373300375, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5204802', 'n_tokens_mistral': 1340, 'n_tokens_neox': 1293, 'n_words': 305}