Article 8

Data-Driven Installations in TouchDesigner

Connect TouchDesigner to REST APIs, WebSockets, MQTT IoT sensors, and databases to create permanent installations driven by real-world data streams.

⏱ 50 min data api websocket mqtt python installation

Permanent data-driven installations translate real-world information — air quality, stock prices, seismic activity, crowd density — into generative visuals that evolve continuously. The technical challenge is less about the visuals and more about building a data pipeline that stays reliable over weeks and months of unattended operation.

REST API Polling with Web Client DAT

The Web Client DAT (Operators > DAT > Web Client) makes HTTP requests on demand or on a timer. For a REST API that returns JSON:

Timer CHOP (pulse every 60s) → Web Client DAT → onReceive callback → JSON DAT → Table DAT

Configure the Web Client DAT:

URL: the API endpoint, with query parameters if needed
Request Method: GET (most public APIs)
Custom Headers: add Authorization: Bearer YOUR_TOKEN in the Headers parameter if required
Timeout: set to 10 seconds — an unresponsive server will block TD’s cook thread if no timeout is set

The onReceive callback fires when the response arrives:

# In the Web Client DAT's Callbacks DAT
def onHTTPResponse(webClientDAT, statusCode, headerDict, data, id):
    if statusCode == 200:
        # Write the JSON response to a Text DAT for the JSON DAT to parse
        op('api_raw_json').text = data.decode('utf-8')
        op('api_json').cook(force=True)  # Force JSON DAT to re-parse
    else:
        print(f'[API] HTTP {statusCode} — using cached data')
        # Don't update — the JSON DAT keeps its last good parse

Parsing JSON with the JSON DAT

The JSON DAT parses a JSON string (from a Text DAT or directly from the Web Client DAT output) into a Table DAT. For nested JSON, use the JSON Path parameter to specify which nested object to flatten:

For an air quality API response like:

{
  "status": "ok",
  "data": {
    "aqi": 47,
    "iaqi": {
      "co2": {"v": 412},
      "no2": {"v": 23.5},
      "pm25": {"v": 8.2}
    },
    "time": {"iso": "2026-07-17T14:00:00Z"}
  }
}

Set JSON Path to data.iaqi and the JSON DAT produces a table with columns co2.v, no2.v, pm25.v. Extract individual values with a Select DAT or Python:

# In a Script CHOP or Execute DAT
def get_air_quality():
    json_dat = op('api_json')
    return {
        'co2':  float(json_dat['co2.v', 1]),    # row 1 = values row
        'no2':  float(json_dat['no2.v', 1]),
        'pm25': float(json_dat['pm25.v', 1]),
    }

WebSocket DAT for Push-Based Data

When the data source supports WebSocket (real-time push rather than request/response polling), the WebSocket DAT is significantly more efficient — no polling overhead, no latency from request cycles.

# WebSocket DAT Callbacks DAT

def onOpen(webSocketDAT):
    print('[WS] Connected')
    # Send subscription message if the server requires it
    webSocketDAT.sendText('{"action":"subscribe","channel":"air_quality_realtime"}')

def onReceiveText(webSocketDAT, data):
    import json
    try:
        payload = json.loads(data)
        if payload.get('type') == 'air_quality':
            store = op('realtime_store')  # a Storage DAT or global dict
            store['co2']  = payload['co2']
            store['no2']  = payload['no2']
            store['pm25'] = payload['pm25']
            store['ts']   = absTime.seconds  # timestamp for staleness detection
    except json.JSONDecodeError as e:
        print(f'[WS] Parse error: {e}')

def onClose(webSocketDAT, info):
    print(f'[WS] Disconnected: {info}')
    # Auto-reconnect via Timer CHOP watching connection state

Add a Timer CHOP that pulses every 30 seconds and checks whether the WebSocket is still connected:

def onTimerPulse(timerCHOP):
    ws = op('websocket_dat')
    if ws.par.Active.val == False:
        print('[WS] Reconnecting...')
        ws.par.Active = False
        ws.par.Active = True

Database Connections via Python

For installations pulling from a local or LAN database:

# SQLite — embedded, no server required
import sqlite3

def query_local_db(query, params=()):
    conn = sqlite3.connect('/data/installation.db')
    conn.row_factory = sqlite3.Row
    cursor = conn.execute(query, params)
    rows = [dict(row) for row in cursor.fetchall()]
    conn.close()
    return rows

# PostgreSQL — requires psycopg2 installed in TD's Python
import psycopg2

_pg_conn = None

def get_pg_connection():
    global _pg_conn
    if _pg_conn is None or _pg_conn.closed:
        _pg_conn = psycopg2.connect(
            host='192.168.1.50',
            database='installation_data',
            user='td_reader',
            password='redacted',
            connect_timeout=5
        )
    return _pg_conn

def query_sensor_history(hours=24):
    conn = get_pg_connection()
    with conn.cursor() as cur:
        cur.execute("""
            SELECT ts, co2, no2, pm25
            FROM air_quality
            WHERE ts > NOW() - INTERVAL '%s hours'
            ORDER BY ts DESC LIMIT 1000
        """, (hours,))
        return cur.fetchall()

Keep the database connection at module level and check conn.closed before each use — PostgreSQL connections time out after idle periods and must be re-established.

IoT Sensors via MQTT

MQTT is the de-facto protocol for IoT sensor networks. Install paho-mqtt into TD’s Python environment (copy the package to %USERPROFILE%\Documents\Derivative\Packages\).

# In a startup Execute DAT or module-level code

import paho.mqtt.client as mqtt
import threading

_mqtt_client = None
_sensor_data = {}
_lock = threading.Lock()

def start_mqtt(broker_ip='192.168.1.10', port=1883):
    global _mqtt_client

    def on_connect(client, userdata, flags, rc):
        print(f'[MQTT] Connected with code {rc}')
        client.subscribe('sensors/#')  # Subscribe to all sensor topics

    def on_message(client, userdata, msg):
        topic  = msg.topic
        value  = float(msg.payload.decode())
        with _lock:
            _sensor_data[topic] = {'value': value, 'ts': absTime.seconds}

    _mqtt_client = mqtt.Client()
    _mqtt_client.on_connect = on_connect
    _mqtt_client.on_message = on_message
    _mqtt_client.connect(broker_ip, port, keepalive=60)

    # Run in background thread — never block TD's main thread
    thread = threading.Thread(target=_mqtt_client.loop_forever, daemon=True)
    thread.start()

def get_sensor(topic, default=0.0, stale_threshold=30.0):
    """Return sensor value, or default if data is stale or missing."""
    with _lock:
        entry = _sensor_data.get(topic)
        if entry is None:
            return default
        if absTime.seconds - entry['ts'] > stale_threshold:
            return default  # Data is stale
        return entry['value']

Normalisation and Smoothing

Raw sensor data arrives in domain-specific units with arbitrary ranges. Normalise to 0–1 before feeding into visual parameters:

def normalize(value, src_min, src_max, clamp=True):
    n = (value - src_min) / (src_max - src_min)
    return max(0.0, min(1.0, n)) if clamp else n

# CO2: 400 ppm (clean outdoor air) to 2000 ppm (very poor indoor air)
co2_norm = normalize(get_sensor('sensors/co2'), 400, 2000)

Route normalised values through a Filter CHOP (lag=2.0 seconds) before driving visual parameters — raw sensor data often jumps discretely between readings, which produces visual discontinuities. The Filter CHOP smooths these into continuous transitions appropriate for generative visuals.

Handling Data Dropouts

Long-running installations will experience data outages — API rate limits, network blips, sensor firmware crashes. Design defensively:

# Module-level state
_last_good_data = {'co2': 500.0, 'no2': 10.0, 'pm25': 5.0}
_last_update_time = 0.0
DATA_TIMEOUT = 120.0  # 2 minutes before fallback

def get_safe_data():
    """Return latest data if fresh, or last good data if stale."""
    global _last_good_data, _last_update_time

    fresh = fetch_from_api()  # your API polling function
    if fresh is not None:
        _last_good_data = fresh
        _last_update_time = absTime.seconds
        op('data_status').par.Data = 'live'
        return fresh

    elapsed = absTime.seconds - _last_update_time
    if elapsed > DATA_TIMEOUT:
        op('data_status').par.Data = 'fallback'
        return _last_good_data  # Return last known values
    else:
        op('data_status').par.Data = f'cached ({elapsed:.0f}s old)'
        return _last_good_data

Full Example: Air Quality Generative Sky

The complete pipeline for a permanent installation where air quality data drives a procedural sky:

Data layer: Web Client DAT polls the AQI API every 5 minutes. WebSocket DAT provides real-time updates when available. Python normalises CO2 (cloud density), NO2 (colour temperature), PM2.5 (turbulence).

Visual layer: A GLSL TOP generates a sky using Perlin noise with three parameters driven by the sensor data:

uniform float uCloudDensity;    // CO2 mapped 0–1
uniform float uColorTemp;       // NO2 mapped 0–1 (cool → warm)
uniform float uTurbulence;      // PM2.5 mapped 0–1

// [noise and sky shader code here — references TDSimplexNoise()]

void main()
{
    vec2 uv = vUV.st + vec2(uTD.frame * 0.0002, 0.0); // drift
    float cloud = TDSimplexNoise(uv * (1.0 + uTurbulence * 3.0) + uTD.frame * 0.0003 * uTurbulence);
    cloud = smoothstep(1.0 - uCloudDensity, 1.0, cloud * 0.5 + 0.5);

    vec3 skyBlue  = vec3(0.3, 0.5, 0.9);
    vec3 skyWarm  = vec3(0.8, 0.6, 0.4);
    vec3 sky = mix(skyBlue, skyWarm, uColorTemp);
    vec3 col = mix(sky, vec3(0.95, 0.97, 1.0), cloud);

    fragColor = vec4(col, 1.0);
}

Long-term stability: the Python database connection uses a context manager that closes and reopens the connection after each query (avoiding idle timeout issues). The MQTT client runs in a daemon thread that the GC cannot collect. All sensor data writes go through a threading.Lock. Memory is profiled weekly with TD’s built-in diagnostics (Dialogs > Monitors > Performance Monitor) to catch any gradual leaks.