ONNX Pipeline Runner for thin-edge.io

A generic, reusable ML inference engine for edge devices managed by Cumulocity via thin-edge.io.

The core idea: install the runner once, then swap use cases by pushing configuration files — no redeployment needed.

The runner executes a standard three-step pipeline on every cycle:

  Preprocess  ──>  ONNX Inference  ──>  Postprocess

It doesn't care whether you're doing thermal imaging, anomaly detection, or predictive maintenance. The use case is defined entirely by the configuration files you push to the device:

File	Purpose
pipeline.json	Pipeline settings, device info, thresholds
preprocessor.py	Reads raw data, outputs model-ready tensor
model.onnx	Your trained ONNX model
postprocessor.py	Interprets model output, creates alerts/events

All four files are managed via Cumulocity's Configuration Repository and pushed directly to the device — no SSH required for day-to-day operations.

---

Architecture

More Details -> TechCommunity Article

Cumulocity
├── Software Repository
│   └── tedge-pipeline-runner (apt)            <-- install once
│
├── Configuration Management (push to device)
│   ├── pipeline-config         pipeline.json
│   ├── pipeline-preprocessor   preprocessor.py
│   ├── pipeline-postprocessor  postprocessor.py
│   └── pipeline-model          model.onnx
│
└── Device Dashboard
    ├── Measurements   inference_time, cycle_time, model_score, is_alert [TODO]
    ├── Events         alerts, lifecycle events
    └── Services       tedge-pipeline-runner (up / down)

Edge Device — Raspberry Pi (/opt/tedge-pipeline/)
├── pipeline_runner.py            <-- from .deb (generic, never changes)
├── config/
│   └── pipeline.json             <-- pushed via Cumulocity
├── processors/
│   ├── preprocessor.py           <-- pushed via Cumulocity
│   └── postprocessor.py          <-- pushed via Cumulocity
├── models/
│   └── model.onnx                <-- pushed via Cumulocity
└── data/
    └── frames/                   <-- from simulator or real sensor

---

Prerequisites

On your laptop (build machine):

• Python 3.8+ with numpy (for building ONNX models)
• dpkg-deb for building the Debian package (macOS: brew install dpkg)

On the edge device (Raspberry Pi or similar):

• thin-edge.io installed and connected to Cumulocity
• Python 3 (comes with Raspberry Pi OS)

---

Quick Start

1. Clone the repo

git clone https://github.com/Cumulocity-IoT/onnx-pipeline-runner.git
cd onnx-pipeline-runner

2. Build the .deb package

bash build_runner_deb.sh

This produces tedge-pipeline-runner_1.0.0_all.deb. The .deb installs the generic runner and registers the four configuration types with thin-edge.io's configuration plugin.

3. Build the ONNX model (thermal demo)

cd demos/thermal-demo
python3 build_thermal_model.py --output model.onnx

4. Upload the .deb to Cumulocity

1. Go to Management > Software Repository > Add software
2. Name: tedge-pipeline-runner, Version: 1.0.0, Type: apt
3. Upload the .deb file
4. Click Add software

5. Install on your device

1. Go to Device Management > (your device) > Software tab
2. Search tedge-pipeline-runner > Install > Apply changes
3. Wait 1-2 minutes (installs Python dependencies automatically)

6. Generate demo data on the device

SSH into the device once to run the simulator:

# Copy the simulator to the device
scp demos/thermal-demo/simulator/thermal_simulator.py pi@<DEVICE_IP>:~/

# SSH in and run it
ssh pi@<DEVICE_IP>
python3 thermal_simulator.py --output /opt/tedge-pipeline/data/frames

This creates synthetic .npy frames. In production, your real sensor writes to the same directory — the preprocessor reads from it either way.

7. Push configuration files via Cumulocity

Go to your device > Configuration tab. Upload each file to its matching configuration type:

Configuration Type	File to Upload
pipeline-config	demos/thermal-demo/config/pipeline.json
pipeline-preprocessor	demos/thermal-demo/processors/preprocessor.py
pipeline-postprocessor	demos/thermal-demo/processors/postprocessor.py
pipeline-model	demos/thermal-demo/model.onnx

For each one: select the type, upload the file, click Send configuration to device.

8. Start the service

ssh pi@<DEVICE_IP>
sudo systemctl start tedge-pipeline-runner
sudo journalctl -u tedge-pipeline-runner -f

You should see output like:

TEDGE PIPELINE RUNNER
  Pipeline:      thermal-monitoring
  Model:         model.onnx (540 bytes)
Cycle 1 | NORMAL | pre=2ms inf=1ms post=0ms
Cycle 2 | NORMAL | ...
...
Cycle 7 | ALERT  | pre=1ms inf=0ms post=0ms    <-- event sent!
Cycle 8 | ALERT  | ...
Cycle 9 | NORMAL | ...

9. Check Cumulocity

• Events tab: c8y_ThermalAlert events with annotated image and bounding box data
• Services tab: tedge-pipeline-runner status shown as up
• Services tab > tedge-pipeline-runner > Measurements: inference_time, cycle_time, model_score, score_threshold, is_alert [TODO]

---

Included Demos

Thermal Monitoring (demos/thermal-demo/)

Detects temperature hotspots in thermal camera frames. The model takes a 120x160 thermal image, finds the maximum temperature, and produces a heat grid. When a grid cell exceeds the threshold, the postprocessor sends a c8y_ThermalAlert event with an annotated image and bounding box coordinates.

Model input: [1, 1, 120, 160] (thermal image in Celsius) Model output: max_temp [1,1], heat_grid [1, 1, 6, 8]

Anomaly Detection (demos/anomaly-demo/)

Detects anomalous spikes in time-series sensor data. The model takes a sliding window of 30 temperature readings, computes an anomaly score (max minus mean deviation), and flags readings above a configurable threshold.

Model input: [1, 1, 30] (window of sensor readings) Model output: anomaly_score [1,1], smoothed [1, 1, 6]

To switch from thermal to anomaly on a running device:

# 1. Generate anomaly demo data
scp demos/anomaly-demo/simulator/anomaly_simulator.py pi@<DEVICE_IP>:~/
ssh pi@<DEVICE_IP>
rm -f /opt/tedge-pipeline/data/frames/*.npy
python3 anomaly_simulator.py --output /opt/tedge-pipeline/data/frames

# 2. Stop the pipeline
sudo systemctl stop tedge-pipeline-runner

Then push the four anomaly files via the Cumulocity Configuration tab (same types — they overwrite the thermal versions):

Configuration Type	File to Upload
pipeline-config	demos/anomaly-demo/config/pipeline.json
pipeline-preprocessor	demos/anomaly-demo/processors/preprocessor.py
pipeline-postprocessor	demos/anomaly-demo/processors/postprocessor.py
pipeline-model	demos/anomaly-demo/model.onnx

Start the pipeline again:

sudo systemctl start tedge-pipeline-runner

Same runner, completely different use case.

---

Bring Your Own Use Case

The runner is designed to be extended. To add your own pipeline:

1. Create your preprocessor (preprocessor.py)

Must implement one function:

def get_input(config, cycle_count):
    """
    Read raw data and prepare it for the ONNX model.

    Args:
        config:      dict - full pipeline.json contents (use config["settings"] for your custom fields)
        cycle_count: int  - current cycle number (1-based), useful for rotating through data files

    Returns:
        dict with two keys:
            "input":    np.ndarray (float32) - model-ready tensor, shape must match your ONNX model's input
            "metadata": dict - arbitrary data passed through to the postprocessor (timestamps, raw values, file paths, etc.)
    """

2. Train your ONNX model (model.onnx)

Any ONNX model works. Export from PyTorch, TensorFlow, scikit-learn, or build from scratch. The runner calls onnxruntime.InferenceSession and passes your preprocessor's output tensor to it.

3. Create your postprocessor (postprocessor.py)

Must implement one function:

def handle_output(config, model_outputs, metadata, mqtt_client):
    """
    Interpret model results and take action (send alerts, publish metrics).

    Args:
        config:        dict           - full pipeline.json contents
        model_outputs: list[ndarray]  - raw ONNX output arrays
        metadata:      dict           - whatever your preprocessor returned in "metadata"
        mqtt_client:   MQTTPublisher  - use mqtt_client.publish_event() and mqtt_client.publish_measurement()

    Returns:
        dict with two keys:
            "status":  str  - e.g. "normal", "alert", "anomaly" (logged each cycle)
            "metrics": dict - published to Cumulocity under the service's Measurements tab.
                              Use these three standardized names so all pipelines are consistent:
                              {
                                  "model_score":     {"value": 42.0, "unit": ""},
                                  "score_threshold": {"value": 30.0, "unit": ""},
                                  "is_alert":        {"value": 1,    "unit": ""},
                              }
    """

MQTT client methods available in your postprocessor:

• mqtt_client.publish_event(event_type, text, extras_dict) — creates a Cumulocity event
• mqtt_client.publish_measurement(measurement_type, values_dict) — creates a Cumulocity measurement

Standardized metrics: Both included demos return the same three metric names (model_score, score_threshold, is_alert). The runner also adds inference_time and cycle_time automatically. This consistency means dashboards and smart rules work across use cases without reconfiguration.

4. Configure your pipeline (pipeline.json)

{
  "pipeline_name": "my-use-case",
  "capture_interval_sec": 10,

  "preprocessor_path": "/opt/tedge-pipeline/processors/preprocessor.py",
  "postprocessor_path": "/opt/tedge-pipeline/processors/postprocessor.py",
  "model_path": "/opt/tedge-pipeline/models/model.onnx",
  "data_dir": "/opt/tedge-pipeline/data",

  "mqtt_host": "localhost",
  "mqtt_port": 1883,
  "service_name": "tedge-pipeline-runner",

  "settings": {
    "device_id": "<YOUR-DEVICE-ID>",
    "device_name": "<YOUR-DEVICE-NAME>",
    "my_threshold": 50.0,
    "c8y_event_type": "c8y_MyCustomEvent",
    "c8y_event_text": "Something interesting happened"
  }
}

The settings block is freeform — put whatever your preprocessor and postprocessor need in there. The runner passes the entire config dict to both functions.

5. Deploy

Push all four files via Cumulocity Configuration Management. The runner picks them up automatically. No rebuild, no redeployment.

---

Day-to-Day Operations (no SSH needed)

Task	How
Change threshold	Edit pipeline.json > push via Configuration tab
Update model	Retrain > push new model.onnx via Configuration tab
Change preprocessing	Edit preprocessor.py > push via Configuration tab
Change alert logic	Edit postprocessor.py > push via Configuration tab
Switch use case entirely	Push a different set of all four files

The runner detects file changes automatically via MD5 hashing and reloads without restart.

---

Repository Structure

onnx-pipeline-runner/
├── README.md                                   <-- you are here
├── .gitignore
├── build_runner_deb.sh                         <-- builds the .deb package
│
├── runner/                                     <-- generic engine (packaged in the .deb)
│   ├── pipeline_runner.py                          main orchestrator
│   ├── tedge-pipeline-runner.service               systemd unit file
│   └── tedge-configuration-plugin.toml             config plugin entries (reference)
│
└── demos/                                      <-- example use cases (add your own here)
    ├── thermal-demo/                               thermal monitoring
    │   ├── build_thermal_model.py                      builds model.onnx on your laptop
    │   ├── config/
    │   │   └── pipeline.json                           pipeline settings
    │   ├── processors/
    │   │   ├── preprocessor.py                         reads thermal .npy frames
    │   │   └── postprocessor.py                        detects hotspots, sends alerts
    │   └── simulator/
    │       └── thermal_simulator.py                    generates synthetic thermal frames
    │
    └── anomaly-demo/                               anomaly detection
        ├── build_anomaly_model.py                      builds model.onnx on your laptop
        ├── config/
        │   └── pipeline.json                           pipeline settings
        ├── processors/
        │   ├── preprocessor.py                         reads sensor window .npy files
        │   └── postprocessor.py                        detects anomalies, sends alerts
        └── simulator/
            └── anomaly_simulator.py                    generates synthetic sensor data

---

How It Works Under the Hood

1. Install: The .deb places pipeline_runner.py at /opt/tedge-pipeline/ and registers a systemd service. The post-install script appends four configuration types to thin-edge.io's configuration plugin, so they appear in the Cumulocity Configuration tab.

2. Configure: Push pipeline.json, preprocessor.py, postprocessor.py, and model.onnx via Cumulocity. They land at their expected paths on the device.

3. Run: The service starts the runner, which loads the config, dynamically imports the preprocessor and postprocessor modules, loads the ONNX model, and enters its loop.

4. Each cycle: preprocessor.get_input() reads raw data and returns a tensor. The runner feeds it to the ONNX model. postprocessor.handle_output() interprets the result and publishes events/measurements via thin-edge MQTT.

5. Hot reload: Every cycle, the runner checks MD5 hashes of the config files. If anything changed (because you pushed an update via Cumulocity), it reloads automatically.

6. Health: The runner publishes its service status (up/down) and lifecycle events (started, stopped, error) to Cumulocity.

---

License

Licensed under the Apache License, Version 2.0. See LICENSE for details.

These tools are provided as-is and without warranty or support. They do not constitute part of the Cumulocity product suite. Users are free to use, fork and modify them, subject to the license agreement. While Cumulocity GmbH welcomes contributions, we cannot guarantee to include every contribution in the master project.