---
pipeline:
  name: parQuantumLine_level1_group_consolidate_srf
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-levl1-grp-cons-srf:v2.1.0
  cmd:
  - sh
  - "-c"
  - |-
    /bin/bash <<'EOF'
    # Use bash-scrict mode. See http://redsymbol.net/articles/unofficial-bash-strict-mode/
    set -euo pipefail
    IFS=$'\n\t'
    #
    # Refresh interim directories with each datum (otherwise they persist and cause probs)
    rm -rf /tmp/interimA
    rm -rf /tmp/interimB
    rm -rf /tmp/pfs/interimC
    mkdir /tmp/interimA
    mkdir /tmp/interimB
    mkdir -p /tmp/pfs/interimC
    #
    # Set some environment variables for the first module
    export OUT_PATH=$OUT_PATH_1
    #
    # Run first module - filter-joiner (using environment variables below as input parameters)
    python3 -m filter_joiner.filter_joiner_main
    # Set some environment variables for the second module
    export OUT_PATH=$OUT_PATH_2
    export CONFIG=$CONFIG2
    #
    # Run second module - filter-joiner to bring in SRF (using environment variables below as input parameters)
    python3 -m filter_joiner.filter_joiner_main
    # Clean up 1st interim directory (this is the only one we can clean up bc the rest use symlinks)
    rm -rf /tmp/interimA  
    # Set some environment variables for the 3rd module
    export OUT_PATH=$OUT_PATH_3
    #
    # Run third module - level 1 consolidate (using environment variables below as input parameters)
    python3 -m level1_consolidate.level1_consolidate_main
    #
    # Run fourth module - pub workbook loader (using environment variables below as input parameters)
    python3 -m pub_workbook_loader.pub_workbook_loader_main
    #
    # Run fifth and final module - create pub tables and apply science review flags (if any)
    Rscript ./flow.pub.tabl.srf.R \
      DirIn=/tmp/pfs/interimC \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      "DirData=stats|quality_metrics" \
      PathPubWb=$PUB_WORKBOOKS \
      "DirSubCopy=science_review_flags|group|location"
    #
    #
    # Export Level 1 data to bucket
    export OUT_PATH=/pfs/out
    linkdir=$(mktemp -d)
    shopt -s globstar
    out_parquet_glob="${OUT_PATH}/**/*.parquet"
    # Example: /2024/01/18/par-quantum-line_UKFS001000/data/par-quantum-line_UKFS001000_2024-01-18_PARQL_1min_001.parquet
    echo "Linking output files to ${linkdir}"
    #set -x # Echo commands to output for debugging
    fname=""
    for f in $out_parquet_glob; do
      if [[ -f "$f" ]]; then
        # Parse the path
        [[ "$f" =~ ^$OUT_PATH/([0-9]+)/([0-9]+)/([0-9]+)/(${GROUP_PREFIX}_[A-Za-z0-9]+)/data/(.*)$ ]]
        fyear="${BASH_REMATCH[1]}"
        fmonth="${BASH_REMATCH[2]}"
        fday="${BASH_REMATCH[3]}"
        fgroup="${BASH_REMATCH[4]}"
        fname="${BASH_REMATCH[5]}"
        # Now get the timing index from the file name
        [[ "$fname" =~ ^${GROUP_PREFIX}_[A-Za-z0-9]+_${fyear}-${fmonth}-${fday}_[A-Za-z0-9]+_([A-Za-z0-9]+)_([A-Za-z0-9]+).parquet ]]
        avg_int="${BASH_REMATCH[1]}"
        #Form the output path and link
        outdir="${linkdir}/v2/${GROUP_PREFIX}/${avg_int}/group=${fgroup}/ms=${fyear}-${fmonth}"
        mkdir -p "${outdir}"
        ln -s "${f}" "${outdir}/${fname}"
      fi
    done
    #set +x
    if [[ "${fname}" ]]; then
      echo "Syncing files to bucket"
      rclone \
        --no-check-dest \
        --copy-links \
        --gcs-bucket-policy-only \
        --gcs-no-check-bucket \
        copy \
        "${linkdir}" \
        ":gcs://${BUCKET_NAME}"
      echo "Removing temporary files"
      rm -rf $linkdir
    fi
    EOF
  env:
    # Environment variables for 1st filter-joiner. Need to join by day again here because an outer join was used on 
    # these repos in order to pull them in with or without the SRF
    CONFIG: |
      ---
      # In Pachyderm root will be index 0, 'pfs' index 1, and the repo name index 2.
      # Metadata indices will typically begin at index 3.
      # Use unix-style glob pattern to select the desired directories in each repo 
      input_paths:
        - path:
            name: QUALITY_METRICS_PATH
            # Filter for data directory
            glob_pattern: /pfs/QUALITY_METRICS_PATH/*/*/*/*/**
            # Join on Y/M/D/group ID 
            join_indices: [3,4,5,6]
        - path:
            name: STATISTICS_PATH
            # Filter for data directory
            glob_pattern: /pfs/STATISTICS_PATH/*/*/*/*/**
            # Join on Y/M/D/group ID 
            join_indices: [3,4,5,6]
        - path:
            name: GROUP_PATH
            # Grab group information
            glob_pattern: /pfs/GROUP_PATH/*/*/*/*/group/**
            # Join on Y/M/D/group ID 
            join_indices: [3,4,5,6]
        - path:
            name: GROUP_PATH
            # Grab location information
            glob_pattern: /pfs/GROUP_PATH/*/*/*/*/*/*/location/**
            # Join on Y/M/D/group ID 
            join_indices: [3,4,5,6]
    OUT_PATH_1: /tmp/interimA # Transfered to OUT_PATH for the first module
    RELATIVE_PATH_INDEX: "3" # This is shared among the 2 filter joiners and consolidation module
    LINK_TYPE: COPY # options are COPY or SYMLINK. Use COPY for combined modules. Also shared with 2nd & 3rd modules 
    LOG_LEVEL: INFO # Shared among all modules

# Below are the environment variables for 2nd filter-joiner bringing in the Science review flags 
# Can't do this in first filter-joiner bc there are only data in the srf assignment
# repo for groups that have applicable SRFs for the day. Need to pass through the
# consolidated output with an outer join.
    CONFIG2: |
      ---
      # In Pachyderm root will be index 0, 'pfs' index 1, and the repo name index 2.
      # Metadata indices will typically begin at index 3.
      # Use unix-style glob pattern to select the desired directories in each repo 
      input_paths:
        - path:
            name: CONSOLIDATED_PATH
            # Filter for data directory
            glob_pattern: /tmp/interimA/*/*/*/*/**
            # Join on group ID (already joined below by day)
            join_indices: [6]
            outer_join: True
        - path:
            name: SRF_PATH
            # Filter for data directory
            glob_pattern: /pfs/SRF_PATH/*/*/*/*/**
            # Join on group ID(already joined below by day)
            join_indices: [6]
    OUT_PATH_2: /tmp/interimB # This will be transfered to OUT_PATH for the this module

# Environment variables for level 1 consolidation
    IN_PATH: /tmp/interimB
    OUT_PATH_3: /tmp/pfs/interimC # This will be transfered to OUT_PATH for the second module
    GROUP_INDEX: "6" # path index of names of group-level metadata to include in the output
    GROUP_METADATA_INDEX: "7"
    GROUP_METADATA_NAMES: group,science_review_flags
    # path index of names of directories to include in the output
    DATA_TYPE_INDEX: "9"
    DATA_TYPE_NAMES: location,stats,quality_metrics
    
# Environment variables for pub_workbook_loader
    OUT_PATH_WORKBOOK: /tmp/pub_workbooks
    PRODUCTS: NEON.DOM.SITE.DP1.00066.001 # Format: NEON.DOM.SITE.DPX.XXXXX.XXX,NEON.DOM.SITE.DPX.XXXXX.XXX,etc
  
# Environment variables for pub table and srf module
    PUB_WORKBOOKS: /tmp/pub_workbooks
    PARALLELIZATION_INTERNAL: '2'

# Environment variables for the L1 archiver
    GROUP_PREFIX: par-quantum-line # no ending "_"

  secrets:
  - name: pdr-secret
    mount_path: /var/db_secret
  - name: l1-bucket
    env_var: BUCKET_NAME
    key: L1_BUCKET
    
input:
  join:
  - pfs:
      name: QUALITY_METRICS_PATH
      repo: parQuantumLine_qm_group_and_compute
      glob: /(*/*/*)
      joinOn: $1
      outer_join: true # Need outer join to pull in with or without SRFs
      empty_files: false # Make sure this is false for LINK_TYPE=COPY
  - pfs:
      name: STATISTICS_PATH
      repo: parQuantumLine_stats_group_and_compute
      glob: /(*/*/*)
      joinOn: $1
      outer_join: true # Need outer join to pull in with or without SRFs
      empty_files: false # Make sure this is false for LINK_TYPE=COPY
  - pfs:
      name: GROUP_PATH
      repo: parQuantumLine_group_path
      glob: /(*/*/*)
      joinOn: $1
      outer_join: true # Need outer join to pull in with or without SRFs
      empty_files: false # Make sure this is false for LINK_TYPE=COPY
  - pfs:
      name: SRF_PATH
      repo: parQuantumLine_srf_assignment
      glob: /(*/*/*)
      joinOn: $1
      empty_files: false # Make sure this is false for LINK_TYPE=COPY
parallelism_spec:
  constant: 5
autoscaling: true
resource_requests:
  memory: 1G
  cpu: 2.2
resource_limits:
  memory: 2G
  cpu: 3.5
sidecar_resource_requests:
  memory: 3G
  cpu: 0.5
datum_set_spec:
  number: 1
scheduling_spec:
  node_selector:
    cloud.google.com/gke-ephemeral-storage-local-ssd: "true"
    nodepool.neonscience.org/pipeline: "yes"
    cloud.google.com/gke-spot: "true"
pod_spec: |-
  { "tolerations": [
    {
      "key": "nodepool.neonscience.org/pipeline",
      "operator": "Exists"
    },
    {
      "effect": "NoSchedule",
      "key": "cloud.google.com/gke-spot",
      "operator": "Exists"
    }  
  ] }
