Solar Telemetry Inverter Anomaly Sentinel and Maintenance Dispatch

id: solar-telemetry-inverter-anomaly-sentinel
namespace: company.energy

description: |
  Ingest utility and commercial solar photovoltaic (PV) SCADA telemetry, calculate
  plane-of-array (POA) temperature-corrected performance baselines using DuckDB SQL,
  detect thermal derating, string MPPT faults, soiling, and offline inverter trips,
  generate prioritized CMMS work orders, and route incidents through optional AI
  technician triage, CMMS REST dispatch, and Slack alerts.

triggers:
  - id: hourly_telemetry_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Hourly SCADA telemetry reconciliation sweep during generation
      daylight hours. Shipped disabled by default.
    cron: "0 6-19 * * *"
    disabled: true

  - id: scada_telemetry_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Webhook trigger for event-driven IoT edge gateways or plant SCADA
      batch payloads.
    key: "{{ secret('SOLAR_WEBHOOK_KEY') }}"

inputs:
  - id: inject_anomalies
    type: BOOL
    defaults: true
    description: When true, evaluates the full 6-inverter demo fleet including 4
      realistic operational faults (trip, thermal derating, soiling, MPPT
      fault). Set to false to simulate a 100% healthy baseline fleet.

  - id: irradiance_threshold_w_m2
    type: FLOAT
    defaults: 100.0
    description: Minimum plane-of-array (POA) solar irradiance in W/m2 required to
      evaluate generation performance (filters low-light dawn/dusk transients).

  - id: soiling_deficit_threshold_pct
    type: FLOAT
    defaults: 15.0
    description: Efficiency deficit percentage below the expected yield before
      flagging panel soiling and module washing requirements.

  - id: temp_derate_threshold_c
    type: FLOAT
    defaults: 65.0
    description: Core IGBT/heatsink internal temperature in Celsius triggering
      thermal curtailment warnings.

  - id: ppa_energy_tariff_usd_mwh
    type: FLOAT
    defaults: 65.0
    description: Power Purchase Agreement (PPA) energy export rate in USD per MWh
      used to quantify hourly revenue leakage.

  - id: send_slack_alert
    type: BOOL
    defaults: false
    description: Enable optional Slack notification to the Solar Operations Center
      channel when critical or high anomalies are detected (requires
      SLACK_WEBHOOK_URL secret).

  - id: dispatch_work_order
    type: BOOL
    defaults: false
    description: Automatically dispatch prioritized maintenance tickets to the field
      engineering CMMS API endpoint (requires CMMS_API_KEY secret and valid
      cmms_webhook_url).

  - id: cmms_webhook_url
    type: STRING
    defaults: "https://api.example-cmms.com/v1/work-orders/batch"
    description: REST API endpoint or webhook URL for field engineering CMMS (e.g.,
      UpKeep, Fiix, Maximo) receiving maintenance tickets.

  - id: enable_ai_diagnostics
    type: BOOL
    defaults: false
    description: Enable Google Gemini AI to analyze raw inverter metrics and
      formulate step-by-step field technician troubleshooting instructions
      (requires GEMINI_API_KEY secret).

  - id: model_name
    type: STRING
    defaults: gemini-2.5-flash
    description: Google Gemini model used when enable_ai_diagnostics is set to true.

  - id: weather_sensor_data
    type: JSON
    defaults: |
      [
        {
          "sensor_id": "MET-01",
          "plant_id": "SOLARIA-01",
          "timestamp": "2026-10-14T11:00:00Z",
          "poa_irradiance_w_m2": 850.0,
          "ambient_temp_c": 31.5,
          "wind_speed_m_s": 3.2
        }
      ]
    description: JSON array containing weather station metrics (Plane of Array
      irradiance, ambient temperature, wind speed).

  - id: inverter_telemetry_batch
    type: JSON
    defaults: |
      [
        {
          "inverter_id": "INV-01",
          "plant_id": "SOLARIA-01",
          "zone": "Zone-A",
          "dc_voltage_v": 755.0,
          "dc_current_a": 1642.0,
          "dc_power_kw": 1239.7,
          "ac_power_kw": 1215.0,
          "internal_temp_c": 48.2,
          "rated_capacity_kw": 1250.0
        },
        {
          "inverter_id": "INV-02",
          "plant_id": "SOLARIA-01",
          "zone": "Zone-A",
          "dc_voltage_v": 810.0,
          "dc_current_a": 0.0,
          "dc_power_kw": 0.0,
          "ac_power_kw": 0.0,
          "internal_temp_c": 28.1,
          "rated_capacity_kw": 1250.0
        },
        {
          "inverter_id": "INV-03",
          "plant_id": "SOLARIA-01",
          "zone": "Zone-A",
          "dc_voltage_v": 740.0,
          "dc_current_a": 1675.0,
          "dc_power_kw": 1239.5,
          "ac_power_kw": 720.0,
          "internal_temp_c": 78.5,
          "rated_capacity_kw": 1250.0
        },
        {
          "inverter_id": "INV-04",
          "plant_id": "SOLARIA-01",
          "zone": "Zone-A",
          "dc_voltage_v": 730.0,
          "dc_current_a": 1178.0,
          "dc_power_kw": 860.0,
          "ac_power_kw": 840.0,
          "internal_temp_c": 46.8,
          "rated_capacity_kw": 1250.0
        },
        {
          "inverter_id": "INV-05",
          "plant_id": "SOLARIA-01",
          "zone": "Zone-A",
          "dc_voltage_v": 420.0,
          "dc_current_a": 1100.0,
          "dc_power_kw": 462.0,
          "ac_power_kw": 445.0,
          "internal_temp_c": 49.0,
          "rated_capacity_kw": 1250.0
        },
        {
          "inverter_id": "INV-06",
          "plant_id": "SOLARIA-01",
          "zone": "Zone-A",
          "dc_voltage_v": 758.0,
          "dc_current_a": 1650.0,
          "dc_power_kw": 1250.7,
          "ac_power_kw": 1222.0,
          "internal_temp_c": 49.5,
          "rated_capacity_kw": 1250.0
        }
      ]
    description: JSON array containing string/central inverter electrical and
      thermal telemetry.

tasks:
  - id: stage_telemetry_data
    type: io.kestra.plugin.scripts.python.Script
    description: Parse and stage inverter telemetry and environmental weather
      metrics into CSV tables for DuckDB analysis.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      image: python:3.11-slim
    inputFiles:
      inverters_raw.json: "{{ inputs.inverter_telemetry_batch | toJson }}"
      weather_raw.json: "{{ inputs.weather_sensor_data | toJson }}"
    outputFiles:
      - inverters.csv
      - weather.csv
    env:
      INJECT_ANOMALIES: "{{ inputs.inject_anomalies }}"
    script: |
      import json
      import csv
      import os

      inject_anomalies = os.environ.get("INJECT_ANOMALIES", "true").lower() in ("true", "1", "yes")

      with open("inverters_raw.json", "r", encoding="utf-8") as f:
          inverters = json.load(f)

      with open("weather_raw.json", "r", encoding="utf-8") as f:
          weather = json.load(f)

      # If simulating clean baseline, normalize all units to nominal operating values
      if not inject_anomalies:
          clean_inverters = []
          for inv in inverters:
              clean_inv = dict(inv)
              clean_inv["dc_voltage_v"] = 755.0
              clean_inv["dc_current_a"] = 1645.0
              clean_inv["dc_power_kw"] = 1242.0
              clean_inv["ac_power_kw"] = 1218.0
              clean_inv["internal_temp_c"] = 48.0
              clean_inverters.append(clean_inv)
          inverters = clean_inverters

      with open("inverters.csv", "w", newline="", encoding="utf-8") as f:
          writer = csv.DictWriter(f, fieldnames=list(inverters[0].keys()))
          writer.writeheader()
          writer.writerows(inverters)

      with open("weather.csv", "w", newline="", encoding="utf-8") as f:
          writer = csv.DictWriter(f, fieldnames=list(weather[0].keys()))
          writer.writeheader()
          writer.writerows(weather)

      print(f"Staged {len(inverters)} inverter records and {len(weather)} weather records (inject_anomalies={inject_anomalies}).")

  - id: compute_inverter_analytics
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Execute DuckDB SQL analytics to compute temperature-adjusted
      expected yield, peer fleet baselines, anomaly classifications, and hourly
      revenue leakage.
    inputFiles:
      inverters.csv: "{{ outputs.stage_telemetry_data.outputFiles['inverters.csv'] }}"
      weather.csv: "{{ outputs.stage_telemetry_data.outputFiles['weather.csv'] }}"
    outputFiles:
      - inverter_performance_analysis.csv
    sql: |
      COPY (
        WITH raw_telemetry AS (
          SELECT
            inv.inverter_id,
            inv.plant_id,
            inv.zone,
            inv.rated_capacity_kw,
            inv.dc_voltage_v,
            inv.dc_current_a,
            inv.dc_power_kw,
            inv.ac_power_kw,
            inv.internal_temp_c,
            met.poa_irradiance_w_m2,
            met.ambient_temp_c,
            ROUND(CASE WHEN inv.dc_power_kw > 0 THEN (inv.ac_power_kw / inv.dc_power_kw) * 100.0 ELSE 0.0 END, 2) AS conversion_efficiency_pct,
            -- Expected AC power output under current POA irradiance with PV cell temperature coefficient (-0.35%/C above 25C STC)
            ROUND(inv.rated_capacity_kw * (met.poa_irradiance_w_m2 / 1000.0) * (1.0 - (0.0035 * GREATEST(0.0, met.ambient_temp_c - 25.0))), 2) AS expected_power_kw
          FROM read_csv_auto('inverters.csv') inv
          CROSS JOIN read_csv_auto('weather.csv') met
        ),
        performance_calculations AS (
          SELECT
            *,
            -- Generation deficit percentage compared to temperature-compensated expected power
            ROUND(CASE
              WHEN expected_power_kw > 0
              THEN ((expected_power_kw - ac_power_kw) / expected_power_kw) * 100.0
              ELSE 0.0
            END, 2) AS expected_deficit_pct
          FROM raw_telemetry
        ),
        classified_anomalies AS (
          SELECT
            inverter_id,
            plant_id,
            zone,
            poa_irradiance_w_m2,
            rated_capacity_kw,
            dc_voltage_v,
            dc_current_a,
            dc_power_kw,
            ac_power_kw,
            internal_temp_c,
            conversion_efficiency_pct,
            expected_power_kw,
            expected_deficit_pct,
            CASE
              -- 1. Complete zero-output trip / breaker open during full daylight
              WHEN poa_irradiance_w_m2 >= {{ inputs.irradiance_threshold_w_m2 }} AND ac_power_kw <= 5.0
                THEN 'INVERTER_TRIP_OFFLINE'
              -- 2. Thermal Derating: Heatsink core temperature exceeding threshold with generation drop or severe efficiency throttling
              WHEN internal_temp_c >= {{ inputs.temp_derate_threshold_c }} AND (conversion_efficiency_pct < 85.0 OR expected_deficit_pct >= 15.0)
                THEN 'THERMAL_DERATING'
              -- 3. MPPT Tracking Fault: Abnormal low DC voltage under load
              WHEN dc_voltage_v < 550.0 AND expected_deficit_pct >= 20.0
                THEN 'MPPT_TRACKER_FAULT'
              -- 4. Soiling / Module Degradation: Normal efficiency, but output deficit exceeds threshold
              WHEN expected_deficit_pct >= {{ inputs.soiling_deficit_threshold_pct }} AND conversion_efficiency_pct >= 95.0
                THEN 'SOILING_DEGRADATION'
              ELSE 'NOMINAL'
            END AS anomaly_classification,
            CASE
              WHEN poa_irradiance_w_m2 >= {{ inputs.irradiance_threshold_w_m2 }} AND ac_power_kw <= 5.0 THEN 'CRITICAL'
              WHEN internal_temp_c >= {{ inputs.temp_derate_threshold_c }} AND (conversion_efficiency_pct < 85.0 OR expected_deficit_pct >= 15.0) THEN 'HIGH'
              WHEN dc_voltage_v < 550.0 AND expected_deficit_pct >= 20.0 THEN 'HIGH'
              WHEN expected_deficit_pct >= {{ inputs.soiling_deficit_threshold_pct }} AND conversion_efficiency_pct >= 95.0 THEN 'MEDIUM'
              ELSE 'INFO'
            END AS severity_level,
            -- Lost generation energy in 1 hour (kWh)
            ROUND(GREATEST(0.0, expected_power_kw - ac_power_kw), 2) AS lost_energy_kwh_1h,
            -- Estimated revenue loss in USD for this hour based on PPA tariff
            ROUND(GREATEST(0.0, expected_power_kw - ac_power_kw) * ({{ inputs.ppa_energy_tariff_usd_mwh }} / 1000.0), 2) AS revenue_loss_usd_1h,
            CASE
              WHEN poa_irradiance_w_m2 >= {{ inputs.irradiance_threshold_w_m2 }} AND ac_power_kw <= 5.0
                THEN 'Inverter zero-output trip under ' || CAST(ROUND(poa_irradiance_w_m2, 0) AS VARCHAR) || ' W/m2 irradiance. Inspect AC breaker, DC contactors, and protection relays.'
              WHEN internal_temp_c >= {{ inputs.temp_derate_threshold_c }} AND (conversion_efficiency_pct < 85.0 OR expected_deficit_pct >= 15.0)
                THEN 'Core IGBT temperature (' || CAST(ROUND(internal_temp_c, 1) AS VARCHAR) || ' C) exceeds ' || CAST({{ inputs.temp_derate_threshold_c }} AS VARCHAR) || ' C limit. Clean heatsink dust filters and check cooling fan operation.'
              WHEN dc_voltage_v < 550.0 AND expected_deficit_pct >= 20.0
                THEN 'Abnormal DC string voltage (' || CAST(ROUND(dc_voltage_v, 1) AS VARCHAR) || ' V). Inspect combiner box string fuses and check for blown diodes or disconnected strings.'
              WHEN expected_deficit_pct >= {{ inputs.soiling_deficit_threshold_pct }} AND conversion_efficiency_pct >= 95.0
                THEN 'Generation is ' || CAST(ROUND(expected_deficit_pct, 1) AS VARCHAR) || '% below expected yield despite nominal conversion efficiency. Schedule robotic or manual module washing.'
              ELSE 'Operating within nominal operating parameters.'
            END AS recommended_action
          FROM performance_calculations
        )
        SELECT * FROM classified_anomalies ORDER BY lost_energy_kwh_1h DESC
      ) TO 'inverter_performance_analysis.csv' (HEADER, DELIMITER ',');

  - id: generate_work_orders_and_report
    type: io.kestra.plugin.scripts.python.Script
    description: Split analyzed telemetry into actionable CMMS work order tickets,
      build executive Markdown health report, and emit structured Kestra
      metrics.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      image: python:3.11-slim
    inputFiles:
      inverter_performance_analysis.csv: "{{
        outputs.compute_inverter_analytics.outputFiles['inverter_performance_an\
        alysis.csv'] }}"
    outputFiles:
      - priority_maintenance_tickets.csv
      - priority_maintenance_tickets.json
      - nominal_inverters_report.csv
      - fleet_summary_metrics.json
      - solar_fleet_health_report.md
    script: |
      import csv
      import json

      with open("inverter_performance_analysis.csv", "r", encoding="utf-8") as f:
          reader = list(csv.DictReader(f))

      tickets = []
      nominal = []
      total_inverters = len(reader)
      critical_trips = 0
      high_anomalies = 0
      medium_anomalies = 0
      total_lost_kwh = 0.0
      total_revenue_loss_usd = 0.0

      for row in reader:
          classification = row.get("anomaly_classification")
          severity = row.get("severity_level")
          lost_kwh = float(row.get("lost_energy_kwh_1h", 0.0))
          loss_usd = float(row.get("revenue_loss_usd_1h", 0.0))

          if classification != "NOMINAL":
              tickets.append(row)
              total_lost_kwh += lost_kwh
              total_revenue_loss_usd += loss_usd
              if severity == "CRITICAL":
                  critical_trips += 1
              elif severity == "HIGH":
                  high_anomalies += 1
              elif severity == "MEDIUM":
                  medium_anomalies += 1
          else:
              nominal.append(row)

      # 1. Write priority maintenance tickets (CSV and JSON)
      ticket_fieldnames = [
          "inverter_id", "plant_id", "zone", "severity_level",
          "anomaly_classification", "lost_energy_kwh_1h", "revenue_loss_usd_1h",
          "internal_temp_c", "dc_voltage_v", "ac_power_kw", "expected_power_kw", "recommended_action"
      ]
      with open("priority_maintenance_tickets.csv", "w", newline="", encoding="utf-8") as f:
          writer = csv.DictWriter(f, fieldnames=ticket_fieldnames, extrasaction="ignore")
          writer.writeheader()
          writer.writerows(tickets)

      with open("priority_maintenance_tickets.json", "w", encoding="utf-8") as f:
          json.dump(tickets, f, indent=2)

      # 2. Write nominal units
      with open("nominal_inverters_report.csv", "w", newline="", encoding="utf-8") as f:
          if nominal:
              writer = csv.DictWriter(f, fieldnames=list(nominal[0].keys()))
              writer.writeheader()
              writer.writerows(nominal)
          else:
              f.write("inverter_id,plant_id,zone,status\n")

      # 3. Write JSON metrics
      metrics = {
          "total_inverters": total_inverters,
          "nominal_inverters": len(nominal),
          "anomalies_detected": len(tickets),
          "critical_trips": critical_trips,
          "high_anomalies": high_anomalies,
          "medium_anomalies": medium_anomalies,
          "total_lost_energy_kwh_1h": round(total_lost_kwh, 2),
          "total_revenue_loss_usd_1h": round(total_revenue_loss_usd, 2),
          "projected_daily_revenue_loss_usd": round(total_revenue_loss_usd * 24.0, 2)
      }
      with open("fleet_summary_metrics.json", "w", encoding="utf-8") as f:
          json.dump(metrics, f, indent=2)

      # 4. Generate Executive Markdown Audit Report
      md = []
      md.append("# ☀️ Solar Fleet Telemetry & Inverter Health Audit Report")
      md.append(f"**Plant ID:** `SOLARIA-01` | **Status:** {'⚠️ ACTION REQUIRED' if tickets else '✅ FLEET NOMINAL'}\n")
      md.append("## Executive Fleet Summary\n")
      md.append("| Metric | Value |")
      md.append("| :--- | :--- |")
      md.append(f"| **Total Monitored Inverters** | `{total_inverters}` |")
      md.append(f"| **Nominal Inverters** | `{len(nominal)}` |")
      md.append(f"| **Anomalous Inverters** | `{len(tickets)}` |")
      md.append(f"| **Critical Breaker Trips** | `{critical_trips}` |")
      md.append(f"| **Hourly Generation Deficit** | `{total_lost_kwh:,.2f} kWh` |")
      md.append(f"| **Hourly Revenue Loss** | `${total_revenue_loss_usd:,.2f} USD` |")
      md.append(f"| **Projected 24h Revenue Exposure** | `${total_revenue_loss_usd * 24.0:,.2f} USD` |\n")

      if tickets:
          md.append("## Priority Maintenance Tickets\n")
          md.append("| Inverter ID | Severity | Fault Classification | Actual / Expected kW | Deficit (kWh) | Revenue Loss (1h) | Recommended Field Action |")
          md.append("| :--- | :--- | :--- | :--- | :--- | :--- | :--- |")
          for t in tickets:
              actual_exp = f"{float(t['ac_power_kw']):,.0f} / {float(t['expected_power_kw']):,.0f}"
              md.append(f"| **{t['inverter_id']}** | `{t['severity_level']}` | `{t['anomaly_classification']}` | {actual_exp} | `{float(t['lost_energy_kwh_1h']):,.1f}` | `${float(t['revenue_loss_usd_1h']):,.2f}` | {t['recommended_action']} |")
      else:
          md.append("## Fleet Status\n")
          md.append("All inverters are performing within expected temperature-compensated irradiance baselines. No work orders required.")

      with open("solar_fleet_health_report.md", "w", encoding="utf-8") as f:
          f.write("\n".join(md))

      # 5. Emit Kestra outputs
      print(f"::{{'outputs': {json.dumps(metrics)}}}::")

  - id: expose_fleet_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Expose key generation and revenue risk metrics to Kestra execution
      outputs and downstream monitoring.
    values:
      total_inverters: "{{ outputs.generate_work_orders_and_report.total_inverters }}"
      nominal_inverters: "{{ outputs.generate_work_orders_and_report.nominal_inverters }}"
      anomalies_detected: "{{ outputs.generate_work_orders_and_report.anomalies_detected }}"
      critical_trips: "{{ outputs.generate_work_orders_and_report.critical_trips }}"
      total_lost_energy_kwh_1h: "{{ outputs.generate_work_orders_and_report.total_lost_energy_kwh_1h }}"
      total_revenue_loss_usd_1h: "{{ outputs.generate_work_orders_and_report.total_revenue_loss_usd_1h }}"
      projected_daily_revenue_loss_usd: "{{
        outputs.generate_work_orders_and_report.projected_daily_revenue_loss_usd
        }}"

  - id: route_incident_escalation
    type: io.kestra.plugin.core.flow.If
    description: Branch on detected anomalies. Escalate maintenance dispatch,
      optional AI root-cause analysis, and notification alerts when faults
      exist.
    condition: "{{ outputs.expose_fleet_metrics.values.anomalies_detected > 0 }}"
    then:
      - id: log_anomaly_alert
        type: io.kestra.plugin.core.log.Log
        message: >
          [INCIDENT ESCALATION] {{
          outputs.expose_fleet_metrics.values.anomalies_detected }} solar
          inverters flagged with performance anomalies. Critical Trips: {{
          outputs.expose_fleet_metrics.values.critical_trips }} | Hourly Revenue
          Loss: ${{
          outputs.expose_fleet_metrics.values.total_revenue_loss_usd_1h }} USD.

      - id: check_slack_notification_enabled
        type: io.kestra.plugin.core.flow.If
        description: Dispatch Slack incident alert card when send_slack_alert input is true.
        condition: "{{ inputs.send_slack_alert }}"
        then:
          - id: send_slack_incident_alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "channel": "#solar-operations-center",
                "text": "⚠️ *Solar PV Fleet Incident Alert - Plant SOLARIA-01*",
                "attachments": [
                  {
                    "color": "#D32F2F",
                    "fields": [
                      {"title": "Anomalies Detected", "value": "{{ outputs.expose_fleet_metrics.values.anomalies_detected }}", "short": true},
                      {"title": "Critical Trips", "value": "{{ outputs.expose_fleet_metrics.values.critical_trips }}", "short": true},
                      {"title": "Hourly Generation Loss", "value": "{{ outputs.expose_fleet_metrics.values.total_lost_energy_kwh_1h }} kWh", "short": true},
                      {"title": "Hourly Revenue Leakage", "value": "${{ outputs.expose_fleet_metrics.values.total_revenue_loss_usd_1h }} USD", "short": true}
                    ],
                    "footer": "Kestra Solar Telemetry Sentinel"
                  }
                ]
              }

      - id: check_cmms_dispatch_enabled
        type: io.kestra.plugin.core.flow.If
        description: Automatically dispatch priority maintenance tickets to the field
          engineering CMMS API endpoint when dispatch_work_order is true.
        condition: "{{ inputs.dispatch_work_order }}"
        then:
          - id: dispatch_cmms_work_orders
            type: io.kestra.plugin.core.http.Request
            description: Post prioritized maintenance tickets to the external CMMS work
              order ingestion endpoint.
            uri: "{{ inputs.cmms_webhook_url }}"
            method: POST
            contentType: application/json
            headers:
              Authorization: "Bearer {{ secret('CMMS_API_KEY') }}"
            body: |
              {
                "plant_id": "SOLARIA-01",
                "timestamp": "{{ now() }}",
                "source": "kestra-solar-telemetry-sentinel",
                "critical_trips": {{ outputs.expose_fleet_metrics.values.critical_trips }},
                "anomalies_count": {{ outputs.expose_fleet_metrics.values.anomalies_detected }},
                "estimated_revenue_loss_usd_1h": {{ outputs.expose_fleet_metrics.values.total_revenue_loss_usd_1h }},
                "tickets": {{ outputs.generate_work_orders_and_report.outputFiles['priority_maintenance_tickets.json'] | readString }}
              }

          - id: log_cmms_dispatch_success
            type: io.kestra.plugin.core.log.Log
            message: "Dispatched {{ outputs.expose_fleet_metrics.values.anomalies_detected
              }} priority tickets to CMMS at {{ inputs.cmms_webhook_url }}.
              Response status: {{ outputs.dispatch_cmms_work_orders.code }}."

      - id: check_ai_diagnostics_enabled
        type: io.kestra.plugin.core.flow.If
        description: Invoke Google Gemini to synthesize root-cause diagnostics and
          technician field safety instructions.
        condition: "{{ inputs.enable_ai_diagnostics }}"
        then:
          - id: generate_ai_field_troubleshooting_guide
            type: io.kestra.plugin.ai.completion.ChatCompletion
            description: Uses Google Gemini to synthesize root-cause diagnostics and field
              technician troubleshooting instructions.
            provider:
              type: io.kestra.plugin.ai.provider.GoogleGemini
              apiKey: "{{ secret('GEMINI_API_KEY') }}"
              modelName: "{{ inputs.model_name }}"
            messages:
              - type: SYSTEM
                content: |
                  You are a senior solar PV field operations and electrical engineering specialist. Analyze inverter telemetry faults and write clear, concise step-by-step troubleshooting instructions for field technicians.
              - type: USER
                content: |
                  Review the following flagged solar inverter telemetry tickets from Plant SOLARIA-01:
                  {{ outputs.generate_work_orders_and_report.outputFiles['priority_maintenance_tickets.csv'] | readString }}

                  Please provide:
                  1. Root-cause diagnosis for each flagged inverter (Trip, Thermal Derating, MPPT Fault, Soiling).
                  2. Priority sequence for field technician dispatch based on revenue leakage.
                  3. Key electrical safety precautions (DC lock-out / tag-out, thermal cooling times).

          - id: save_ai_diagnostics_guide
            type: io.kestra.plugin.core.storage.Write
            description: Persist the AI troubleshooting guide as a Markdown artifact in
              Kestra internal storage.
            extension: .md
            content: "{{ outputs.generate_ai_field_troubleshooting_guide.textOutput }}"

    else:
      - id: log_fleet_nominal
        type: io.kestra.plugin.core.log.Log
        message: "All {{ outputs.expose_fleet_metrics.values.total_inverters }}
          monitored inverters operating nominally within expected irradiance
          baselines."