OpenDAQ to Quix Integration Template

This repository contains scripts that demonstrate integration between openDAQ (a data acquisition library) and Quix (a platform for engineering
data management).

Overview

The publish_simdevice_data_to_quix.py script is designed to work with the OpenDAQ VirtualBox Device simulator and the standalone simulator script (Python).

It the connects to the simulated device, reads sensor data from multiple signals, and streams that data the Quix platform. This script enables you to get data straight out of the DAQ and stream it into the cloud, instead of exporting and and uploading large data files.

You can also use Quix Python functions to continuously analyze data as soon as it arrives and perform tasks such as normalization, filtering and enrichment in real-time

Prerequisites

• Python 3.8+
• openDAQ device simulator running and accessible on the network
• python-dotenv package for loading environment variables
• Quix Streams SDK credentials (SDK token and workspace configuration)

Install dependencies:

pip install python-dotenv opendaq quixstreams

Configuration

Environment variables are loaded from a .env file in the same directory. The file should contain:

QUIX_SDK_TOKEN=your-sdk-token-here
Quix__Workspace__Id=your-workspace-id
Quix__Portal__Api=https://portal-api.cloud.quix.io

Note: Do not commit the .env file to version control as it contains sensitive credentials. The file is listed in .gitignore.

Script Workflow

publish_simdevice_data_to_quix.py

This script follows a four-stage pipeline:

1. Connect to openDAQ Device

• Creates an openDAQ instance
• Searches for the "Reference device simulator" in available devices
• Establishes a connection to the device
• Fails gracefully if the device is not found or not accessible

2. Prepare for Streaming

• Retrieves all signals from the connected device using a recursive search filter
• Creates am openDAQ StreamReader for each signal that has domain information (timestamp data)
• Extracts and caches timing metadata:
  • Domain origin (timestamp offset)
  • Tick resolution (numerator and denominator for converting ticks to time)

3. Setup Quix Streams Producer

• Initializes a Quix Streams application with the SDK token
• Creates or connects to the "opendaq-telemetry" topic configured for JSON serialization
• Prepares the producer to send messages to Kafka

4. Start the Streaming Loop

• Continuously reads data from all signal readers (batches of up to 100 samples)
• For each sample, constructs a payload containing:
  • timestamp_ticks: The raw timestamp value (converted from NumPy int64)
  • value: The sensor measurement (converted from NumPy float64)
  • signal_id: Unique identifier of the signal source
  • origin: Domain origin timestamp offset
  • tick_resolution_num/den: Tick resolution as a fraction (numerator/denominator)
• Serializes the payload to Kafka message format
• Produces the message to the Quix topic
• Sleeps briefly when no data is available to avoid busy-waiting

Graceful Shutdown

The script handles termination signals (SIGINT and SIGTERM) to cleanly shut down:

• Sets the running flag to False
• Exits the streaming loop
• Closes the producer connection
• Provides user feedback via console output

Data Format

Each message published to Quix contains a JSON payload structured as:

{
  "timestamp_ticks": 12345678,
  "value": 42.5,
  "signal_id": "Device/Channel0/Signal1",
  "origin": 0,
  "tick_resolution_num": 1,
  "tick_resolution_den": 1000000
}

Key Considerations

• NumPy Type Conversion: Values are explicitly converted from NumPy types (int64, float64) to Python native types to ensure proper JSON serialization
• Batch Reading: The script reads up to 100 samples at a time to optimize throughput
• Domain Information: Timestamps are preserved with their resolution metadata, allowing accurate time reconstruction downstream

Usage

python publish_simdevice_data_to_quix.py

The script will:

1. Search for and connect to the openDAQ device simulator
2. Begin streaming all available signals
3. Display progress on the console
4. Continue until interrupted with Ctrl+C

Troubleshooting

Issue	Solution
"Device simulator not found"	Ensure the openDAQ device simulator is running and accessible
"No signals found on the device"	Check that the device has signals with domain information configured
"QUIX_SDK_TOKEN environment variable must be set"	Ensure the .env file exists and contains a valid QUIX_SDK_TOKEN
ImportError for dotenv	Install python-dotenv: pip install python-dotenv

Related Scripts

• connect_to_simdevice.py: Example of connecting to an openDAQ device simulator
• read_data_from_simdevice.py: Example of reading data from a device simulator without streaming to Quix

Resources

• openDAQ Documentation
• Quix Streams Documentation