---
pipeline:
  name: relHumidity_stats_group_and_compute
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-stat-basc-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 -rf /tmp/pfs/interimData
    mkdir -p /tmp/pfs/interimData
    # Determine which group type we have
    if [ ${FILE_SCHEMA_STATS_MET+x} ]
    then export GROUP_TYPE=met
    elif [ ${FILE_SCHEMA_STATS_BUOY+x} ]
    then export GROUP_TYPE=buoy
    else echo "FATAL - Input repository names do not match expectations. Check pipeline specification."
    fi
    echo "Group type detected = $GROUP_TYPE"
    # 
    # Run first module - filter-joiner (using environment variables below as input parameters)
    python3 -m filter_joiner.filter_joiner_main
    # Run second module - basic stats
    # If met... (make sure both paths are present - not true partial timeseries pads
    if [ $GROUP_TYPE = 'met' ] && [ ${QAQC_PATH+x} ] && [ ${UNCERTAINTY_PATH+x} ]
    then 
      Rscript ./flow.stat.basc.R \
        DirIn=/tmp/pfs/interimData \
        DirOut=/pfs/out \
        DirErr=/pfs/out/errored_datums \
        FileSchmStat=$FILE_SCHEMA_STATS_MET \
        "WndwAgr=001|030" \
        "TermStat1=relativeHumidity:mean|minimum|maximum|variance|numPts|stdEr|expUncert(wrap.ucrt.dp01.cal.cnst)" \
        "TermStat2=temperature:mean|minimum|maximum|variance|numPts|stdEr|expUncert(wrap.ucrt.dp01.cal.cnst)" \
        "TermStat3=dewPoint:mean|minimum|maximum|variance|numPts|stdEr|expUncert(wrap.ucrt.dp01.rh.dew.frst.pt)"
    elif [ $GROUP_TYPE = 'buoy' ] && [ ${QAQC_PATH+x} ] && [ ${UNCERTAINTY_PATH+x} ]
    then 
      Rscript ./flow.stat.basc.R \
        DirIn=/tmp/pfs/interimData \
        DirOut=/pfs/out \
        DirErr=/pfs/out/errored_datums \
        FileSchmStat=$FILE_SCHEMA_STATS_BUOY \
        WndwAgr=030 \
        "TermStat1=relativeHumidity:mean|minimum|maximum|variance|numPts|stdEr|expUncert(wrap.ucrt.dp01.cal.cnst)" \
        "TermStat2=temperature:mean|minimum|maximum|variance|numPts|stdEr|expUncert(wrap.ucrt.dp01.cal.cnst)" \
        "TermStat3=dewPoint:mean|minimum|maximum|variance|numPts|stdEr|expUncert(wrap.ucrt.dp01.rh.dew.frst.pt)"
      Rscript ./flow.data.comb.ts.R \
        DirIn=/tmp/pfs/interimData \
        DirOut=/pfs/out \
        DirErr=/pfs/out/errored_datums \
        FileSchmComb=$FILE_SCHEMA_STATS_INST_BUOY \
        "DirComb=data|uncertainty_data" \
        NameDirCombOut=stats \
        NameVarTime=readout_time \
        "ColKeep=readout_time|readout_time|relativeHumidity|relativeHumidity_ucrtExpn|temperature|temperature_ucrtExpn|dewPoint|dewPoint_ucrtExpn" \
        NameFileSufx=_basicStats_001
    else echo "FATAL - Cannot determine group_type as met or buoy. Aborting..."
    fi
    EOF
  env:
    # Environment variables for filter-joiner
    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: QAQC_PATH
            # Filter for data & uncertainty_data directories
            glob_pattern: /pfs/QAQC_PATH/*/*/*/*/hmp155/*/data/**
            # Join on named location (already joined below by day)
            join_indices: [3,4,5,6]
        - path:
            name: UNCERTAINTY_PATH
            # Filter for data directory
            glob_pattern: /pfs/UNCERTAINTY_PATH/*/*/*/*/hmp155/*/uncertainty*/**
            # Join on named location (already joined below by day)
            join_indices: [3,4,5,6]
    OUT_PATH: /tmp/pfs/interimData
    LOG_LEVEL: INFO
    RELATIVE_PATH_INDEX: "3"
    LINK_TYPE: COPY # options are COPY or SYMLINK. Use COPY for combined module.
    # Environment variables for calibration module
    PARALLELIZATION_INTERNAL: '4' # Option for stats module
input:
# We're going to send in the buoy locations separately from the tower/soil/met-station locations
# The 'separate' part is achieved by the union. Each datum in a union is sent in separately to the container. 
# Each datum consists of a cross between the data and it's associated avro schema 
# for the output. The group nested within this cross is what allows us to send in all the buoy data for 
# each day in as one datum and all the tower/soil/met-station data for one day as a datum. It groups by day AND group type.
# We need to do some distinguishing between these gorup types when we execute the code above. We'll 
# determine which group type we have by seeing which FILE_SCHEMA_STATS_[GROUP_TYPE] environment variable is populated 
# in the container. 
  union:
  - cross:
    - pfs:
        name: FILE_SCHEMA_STATS_MET
        repo: relHumidity_avro_schemas
        glob: /relHumidity/relHumidity_dp01_stats.avsc
    - group:
      - pfs:
          name: QAQC_PATH
          repo: relHumidity_analyze_pad_and_qaqc_plau
          glob: /(*/*/*/rel-humidity_)*
          group_by: $1
          empty_files: false # Make sure this is false for LINK_TYPE=COPY
      - pfs:
          name: UNCERTAINTY_PATH
          repo: relHumidity_pre_stats_qm
          glob: /(*/*/*/rel-humidity_)*
          group_by: $1
          empty_files: false # Make sure this is false for LINK_TYPE=COPY
  - cross:
    - pfs:
        name: FILE_SCHEMA_STATS_BUOY
        repo: relHumidity_avro_schemas
        glob: /relHumidityBuoy/relHumidityBuoy_dp01_stats.avsc
    - pfs:
        name: FILE_SCHEMA_STATS_INST_BUOY
        repo: relHumidity_avro_schemas
        glob: /relHumidityBuoy/relHumidityBuoy_dp01_stats_instantaneous.avsc
    - group:
      - pfs:
          name: QAQC_PATH
          repo: relHumidity_analyze_pad_and_qaqc_plau
          glob: /(*/*/*/rel-humidity-buoy_)*
          group_by: $1
          empty_files: false # Make sure this is false for LINK_TYPE=COPY
      - pfs:
          name: UNCERTAINTY_PATH
          repo: relHumidity_pre_stats_qm
          glob: /(*/*/*/rel-humidity-buoy_)*
          group_by: $1
          empty_files: false # Make sure this is false for LINK_TYPE=COPY
parallelism_spec:
  constant: 2
autoscaling: true
resource_requests:
  memory: 2.5G
  cpu: 4.4
resource_limits:
  memory: 4G
  cpu: 6
sidecar_resource_requests:
  memory: 2G
  cpu: 0.5
sidecar_resource_limits:
  memory: 3Gi
  cpu: 1.2
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"
    }  
  ] }
pod_patch: |-
  [
    { "op": "add",
      "path": "/containers/1/env/-",
      "value": {
        "name": "GOMEMLIMIT",
        "value": "2000000000"
      }
    }
  ]
