Implementation Examples
This section provides concise examples for various programming languages. Fundamentally, any language that supports WebSockets or HTTP can be used to communicate with the device.
The ID command (/id) is used here to demonstrate how to retrieve identification information from a connected Zahner IM7 Workstation. The following examples show how to send this command using raw WebSocket connections across different languages.
Each example produces the following JSON response from the IM7 device:
data received:
{
"type": "REPLY",
"status": "SUCCESS",
"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
"workstation": {
"protocol_version": 2,
"serial_number": "73000000",
"model_name": "IM7x",
"firmware_version": "p2",
"cpu_card_uuid": "106096c6-203d-47a1-86e8-d2e671c93b9e",
"system_mac": "00C06AFFFFFE",
"system_name": "My Zahner Workstation"
}
}
All languages use at least the ID command for which the response was listed as an example. If additional commands have been implemented as examples, their responses are displayed separately as needed.
Python WebSocket
ID Command
This example demonstrates raw WebSocket communication in Python to illustrate the general procedure. For easier access to the IM7 API, we recommend using the provided Python library.
import json
from websocket import create_connection
id_command = {
"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
"do": "/id",
}
socket_url = "ws://localhost:1994"
web_socket = create_connection(socket_url)
web_socket.send(json.dumps(id_command))
result = web_socket.recv()
result_data = json.loads(result)
print(f"data received:\n{json.dumps(result_data, indent=2)}")
web_socket.close()
Stop Command
It is also possible to stop jobs via WebSocket by sending the /stop command. The job that was stopped then has the status: “status_detail”: “STOPPED_BY_USER”.
import json
from websocket import create_connection
id_command = {
"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
"do": "/stop",
}
socket_url = "ws://localhost:1994"
web_socket = create_connection(socket_url)
web_socket.send(json.dumps(id_command))
result = web_socket.recv()
result_data = json.loads(result)
print(f"data received:\n{json.dumps(result_data, indent=2)}")
web_socket.close()
Measurement Data Retrieval and Decoding
A finished measurement job does not return its data inline. Instead, every finished job stores its result as a resource (type: JOB_RESULT) in the workstation’s resource pool, which you then download as a compact binary blob.
The same mechanism serves both DC and EIS measurements; the resource’s sub_type selects the payload format:
DATASET_DC- DC methods (CV, ramps, OCV, arbitrary waves, …): a single data table.DATASET_AC- EIS and calibration jobs: a data table plus a waveform region.
Note
The resource pool has a limited capacity and evicts older results, so download a dataset soon after its job has finished.
Retrieval always takes two steps:
Locate the resource.
/JOB_RESULT/listreturns every result resource; pick the one whosejob_idmatches your finished job. Each entry carriesresource_id,sub_type,size(the payload size in bytes) and aheaderobject holdingnumber_of_rowsand the structure descriptionjob_data_header. (/JOB_RESULT/infowith aresource_idreturns a single entry.)Download the payload.
/JOB_RESULT/downloadwith theresource_idreplies with the usual JSON success message (a text frame) followed by exactly one binary WebSocket message that contains the raw payload - no Base64, no additional framing. (For plain HTTP or resumable transfers,/JOB_RESULT/getreturns the same bytes as Base64 chunks selected viaoffset/limit.)
The payload obeys the same rules for both formats:
Every value is an IEEE-754 64-bit float (
double) in little-endian byte order.Tables are stored column-major: each column is one contiguous run of
Rdoubles, whereRisnumber_of_rows. The cell in columnc, rowrsits at byte offset(c * R + r) * 8.Integer quantities (periods, gain, filter, shunt, …) are stored as exact doubles. Values may be
NaNor±Inf.The
headerdescribes the structure completely - parse against it instead of hardcoding column counts. User calibration polynomials are already applied by the device, so the values are final.
The examples below use only the Python standard library (plus websocket-client) and are written to be easy to port to other languages.
DC Datasets (DATASET_DC)
A DC dataset is a single column-major table. job_data_header.columns holds three parallel arrays - entry i of each describes column i: dimensions (names), units and urns (machine-readable identifiers).
"columns": {
"dimensions": ["time", "voltage", "current", "shunt"],
"units": ["s", "V", "A", "index"],
"urns": ["time", "37128:POT:U~43639:PAD_U", "37128:POT:I~43639:PAD_I", "shunt"]
}
With C = len(dimensions) columns and R = number_of_rows, the payload is exactly C * R * 8 bytes; column c occupies the byte range [c * R * 8, (c + 1) * R * 8).
import json
import struct
import uuid
from websocket import create_connection
DOUBLE = 8 # bytes per IEEE-754 double value
socket_url = "ws://localhost:1994"
job_id = "2c6a7cf1-aa09-414e-b4f4-3a220bc864ed" # a finished DC job
ws = create_connection(socket_url)
# 1. Find the result resource produced by our job.
ws.send(json.dumps({"do": "/JOB_RESULT/list", "request_id": str(uuid.uuid4())}))
resources = json.loads(ws.recv())["resources"]
resource = next(r for r in resources if r["job_id"] == job_id)
resource_id = resource["resource_id"]
header = resource["header"]["job_data_header"]
rows = resource["header"]["number_of_rows"]
# 2. Download the payload: one JSON reply (text frame) + one binary frame.
ws.send(json.dumps({"do": "/JOB_RESULT/download",
"resource_id": resource_id,
"request_id": str(uuid.uuid4())}))
ws.recv() # text frame: {"type": "REPLY", "status": "SUCCESS", ...}
payload = ws.recv() # binary frame: raw little-endian doubles
ws.close()
# 3. Decode the column-major table into named tracks.
names = header["columns"]["dimensions"]
units = header["columns"]["units"]
values = struct.unpack(f"<{len(payload) // DOUBLE}d", payload) # '<' = little-endian
tracks = {}
for c, name in enumerate(names):
tracks[name] = values[c * rows:(c + 1) * rows]
print(f"{rows} rows, {len(names)} columns, {len(payload)} bytes")
for name, unit in zip(names, units):
print(f" {name} [{unit}]: {tracks[name][:3]} ...")
50 rows, 4 columns, 1600 bytes
time [s]: (0.04, 0.08, 0.12) ...
voltage [V]: (-0.0001446000761931741, -0.00015700776869875528, -0.00016941546120433646) ...
current [A]: (3.225288779627037e-07, 1.7114593607903328e-07, 1.9762994195362863e-08) ...
shunt [index]: (5.0, 6.0, 6.0) ...
EIS Datasets (DATASET_AC)
An EIS row is much wider than a DC row: each row (one measured frequency point) carries values for several groups of channels, plus an averaged waveform per synchronous path. The binary layout keeps all scalar values in one column-major main table (exactly like DC) and appends the waveforms afterwards.
The header declares the groups. Their number and order define the number and order of the columns and wave tables:
Header key |
Count |
Columns per group (in payload order) |
Notes |
|---|---|---|---|
|
1 |
|
always present |
|
|
|
carries numerator / denominator path URIs |
|
|
|
each has a |
|
|
|
no waveform |
|
|
|
|
The main table therefore has C = 3 + 6*I + 6*P + 1*A + 4*T columns - or, robustly, len(meta.dimensions) plus the sum of len(group.dimensions) over all groups.
After the main table (starting at byte C * R * 8) follows one wave table per synchronous path, in header order; each holds R waves back to back, every wave wave_size doubles (layout [row][sample]). The total payload size is thus C * R * 8 plus the sum of R * wave_size * 8 over all synchronous paths.
import cmath
import json
import struct
import uuid
from websocket import create_connection
DOUBLE = 8 # bytes per IEEE-754 double value
socket_url = "ws://localhost:1994"
job_id = "17e104de-b2e9-40b0-a72a-9f8ceaf96d27" # a finished EIS job
ws = create_connection(socket_url)
# 1. Locate the DATASET_AC resource produced by our EIS job.
ws.send(json.dumps({"do": "/JOB_RESULT/list", "request_id": str(uuid.uuid4())}))
resources = json.loads(ws.recv())["resources"]
resource = next(r for r in resources if r["job_id"] == job_id)
resource_id = resource["resource_id"]
header = resource["header"]["job_data_header"]
rows = resource["header"]["number_of_rows"]
# 2. Download the payload: JSON reply (text frame) + one binary frame.
ws.send(json.dumps({"do": "/JOB_RESULT/download",
"resource_id": resource_id,
"request_id": str(uuid.uuid4())}))
ws.recv() # text frame: {"type": "REPLY", "status": "SUCCESS", ...}
payload = ws.recv() # binary frame: raw little-endian doubles
ws.close()
values = struct.unpack(f"<{len(payload) // DOUBLE}d", payload) # '<' = little-endian
# 3. Column counts come from the header groups; never hardcode them.
meta = header["meta"]["dimensions"]
impedances = header["impedances"]
paths = header["paths"] # sync paths: each carries a waveform
async_paths = header["async_paths"]
potentiostats = header["potentiostats"]
groups = impedances + paths + async_paths + potentiostats
main_columns = len(meta) + sum(len(g["dimensions"]) for g in groups)
# 4. Slice the column-major main table into its groups, in header order.
def column(c):
return values[c * rows:(c + 1) * rows]
cursor = 0
def take(count):
global cursor
block = [column(cursor + i) for i in range(count)]
cursor += count
return block
meta_cols = take(len(meta))
impedance_cols = [take(len(g["dimensions"])) for g in impedances]
path_cols = [take(len(g["dimensions"])) for g in paths]
async_cols = [take(len(g["dimensions"])) for g in async_paths]
pot_cols = [take(len(g["dimensions"])) for g in potentiostats]
# 5. The wave region follows the main table: one wave table per sync path.
wave_offset = main_columns * rows
waves = []
for path in paths:
width = path["wave_size"]
per_path = [values[wave_offset + r * width: wave_offset + (r + 1) * width]
for r in range(rows)]
waves.append(per_path)
wave_offset += rows * width
# 6. Results: complex impedance per frequency + waveform shapes.
frequency = meta_cols[0]
z_abs, z_phase = impedance_cols[0][0], impedance_cols[0][1]
print(f"{rows} rows, {main_columns} main columns, {len(payload)} bytes")
for r in range(rows):
z = z_abs[r] * cmath.exp(1j * z_phase[r])
print(f" f = {frequency[r]:8.1f} Hz -> Z = {z.real:8.3f} {z.imag:+8.3f}j Ohm")
for index, path in enumerate(paths):
print(f" path {path['name']!r}: {len(waves[index])} waves x {path['wave_size']} samples")
3 rows, 25 main columns, 12888 bytes
f = 10000.0 Hz -> Z = 59.802 -50.227j Ohm
f = 1000.0 Hz -> Z = 247.535 -134.213j Ohm
f = 100.0 Hz -> Z = 428.021 -281.065j Ohm
path 'voltage': 3 waves x 256 samples
path 'current': 3 waves x 256 samples
A few semantic points:
Complex impedance of a row is
Z = |impedance| * exp(1j * phase)from the first two columns of an impedance group.Zero-filled cells. Groups that are declared in the header but produce no value in a given row are
0.0(e.g. declared async channels without data, or paths not participating in a measurement). This is normal.Multi-sine. For multi-sine measurements there is one row per harmonic per point, and the per-row waveforms are all-zero placeholders of full length.
Empty dataset. A job that produced no rows yields
size = 0andnumber_of_rows = 0.
Offline Snapshot Files (.zrb)
The resource browser on the IM7 web frontend can save a result together with its metadata in a single .zrb file - the “With Header” download.
Such a file is a self-contained, durable copy of the dataset: it can be fetched from the device after a measurement and
opened in client software without a live connection - a useful failsafe should the connection drop mid-measurement.
The container is just the resource metadata (JSON, UTF-8), a single 0x00 separator byte, and the raw payload, concatenated in that order:
Segment |
Encoding |
Length |
|---|---|---|
Resource metadata JSON |
UTF-8 text |
variable |
Separator |
a single |
1 B |
Raw binary payload |
little-endian doubles |
exactly |
To read it, split at the first 0x00 byte: valid JSON never contains a raw 0x00 byte, so the split is unambiguous. Decode the front part as UTF-8 - unit strings contain characters such as Ω. The metadata is the same object as a /JOB_RESULT/list entry, plus a workstation_info object identifying the producing device (protocol_version, serial_number, model_name, …); use protocol_version to detect format revisions.
import json
raw = open("eis_job_result.zrb", "rb").read()
separator = raw.index(b"\x00")
info = json.loads(raw[:separator]) # UTF-8 JSON metadata
payload = raw[separator + 1:] # raw little-endian doubles
assert len(payload) == info["size"] # reject the file on mismatch
header = info["header"]["job_data_header"]
rows = info["header"]["number_of_rows"]
# info["sub_type"] is "DATASET_DC" or "DATASET_AC" -> decode as shown above.
Quick Reference
Quantity |
DC ( |
EIS ( |
|---|---|---|
Column count |
|
|
Row count |
|
|
Cell |
byte |
byte |
Wave region |
after the main table; per sync path |
|
Total size |
|
|
Byte order / type |
little-endian |
little-endian |
Endpoint |
Purpose |
|---|---|
|
list all result resources (match by |
|
metadata of a single resource (by |
|
JSON reply + one raw binary message (WebSocket only) |
JavaScript WebSocket
const socket_url = 'ws://localhost:1994';
const socket = new WebSocket(socket_url);
const id_command = {
"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
"do": "/id",
};
socket.onopen = function(event) {
// Send the command as a JSON string
socket.send(JSON.stringify(id_command));
};
socket.onmessage = function(event) {
// Parse the received JSON string
const result_data = JSON.parse(event.data);
// Print formatted JSON
console.log(`data received:\n${JSON.stringify(result_data, null, 2)}`);
socket.close();
};
PowerShell WebSocket
$socketUrl = "ws://localhost:1994"
$ws = New-Object System.Net.WebSockets.ClientWebSocket
$cts = New-Object System.Threading.CancellationTokenSource
# Connect to the WebSocket
$ws.ConnectAsync($socketUrl, $cts.Token).Wait()
# Create the JSON command
$idCommand = @{
request_id = "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb"
do = "/id"
}
$jsonString = $idCommand | ConvertTo-Json -Compress
$buffer = [System.Text.Encoding]::UTF8.GetBytes($jsonString)
# Send the command
$ws.SendAsync([ArraySegment[byte]]$buffer, [System.Net.WebSockets.WebSocketMessageType]::Text, $true, $cts.Token).Wait()
# Receive the response
$buffer = New-Object byte[] 4096
$result = $ws.ReceiveAsync([ArraySegment[byte]]$buffer, $cts.Token)
$result.Wait()
# Decode and print the response
$responseString = [System.Text.Encoding]::UTF8.GetString($buffer, 0, $result.Result.Count)
$responseData = $responseString | ConvertFrom-Json
Write-Host "data received:"
$responseData | ConvertTo-Json -Depth 5
# Close the connection
$ws.CloseAsync([System.Net.WebSockets.WebSocketCloseStatus]::NormalClosure, "Closing", $cts.Token).Wait()
Bash WebSocket using websocat and jq
SOCKET_URL="ws://localhost:1994"
ID_COMMAND='{"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb", "do": "/id"}'
RESPONSE=$(echo $ID_COMMAND | websocat -1 $SOCKET_URL)
echo "data received:"
echo $RESPONSE | jq .
Bash HTTP using curl and jq
The IM7 also supports HTTP requests but we recommend using WebSockets for full functionality.
ID Command
URL="http://localhost:1994/id"
RESPONSE=$(curl -s $URL)
echo "data received:"
echo $RESPONSE | jq .
Switch On
As an addition for example the Switch On command via HTTP with the required parameters.
URL="http://host.docker.internal:1994/job/start"
PAYLOAD='{
"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
"job": {
"type": "switch_on",
"parameters": {
"potentiostat": "MAIN:1:POT",
"coupling": "POTENTIOSTATIC",
"bias": 0,
"voltage_range_index": 0,
"compliance_range_index": 0
}
}
}'
RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" -d "$PAYLOAD" $URL)
echo "data received:"
echo $RESPONSE | jq .
The output shown below represents only the initial response, indicating that the job has been successfully submitted but has not yet completed. As detailed in the main WebSocket documentation, the job is finished only when a message with “type”:”JOB_DONE” is received. Alternatively, the job status in the job list can be polled until completion.
data received:
{
"type": "REPLY",
"status": "SUCCESS",
"request_id": "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
"job_info": {
"job_id": "27a6688c-6ba7-48ae-ab2c-f58d378b68ff",
"status": "PENDING",
"status_detail": "NOT_FINISHED",
"job_meta": {},
"created": "2025-12-17T10:14:05.807353Z",
"type": "switch_on",
"parameters": {
"potentiostat": "MAIN:1:POT",
"coupling": "POTENTIOSTATIC",
"bias": 0,
"voltage_range_index": 0,
"compliance_range_index": 0
},
"started": "2025-12-17T10:14:05.807353Z",
"error": {
"number": 0,
"code": "NONE",
"message": "no error occoured",
"message_parameters": []
}
}
}
Job List
Use the following HTTP GET command to retrieve the complete list of jobs and their statuses.
URL="http://host.docker.internal:1994/job/list"
RESPONSE=$(curl -s $URL)
echo "data received:"
echo $RESPONSE | jq .
data received:
{
"type": "REPLY",
"status": "SUCCESS",
"jobs": [
{
"job_id": "0a6b9787-1328-43aa-b5c5-6011f7b5a769",
"status": "FINISHED",
"status_detail": "RUN_TO_COMPLETION",
"job_meta": {},
"created": "2025-12-17T10:41:21.467971Z",
"type": "switch_on",
"parameters": {
"potentiostat": "MAIN:1:POT",
"coupling": "POTENTIOSTATIC",
"bias": 0,
"voltage_range_index": 0,
"compliance_range_index": 0
},
"started": "2025-12-17T10:41:21.467971Z",
"ended": "2025-12-17T10:41:21.858092Z",
"error": {
"number": 0,
"code": "NONE",
"message": "no error occoured",
"message_parameters": []
}
}
]
}
Stop Command
The following example demonstrates how to stop a running job using the /stop HTTP command. The job that was stopped has the following status: “status_detail”: “STOPPED_BY_USER”.
URL="http://host.docker.internal:1994/stop"
RESPONSE=$(curl -s $URL)
echo "data received:"
echo $RESPONSE | jq .
C# WebSocket
using System;
using System.IO;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Text.Json;
class Program
{
static async Task Main(string[] args)
{
using (ClientWebSocket ws = new ClientWebSocket())
{
Uri serverUri = new Uri("ws://localhost:1994");
CancellationTokenSource cts = new CancellationTokenSource();
await ws.ConnectAsync(serverUri, cts.Token);
var idCommand = new
{
request_id = "3bcbaaf4-c27e-4bad-ae23-de30a1a0d8fb",
@do = "/id"
};
string jsonString = JsonSerializer.Serialize(idCommand);
byte[] bytesToSend = Encoding.UTF8.GetBytes(jsonString);
await ws.SendAsync(new ArraySegment<byte>(bytesToSend), WebSocketMessageType.Text, true, cts.Token);
using (var ms = new MemoryStream())
{
var buffer = new byte[1024];
WebSocketReceiveResult result;
do
{
result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer), cts.Token);
ms.Write(buffer, 0, result.Count);
} while (!result.EndOfMessage);
string responseString = Encoding.UTF8.GetString(ms.ToArray());
using (JsonDocument doc = JsonDocument.Parse(responseString))
{
var options = new JsonSerializerOptions { WriteIndented = true };
Console.WriteLine("data received:\n" + JsonSerializer.Serialize(doc.RootElement, options));
}
}
await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", cts.Token);
}
}
}