Cold-Chain Temperature Excursions with Kestra and DuckDB

id: cold-chain-temperature-excursions
namespace: company.team
description: |
  Turn raw fridge, freezer and storage-room temperature logs into a list of
  cold-chain excursions: when each unit went out of range, for how long, how
  hot or cold it got, whether a sensor stopped reporting, and which readings
  could not be read, so the affected stock can be quarantined and checked.

inputs:
  - id: readings_file
    type: FILE
    required: false
    description: CSV with the columns site, unit_id, unit_type (FRIDGE, FREEZER or
      AMBIENT), reading_time (YYYY-MM-DD HH:MM) and temperature_c. Leave empty
      to run against the built-in sample sensor log.
  - id: min_excursion_minutes
    type: INT
    defaults: 30
    description: Out-of-range periods shorter than this are reported as BRIEF (for
      example a door left open while loading) instead of CRITICAL.
  - id: max_gap_minutes
    type: INT
    defaults: 60
    description: A unit that sends no reading for longer than this is reported as a
      SENSOR_GAP, because nobody knows what the temperature was during that
      time.
  - id: notify_slack
    type: BOOL
    defaults: false
    description: Post the findings to Slack when there is at least one critical
      excursion, sensor gap or data error. Requires the SLACK_WEBHOOK_URL
      secret.

triggers:
  - id: daily_cold_chain_review
    type: io.kestra.plugin.core.trigger.Schedule
    description: Review the previous day's sensor log every morning at 07:00 IST,
      before stock from the affected units is picked or dispatched. Shipped
      disabled. Enable it once the flow reads your real sensor export instead of
      the sample.
    cron: "0 7 * * *"
    timezone: Asia/Kolkata
    disabled: true

tasks:
  - id: sample_readings
    type: io.kestra.plugin.core.storage.Write
    description: A 12-hour sensor log for four units, one reading every 15 minutes,
      built with a Pebble loop. It contains a long high excursion, a brief
      door-open spike, a freezing excursion, a sensor gap, an unreadable reading
      and an unknown unit type. Only used when no readings_file is provided.
    extension: .csv
    content: |
      site,unit_id,unit_type,reading_time,temperature_c
      {% set hot = [8.6, 9.1, 9.7, 10.4, 11.2, 10.6, 9.4, 8.3] -%}
      {% set cold = [1.4, 0.8, 1.1] -%}
      {% for i in range(0, 47) -%}
      {% set t = "2026-10-01 " ~ ((i / 4) | numberFormat("00")) ~ ":" ~ ((i % 4 * 15) | numberFormat("00")) -%}
      {% set fridge1 = (i >= 20 and i <= 27) ? hot[i - 20] : 4.0 + (i % 5) * 0.3 -%}
      {% set fridge2 = (i == 14) ? 8.9 : ((i >= 32 and i <= 34) ? cold[i - 32] : 5.0 + (i % 3) * 0.2) -%}
      Pune DC,FRG-01,FRIDGE,{{ t }},{{ fridge1 | numberFormat("0.0") }}
      Pune DC,FRG-02,FRIDGE,{{ t }},{{ fridge2 | numberFormat("0.0") }}
      {% if i < 16 or i > 25 -%}
      Pune DC,FRZ-01,FREEZER,{{ t }},{{ (-20.0 + (i % 4) * 0.5) | numberFormat("0.0") }}
      {% endif -%}
      Pune DC,AMB-01,AMBIENT,{{ t }},{{ (22.0 + (i % 6) * 0.4) | numberFormat("0.0") }}
      {% endfor -%}
      Pune DC,FRG-01,FRIDGE,2026-10-01 06:07,ERR
      Pune DC,FRG-03,CHILLER,2026-10-01 06:00,4.5

  - id: find_excursions
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Match every reading to the allowed range of its unit type, group
      consecutive out-of-range readings into excursions, find gaps between
      readings, and collect unreadable rows in DuckDB. The event report is
      exported to CSV and the last query returns the summary.
    inputFiles:
      readings.csv: "{{ inputs.readings_file ?? outputs.sample_readings.uri }}"
    outputFiles:
      - excursion_report.csv
    fetchType: FETCH_ONE
    sql: |
      CREATE TABLE ranges AS
      SELECT * FROM (VALUES
        ('FRIDGE', 2.0, 8.0),
        ('FREEZER', -25.0, -15.0),
        ('AMBIENT', 15.0, 25.0)
      ) AS t(unit_type, min_c, max_c);

      CREATE TABLE raw AS
      SELECT
        trim(site) AS site,
        trim(unit_id) AS unit_id,
        upper(trim(coalesce(unit_type, ''))) AS unit_type,
        trim(coalesce(reading_time, '')) AS reading_time_raw,
        trim(coalesce(temperature_c, '')) AS temperature_raw,
        try_cast(trim(reading_time) AS TIMESTAMP) AS ts,
        try_cast(trim(temperature_c) AS DOUBLE) AS temp
      FROM read_csv('{{ workingDir }}/readings.csv', header = true, all_varchar = true);

      CREATE TABLE readings AS
      SELECT
        r.*,
        g.min_c,
        g.max_c,
        CASE WHEN r.temp > g.max_c THEN 'HIGH' WHEN r.temp < g.min_c THEN 'LOW' ELSE 'IN' END AS band
      FROM raw r
      JOIN ranges g ON g.unit_type = r.unit_type
      WHERE r.ts IS NOT NULL AND r.temp IS NOT NULL;

      CREATE TABLE sequenced AS
      SELECT
        *,
        lead(ts) OVER (PARTITION BY site, unit_id ORDER BY ts) AS next_ts,
        row_number() OVER (PARTITION BY site, unit_id ORDER BY ts)
          - row_number() OVER (PARTITION BY site, unit_id, band ORDER BY ts) AS island
      FROM readings;

      CREATE TABLE excursions AS
      SELECT
        site,
        unit_id,
        unit_type,
        CASE WHEN band = 'HIGH' THEN 'HIGH_EXCURSION' ELSE 'LOW_EXCURSION' END AS event,
        min(ts) AS start_time,
        coalesce(max_by(next_ts, ts), max(ts)) AS end_time,
        count(*) AS readings,
        CASE WHEN band = 'HIGH' THEN max(temp) ELSE min(temp) END AS peak_c,
        any_value(min_c) AS min_c,
        any_value(max_c) AS max_c
      FROM sequenced
      WHERE band <> 'IN'
      GROUP BY site, unit_id, unit_type, band, island;

      CREATE TABLE gaps AS
      SELECT site, unit_id, unit_type, 'SENSOR_GAP' AS event, ts AS start_time, next_ts AS end_time, 0 AS readings, NULL::DOUBLE AS peak_c, min_c, max_c
      FROM sequenced
      WHERE date_diff('minute', ts, next_ts) > {{ inputs.max_gap_minutes }}
      UNION ALL
      SELECT site, unit_id, any_value(unit_type), 'SENSOR_GAP', max(ts), (SELECT max(ts) FROM readings), 0, NULL::DOUBLE, any_value(min_c), any_value(max_c)
      FROM readings
      GROUP BY site, unit_id
      HAVING date_diff('minute', max(ts), (SELECT max(ts) FROM readings)) > {{ inputs.max_gap_minutes }};

      CREATE TABLE bad_rows AS
      SELECT
        site,
        unit_id,
        unit_type,
        list_filter([
          CASE WHEN unit_type NOT IN (SELECT unit_type FROM ranges) THEN 'unknown unit_type "' || unit_type || '" (use FRIDGE, FREEZER or AMBIENT)' END,
          CASE WHEN ts IS NULL THEN 'reading_time "' || reading_time_raw || '" is not a timestamp' END,
          CASE WHEN temp IS NULL THEN 'temperature "' || temperature_raw || '" is not a number' END
        ], lambda x: x IS NOT NULL) AS problems
      FROM raw;

      CREATE TABLE report AS
      SELECT * FROM (
      SELECT
        site,
        unit_id,
        unit_type,
        event,
        status,
        start_time,
        end_time,
        duration_minutes,
        readings,
        peak_c,
        allowed_range,
        action
      FROM (
        SELECT
          *,
          CASE
            WHEN event = 'SENSOR_GAP' THEN 'SENSOR_GAP'
            WHEN duration_minutes >= {{ inputs.min_excursion_minutes }} THEN 'CRITICAL'
            ELSE 'BRIEF'
          END AS status,
          CASE
            WHEN event = 'SENSOR_GAP' THEN 'No reading for ' || duration_minutes || ' min: check the logger and treat the stock as unverified for this period'
            WHEN event = 'HIGH_EXCURSION' AND duration_minutes >= {{ inputs.min_excursion_minutes }} THEN 'Above ' || max_c || ' C for ' || duration_minutes || ' min, peak ' || peak_c || ' C: quarantine the stock and check stability data before use'
            WHEN event = 'LOW_EXCURSION' AND duration_minutes >= {{ inputs.min_excursion_minutes }} THEN 'Below ' || min_c || ' C for ' || duration_minutes || ' min, lowest ' || peak_c || ' C: quarantine the stock, freeze-sensitive products may be damaged'
            ELSE 'Out of range for ' || duration_minutes || ' min (peak ' || peak_c || ' C): likely a door opening, check if it repeats'
          END AS action
        FROM (
          SELECT
            *,
            date_diff('minute', start_time, end_time) AS duration_minutes,
            min_c || ' to ' || max_c || ' C' AS allowed_range
          FROM (SELECT * FROM excursions UNION ALL SELECT * FROM gaps)
        )
      )
      UNION ALL
      SELECT
        site,
        unit_id,
        unit_type,
        'UNREADABLE' AS event,
        'DATA_ERROR' AS status,
        NULL AS start_time,
        NULL AS end_time,
        NULL AS duration_minutes,
        count(*) AS readings,
        NULL AS peak_c,
        NULL AS allowed_range,
        count(*) || ' row(s) skipped: ' || array_to_string(list_distinct(flatten(list(problems))), ', ') AS action
      FROM bad_rows
      WHERE len(problems) > 0
      GROUP BY site, unit_id, unit_type
      )
      ORDER BY
        CASE status WHEN 'CRITICAL' THEN 0 WHEN 'SENSOR_GAP' THEN 1 WHEN 'DATA_ERROR' THEN 2 ELSE 3 END,
        duration_minutes DESC NULLS LAST,
        site,
        unit_id;

      COPY report TO '{{ outputFiles["excursion_report.csv"] }}' (HEADER, DELIMITER ',');

      SELECT
        (SELECT count(*) FROM raw) AS total_readings,
        (SELECT count(DISTINCT site || '/' || unit_id) FROM readings) AS units_checked,
        (SELECT count(DISTINCT site || '/' || unit_id) FROM readings)
          - (SELECT count(DISTINCT site || '/' || unit_id) FROM report WHERE status IN ('CRITICAL', 'SENSOR_GAP', 'DATA_ERROR') AND unit_type IN (SELECT unit_type FROM ranges)) AS units_ok,
        count(*) FILTER (WHERE status = 'CRITICAL') AS critical_count,
        count(*) FILTER (WHERE status = 'BRIEF') AS brief_count,
        count(*) FILTER (WHERE status = 'SENSOR_GAP') AS gap_count,
        count(*) FILTER (WHERE status = 'DATA_ERROR') AS data_error_count,
        coalesce(max(duration_minutes) FILTER (WHERE status = 'CRITICAL'), 0) AS longest_excursion_minutes,
        coalesce(string_agg(site || ' / ' || unit_id || ' from ' || strftime(start_time, '%Y-%m-%d %H:%M') || ': ' || action, chr(10)) FILTER (WHERE status = 'CRITICAL'), '') AS critical_details,
        coalesce(string_agg(site || ' / ' || unit_id || ' from ' || strftime(start_time, '%Y-%m-%d %H:%M') || ': ' || action, chr(10)) FILTER (WHERE status = 'SENSOR_GAP'), '') AS gap_details,
        coalesce(string_agg(site || ' / ' || unit_id || ': ' || action, chr(10)) FILTER (WHERE status = 'DATA_ERROR'), '') AS data_error_details,
        coalesce(string_agg(site || ' / ' || unit_id || ' from ' || strftime(start_time, '%Y-%m-%d %H:%M') || ': ' || action, chr(10)) FILTER (WHERE status = 'BRIEF'), '') AS brief_details
      FROM report;

  - id: route_findings
    type: io.kestra.plugin.core.flow.If
    description: Branch on the result. Critical excursions, sensor gaps and
      unreadable data mean some stock may be compromised, so they are logged as
      warnings and optionally sent to Slack. A clean log is recorded too,
      together with any brief spikes.
    condition: "{{ outputs.find_excursions.outputs[0].row.critical_count +
      outputs.find_excursions.outputs[0].row.gap_count +
      outputs.find_excursions.outputs[0].row.data_error_count > 0 }}"
    then:
      - id: log_findings
        type: io.kestra.plugin.core.log.Log
        level: WARN
        description: Write every critical excursion, sensor gap, data error and brief
          spike to the execution logs with the action to take.
        message: |
          Cold chain review: {{ outputs.find_excursions.outputs[0].row.critical_count }} critical excursion(s), {{ outputs.find_excursions.outputs[0].row.gap_count }} sensor gap(s), {{ outputs.find_excursions.outputs[0].row.data_error_count }} unit(s) with unreadable data, {{ outputs.find_excursions.outputs[0].row.brief_count }} brief spike(s). {{ outputs.find_excursions.outputs[0].row.units_ok }} of {{ outputs.find_excursions.outputs[0].row.units_checked }} units clean, {{ outputs.find_excursions.outputs[0].row.total_readings }} readings.
          Critical:
          {{ outputs.find_excursions.outputs[0].row.critical_details }}
          Sensor gaps:
          {{ outputs.find_excursions.outputs[0].row.gap_details }}
          Data errors:
          {{ outputs.find_excursions.outputs[0].row.data_error_details }}
          Brief spikes:
          {{ outputs.find_excursions.outputs[0].row.brief_details }}

      - id: slack_enabled
        type: io.kestra.plugin.core.flow.If
        description: Only call Slack when the notify_slack input is true, so the flow
          runs out of the box without any secret.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: alert_quality_team
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Tell the warehouse and quality team which units to check and which
              stock to quarantine.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              :thermometer: Cold chain review: {{ outputs.find_excursions.outputs[0].row.critical_count }} critical excursion(s), {{ outputs.find_excursions.outputs[0].row.gap_count }} sensor gap(s), {{ outputs.find_excursions.outputs[0].row.data_error_count }} unit(s) with unreadable data.
              {{ outputs.find_excursions.outputs[0].row.critical_details }}
              {{ outputs.find_excursions.outputs[0].row.gap_details }}
              Full report: execution {{ execution.id }} in flow {{ flow.namespace }}.{{ flow.id }}
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        description: Record a clean review, including any brief spikes, so the execution
          history shows the cold chain was checked.
        message: "Cold chain review passed: {{
          outputs.find_excursions.outputs[0].row.units_checked }} units, {{
          outputs.find_excursions.outputs[0].row.total_readings }} readings, no
          critical excursion or sensor gap, {{
          outputs.find_excursions.outputs[0].row.brief_count }} brief spike(s)."

errors:
  - id: log_review_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    description: Make a failed review loud. If the file is missing or malformed, no
      reading was checked, and a silent failure looks like a cold chain with no
      excursions.
    message: "Cold chain review FAILED in {{ flow.namespace }}.{{ flow.id }}
      (execution {{ execution.id }}). No readings were checked. Make sure
      readings_file is a CSV with the columns site, unit_id, unit_type,
      reading_time and temperature_c."

outputs:
  - id: excursion_report
    type: FILE
    description: One row per event (HIGH_EXCURSION, LOW_EXCURSION, SENSOR_GAP or
      UNREADABLE) with the status (CRITICAL, BRIEF, SENSOR_GAP or DATA_ERROR),
      start and end time, duration, peak temperature, allowed range and the
      action to take.
    value: "{{ outputs.find_excursions.outputFiles['excursion_report.csv'] }}"
  - id: excursion_summary
    type: JSON
    description: 'Counts and details, e.g. {"units_checked": 4, "critical_count": 2,
      "gap_count": 1, "data_error_count": 2, "longest_excursion_minutes": 120,
      ...}.'
    value: "{{ outputs.find_excursions.outputs[0].row | toJson }}"

Vaccines, insulin, biologics and chilled food are only safe if they stayed in their temperature range the whole time. Data loggers record a reading every few minutes, but a CSV with thousands of rows does not tell anyone that a fridge sat at 11 C for two hours last night, or that a freezer logger went quiet. This blueprint reviews the sensor log on a schedule, turns it into a short list of excursions with their duration and peak temperature, and tells the quality team which stock to quarantine. Every check runs locally in DuckDB: no IoT platform, no paid API, and no data leaving your Kestra instance.

How it works

sample_readings (io.kestra.plugin.core.storage.Write) writes a 12-hour sensor log for four units with a Pebble loop, one reading every 15 minutes. It contains one example of every problem and is used only when the readings_file input is empty, so the flow runs out of the box.
find_excursions (io.kestra.plugin.jdbc.duckdb.Queries) loads the CSV through inputFiles and:
- matches each reading to the allowed range of its unit type: FRIDGE 2 to 8 C, FREEZER -25 to -15 C, AMBIENT 15 to 25 C;
- groups consecutive out-of-range readings into one excursion with window functions (the gaps-and-islands pattern), and measures it from the first bad reading to the next good one;
- marks an excursion CRITICAL when it lasts at least min_excursion_minutes, and BRIEF when it is shorter (typically a door opening);
- reports a SENSOR_GAP when a unit sends no reading for longer than max_gap_minutes, including a unit that stopped reporting before the end of the log;
- reports DATA_ERROR for rows with an unknown unit type, a timestamp that cannot be read, or a temperature that is not a number.
The event report is written to excursion_report.csv with outputFiles, and the final SELECT returns the summary row through fetchType: FETCH_ONE, available as outputs.find_excursions.outputs[0].row.
route_findings (io.kestra.plugin.core.flow.If) logs every critical excursion, sensor gap, data error and brief spike when anything needs action, and a nested If posts the findings to Slack when notify_slack is true. A clean review is logged as well.
The errors block logs a failed review loudly, because a review that silently does not run looks the same as a cold chain with no excursions.
The daily_cold_chain_review Schedule trigger (every day at 07:00 IST, shipped disabled) reviews the previous day before stock is picked or dispatched.

What you get

outputs.excursion_report: CSV with site, unit_id, unit_type, event, status, start_time, end_time, duration_minutes, readings, peak_c, allowed_range and action, sorted with the most serious events first.
outputs.excursion_summary: JSON with total_readings, units_checked, units_ok, critical_count, brief_count, gap_count, data_error_count, longest_excursion_minutes and the details for each group.
Plain-language actions such as "Above 8.0 C for 120 min, peak 11.2 C: quarantine the stock and check stability data before use" or "No reading for 165 min: check the logger and treat the stock as unverified for this period".
With the sample data: 2 critical excursions (a fridge above 8 C for 120 minutes and a fridge below 2 C for 45 minutes), 1 sensor gap of 165 minutes on the freezer, 1 brief 15-minute spike, and 2 data errors (an ERR reading and an unknown unit type CHILLER). The ambient store is clean.

Who it's for

Pharmacies, hospitals, vaccine stores and clinical trial depots that keep temperature-sensitive medicine in fridges and freezers.
Food and grocery warehouses, dark stores and cold storage operators.
Quality and data teams that receive logger exports and want a repeatable excursion review with an audit trail.

Why orchestrate this with Kestra

Most data loggers can raise an alarm, but nobody reviews the full history every day, sensor gaps go unnoticed, and the review is not recorded anywhere. Kestra runs the same review on a schedule, keeps every report in the execution history as evidence that the cold chain was checked, branches on the result, and alerts the people who act on it. Because the input is just a file, the same flow can be triggered when a new logger export lands in S3, SFTP or Google Drive, or run once per site with a ForEach.

Prerequisites

Nothing for the demo: the sample sensor log is built by the flow.
For your own data: a CSV export from your data loggers or monitoring system with the columns site, unit_id, unit_type, reading_time and temperature_c.

Inputs

readings_file (FILE, optional): your sensor log CSV. When empty, the built-in sample is used.
min_excursion_minutes (INT, default 30): excursions shorter than this are reported as BRIEF.
max_gap_minutes (INT, default 60): a longer silence between two readings is a SENSOR_GAP.
notify_slack (BOOL, default false): post the findings to Slack when anything needs action.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook URL. Only needed when notify_slack is true.

Quick start

Save the flow and run it with the default inputs.
Open the logs of log_findings and download excursion_report from the Outputs tab to see the result for the sample sensor log.
Run it again and upload your own logger export as readings_file.
Adjust the ranges in the ranges table of find_excursions if your products need different limits, and set max_gap_minutes to a few times your logger interval.
Optional: add the SLACK_WEBHOOK_URL secret and set notify_slack to true.
Replace the input with your real source and set disabled: false on daily_cold_chain_review.

Common pitfalls

The allowed ranges are typical defaults, not product rules. Always follow the storage conditions on the product label and your stability data before releasing stock after an excursion.
Use one time zone for all readings. Mixing local time and UTC creates fake gaps and overlaps.
An excursion is measured until the next reading in range. With a long logging interval, durations are rounded up to that interval.
Readings that cannot be parsed are skipped and reported as DATA_ERROR, so a logger that writes ERR or an empty value never hides inside an average.

How to extend

Read the readings straight from your monitoring database or time-series store instead of a file, and write report back to a table.
Trigger the review when a new logger export lands, with the S3, SFTP or Google Drive triggers.
Add a per-product range table and join it on unit_id to check each unit against what it actually stores.
Open a deviation ticket in Jira or ServiceNow for every critical excursion.

Find Cold-Chain Temperature Excursions in Fridge and Freezer Sensor Logs

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Inputs

Secrets

Quick start

Common pitfalls

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Find Cold-Chain Temperature Excursions in Fridge and Freezer Sensor Logs

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Inputs

Secrets

Quick start

Common pitfalls

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Business Blueprints

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

Generate AI Blog Posts in Your Brand Voice and Publish to WordPress as Draft

Parallel Multilingual Markdown Translation with OpenAI

AI Contract Clause Risk Analyzer and Legal Sign-Off Gate

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

AI Inbound Lead Enrichment, Scoring and VIP Routing Gate

More Data Blueprints

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Automate Cross-Cloud ETL from Azure Blob Storage to BigQuery with dbt Cloud

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

AI Data Quality Anomaly Advisor and SQL Remediation Generator

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

New to Kestra?