---
pipeline:
  name: surfacewaterPhysical_qm_group_and_compute
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-qaqc-qm-grp:v2.0.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 -r -f /tmp/pfs/filter_joined
    mkdir -p /tmp/pfs/filter_joined
    
    # Run filter-joiner for Aquatroll data (using environment variables below as input parameters)
    export CONFIG=$CONFIG_AQUATROLL
    python3 -m filter_joiner.filter_joiner_main
    
    # Run quality metrics for aquatroll
    Rscript ./flow.qaqc.qm.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmQm=$FILE_SCHEMAS/surfacewaterPhysical_aquatroll200_dp01_quality_metrics.avsc \
      "WndwAgr=005|030" \
      "WghtAlphBeta=2|1" \
      Thsh=0.2 \
      "GrpQfAlph1=pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBeta1=pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBetaIgnr1=pressure:pressureNullQF|pressureGapQF" \
      "GrpQfAlph2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperaturePersistenceQF|zeroPressureQF" \
      "GrpQfBeta2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperaturePersistenceQF" \
      "GrpQfBetaIgnr2=temperature:temperatureNullQF|temperatureGapQF" \
      "GrpQfAlph3=conductivity:conductivityNullQF|conductivityGapQF|conductivityRangeQF|conductivityStepQF|conductivityPersistenceQF|missingTempQF|zeroPressureQF" \
      "GrpQfBeta3=conductivity:conductivityNullQF|conductivityGapQF|conductivityRangeQF|conductivityStepQF|conductivityPersistenceQF" \
      "GrpQfBetaIgnr3=conductivity:conductivityNullQF|conductivityGapQF" \
      "QfForc1=zeroPressureQF:1:pressureRangeQF|pressureStepQF|pressurePersistenceQF|pressureSpikeQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|conductivityRangeQF|conductivityStepQF|conductivitySpikeQF|conductivityPersistenceQF:-1" \
      "QfForc2=missingTempQF:1:temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|conductivityRangeQF|conductivityStepQF|conductivitySpikeQF|conductivityPersistenceQF:-1"
    
    # ---- Run third module - quality metrics (instantaneous) - only for elevation/water column height for DSC---- 
    Rscript ./flow.qaqc.qm.dp0p.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmQm=$FILE_SCHEMAS/surfacewaterPhysical_aquatroll200_dp01_quality_metrics_inst.avsc \
      Tmi=001 \
      VarTimeBgn=readout_time \
      VarTimeEnd=readout_time \
      "GrpQfAlph1=pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBeta1=pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBetaIgnr1=pressure:pressureNullQF|pressureGapQF" \
      "QfForc1=zeroPressureQF:1:pressureRangeQF|pressureStepQF|pressurePersistenceQF|pressureSpikeQF:-1" \
    
    # We are done with aquatroll. Clean up interim directories before doing leveltroll
    rm -r -f /tmp/pfs/filter_joined
    mkdir -p /tmp/pfs/filter_joined
    
    # Now do the same for Leveltroll: repeat above but with different options...
    
    # Run filter-joiner for Leveltroll data (using environment variables below as input parameters)
    export CONFIG=$CONFIG_LEVELTROLL
    python3 -m filter_joiner.filter_joiner_main
    
    # Run quality metrics for leveltroll
    Rscript ./flow.qaqc.qm.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmQm=$FILE_SCHEMAS/surfacewaterPhysical_leveltroll500_dp01_quality_metrics.avsc \
      "WndwAgr=005|030" \
      "WghtAlphBeta=2|1" \
      Thsh=0.2 \
      "GrpQfAlph1=pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBeta1=pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBetaIgnr1=pressure:pressureNullQF|pressureGapQF" \
      "GrpQfAlph2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperaturePersistenceQF|zeroPressureQF" \
      "GrpQfBeta2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperaturePersistenceQF" \
      "GrpQfBetaIgnr2=temperature:temperatureNullQF|temperatureGapQF" \
      "QfForc1=zeroPressureQF:1:pressureRangeQF|pressureStepQF|pressurePersistenceQF|pressureSpikeQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF:-1"
    # ---- Run third module - quality metrics (instantaneous) - only for elevation/water column height for DSC---- 
    Rscript ./flow.qaqc.qm.dp0p.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmQm=$FILE_SCHEMAS/surfacewaterPhysical_leveltroll500_dp01_quality_metrics_inst.avsc \
      Tmi=001 \
      VarTimeBgn=readout_time \
      VarTimeEnd=readout_time \
      "GrpQfAlph1=pressure:pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBeta1=pressure:pressure:pressureNullQF|pressureGapQF|pressureRangeQF|pressureStepQF|pressurePersistenceQF|zeroPressureQF" \
      "GrpQfBetaIgnr1=pressure:pressureNullQF|pressureGapQF" \
      "QfForc1=zeroPressureQF:1:pressureRangeQF|pressureStepQF|pressurePersistenceQF|pressureSpikeQF:-1" \
    EOF
  env:
    # Environment variables for filter-joiner
    CONFIG_AQUATROLL: |
      ---
      # 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: QAQC_PLAUSIBILITY_PATH
            # Filter for flags directory
            glob_pattern: /pfs/QAQC_PLAUSIBILITY_PATH/*/*/*/*/aquatroll200/*/flags/**
            # Join on named location (already joined below by day)
            join_indices: [6]
        - path:
            name: REGULARIZED_FLAGS_PATH
            # Filter for flags directory
            glob_pattern: /pfs/REGULARIZED_FLAGS_PATH/*/*/*/*/aquatroll200/*/flags/**
            # Join on named location (already joined below by day
            join_indices: [6]
    CONFIG_LEVELTROLL: |
      ---
      # 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: QAQC_PLAUSIBILITY_PATH
            # Filter for flags directory
            glob_pattern: /pfs/QAQC_PLAUSIBILITY_PATH/*/*/*/*/leveltroll500/*/flags/**
            # Join on named location (already joined below by day)
            join_indices: [6]
        - path:
            name: REGULARIZED_FLAGS_PATH
            # Filter for flags directory
            glob_pattern: /pfs/REGULARIZED_FLAGS_PATH/*/*/*/*/leveltroll500/*/flags/**
            # Join on named location (already joined below by day
            join_indices: [6]
    OUT_PATH: /tmp/pfs/filter_joined
    LOG_LEVEL: INFO
    RELATIVE_PATH_INDEX: "3"
    LINK_TYPE: COPY # options are COPY or SYMLINK. Use COPY for combined modules.
    PARALLELIZATION_INTERNAL: '2' # Option for quality metrics module
input:
# We're going to send in all the data at once and partition out the aquatroll vs. leveltroll data using the
# filter-joiner.  
  cross:
  - pfs:
      name: FILE_SCHEMAS
      repo: surfacewaterPhysical_avro_schemas
      glob: /surfacewaterPhysical/quality_metrics_schemas
  - join:
    - pfs:
        name: QAQC_PLAUSIBILITY_PATH
        repo: surfacewaterPhysical_analyze_pad_and_qaqc_plau
        glob: /(*/*/*)
        joinOn: $1
        empty_files: false # Make sure this is false for LINK_TYPE=COPY
    - pfs:
        name: REGULARIZED_FLAGS_PATH
        repo: surfacewaterPhysical_group_path
        glob: /(*/*/*)
        joinOn: $1
        empty_files: false # Make sure this is false for LINK_TYPE=COPY
parallelism_spec:
  constant: 5
autoscaling: true
resource_requests: 
  memory: 800M
  cpu: 2.2
resource_limits:
  memory: 1.5G
  cpu: 4
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"
    }  
  ] }

