"Tutorial: Cold Chain Monitoring for Pharmaceuticals"

June 2, 2026 | 25 min read

Introduction

Pharmaceutical cold chains are among the most heavily regulated edge-computing domains. A single temperature excursion can render a batch of vaccines unusable, while a leaked patient identifier can trigger a GDPR fine. This tutorial builds a complete GDP-compliant cold-chain monitor using Pyvorin Edge: temperature sensors, privacy redaction, tamper-evident audit logging, signed bundles, and offline-capable cloud sync.

Regulatory Context

Two frameworks dominate pharmaceutical transport:

  • EU GDP (Good Distribution Practice): Requires continuous temperature monitoring, alarm records, and documented corrective actions.
  • WHO PQS (Performance, Quality & Safety): Defines temperature ranges for vaccine cold boxes: +2 °C to +8 °C for standard refrigerators, −25 °C to −15 °C for freezer transport.

Our pipeline will enforce the +2 °C to +8 °C window and log every excursion with a cryptographically signed audit record.

Sensor Setup

We use two temperature sensors: one inside the refrigerator compartment and one ambient sensor to detect door-open events caused by high ambient influx. 


from pyvorin_edge.pipeline import Pipeline, WindowConfig, RuleConfig
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading

pipeline = Pipeline(name="cold_chain_pharma")

# Refrigerator core temperature
pipeline.add_sensor(
    Sensor(
        name="fridge_temp",
        sensor_type=SensorType.TEMPERATURE,
        unit="C",
        normal_range=(2.0, 8.0),
        location="Refrigerator_Compartment",
    )
)

# Ambient temperature for contextual analysis
pipeline.add_sensor(
    Sensor(
        name="ambient_temp",
        sensor_type=SensorType.TEMPERATURE,
        unit="C",
        normal_range=(15.0, 25.0),
        location="Vehicle_Cabin",
    )
)

# 60-second rolling window for trend analysis
pipeline.add_window(
    WindowConfig(
        duration_seconds=60.0,
        sensor_name="fridge_temp",
        window_type="rolling",
    )
)

# GDP-critical excursion rules (no cooldown — every second matters)
pipeline.add_rule(
    RuleConfig(
        name="temp_excursion_low",
        condition_expr="ctx.value < 2.0",
        severity="critical",
        cooldown_seconds=0.0,
    )
)
pipeline.add_rule(
    RuleConfig(
        name="temp_excursion_high",
        condition_expr="ctx.value > 8.0",
        severity="critical",
        cooldown_seconds=0.0,
    )
)

Privacy Redaction

Transport manifests often contain patient identifiers, batch numbers, and trial codes. The PrivacyPolicy class in edge_sdk/pyvorin_edge/policies.py lets you declare which fields are redacted before any data leaves the device. 


from pyvorin_edge.policies import PrivacyPolicy

privacy = PrivacyPolicy(
    redact_fields=["patient_id", "shipment_id", "trial_code"],
    retain_hours=72.0,
    pseudonymize_ids=True,
    local_only_sensors=["gps_location"],
)
pipeline.add_policy(privacy)

Signed Audit Reports

Every privacy action and temperature excursion must be auditable. The PrivacyAudit class in edge_runtime/pyv_edge_agent/privacy_firewall/audit.py maintains a SHA-256 hash chain in SQLite, making tampering evident. 


from pyv_edge_agent.privacy_firewall.audit import PrivacyAudit
import time

audit = PrivacyAudit(db_path="edge_store.db")

# Log policy activation
audit.log_action(
    action="PRIVACY_POLICY_LOADED",
    details={
        "redact_fields": privacy.redact_fields,
        "pseudonymize_ids": privacy.pseudonymize_ids,
        "purpose": "GDP compliance",
        "data_subjects": ["patients"],
        "data_categories": ["temperature", "shipment_manifest"],
    },
)

# Simulate an excursion and log it
excursion_reading = SensorReading(
    sensor_name="fridge_temp",
    timestamp=time.time(),
    value=10.5,
    unit="C",
    metadata={"batch": "B12345", "patient_id": "REDACTED"},
)
audit.log_action(
    action="TEMPERATURE_EXCURSION",
    details={
        "sensor": excursion_reading.sensor_name,
        "value": excursion_reading.value,
        "threshold_breached": "upper",
        "severity": "critical",
    },
)

# Verify chain integrity
assert audit.verify_chain(), "Audit chain has been tampered with!"
print(f"Audit records: {len(audit.get_audit_chain())}")

To prove bundle integrity to a third-party auditor, sign the deployment directory with BundleSigner from edge_sdk/pyvorin_edge/packaging/signer.py. 


from pyvorin_edge.packaging.signer import BundleSigner
from pathlib import Path

# Generate an Ed25519 keypair (do this once and protect the private key)
private_key, public_key = BundleSigner.generate_keypair()

# Sign the entire deployment bundle
bundle_dir = Path("./cold_chain_bundle")
BundleSigner.sign_bundle(
    bundle_dir=bundle_dir,
    private_key=private_key,
    output_manifest=bundle_dir / "manifest.json",
)

# Verify on the auditor's side
is_valid = BundleSigner.verify_bundle(
    bundle_dir=bundle_dir,
    manifest_path=bundle_dir / "manifest.json",
    public_key=public_key,
)
print(f"Bundle integrity: {is_valid}")

Offline Operation Mode

Refrigerated trucks lose cellular coverage in rural areas and underground loading bays. The CloudSyncQueue in edge_runtime/pyv_edge_agent/cloud_sync/queue.py persists every item to SQLite and retries with exponential backoff when the link returns. 


from pyv_edge_agent.cloud_sync.queue import CloudSyncQueue, Priority
import json

queue = CloudSyncQueue(db_path="sync_queue.db")

# Enqueue a critical excursion payload
payload = {
    "pipeline": "cold_chain_pharma",
    "timestamp": time.time(),
    "event": "TEMPERATURE_EXCURSION",
    "value": 10.5,
    "unit": "C",
}
item_id = queue.enqueue(payload, priority=Priority.CRITICAL, ttl_seconds=172800)
print(f"Enqueued item {item_id}, pending: {queue.pending_count()}")

# On reconnect, flush the queue
# uploader = HTTPCloudUploader(endpoint="...", api_key="...")
# flushed = queue.maybe_flush(uploader=uploader)
# print(f"Flushed {flushed} items")

Complete Working Script

Save the following as cold_chain.py. It creates the pipeline, loads privacy policies, logs audit events, signs the bundle, and demonstrates offline queue behaviour. 


#!/usr/bin/env python3
"""Cold Chain Monitoring — complete working example."""

import time
from pathlib import Path
from pyvorin_edge.pipeline import Pipeline, WindowConfig, RuleConfig
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading
from pyvorin_edge.policies import PrivacyPolicy, CloudPolicy
from pyv_edge_agent.privacy_firewall.audit import PrivacyAudit
from pyvorin_edge.packaging.signer import BundleSigner
from pyv_edge_agent.cloud_sync.queue import CloudSyncQueue, Priority


def main():
    pipeline = Pipeline(name="cold_chain_pharma")

    # Sensors
    pipeline.add_sensor(
        Sensor(name="fridge_temp", sensor_type=SensorType.TEMPERATURE, unit="C",
               normal_range=(2.0, 8.0), location="Refrigerator_Compartment")
    )
    pipeline.add_sensor(
        Sensor(name="ambient_temp", sensor_type=SensorType.TEMPERATURE, unit="C",
               normal_range=(15.0, 25.0), location="Vehicle_Cabin")
    )

    # Windows
    pipeline.add_window(
        WindowConfig(duration_seconds=60.0, sensor_name="fridge_temp", window_type="rolling")
    )

    # Rules
    pipeline.add_rule(
        RuleConfig(name="temp_excursion_low", condition_expr="ctx.value < 2.0",
                   severity="critical", cooldown_seconds=0.0)
    )
    pipeline.add_rule(
        RuleConfig(name="temp_excursion_high", condition_expr="ctx.value > 8.0",
                   severity="critical", cooldown_seconds=0.0)
    )

    # Privacy policy
    privacy = PrivacyPolicy(
        redact_fields=["patient_id", "shipment_id", "trial_code"],
        retain_hours=72.0,
        pseudonymize_ids=True,
        local_only_sensors=["gps_location"],
    )
    pipeline.add_policy(privacy)

    # Cloud policy
    pipeline.add_policy(
        CloudPolicy(
            endpoint_url="https://api.pyvorin.com/v1/ingest",
            protocol="mqtt",
            max_messages_per_minute=10,
            summary_interval_seconds=60.0,
        )
    )

    # Audit
    audit = PrivacyAudit(db_path="edge_store.db")
    audit.log_action(
        action="PRIVACY_POLICY_LOADED",
        details={
            "redact_fields": privacy.redact_fields,
            "purpose": "GDP compliance",
            "data_subjects": ["patients"],
        },
    )

    # Simulate readings and run
    now = time.time()
    readings = [
        SensorReading(sensor_name="fridge_temp", timestamp=now, value=1.5, unit="C",
                      metadata={"batch": "B12345", "patient_id": "REDACTED"}),
        SensorReading(sensor_name="fridge_temp", timestamp=now + 5.0, value=10.5, unit="C",
                      metadata={"batch": "B12345", "patient_id": "REDACTED"}),
    ]
    result = pipeline.run(readings)
    print(f"Processed {result.readings_processed} readings, {len(result.events)} events")
    for ev in result.events:
        print(f"  [{ev.severity}] {ev.rule_name}")
        audit.log_action(
            action="TEMPERATURE_EXCURSION",
            details={"sensor": ev.rule_name, "value": ev.sensor_readings[0].value},
        )

    assert audit.verify_chain(), "Audit chain tampered!"

    # Bundle signing
    bundle_dir = Path("./cold_chain_bundle")
    bundle_dir.mkdir(exist_ok=True)
    private_key, public_key = BundleSigner.generate_keypair()
    BundleSigner.sign_bundle(
        bundle_dir=bundle_dir,
        private_key=private_key,
        output_manifest=bundle_dir / "manifest.json",
    )
    print("Bundle signed.")

    # Offline queue
    queue = CloudSyncQueue(db_path="sync_queue.db")
    payload = {
        "pipeline": "cold_chain_pharma",
        "timestamp": time.time(),
        "event": "TEMPERATURE_EXCURSION",
        "value": 10.5,
    }
    queue.enqueue(payload, priority=Priority.CRITICAL, ttl_seconds=172800)
    print(f"Queue depth: {queue.pending_count()}")


if __name__ == "__main__":
    main()

Summary

You now have a cold-chain pipeline that enforces GDP temperature limits, redacts sensitive identifiers, maintains a tamper-evident audit chain, signs deployment bundles with Ed25519, and buffers critical events during network outages. In production, pair this with a calibrated RTD or thermocouple probe and validate against your regulatory auditor's checklist.