Python Client

Use the Python client for analysis scripts, live monitors, and applications that write data into an engine. Calls are synchronous; telemetry callbacks run on a receiver thread.

Download and install locally

Download Python Client from Downloads and extract it. The client is distributed as a release archive, not through PyPI. Use Python 3.10 or newer and create a virtual environment in your application directory:

python -m venv .venv

Install the extracted directory containing pyproject.toml. Replace the example path and version with your downloaded release. On Windows:

./.venv/Scripts/python.exe -m pip install ./dartwic-python-client-1.1.0

On macOS/Linux:

./.venv/bin/python -m pip install ./dartwic-python-client-1.1.0

These commands install a local package; pip install dartwic-client is not the installation method. Run your scripts with the same virtual environment’s Python. The release bundles TEMPEST, and pip installs pyzmq. Platforms without a compatible pyzmq wheel need its native build dependencies.

Connect

Set DARTWIC_PASSWORD to your engine password. Put the examples below inside this connected scope, replacing the example channel names with yours:

import os
from dartwic_client import DartwicClient

with DartwicClient() as client:
    client.connect("127.0.0.1", 7000, os.environ["DARTWIC_PASSWORD"])
    # Query, write, or subscribe here.

The base port carries operations; base port + 1 carries telemetry. connect registers the client and checks engine compatibility. The context manager disconnects even if your script raises an exception.

Query channels

Read one value, a metadata field, or a group of current channel records:

pressure = client.query_channel("tank_pressure")
units = client.query_channel("tank_pressure", field="units")
records = client.get_channels(["tank_pressure", "tank_temperature"])
for name, record in records.items():
    if record["exists"]:
        print(name, record["channel_data"]["value"])

A single read raises KeyError for a missing channel or field. Batch reads expose exists; absence is not zero. This live-read endpoint rounds value to three decimals. Recorded queries below retain the recorded samples.

Search channels

matches = client.get_channel_keys(query="tank", limit=25)["channels"]
print(matches)
if matches:
    print(client.get_channels(matches))

Search discovers live channel names. Use dataframe search in the analysis example to discover recorded data.

Upsert channels

client.upsert_channel("analysis_result", 42.0)
client.upsert_channel("analysis_result", "bar", field="units")

For a device adapter that has just read several sensors, send the readings with a simple loop:

readings = {"adapter_pressure": 2.5, "adapter_temperature": 23.5}
for channel, value in readings.items():
    client.upsert_channel(channel, value)

Each call is an independent acknowledged write, not an atomic bulk transaction. If a later call fails, earlier writes remain applied. These writes obey engine authority and do not take operator manual override. For a remote driver that owns its channels, publishes timestamped batches, and receives commands, use DARTWIC Share.

Subscribe and receive telemetry

This live monitor receives a snapshot and subsequent updates for ten seconds:

import time

def receive(snapshot):
    if snapshot["exists"]:
        print(snapshot["channel_name"], snapshot["channel_data"]["value"])

stop = client.subscribe_channel("tank_pressure", receive)
try:
    time.sleep(10)
finally:
    stop()

The initial callback runs on the subscribing thread; updates arrive on the telemetry thread and may overlap setup. Protect shared application state and keep callbacks short. Multiple listeners share one server reference; the returned stop function is idempotent. Update callback errors are isolated; an initial callback error fails setup.

For a raw topic handler, bind before enabling the channels you need:

def receive(message, suffix):
    print(suffix, message["payload"])

stop_raw = client.subscribe_telemetry("rapid/channels/", receive)
stop_channel = client.subscribe_channel("tank_pressure", lambda snapshot: None)
try:
    time.sleep(10)
finally:
    stop_channel()
    stop_raw()

A raw binding alone does not ask the engine to stream a channel.

Example: find recorded channels and plot their history

Install matplotlib in the same environment as the client. While connected, find a recording group, select its recorded pressure channels, and retrieve all available samples for those channels in that dataframe:

import matplotlib.pyplot as plt

frames = client.search_channel_dataframes(query="tank_run")["dataframes"]
frame = frames.get("tank_run")
if frame is None:
    raise RuntimeError("No matching dataframe: tank_run")
selected = [name for name in frame["recorded_channels"] if "pressure" in name]
if not selected:
    raise RuntimeError("This dataframe has no recorded pressure channels")

history = client.query_channel_range(selected, data_frame=frame["name"])
for name, channel in history["channels"].items():
    if not channel["timestamps"]:
        continue
    start = channel["timestamps"][0]
    seconds = [(timestamp - start) / 1_000_000_000 for timestamp in channel["timestamps"]]
    plt.plot(seconds, channel["values"], label=name)
plt.xlabel("Seconds since each channel's first sample")
plt.ylabel("Recorded value")
plt.legend()
plt.show()

No bounds or buckets are supplied: this requests raw recorded history, subject to the engine’s point limit. If the limit is exceeded, narrow the interval with inclusive from_timestamp/to_timestamp nanoseconds. For a preview, explicitly request bucket_count=500 and bucket_mode="average" or "extrema"; that is aggregated data. Different channels can have different sample times, so align them explicitly before calculating across series.

The result is keyed by channel reference under history["channels"], with points, timestamps, values, UTC datetimes, and counts. Integer timestamps preserve nanoseconds; Python datetime has microsecond resolution. Use integer timestamps for precise joins and exports.

Call other operations

response = client.operation("dartwic/get-runtime-metadata")
response.raise_for_error()
print(response.payload["engineVersion"])

Helpers raise on errors themselves. Raw operations return a TempestResponse; check it before using its payload.

Connection loss and analysis details

is_connected() reports local state. A registered heartbeat probe detects connection or session loss; detection is not instantaneous. This client does not reconnect automatically. Call disconnect(), connect again, then recreate managed channel subscriptions. Serialize lifecycle calls and disconnect outside telemetry callbacks. Parallel operation calls are supported; each uses an independent socket so a late reply cannot satisfy a later request.

A timeout does not cancel work on the engine, and writes are never replayed automatically. Query resulting state before retrying a write.

See the generated DartwicClient reference for all signatures and Telemetry and History for recording behavior.