Ingest PLC Telemetry Only When Every Reading Is Trustworthy

id: opcua-modbus-telemetry-ingest
namespace: company.team
description: |
  Pull a batch of PLC readings over OPC UA and Modbus, either from a five
  minute poll that ships disabled or on demand from a keyed webhook, and
  decide whether the batch is trustworthy enough to land in the time series
  database. A python poller captures a deterministic register snapshot of
  temperature, pressure and rpm points, a normalizer rewrites it into the
  canonical tag, value, unit, quality and timestamp shape and checks every
  reading against its quality flag and its engineering bounds. A single bad
  quality flag or out of bounds value routes the batch to a Slack alarm and
  a GitHub issue instead of the TSDB, while a clean batch is POSTed to the
  ingest endpoint, and every execution publishes the verdict and counts as
  outputs.

triggers:
  - id: five_minute_poll
    type: io.kestra.plugin.core.trigger.Schedule
    description: Standing five minute register poll, the safety net that ships
      disabled until an operator points the flow at a live endpoint and switches
      it on.
    cron: "*/5 * * * *"
    disabled: true
  - id: on_demand_poll
    type: io.kestra.plugin.core.trigger.Webhook
    description: POST here to poll and ingest on demand, for example from the edge
      gateway after a line changeover or from an operator debugging a suspect
      register.
    key: "{{ secret('PLC_WEBHOOK_KEY') }}"

inputs:
  - id: plc_endpoint
    type: STRING
    defaults: "opc.tcp://line3-plc.local:4840"
    description: Endpoint the poller reads, carried into the Slack alarm, the GitHub
      issue and the TSDB batch metadata.
  - id: temperature_limit_c
    type: FLOAT
    defaults: 120.0
    description: High bound in Celsius applied to every temperature register, a
      reading above it counts as out of bounds.
  - id: github_repository
    type: STRING
    defaults: your-org/plant-telemetry
    description: Repository in owner/repo form that receives the deviation issue
      when a batch is withheld from the TSDB.

tasks:
  # --- 1. poll and normalize ---
  - id: telemetry_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the register snapshot written by the poller
      is the exact file the normalizer reads.
    tasks:
      - id: poll_plc_registers
        type: io.kestra.plugin.scripts.python.Script
        description: Simulate the OPC UA and Modbus poll by capturing eight registers -
          three temperature, two pressure and three rpm - with node ids,
          register addresses, engineering values, units and OPC UA status names
          into opcua-points.json, then read the file back and emit the raw count
          and the poll timestamp.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          PLC_ENDPOINT: "{{ inputs.plc_endpoint }}"
        script: |
          import json
          import os
          from datetime import datetime, timezone

          endpoint = os.environ.get("PLC_ENDPOINT", "unknown-endpoint")
          poll_ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

          snapshot = [
              {"node": "ns=3;s=Line3.Temp.Inlet", "register": 40101, "value": 78.4, "unit": "C", "quality": "Good"},
              {"node": "ns=3;s=Line3.Temp.Outlet", "register": 40102, "value": 91.2, "unit": "C", "quality": "Good"},
              {"node": "ns=3;s=Line3.Temp.Bearing", "register": 40103, "value": 64.9, "unit": "C", "quality": "Good"},
              {"node": "ns=3;s=Line3.Press.Hydraulic", "register": 40201, "value": 168.5, "unit": "bar", "quality": "Good"},
              {"node": "ns=3;s=Line3.Press.Supply", "register": 40202, "value": 4.2, "unit": "bar", "quality": "Good"},
              {"node": "ns=3;s=Line3.Rpm.Spindle", "register": 40301, "value": 1840.0, "unit": "rpm", "quality": "Good"},
              {"node": "ns=3;s=Line3.Rpm.Conveyor", "register": 40302, "value": 62.0, "unit": "rpm", "quality": "Good"},
              {"node": "ns=3;s=Line3.Rpm.Fan", "register": 40303, "value": 1450.0, "unit": "rpm", "quality": "Good"}
          ]

          with open("opcua-points.json", "w") as handle:
              json.dump(snapshot, handle, indent=2)

          with open("opcua-points.json", "r") as handle:
              reloaded = json.load(handle)

          print("polled %d registers from %s at %s" % (len(reloaded), endpoint, poll_ts))
          print("::" + json.dumps({"outputs": {"raw_count": len(reloaded), "poll_ts": poll_ts, "endpoint": endpoint}}) + "::")

      - id: normalize_readings
        type: io.kestra.plugin.scripts.python.Script
        description: Read opcua-points.json, normalize every register into the canonical
          tag, value, unit, quality and ts shape, flag any non-good quality flag
          or out of bounds value, and emit the verdict, the ingest count, the
          flagged count, the maximum temperature and the JSON batch through the
          outputs protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          PLC_ENDPOINT: "{{ inputs.plc_endpoint }}"
          TEMPERATURE_LIMIT_C: "{{ inputs.temperature_limit_c }}"
          POLL_TS: "{{ outputs.poll_plc_registers.vars.poll_ts }}"
        script: |
          import json
          import os
          from datetime import datetime, timezone

          POINTS_FILE = "opcua-points.json"
          poll_ts = os.environ.get("POLL_TS") or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
          endpoint = os.environ.get("PLC_ENDPOINT", "unknown-endpoint")

          try:
              temp_limit = float(os.environ.get("TEMPERATURE_LIMIT_C", "120"))
          except ValueError:
              temp_limit = 120.0

          BOUNDS = {"c": (0.0, temp_limit), "bar": (0.0, 250.0), "rpm": (0.0, 3600.0)}

          raw = []
          if os.path.isfile(POINTS_FILE) and os.path.getsize(POINTS_FILE) > 0:
              try:
                  with open(POINTS_FILE, "r") as handle:
                      loaded = json.load(handle)
                  if isinstance(loaded, list):
                      raw = loaded
              except ValueError:
                  raw = []

          def tag_of(node):
              segment = str(node or "").split(";s=")[-1]
              parts = [p for p in segment.split(".") if p]
              return "_".join(p.lower() for p in parts) if parts else "unknown_tag"

          def normalize_quality(status):
              name = str(status or "bad").strip().lower()
              if name.startswith("good"):
                  return "good"
              if name.startswith("uncert"):
                  return "uncertain"
              return "bad"

          points = []
          flagged = []
          for entry in raw:
              if not isinstance(entry, dict):
                  continue
              tag = tag_of(entry.get("node") or entry.get("tag"))
              unit = str(entry.get("unit") or "").strip()
              quality = normalize_quality(entry.get("quality"))
              try:
                  value = float(entry.get("value"))
              except (TypeError, ValueError):
                  value = None
              ts = str(entry.get("ts") or poll_ts)
              reason = None
              if value is None:
                  quality = "bad"
                  reason = "register returned a non-numeric value"
              elif quality != "good":
                  reason = "quality flag is " + quality
              else:
                  bounds = BOUNDS.get(unit.lower())
                  if bounds is not None and not (bounds[0] <= value <= bounds[1]):
                      quality = "bad"
                      reason = "%s value %.2f sits outside %.2f to %.2f" % (unit, value, bounds[0], bounds[1])
              point = {"tag": tag, "value": value, "unit": unit, "quality": quality, "ts": ts}
              points.append(point)
              if reason is not None:
                  flagged.append({"tag": tag, "value": value, "unit": unit, "quality": quality, "reason": reason})

          if not points:
              flagged.append({"tag": "poll", "value": None, "unit": "", "quality": "bad",
                              "reason": "the poll returned no readable registers"})

          temps = [p["value"] for p in points if p["unit"].lower() == "c" and p["value"] is not None]
          max_temp = round(max(temps), 2) if temps else 0.0
          ingestable = [p for p in points if p["quality"] == "good"]
          verdict = "ALARM" if flagged else "CLEAN"
          points_ingested = len(ingestable) if verdict == "CLEAN" else 0
          flagged_md = "\n".join("- " + f["tag"] + " (" + f["unit"] + ") " + f["reason"] for f in flagged)
          if not flagged_md:
              flagged_md = "No reading was flagged by the quality or bounds checks."

          payload = {
              "verdict": verdict,
              "points_ingested": points_ingested,
              "bad_quality": len(flagged),
              "max_temp": max_temp,
              "raw_count": len(points),
              "poll_ts": poll_ts,
              "endpoint": endpoint,
              "flagged": flagged,
              "flagged_md": flagged_md,
              "batch_json": json.dumps(ingestable),
          }
          print("endpoint=%s raw=%d ingested=%d flagged=%d max_temp=%.2f verdict=%s" % (
              endpoint, len(points), points_ingested, len(flagged), max_temp, verdict))
          print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 2. verdict gate ---
  - id: check_verdict
    type: io.kestra.plugin.core.flow.If
    description: Escalate only when at least one reading carried a non-good quality
      flag or sat outside its engineering bounds, otherwise ship the clean batch
      to the TSDB.
    condition: "{{ outputs.normalize_readings.vars.verdict == 'ALARM' }}"
    then:
      - id: alert_plc_deviation
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the plant operations channel how many readings were flagged,
          what the peak temperature was, and that the batch was withheld from
          the TSDB.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "PLC telemetry ALARM on {{ inputs.plc_endpoint }} - {{ outputs.normalize_readings.vars.bad_quality }} of {{ outputs.normalize_readings.vars.raw_count }} readings flagged, max temperature {{ outputs.normalize_readings.vars.max_temp }} C, batch withheld from the TSDB. Execution {{ execution.id }}."
          }
      - id: open_telemetry_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled GitHub issue listing every flagged tag with its unit
          and reason, so sensor and register work is tracked where the team
          already works.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.github_repository }}"
        title: "PLC telemetry deviation on {{ inputs.plc_endpoint }}"
        body: |
          ## PLC telemetry quality alarm

          Endpoint **{{ inputs.plc_endpoint }}** returned **{{ outputs.normalize_readings.vars.bad_quality }}** flagged reading(s) out of **{{ outputs.normalize_readings.vars.raw_count }}** in the poll at {{ outputs.normalize_readings.vars.poll_ts }}. Max temperature **{{ outputs.normalize_readings.vars.max_temp }} C**. The normalized batch was withheld from the TSDB.

          ### Flagged readings

          {{ outputs.normalize_readings.vars.flagged_md }}

          Execution {{ execution.id }}

          Check the register, the wiring and the sensor behind each flagged tag, then close this issue.
        labels:
          - opcua
          - modbus
          - telemetry
    else:
      - id: post_batch_to_tsdb
        type: io.kestra.plugin.core.http.Request
        description: POST the normalized batch to the TSDB ingest endpoint with the
          endpoint, execution and poll metadata, retrying transient failures so
          a blip does not drop a clean poll.
        method: POST
        uri: "{{ secret('TSDB_INGEST_URL') }}"
        headers:
          Content-Type: application/json
        body: |
          {"measurement": "plc_telemetry", "endpoint": "{{ inputs.plc_endpoint }}", "execution": "{{ execution.id }}", "poll_ts": "{{ outputs.normalize_readings.vars.poll_ts }}", "points": {{ outputs.normalize_readings.vars.batch_json }}}
        retry:
          type: exponential
          interval: PT2S
          maxInterval: PT10S
          maxAttempts: 3
      - id: log_all_clear
        type: io.kestra.plugin.core.log.Log
        description: Record the clean poll in the execution history so it doubles as
          ingest evidence.
        message: "PLC telemetry clean on {{ inputs.plc_endpoint }} - {{
          outputs.normalize_readings.vars.points_ingested }} of {{
          outputs.normalize_readings.vars.raw_count }} readings POSTed to the
          TSDB, max temperature {{ outputs.normalize_readings.vars.max_temp }}
          C."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: CLEAN when every reading passed the quality and bounds checks,
      ALARM when the batch was withheld.
    value: "{{ outputs.normalize_readings.vars.verdict }}"
  - id: points_ingested
    type: INT
    description: Number of normalized readings POSTed to the TSDB, zero whenever the
      batch was withheld.
    value: "{{ outputs.normalize_readings.vars.points_ingested }}"
  - id: bad_quality
    type: INT
    description: Number of readings flagged for a non-good quality flag or an out of
      bounds value, for example 0 on a clean poll.
    value: "{{ outputs.normalize_readings.vars.bad_quality }}"
  - id: max_temp
    type: FLOAT
    description: Highest Celsius reading across the temperature registers in the
      poll, for example 91.2.
    value: "{{ outputs.normalize_readings.vars.max_temp }}"