---
pipeline:
  name: pressureAir_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 first module - filter-joiner (using environment variables below as input parameters)
    python3 -m filter_joiner.filter_joiner_main
    # Run second module - quality metrics
    Rscript ./flow.qaqc.qm.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmQm=$FILE_SCHEMA_QM \
      "WndwAgr=001|030" \
      "WghtAlphBeta=2|1" \
      Thsh=0.2 \
      "GrpQfAlph1=staPres:suspectCalQF|nullQF|gapQF|rangeQF|stepQF|spikeQF|persistenceQF" \
      "GrpQfBeta1=staPres:nullQF|gapQF|rangeQF|stepQF|spikeQF|persistenceQF" \
      "GrpQfBetaIgnr1=staPres:nullQF|gapQF"
    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_PLAUSIBILITY_PATH
            # Filter for flags directory
            glob_pattern: /pfs/QAQC_PLAUSIBILITY_PATH/*/*/*/*/ptb330a/*/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/*/*/*/*/ptb330a/*/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:
  cross:
  - pfs:
      name: FILE_SCHEMA_QM
      repo: pressureAir_avro_schemas
      glob: /pressureAir/pressureAir_dp01_quality_metrics.avsc
  - join:
    - pfs:
        name: QAQC_PLAUSIBILITY_PATH
        repo: pressureAir_analyze_pad_and_qaqc_plau
        glob: /(*)/(*)/(*)
        joinOn: $1/$2/$3
        empty_files: false # Make sure this is false for LINK_TYPE=COPY
    - pfs:
        name: REGULARIZED_FLAGS_PATH
        repo: pressureAir_group_path_ptb330a
        glob: /(*)/(*)/(*)
        joinOn: $1/$2/$3
        empty_files: false # Make sure this is false for LINK_TYPE=COPY
parallelism_spec:
  constant: 2
autoscaling: true
resource_requests: 
  memory: 600M
  cpu: 2.2
resource_limits:
  memory: 1.5G
  cpu: 4
sidecar_resource_requests:
  memory: 1G
  cpu: 0.2
sidecar_resource_limits:
  memory: 3Gi
  cpu: 1.3
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"
    }  
  ] }

