---
pipeline:
  name: pqs1_calibration_group_and_convert
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-cal-grp-conv:v1.2.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/kafka_merged
    rm -rf $OUT_PATH
    mkdir -p /tmp/kafka_merged # Filter joiner relies on the same path positions among inputs (i.e. repo name in 2nd position)
    mkdir -p $OUT_PATH # R modules must have pfs in the repo structure
    # Detect if we have data coming from Kafka or trino
    # Note that we run the filter-joiner in sequential if statements rather than an elif statement ...
    # ... so that if there is any overlap in sensor data coming from both Kafka and Trino on the same day, the ...
    # ... trino data wins (filter joiner will not copy a file if it is already in the destination). This scenario ...
    # ... should only arise during initial data load and a site back-streams data from kafka outside the Kafka ...
    # ... retention period for data that have already been loaded from Trino. In normal operations this scenario ...
    # ... should not arise because the only data coming into Pachyderm will be coming from Kafka. ...
    # ... When a conflict does arise, the Trino data will take precedence, which is fine because the ...
    # ... conflict arose because data for the same sensor and day were in both locations.
    if [ ${DATA_PATH_TRINO+x} ]; then 
      # Data from trino. 
      # Set CONFIG for filter-joiner to the trino version and run filter joiner
      export CONFIG=$CONFIG_TRINO
      python3 -m filter_joiner.filter_joiner_main
    fi  
    if [ ${KAFKA_UNMERGED_DATA+x} ]; then 
      # Data from kafka. 
      # Run kafka combiner
      Rscript ./flow.kfka.comb.R \
          DirIn=$KAFKA_UNMERGED_DATA \
          DirOut=/tmp/kafka_merged \
          DirErr=/pfs/out/errored_datums \
          FileSchmL0=$FILE_SCHEMA_L0
      # Set CONFIG for filter-joiner to the kafka version and run filter joiner
      export CONFIG=$CONFIG_KAFKA
      python3 -m filter_joiner.filter_joiner_main
    fi  
    # Run calibration conversion module
    Rscript ./flow.cal.conv.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      FileSchmData=$FILE_SCHEMA_DATA \
      FileSchmQf=$FILE_SCHEMA_FLAGS \
      TermFuncConv=voltage:def.cal.conv.poly \
      TermQf=voltage \
      TermFuncUcrt=voltage:def.ucrt.meas.mult,def.ucrt.fdas.volt.poly \
      FileUcrtFdas=$FILE_UNCERTAINTY_FDAS
    EOF
  env:
    # Environment variables for filter-joiner.
    # There are two configs here, CONFIG_KAFKA is when data comes from kafka, CONFIG_TRINO is when data comes from trino. 
    # The engironment variable CONFIG is set to the appropriate one as detected in the bash script above
    CONFIG_KAFKA: |
      ---
      # Configuration for filter-joiner module that will bring together the data and calibrations
      # In Pachyderm root will be index 0, 'pfs' index 1, and the repo name index 2.
      # Metadata indices will typically begin at index 3.
      input_paths:
        - path:
            name: DATA_PATH_KAFKA
            # Filter for data directory
            glob_pattern: /tmp/kafka_merged/pqs1/*/*/*/*/**
            # Join on named location (already joined below by day)
            join_indices: [7]
            outer_join: true
        - path:
            name: CALIBRATION_PATH
            # Filter for data directory
            glob_pattern: /pfs/CALIBRATION_PATH/pqs1/*/*/*/*/**
            # Join on named location (already joined below by day)
            join_indices: [7]
    CONFIG_TRINO: |
      ---
      # Configuration for filter-joiner module that will bring together the data and calibrations
      # In Pachyderm root will be index 0, 'pfs' index 1, and the repo name index 2.
      # Metadata indices will typically begin at index 3.
      input_paths:
        - path:
            name: DATA_PATH_TRINO
            # Filter for data directory
            glob_pattern: /pfs/DATA_PATH_TRINO/pqs1/*/*/*/*/**
            # Join on named location (already joined below by day)
            join_indices: [7]
            outer_join: true
        - path:
            name: CALIBRATION_PATH
            # Filter for data directory
            glob_pattern: /pfs/CALIBRATION_PATH/pqs1/*/*/*/*/**
            # Join on named location (already joined below by day)
            join_indices: [7]
    OUT_PATH: /tmp/pfs/filter_joined # Note that R modules use "pfs" in the path structure to determine datums
    LOG_LEVEL: INFO
    RELATIVE_PATH_INDEX: "3" # Must be consistent across inputs 
    LINK_TYPE: COPY # options are COPY or SYMLINK. MUST BE SIMLINK IF USING COMBINED MODULE.
    # Environment variables for calibration module
    PARALLELIZATION_INTERNAL: '3' # Option for calibration conversion module
input:
  cross:
  - pfs:
      name: FILE_SCHEMA_L0
      repo: pqs1_avro_schemas
      glob: /pqs1/pqs1.avsc
  - pfs:
      name: FILE_SCHEMA_DATA
      repo: pqs1_avro_schemas
      glob: /pqs1/pqs1_calibrated.avsc
  - pfs:
      name: FILE_SCHEMA_FLAGS
      repo: pqs1_avro_schemas
      glob: /pqs1/flags_calibration_pqs1.avsc
  - pfs:
      name: FILE_UNCERTAINTY_FDAS
      repo: pqs1_uncertainty_fdas
      glob: /fdas_calibration_uncertainty_general.json
  # Outer join all repos so that varying sensors between kafka and trino loaders will all get joined with calibrations. Filter-joiner will narrow down.
  - join:
    - pfs:
        name: CALIBRATION_PATH
        repo: pqs1_calibration_assignment
        glob: /pqs1/(*)/(*)/(*)
        joinOn: $1/$2/$3
        outer_join: true
        empty_files: false # Make sure to use false if LINK_TYPE=COPY. Can also be set to false for LINK_TYPE=SYMLINK.
    - pfs:
        name: DATA_PATH_TRINO
        repo: pqs1_data_source_trino
        glob: /pqs1/(*)/(*)/(*)
        joinOn: $1/$2/$3
        outer_join: true
        empty_files: false # Make sure to use false if LINK_TYPE=COPY. Can also be set to false for LINK_TYPE=SYMLINK.
    - pfs:
        name: KAFKA_UNMERGED_DATA
        repo: pqs1_data_source_kafka
        glob: /pqs1/(*)/(*)/(*)
        joinOn: $1/$2/$3
        outer_join: true
        empty_files: false # Make sure to use false if LINK_TYPE=COPY. Can also be set to false for LINK_TYPE=SYMLINK.
parallelism_spec:
  constant: 3
autoscaling: true
resource_requests:
  memory: 1.5G
  cpu: 3.3
resource_limits:
  memory: 3G
  cpu: 4.5
sidecar_resource_requests:
  memory: 3G
  cpu: 0.6
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"
    }  
  ] }
