JSON signals from a Python generator

What this is about

The example shows the JSON Signal Stream protocol: the simplest way to get data from an Arduino, an ESP32, a Raspberry Pi or your own program into QInsight. The device sends one JSON object per line, e.g. {"name": "sine1", "value": 3.2, "t": 12.05}, and QInsight fills its variables from it.

Because you may not have such a device at hand, QInsight ships with the Python script Qenex_JsonSignal_SineGenerator.py that behaves like one. The generator runs on your PC and:

sends five sine signals sine1 to sine5, each with a different amplitude, offset, frequency and phase, 20 times per second, every sample as one JSON line with a timestamp t,
accepts commands in the same format with which QInsight changes the amplitude, frequency, offset or phase of any signal while it runs, e.g. {"name": "sine1_A", "value": 2} sets the amplitude of sine1 to 2.

The link between QInsight and the generator is either TCP (the generator is a TCP server, QInsight connects to it with the TCP Client driver) or a serial line (the generator opens one end of a COM pair, QInsight the other with the Serial Port driver). The data flow is the same in both cases:

Python generator  ── {"name": "sine1", "value": 3.2, "t": 12.05}  ──►  QInsight (graph, watch table, log)
                  ◄── {"name": "sine1_A", "value": 2}             ──   write from the watch table

There is one project per transport in the Examples folder next to QInsight.exe; the script is in the same folder:

Over TCP — the project Qenex_TcpClient_JsonSignal_PythonSines.qproj, TCP Client driver with ip=127.0.0.1;port=5000; start the generator with
```
python Qenex_JsonSignal_SineGenerator.py
```
Over a serial line — the project Qenex_Serial_JsonSignal_PythonSines.qproj, Serial Port driver with port=COM6;baudRate=115200; the generator runs on the other end of the COM pair:
```
python Qenex_JsonSignal_SineGenerator.py --serial COM3
```

The script is also a template for how your own device should send and receive the lines; it is listed in full at the end of the page.

What the example shows

Reading five sines with a device timestamp: the graph shows the real spacing of the generator's samples, not the moments the PC received them.
Writing to the device: the variables Sine1Amplitude, Sine1Frequency, Sine2Amplitude and Sine2Frequency have direction write. A write from the watch table sends the generator the line {"name": "sine1_A", "value": 2} and the sine changes at once.
The same protocol on two transports: TCP and a serial line; the projects differ only in the driver.

What you need

Python 3 on this or another PC. For the serial mode also pyserial: python -m pip install pyserial.
TCP: nothing else; the generator listens on port 5000 and QInsight connects as a client. On another PC change ip in the driver settings and allow the port in the firewall.
Serial line: a pair of virtual COM ports (e.g. com0com) or two USB/serial converters wired crosswise. The generator runs on one end, QInsight on the other; adjust port in the driver settings to your pair, the baud rate must be the same on both sides.

The project

Driver and protocol

The protocol has no settings. The driver settings are listed above; the TCP driver also has reconnectAttempts=-1 and keepAliveMs=5000, so it reconnects by itself after the generator is stopped and started again.

Variables

Variable	Direction	`name`	Meaning
`sine1` … `sine5`	read	`sine1` … `sine5`	generator signals, O + A·sin(2πft + φ)
`Params/Sine1Amplitude`	write	`sine1_A`	amplitude of sine1
`Params/Sine1Frequency`	write	`sine1_f`	frequency of sine1 in Hz
`Params/Sine2Amplitude`	write	`sine2_A`	amplitude of sine2
`Params/Sine2Frequency`	write	`sine2_f`	frequency of sine2 in Hz

The generator understands names <signal>_<parameter> with the parameters A, O, f and phi, so further writable variables only need a new name.

Workspace

A graph of all five sines, a watch table with the parameters in write mode and five signal controls.

What you will see and what to try

After Start the five sines run within a second. The log shows at most one message about a dropped line: when a serial port opens in the middle of a running stream, the tail of a cut message arrives first; the protocol drops it silently, further unparsable lines would be reported.
In the watch table write Sine1Amplitude = 2 and Sine1Frequency = 0.2: the sine1 curve flattens and slows down, the generator prints sine1.A = 2.0 in its console.
Write an unknown name, e.g. a variable with name="sine9_A": the generator ignores it and nothing happens in QInsight. The other way round, send the generator a line by hand with a name the project does not have and the protocol reports it once in the log.
Stop the generator (Ctrl+C) and start it again: the TCP driver reconnects by itself, the serial port stayed open and the data keep flowing.

The generator script

The whole Qenex_JsonSignal_SineGenerator.py; the line format is in the docstring. For your own device take two things from it: sending {"name": ..., "value": ..., "t": ...}\n and reading lines of the same shape.

#!/usr/bin/env python3
"""Sine signal generator - the counterpart of QInsight's JSON Signal Stream protocol.

Shipped with QInsight in the Examples folder; used by the examples
Qenex_TcpClient_JsonSignal_PythonSines.qproj and Qenex_Serial_JsonSignal_PythonSines.qproj.

Runs as a TCP server (default, QInsight connects with the TCP Client driver) or on a
serial port (QInsight connects with the Serial Port driver):

    python Qenex_JsonSignal_SineGenerator.py                  # TCP server on 0.0.0.0:5000
    python Qenex_JsonSignal_SineGenerator.py --port 6000      # TCP server on another port
    python Qenex_JsonSignal_SineGenerator.py --serial COM7    # serial port, 115200 baud
    python Qenex_JsonSignal_SineGenerator.py --serial /dev/ttyUSB0 --baud 57600

The generator continuously sends 5 messages (one signal each), where each signal value
is O + A * sin(2*pi*f*t + phi). It also listens for lines that change the offset (O),
amplitude (A), frequency (f) or phase (phi) of any signal at runtime.

Wire protocol (newline-delimited JSON, the same shape in both directions):
  Generator -> QInsight: {"name": "sine1", "value": 0.7071, "t": 0.05}\\n   (5x per cycle)
  QInsight -> generator: {"name": "sine1_A", "value": 2.0}\\n
                         The name is "<signal>_<parameter>" with parameter A (amplitude),
                         O (offset), f (frequency [Hz]) or phi (phase [rad]); one line
                         changes one parameter of one signal. This is exactly what the
                         JSON Signal Stream protocol sends for a variable with
                         direction="write" and name="sine1_A".

Serial mode needs pyserial (pip install pyserial).
"""
import argparse
import json
import math
import socket
import threading
import time

SEND_HZ = 20            # how often a full set of samples is sent, per second

# Per-signal initial parameters.
#   A   = amplitude
#   O   = offset
#   f   = frequency [Hz]
#   phi = phase offset [rad]
# The send order follows the order of keys below (dict preserves insertion order).
INITIAL_PARAMS = {
    "sine1": {"A": 10.0, "O": 2.0,  "f": 1.0,  "phi": 0.0},
    "sine2": {"A": 20.0, "O": -5.0, "f": 0.5,  "phi": math.pi / 4},
    "sine3": {"A": 5.0,  "O": 8.0,  "f": 10.0, "phi": math.pi / 2},
    "sine4": {"A": 30.0, "O": 0.5,  "f": 0.25, "phi": math.pi},
    "sine5": {"A": 15.0, "O": 2.0,  "f": 2.0,  "phi": 3 * math.pi / 2},
}
SIGNAL_NAMES = list(INITIAL_PARAMS.keys())
PARAM_NAMES = ("A", "O", "f", "phi")


class SignalStore:
    """Thread-safe store of per-signal parameters."""

    def __init__(self, initial):
        self._lock = threading.Lock()
        # deep-ish copy so INITIAL_PARAMS is not mutated by runtime commands
        self._sig = {n: dict(p) for n, p in initial.items()}

    def value(self, name, t):
        with self._lock:
            p = self._sig[name]
            A, O, f, phi = p["A"], p["O"], p["f"], p["phi"]
        return O + A * math.sin(2.0 * math.pi * f * t + phi)

    def set_param(self, name, param, value):
        with self._lock:
            if name not in self._sig or param not in PARAM_NAMES:
                return False
            self._sig[name][param] = float(value)
            return True


def apply_line(store, raw):
    """One incoming line: {"name": "sine1_A", "value": 2.0} -> sine1.A = 2.0."""
    line = raw.strip()
    if not line:
        return
    try:
        msg = json.loads(line.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError):
        return
    name, value = msg.get("name"), msg.get("value")
    if not isinstance(name, str) or "_" not in name or value is None:
        return
    signal, param = name.rsplit("_", 1)
    if store.set_param(signal, param, value):
        print(f"{signal}.{param} = {value}", flush=True)


def reader_thread(readable, store, stop):
    """Listen for lines from QInsight and apply parameter changes."""
    try:
        for raw in readable:                 # read line by line (socket file / serial port)
            if stop.is_set():
                break
            apply_line(store, raw)
    except (OSError, ValueError):
        pass
    finally:
        stop.set()


def send_loop(write, store, stop):
    """Send all signals SEND_HZ times per second until stop is set or the link breaks."""
    t0 = time.monotonic()
    period = 1.0 / SEND_HZ
    next_t = time.monotonic()
    try:
        while not stop.is_set():
            now = time.monotonic() - t0
            for name in SIGNAL_NAMES:                  # 5 messages, one signal each
                msg = {"name": name, "value": store.value(name, now), "t": round(now, 4)}
                write((json.dumps(msg) + "\n").encode("utf-8"))
            next_t += period
            time.sleep(max(0.0, next_t - time.monotonic()))
    except OSError:
        pass
    except KeyboardInterrupt:
        stop.set()
        raise
    finally:
        stop.set()


def serve_tcp(host, port, store):
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((host, port))
    srv.listen(1)
    # Windows delivers Ctrl+C to the main thread only when it is not stuck in a blocking
    # socket call, so accept() polls with a timeout instead of blocking forever.
    srv.settimeout(0.5)
    print(f"Generator listening on {host}:{port} - connect QInsight with the TCP Client driver (Ctrl+C to stop)", flush=True)
    try:
        while True:
            try:
                conn, addr = srv.accept()
            except socket.timeout:
                continue
            conn.settimeout(None)
            print(f"QInsight connected: {addr}", flush=True)
            stop = threading.Event()
            threading.Thread(target=reader_thread, args=(conn.makefile("rb"), store, stop), daemon=True).start()
            send_loop(conn.sendall, store, stop)
            conn.close()
            print(f"QInsight disconnected: {addr}")
    except KeyboardInterrupt:
        pass
    finally:
        srv.close()


def serve_serial(port, baud, store):
    import serial  # pyserial
    ser = serial.Serial(port, baud, timeout=1)
    print(f"Generator on {port} @ {baud} baud - connect QInsight with the Serial Port driver")
    stop = threading.Event()

    def lines():
        while not stop.is_set():
            raw = ser.readline()
            if raw:
                yield raw

    threading.Thread(target=reader_thread, args=(lines(), store, stop), daemon=True).start()
    try:
        send_loop(ser.write, store, stop)
    except KeyboardInterrupt:
        pass
    finally:
        stop.set()
        ser.close()


def main():
    parser = argparse.ArgumentParser(description="JSON Signal Stream sine generator (TCP server or serial port).")
    parser.add_argument("--host", default="0.0.0.0", help="TCP bind address (default 0.0.0.0)")
    parser.add_argument("--port", type=int, default=5000, help="TCP port (default 5000)")
    parser.add_argument("--serial", metavar="PORT", help="use a serial port instead of TCP, e.g. COM7 or /dev/ttyUSB0")
    parser.add_argument("--baud", type=int, default=115200, help="serial baud rate (default 115200)")
    args = parser.parse_args()

    store = SignalStore(INITIAL_PARAMS)
    if args.serial:
        serve_serial(args.serial, args.baud, store)
    else:
        serve_tcp(args.host, args.port, store)


if __name__ == "__main__":
    main()

JSON Signal Stream in the protocol reference
TCP Client and Serial Port in the driver reference
Writing values to the device