---
pipeline:
  name: relHumidity_qm_group_and_compute
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-qaqc-qm-grp:v2.0.0
  cmd: ["/bin/bash"]
  stdin:
  - "#!/bin/bash"
  - '# 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_QM_MET+x} ]
  - then export GROUP_TYPE=met
  - elif [ ${FILE_SCHEMA_QM_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 - quality metrics'
  - '# If met... (make sure both paths are present - not true partial timeseries pads'
  - if [ $GROUP_TYPE = 'met' ] && [ ${QAQC_PLAUSIBILITY_PATH+x} ]
  - then 
      Rscript ./flow.qaqc.qm.R 
      DirIn=/tmp/pfs/interimData
      DirOut=/pfs/out
      DirErr=/pfs/out/errored_datums
      FileSchmQm=$FILE_SCHEMA_QM_MET
      "WndwAgr=001|030"
      "WghtAlphBeta=2|1"
      Thsh=0.2
      "GrpQfAlph1=relativeHumidity:relativeHumiditySuspectCalQF|relativeHumidityNullQF|relativeHumidityGapQF|relativeHumidityRangeQF|relativeHumidityStepQF|relativeHumiditySpikeQF|relativeHumidityPersistenceQF|errorStateQF"
      "GrpQfBeta1=relativeHumidity:relativeHumidityNullQF|relativeHumidityGapQF|relativeHumidityRangeQF|relativeHumidityStepQF|relativeHumiditySpikeQF|relativeHumidityPersistenceQF|errorStateQF"
      "GrpQfBetaIgnr1=relativeHumidity:relativeHumidityNullQF|relativeHumidityGapQF"
      "GrpQfAlph2=temperature:temperatureSuspectCalQF|temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|temperaturePersistenceQF|errorStateQF"
      "GrpQfBeta2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|temperaturePersistenceQF|errorStateQF"
      "GrpQfBetaIgnr2=temperature:temperatureNullQF|temperatureGapQF"
      "GrpQfAlph3=dewPoint:dewPointSuspectCalQF|dewPointNullQF|dewPointGapQF|dewPointRangeQF|dewPointStepQF|dewPointSpikeQF|dewPointPersistenceQF|errorStateQF"
      "GrpQfBeta3=dewPoint:dewPointNullQF|dewPointGapQF|dewPointRangeQF|dewPointStepQF|dewPointSpikeQF|dewPointPersistenceQF|errorStateQF"
      "GrpQfBetaIgnr3=dewPoint:dewPointNullQF|dewPointGapQF"
  - elif [ $GROUP_TYPE = 'buoy' ] && [ ${QAQC_PLAUSIBILITY_PATH+x} ]
  - then
  -   Rscript ./flow.qaqc.qm.R 
        DirIn=/tmp/pfs/interimData
        DirOut=/pfs/out
        DirErr=/pfs/out/errored_datums
        FileSchmQm=$FILE_SCHEMA_QM_BUOY
        WndwAgr=030
        "WghtAlphBeta=2|1"
        Thsh=0.2
        "GrpQfAlph1=relativeHumidity:relativeHumiditySuspectCalQF|relativeHumidityNullQF|relativeHumidityGapQF|relativeHumidityRangeQF|relativeHumidityStepQF|relativeHumiditySpikeQF|relativeHumidityPersistenceQF"
        "GrpQfBeta1=relativeHumidity:relativeHumidityNullQF|relativeHumidityGapQF|relativeHumidityRangeQF|relativeHumidityStepQF|relativeHumiditySpikeQF|relativeHumidityPersistenceQF"
        "GrpQfBetaIgnr1=relativeHumidity:relativeHumidityNullQF|relativeHumidityGapQF"
        "GrpQfAlph2=temperature:temperatureSuspectCalQF|temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|temperaturePersistenceQF"
        "GrpQfBeta2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|temperaturePersistenceQF"
        "GrpQfBetaIgnr2=temperature:temperatureNullQF|temperatureGapQF"
        "GrpQfAlph3=dewPoint:dewPointSuspectCalQF|dewPointNullQF|dewPointGapQF|dewPointRangeQF|dewPointStepQF|dewPointSpikeQF|dewPointPersistenceQF"
        "GrpQfBeta3=dewPoint:dewPointNullQF|dewPointGapQF|dewPointRangeQF|dewPointStepQF|dewPointSpikeQF|dewPointPersistenceQF"
        "GrpQfBetaIgnr3=dewPoint:dewPointNullQF|dewPointGapQF"
        VarIgnr=errorStateQF
  -   Rscript ./flow.qaqc.qm.dp0p.R
        DirIn=/tmp/pfs/interimData
        DirOut=/pfs/out
        DirErr=/pfs/out/errored_datums
        FileSchmQm=$FILE_SCHEMA_QM_BUOY_INST
        VarTimeBgn=readout_time
        VarTimeEnd=readout_time
        "GrpQfAlph1=relativeHumidity:relativeHumiditySuspectCalQF|relativeHumidityNullQF|relativeHumidityGapQF|relativeHumidityRangeQF|relativeHumidityStepQF|relativeHumiditySpikeQF|relativeHumidityPersistenceQF"
        "GrpQfBeta1=relativeHumidity:relativeHumidityNullQF|relativeHumidityGapQF|relativeHumidityRangeQF|relativeHumidityStepQF|relativeHumiditySpikeQF|relativeHumidityPersistenceQF"
        "GrpQfBetaIgnr1=relativeHumidity:relativeHumidityNullQF|relativeHumidityGapQF"
        "GrpQfAlph2=temperature:temperatureSuspectCalQF|temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|temperaturePersistenceQF"
        "GrpQfBeta2=temperature:temperatureNullQF|temperatureGapQF|temperatureRangeQF|temperatureStepQF|temperatureSpikeQF|temperaturePersistenceQF"
        "GrpQfBetaIgnr2=temperature:temperatureNullQF|temperatureGapQF"
        "GrpQfAlph3=dewPoint:dewPointSuspectCalQF|dewPointNullQF|dewPointGapQF|dewPointRangeQF|dewPointStepQF|dewPointSpikeQF|dewPointPersistenceQF"
        "GrpQfBeta3=dewPoint:dewPointNullQF|dewPointGapQF|dewPointRangeQF|dewPointStepQF|dewPointSpikeQF|dewPointPersistenceQF"
        "GrpQfBetaIgnr3=dewPoint:dewPointNullQF|dewPointGapQF"
        VarIgnr=errorStateQF
        Tmi=001
  - else echo "FATAL - Cannot determine group_type as met or buoy. Aborting..."
  - fi
  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_PLAUSIBILITY_PATH
            # Filter for flags directory
            glob_pattern: /pfs/QAQC_PLAUSIBILITY_PATH/*/*/*/*/hmp155/*/flags/**
            # Join on named location (already joined below by day)
            join_indices: [3,4,5,6]
        - path:
            name: REGULARIZED_FLAGS_PATH
            # Filter for flags directory
            glob_pattern: /pfs/REGULARIZED_FLAGS_PATH/*/*/*/*/hmp155/*/flags/**
            # 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.
    PARALLELIZATION_INTERNAL: '4' # Option for quality metrics 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_QM_[GROUP_TYPE] environment variable is populated 
# in the container. 
  union:
  - cross:
    - pfs:
        name: FILE_SCHEMA_QM_MET
        repo: relHumidity_avro_schemas
        glob: /relHumidity/relHumidity_dp01_quality_metrics.avsc
    - group:
      - pfs:
          name: QAQC_PLAUSIBILITY_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: REGULARIZED_FLAGS_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_QM_BUOY
        repo: relHumidity_avro_schemas
        glob: /relHumidityBuoy/relHumidityBuoy_dp01_quality_metrics.avsc
    - pfs:
        name: FILE_SCHEMA_QM_BUOY_INST
        repo: relHumidity_avro_schemas
        glob: /relHumidityBuoy/relHumidityBuoy_dp01_qm_instantaneous.avsc
    - group:
      - pfs:
          name: QAQC_PLAUSIBILITY_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: REGULARIZED_FLAGS_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"
      }
    }
  ]
