---
pipeline:
  name: surfacewaterPhysical_stats_ucrt_grp_comp
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-troll-stat-ucrt-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_aquatroll
    mkdir -p /tmp/pfs/filter_joined_aquatroll
    rm -r -f /tmp/pfs/filter_joined_leveltroll
    mkdir -p /tmp/pfs/filter_joined_leveltroll
    
    # Run filter-joiner for Aquatroll data (using environment variables below as input parameters)
    export CONFIG=$CONFIG_AQUATROLL
    export OUT_PATH=$OUT_PATH_AQUATROLL
    python3 -m filter_joiner.filter_joiner_main
    
    # Run basic stats for Aquatroll data
    Rscript ./flow.stat.basc.R \
      DirIn=/tmp/pfs/filter_joined_aquatroll \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmStat=$FILE_SCHEMAS/surfacewaterPhysical_aquatroll200_dp01_stats.avsc \
      "WndwAgr=005|030" \
      "TermStat1=temperature:mean|minimum|maximum|variance|numPts|stdEr" \
      "TermStat2=conductivity:mean|minimum|maximum|variance|numPts|stdEr" \
      DirSubCopy=location
      
    # Run third module - troll-specific uncertainty comps for aquatroll data
    Rscript ./flow.troll.uncertainty.R \
      DirIn=/tmp/pfs/filter_joined_aquatroll \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      Context=surfacewater \
      WndwInst=TRUE \
      "WndwAgr=005|030" \
      FileSchmData=$FILE_SCHEMAS/surfacewaterPhysical_dp01_aquatroll200_specific_data.avsc \
      FileSchmUcrtAgr=$FILE_SCHEMAS/surfacewaterPhysical_dp01_aquatroll200_specific_ucrt.avsc \
      FileSchmUcrtInst=$FILE_SCHEMAS/surfacewaterPhysical_dp01_aquatroll200_specific_ucrt_inst.avsc \
      FileSchmSciStats=$FILE_SCHEMAS/surfacewaterPhysical_dp01_troll_specific_sci_stats.avsc
      
    # 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
    export OUT_PATH=$OUT_PATH_LEVELTROLL
    python3 -m filter_joiner.filter_joiner_main
    
    # Run basic stats for Leveltroll
    Rscript ./flow.stat.basc.R \
      DirIn=/tmp/pfs/filter_joined_leveltroll \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmStat=$FILE_SCHEMAS/surfacewaterPhysical_leveltroll500_dp01_stats.avsc \
      "WndwAgr=005|030" \
      "TermStat1=temperature:mean|minimum|maximum|variance|numPts|stdEr" \
      DirSubCopy=location
      
    # Run third module - troll-specific uncertainty comps  for leveltroll data
    Rscript ./flow.troll.uncertainty.R \
      DirIn=/tmp/pfs/filter_joined_leveltroll \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      Context=surfacewater \
      WndwInst=TRUE \
      "WndwAgr=005|030" \
      FileSchmData=$FILE_SCHEMAS/surfacewaterPhysical_dp01_leveltroll500_specific_data.avsc \
      FileSchmUcrtAgr=$FILE_SCHEMAS/surfacewaterPhysical_dp01_leveltroll500_specific_ucrt.avsc \
      FileSchmUcrtInst=$FILE_SCHEMAS/surfacewaterPhysical_dp01_leveltroll500_specific_ucrt_inst.avsc \
      FileSchmSciStats=$FILE_SCHEMAS/surfacewaterPhysical_dp01_troll_specific_sci_stats.avsc
    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_PATH
            # Filter for data & uncertainty_data directories
            glob_pattern: /pfs/QAQC_PATH/*/*/*/*/aquatroll200/*/*/**
            # Join on y/m/d and named location 
            join_indices: [3,4,5,6]
        - path:
            name: UNCERTAINTY_PATH
            # Filter for data directory
            glob_pattern: /pfs/UNCERTAINTY_PATH/*/*/*/*/aquatroll200/*/uncertainty*/**
            # Join on y/m/d and named location 
            join_indices: [3,4,5,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_PATH
            # Filter for data & uncertainty_data directories
            glob_pattern: /pfs/QAQC_PATH/*/*/*/*/leveltroll500/*/*/**
            # Join on y/m/d and named location 
            join_indices: [3,4,5,6]
        - path:
            name: UNCERTAINTY_PATH
            # Filter for data directory
            glob_pattern: /pfs/UNCERTAINTY_PATH/*/*/*/*/leveltroll500/*/uncertainty*/**
            # Join on y/m/d and named location 
            join_indices: [3,4,5,6]
    OUT_PATH_AQUATROLL: /tmp/pfs/filter_joined_aquatroll
    OUT_PATH_LEVELTROLL: /tmp/pfs/filter_joined_leveltroll
    LOG_LEVEL: INFO
    RELATIVE_PATH_INDEX: "3"
    LINK_TYPE: COPY # options are COPY or SYMLINK. Use COPY for combined module.
    PARALLELIZATION_INTERNAL: '2' # Option for stats 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/stats_schemas
  - join:
    - pfs:
        name: QAQC_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: UNCERTAINTY_PATH
        repo: surfacewaterPhysical_group_path
        glob: /(*/*/*)
        joinOn: $1
        empty_files: false # Make sure this is false for LINK_TYPE=COPY
parallelism_spec:
  constant: 5
resource_requests: 
  memory: 800M
  cpu: 2.5
resource_limits:
  memory: 1.5G
  cpu: 3.5
sidecar_resource_requests:
  memory: 3G
  cpu: 0.3
autoscaling: true
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"
    }  
  ] }
