publish_simdevice_data_to_quix.py

import os
import signal
import time
from pathlib import Path
from typing import Dict

import opendaq
import quixstreams as qx
from dotenv import load_dotenv

# Load environment variables from .env file
env_file = Path(__file__).parent / ".env"
load_dotenv(env_file)

# --- Configuration ---
quix_sdk_token = os.environ.get("QUIX_SDK_TOKEN")
if not quix_sdk_token:
    raise RuntimeError("QUIX_SDK_TOKEN environment variable must be set.")

# --- Graceful Shutdown Handler ---
running = True


def shutdown_handler(signum, frame):
    global running
    print("Shutdown signal received, stopping...")
    running = False


signal.signal(signal.SIGINT, shutdown_handler)
signal.signal(signal.SIGTERM, shutdown_handler)


def main():
    # 1. CONNECT TO OPENDAQ DEVICE
    # =============================
    print("Searching for openDAQ device simulator...")
    daq_instance = opendaq.Instance()

    device = None
    for device_info in daq_instance.available_devices:
        if device_info.name == "Reference device simulator":
            device = daq_instance.add_device(device_info.connection_string)
            print(f"Connected to '{device.info.name}'")
            break

    if not device:
        print("Device simulator not found. Please ensure it is running and accessible.")
        return

    # 2. PREPARE FOR STREAMING
    # ========================
    search_filter = opendaq.RecursiveSearchFilter(opendaq.AnySearchFilter())
    signals = device.get_signals(search_filter)

    if not signals:
        print("No signals found on the device.")
        return

    readers: Dict[str, opendaq.StreamReader] = {}
    domain_info: Dict[str, dict] = {}

    print("\nFound signals to stream:")
    for sig in signals:
        if sig.domain_signal:
            global_id = sig.global_id
            print(f"- {global_id}")
            readers[global_id] = opendaq.StreamReader(sig)

            descriptor = sig.domain_signal.descriptor
            resolution = descriptor.tick_resolution
            domain_info[global_id] = {
                "origin": descriptor.origin,
                "resolution_num": resolution.numerator,
                "resolution_den": resolution.denominator,
            }

    # 3. SETUP QUIX STREAMS PRODUCER
    # ==============================
    app = qx.Application(quix_sdk_token=quix_sdk_token)
    output_topic = app.topic("opendaq-telemetry", value_serializer="json")

    with app.get_producer() as producer:
        print(
            f"\nStreaming data to Quix topic '{output_topic.name}'. Press Ctrl+C to stop."
        )

        # 4. START THE STREAMING LOOP
        # ===========================
        while running:
            has_data = False
            for signal_id, reader in readers.items():
                values, domain_values = reader.read_with_domain(100)
                count = len(values)

                if count > 0:
                    has_data = True
                    info = domain_info[signal_id]
                    print(f"Read {count} samples from {signal_id}")

                    for i in range(count):
                        # --- THIS SECTION IS CORRECTED ---
                        # Convert NumPy types to standard Python types before serialization.
                        payload = {
                            "timestamp_ticks": int(
                                domain_values[i]
                            ),  # Cast numpy.int64 to Python int
                            "value": float(
                                values[i]
                            ),  # Cast numpy.float64 to Python float
                            "signal_id": signal_id,
                            "origin": info["origin"],
                            "tick_resolution_num": info["resolution_num"],
                            "tick_resolution_den": info["resolution_den"],
                        }
                        # --- END OF CORRECTION ---

                        kafka_message = output_topic.serialize(
                            key=signal_id, value=payload
                        )

                        producer.produce(
                            topic=output_topic.name,
                            key=kafka_message.key,
                            value=kafka_message.value,
                        )

            if not has_data:
                time.sleep(0.05)

    print("\nStreaming stopped. Producer closed.")


if __name__ == "__main__":
    main()