Error Handling
This example demonstrates how to handle errors when communicating with the IM7 potentiostat using the zahner_link library. Proper error handling is essential for building robust measurement automation applications that can detect problems and recover gracefully.
The library provides two communication class variants:
Exception-based: ZahnerLinkExc - raises ZahnerLinkException on errors, allowing you to use Python’s
try/exceptpattern.Return-value-based: ZahnerLink - returns an ErrorObject that you must check manually after each operation.
This notebook covers:
Runtime errors: Detecting failures when jobs are executed incorrectly (e.g., switching on a potentiostat that is already on)
Connection failures: Handling errors when the device is unreachable
Parameter errors: Catching invalid job parameters before or during execution
Connection loss during measurement: Using callbacks to detect disconnects, reconnecting, and inspecting the job list to check on interrupted measurements
Connection inspection: Querying information about your own connection and listing all active clients
Cancelling pending operations: Using cancel_pending_operations() to cleanly interrupt blocking calls without losing the connection, then re-synchronizing with the device
Stopping device-side measurements: Using send_stop() to abort a running measurement on the IM7 itself
Note: The other example notebooks intentionally omit error handling to keep them focused. In production code, you should always implement appropriate error handling based on your application requirements.
import threading
import subprocess
import time
import sys
import zahner_link as zl
def print_error_object(error_object: zl.ErrorObject):
print(f"ErrorObject: {error_object}")
print(f"ErrorObject==True: {error_object==True}")
print(f"ErrorObject code enum: {error_object.get_error_code_enum()}")
print(f"ErrorObject message: {error_object.get_message_formatted()}")
def print_job_status(job: zl.AbstractMeasurementJob):
print(f"job successfull: {job.was_successful()}")
print(f"job status: {job.get_last_job_status()}")
print(f"job error message: {job.get_last_job_error_message()}")
info = job.get_last_job_info()
print(f"JobInfo object id: {info.get_job_id()}")
print(f"JobInfo object status: {info.get_status()}")
print(f"JobInfo object status detail: {info.get_status_detail()}")
print(f"JobInfo object error message: {info.get_error_message()}")
print(f"JobInfo started: {info.get_start_date()}")
HOST = "10.10.253.150"
PORT = "1994"
Runtime Errors
Runtime errors occur when the IM7 cannot execute a requested operation - for example, when you attempt to switch on a potentiostat that is already on.
Using the Exception-Based Library
We’ll start with ZahnerLinkExc, which raises a ZahnerLinkException when a job fails. The exception object contains an ErrorObject with the error code and a descriptive message.
In this example, we deliberately trigger a runtime error by calling SwitchOnJob twice on the same potentiostat. The second call fails because the potentiostat is already switched on.
The helper functions print_error_object() and print_job_status() (defined above) are used to display error details throughout this notebook.
link = zl.ZahnerLinkExc(HOST, PORT)
try:
link.connect()
print("connected successfully")
except zl.ZahnerLinkException as e:
error_object: zl.ErrorObject = e.error
print(f"failed to connect, status: {error_object.get_error_code_enum()}, message: {error_object.get_message_formatted()}")
sys.exit()
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=1.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
try:
print("\nfirst switch on")
error_object_success: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_success)
print("\nsecond switch on")
error_object_fail: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_fail)
except zl.ZahnerLinkException as e:
print(f"\ncaught exception: {e}")
error_object: zl.ErrorObject = e.error
print_error_object(error_object)
print_job_status(switch_on_job)
link.do_job(switch_off_job)
link.disconnect()
connected successfully
first switch on
ErrorObject: no error occoured
ErrorObject==True: False
ErrorObject code enum: ErrorCodeEnum.NONE
ErrorObject message: no error occoured
second switch on
caught exception: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
ErrorObject: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
ErrorObject==True: True
ErrorObject code enum: ErrorCodeEnum.UNEXPECTED_EXCEPTION
ErrorObject message: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
job successfull: False
job status: JobStatusEnum.FAILED
job error message: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
JobInfo object id: e0c45582-9069-4458-90a5-ceb09087f76a
JobInfo object status: JobStatusEnum.FAILED
JobInfo object status detail: JobStatusDetailEnum.RUNTIME_ERROR
JobInfo object error message: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
JobInfo started: 2026-05-18T06:39:00.953510Z
Using the Return-Value-Based Library
When using ZahnerLink (without exceptions), errors are not raised automatically. Instead, each call to do_job() returns an ErrorObject that you must inspect yourself.
As the output below shows, the except block is never triggered - even though the second SwitchOnJob fails.
You are responsible for checking the return value after every operation to detect errors in your program flow.
link = zl.ZahnerLink(HOST, PORT)
error: zl.ErrorObject = link.connect()
if not error:
print("connected successfully")
else:
print(f"failed to connect, status: {error.get_error_code_enum()}, message: {error.get_message_formatted()}")
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=1.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
try:
print("\nfirst switch on")
error_object_success: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_success)
print("\nsecond switch on")
error_object_fail: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_fail)
except zl.ZahnerLinkException as e:
print(f"\ncaught exception: {e}")
error_object: zl.ErrorObject = e.error
print_error_object(error_object)
print_job_status(switch_on_job)
link.do_job(switch_off_job)
link.disconnect()
connected successfully
first switch on
ErrorObject: no error occoured
ErrorObject==True: False
ErrorObject code enum: ErrorCodeEnum.NONE
ErrorObject message: no error occoured
second switch on
ErrorObject: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
ErrorObject==True: True
ErrorObject code enum: ErrorCodeEnum.UNEXPECTED_EXCEPTION
ErrorObject message: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
Executing Jobs Without a Valid Connection
Another common runtime error is attempting to run a job when the connection to the IM7 has not been established or when the hostname is incorrect. In this example, we intentionally use an invalid hostname to demonstrate the resulting error behavior.
link = zl.ZahnerLink("definitely.wrong.hostname", PORT)
error: zl.ErrorObject = link.connect()
if not error:
print("connected successfully")
else:
print(f"failed to connect, status: {error.get_error_code_enum()}, message: {error.get_message_formatted()}")
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=1.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
try:
print("\nfirst switch on")
error_object_success: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_success)
print("\nsecond switch on")
error_object_fail: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_fail)
except zl.ZahnerLinkException as e:
print(f"\ncaught exception: {e}")
error_object: zl.ErrorObject = e.error
print_error_object(error_object)
link.do_job(switch_off_job)
link.disconnect()
failed to connect, status: ErrorCodeEnum.CONNECTION_DNS_ERROR, message: dns error: definitely.wrong.hostname:1994
first switch on
ErrorObject: no network connection has been established yet
ErrorObject==True: True
ErrorObject code enum: ErrorCodeEnum.CONNECTION_NOT_ESTABLISHED
ErrorObject message: no network connection has been established yet
second switch on
ErrorObject: no network connection has been established yet
ErrorObject==True: True
ErrorObject code enum: ErrorCodeEnum.CONNECTION_NOT_ESTABLISHED
ErrorObject message: no network connection has been established yet
Parameter Errors
The IM7 validates job parameters and returns descriptive errors when invalid values are provided.
In the following example, we use the exception-based library to submit an EisFrequencyTableJob with a negative amplitude value, which is not allowed. The resulting exception includes details about which parameter caused the error.
link = zl.ZahnerLinkExc(HOST, PORT)
try:
link.connect()
print("connected successfully")
except zl.ZahnerLinkException as e:
error_object: zl.ErrorObject = e.error
print(f"failed to connect, status: {error_object.get_error_code_enum()}, message: {error_object.get_message_formatted()}")
sys.exit()
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=1.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
try:
error_object_success: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_success)
eis_dict_table_job = zl.meas.EisFrequencyTableJob(
{
"bias": 0.0,
"spectrum": [
{
"frequency": freq,
"amplitude": -2, # Negative amplitude to provoke an error
"pre_duration": 0.1,
"pre_waves": 1,
"meas_duration": 0.5,
"meas_waves": 3,
}
for freq in [1e3,10e3]
],
}
)
link.do_job(eis_dict_table_job)
except zl.ZahnerLinkException as e:
print(f"\ncaught exception: {e}")
error_object: zl.ErrorObject = e.error
print_error_object(error_object)
print_job_status(eis_dict_table_job)
link.do_job(switch_off_job)
link.disconnect()
connected successfully
ErrorObject: no error occoured
ErrorObject==True: False
ErrorObject code enum: ErrorCodeEnum.NONE
ErrorObject message: no error occoured
caught exception: parameter 'amplitude' is too low: '-2' < '0' must be false
ErrorObject: parameter 'amplitude' is too low: '-2' < '0' must be false
ErrorObject==True: True
ErrorObject code enum: ErrorCodeEnum.PARAMETER_VALUE_TOO_LOW
ErrorObject message: parameter 'amplitude' is too low: '-2' < '0' must be false
job successfull: False
job status: JobStatusEnum.FAILED
job error message: parameter 'amplitude' is too low: '-2' < '0' must be false
JobInfo object id: void
JobInfo object status: JobStatusEnum.FAILED
JobInfo object status detail: JobStatusDetailEnum.FAILED_TO_CREATE
JobInfo object error message: parameter 'amplitude' is too low: '-2' < '0' must be false
JobInfo started:
Handling Connection Loss During a Measurement
In real-world scenarios, the network connection to the IM7 may be lost while a measurement is running - for example, due to a cable disconnect, network timeout, or infrastructure issue. The library supports detecting these events and recovering from them.
You can register a callback using set_connection_status_callback() to be notified immediately when the connection status changes.
Note: In this demo, the connection disruption is simulated by disabling a network adapter via PowerShell (requires administrator privileges). This is purely for demonstration purposes - in production, connection loss would occur naturally due to external factors.
DELAY_BEFORE_DISCONNECT = 5 # seconds before the connection is killed
ADAPTER_NAME = "Ethernet 2"
def disable_adapter():
"""Disables the network adapter via netsh (requires admin, UAC prompt)."""
cmd = f'netsh interface set interface "{ADAPTER_NAME}" admin=disable'
print(f"[NET] Disabling '{ADAPTER_NAME}' ...")
result = subprocess.run(
["powershell", "-Command", f"Start-Process cmd -ArgumentList '/c {cmd}' -Verb RunAs -Wait"],
capture_output=True, text=True,
)
print(f"[NET] rc={result.returncode} stdout={result.stdout.strip()!r} stderr={result.stderr.strip()!r}")
def enable_adapter():
"""Re-enables the network adapter via netsh (requires admin, UAC prompt)."""
cmd = f'netsh interface set interface "{ADAPTER_NAME}" admin=enable'
print(f"[NET] Enabling '{ADAPTER_NAME}' ...")
result = subprocess.run(
["powershell", "-Command", f"Start-Process cmd -ArgumentList '/c {cmd}' -Verb RunAs -Wait"],
capture_output=True, text=True,
)
print(f"[NET] rc={result.returncode} stdout={result.stdout.strip()!r} stderr={result.stderr.strip()!r}")
def disconnect_after_delay(delay: float):
"""Waits `delay` seconds and then disables the network adapter."""
print(f"[DISCONNECT-THREAD] Waiting {delay}s before killing the connection ...")
time.sleep(delay)
print(f"[DISCONNECT-THREAD] Disabling adapter NOW!")
disable_adapter()
def on_connection_status_changed(status):
print(f"[CONNECTION CHANGED] Status changed: {status}")
link = zl.ZahnerLinkExc(HOST, PORT)
try:
link.set_connection_status_callback(on_connection_status_changed)
link.connect()
print("connected successfully")
except zl.ZahnerLinkException as e:
error_object: zl.ErrorObject = e.error
print(f"failed to connect, status: {error_object.get_error_code_enum()}, message: {error_object.get_message_formatted()}")
sys.exit()
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=1.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
eis_dict_table_job = zl.meas.EisFrequencyTableJob(
{
"bias": 0.0,
"spectrum": [
{
"frequency": freq,
"amplitude": 0.1,
"pre_duration": 0.1,
"pre_waves": 1,
"meas_duration": 0.5,
"meas_waves": 3,
}
for freq in [1e3, 10e3, 1, 200e-3]
],
}
)
try:
error_object_success: zl.ErrorObject = link.do_job(switch_on_job)
print_error_object(error_object_success)
# Start disconnect thread - disables adapter after DELAY_BEFORE_DISCONNECT seconds
disconnect_thread = threading.Thread(
target=disconnect_after_delay,
args=(DELAY_BEFORE_DISCONNECT,),
daemon=True,
)
disconnect_thread.start()
print("\nStarting EIS measurement (connection will be killed mid-measurement) ...")
link.do_job(eis_dict_table_job)
print("\nMeasurement unexpectedly completed (connection was not killed fast enough).")
except zl.ZahnerLinkException as e:
print(f"\ncaught exception: {e}")
error_object: zl.ErrorObject = e.error
print_error_object(error_object)
print_job_status(eis_dict_table_job)
finally:
# Always re-enable the adapter
print("\n[CLEANUP] Re-enabling network adapter ...")
enable_adapter()
eis_dict_table_job_id = eis_dict_table_job.get_last_job_info().get_job_id()
print(f"ID of the job during which the connection was lost: {eis_dict_table_job_id}")
connected successfully
[CONNECTION CHANGED] Status changed: ZahnerLinkConnectionStatusEnum.CONNECTED
ErrorObject: no error occoured
ErrorObject==True: False
ErrorObject code enum: ErrorCodeEnum.NONE
ErrorObject message: no error occoured
[DISCONNECT-THREAD] Waiting 5s before killing the connection ...
Starting EIS measurement (connection will be killed mid-measurement) ...
[DISCONNECT-THREAD] Disabling adapter NOW!
[NET] Disabling 'Ethernet 2' ...
[NET] rc=0 stdout='' stderr=''
[CONNECTION CHANGED] Status changed: ZahnerLinkConnectionStatusEnum.DISCONNECTED
caught exception: network connection closed by remote peer
ErrorObject: network connection closed by remote peer
ErrorObject==True: True
ErrorObject code enum: ErrorCodeEnum.CONNECTION_WAS_CLOSED
ErrorObject message: network connection closed by remote peer
job successfull: False
job status: JobStatusEnum.FAILED
job error message: network connection closed by remote peer
JobInfo object id: e92f741d-d0cf-407f-adc9-92ae1bc696fe
JobInfo object status: JobStatusEnum.FAILED
JobInfo object status detail: JobStatusDetailEnum.CONNECTION_LOSS
JobInfo object error message: network connection closed by remote peer
JobInfo started:
[CLEANUP] Re-enabling network adapter ...
[NET] Enabling 'Ethernet 2' ...
[NET] rc=0 stdout='' stderr=''
ID of the job during which the connection was lost: e92f741d-d0cf-407f-adc9-92ae1bc696fe
Reconnecting and Inspecting the Job Status and Job List
After re-establishing the connection, we call get_job_info() to retrieve the state of the interrupted measurement as JobInfo object.
print("\nReconnect Now")
link.connect()
time.sleep(1) # necessary because of prints in the notebook
job_info = link.get_job_info(eis_dict_table_job)
print(f"\nStatus of EIS job: {job_info.get_status()}")
Reconnect Now
[CONNECTION CHANGED] Status changed: ZahnerLinkConnectionStatusEnum.CONNECTED
Status of EIS job: JobStatusEnum.RUNNING
We can also call get_job_info_list() to retrieve the current state of all jobs on the device as JobInfo objects.
The interrupted measurement is identified by matching its job ID (printed above as JobInfo object id) and is marked with ⬅️ in the output. Notice that its status is still JobStatusEnum.RUNNING - the IM7 continues executing the measurement even though the client lost its connection.
print("\nJob list while EIS runs:")
running_jobs = 0
job_list = link.get_job_info_list()
for job_info in job_list:
marker = ""
if job_info.get_job_id() == eis_dict_table_job_id:
marker = "⬅️"
print(f"id: {job_info.get_job_id()} {marker}")
print(f"\tstatus: {job_info.get_status()}")
if job_info.get_status() == zl.JobStatusEnum.RUNNING:
running_jobs += 1
print(f"\terror message: {job_info.get_error_message()}")
print(f"\tstarted: {job_info.get_start_date()}")
print(f"\tended: {job_info.get_end_date()}")
print(f"Number of jobs still running: {running_jobs}")
Job list while EIS runs:
id: e92f741d-d0cf-407f-adc9-92ae1bc696fe ⬅️
status: JobStatusEnum.RUNNING
error message: no error occoured
started: 2026-05-18T06:39:09.875932Z
ended:
id: a9d860f5-8596-4aaf-972a-39e5fe1116ac
status: JobStatusEnum.FAILED
error message: an unexpected exception occured: SystemException: type: 8; message: Potentiostat '37128:POT' must be switched off, before switched on.
started: 2026-05-18T06:39:03.913172Z
ended: 2026-05-18T06:39:03.913500Z
id: c150a6cd-61e3-44c1-add3-3bc507c53c0d
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:03.915614Z
ended: 2026-05-18T06:39:03.915787Z
id: db36a98f-c3d9-4527-82ae-161b30a08aba
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:04.012234Z
ended: 2026-05-18T06:39:06.914370Z
id: 7f776d34-912b-4b36-9eb8-2074b47df56b
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:06.917536Z
ended: 2026-05-18T06:39:06.917732Z
id: 4ec60877-9b73-46b9-b69e-2c1adc02ea19
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:06.971669Z
ended: 2026-05-18T06:39:09.873762Z
Number of jobs still running: 1
Try Retrieving the Data
The next step demonstrates that an exception is triggered when I try to retrieve data from a job that hasn’t finished yet.
try:
data = link.get_job_result_data(eis_dict_table_job)
print(f"Data retrieved for job, frequencies: {data.get_frequencies()}")
except zl.ZahnerLinkException as e:
print(f"\ncaught exception: {e}")
caught exception: could not find job with job_id 'e92f741d-d0cf-407f-adc9-92ae1bc696fe'
Waiting for the Measurement to Complete
After reconnecting, we wait until the interrupted EIS job has actually finished on the device by calling wait_for_job_to_finish().
Once the wait call returns, the measurement is completed and the result data can be retrieved safely with get_job_result_data(). In the example below, we print the measured frequency points to confirm that the dataset was transferred successfully.
The cell then queries get_job_info_list() again and prints the updated status of all jobs. This makes it easy to verify that the job is no longer in JobStatusEnum.RUNNING state.
Finally, the example counts how many jobs are still running after the wait. In this case the value is 0, which shows that the measurement has completed cleanly and that the workflow can continue with result processing or follow-up jobs.
print("\nWait until EIS is finished")
link.wait_for_job_to_finish(eis_dict_table_job)
print("\nEIS measurement finished")
data = link.get_job_result_data(eis_dict_table_job)
print(f"Data retrieved for job, frequencies: {data.get_frequencies()}")
print("\nJob list after waiting for EIS to finish:")
running_jobs = 0
job_list = link.get_job_info_list()
for job_info in job_list:
marker = ""
if job_info.get_job_id() == eis_dict_table_job_id:
marker = "⬅️"
print(f"id: {job_info.get_job_id()} {marker}")
print(f"\tstatus: {job_info.get_status()}")
if job_info.get_status() == zl.JobStatusEnum.RUNNING:
running_jobs += 1
print(f"\terror message: {job_info.get_error_message()}")
print(f"\tstarted: {job_info.get_start_date()}")
print(f"\tended: {job_info.get_end_date()}")
print(f"Number of jobs still running: {running_jobs}")
Wait until EIS is finished
EIS measurement finished
Data retrieved for job, frequencies: [1000.0, 10000.0, 1.0, 0.2]
Job list after waiting for EIS to finish:
id: c150a6cd-61e3-44c1-add3-3bc507c53c0d
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:03.915614Z
ended: 2026-05-18T06:39:03.915787Z
id: db36a98f-c3d9-4527-82ae-161b30a08aba
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:04.012234Z
ended: 2026-05-18T06:39:06.914370Z
id: 7f776d34-912b-4b36-9eb8-2074b47df56b
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:06.917536Z
ended: 2026-05-18T06:39:06.917732Z
id: 4ec60877-9b73-46b9-b69e-2c1adc02ea19
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:06.971669Z
ended: 2026-05-18T06:39:09.873762Z
id: e92f741d-d0cf-407f-adc9-92ae1bc696fe ⬅️
status: JobStatusEnum.FINISHED
error message: no error occoured
started: 2026-05-18T06:39:09.875932Z
ended: 2026-05-18T06:39:40.619604Z
Number of jobs still running: 0
Finally, we switch off the potentiostat using SwitchOffJob, remove the connection status callback, and disconnect from the device.
try:
print("\nSwitch off")
link.do_job(switch_off_job)
print_job_status(switch_off_job)
except zl.ZahnerLinkException as e:
print(f"\nSwitch-off after disconnect failed: {e}")
link.disconnect()
link.set_connection_status_callback(None)
print("\nDisconnected successfully.")
Switch off
job successfull: True
job status: JobStatusEnum.FINISHED
job error message: no error occoured
JobInfo object id: 19ec0197-4ed8-4d4c-a5d1-c27f3999fbfe
JobInfo object status: JobStatusEnum.FINISHED
JobInfo object status detail: JobStatusDetailEnum.RUN_TO_COMPLETION
JobInfo object error message: no error occoured
JobInfo started: 2026-05-18T06:39:40.644764Z
Disconnected successfully.
Connection Inspection
The library provides methods to inspect the WebSocket connection between your client and the ECW (Electrochemical Workstation). This is useful for debugging connection issues, monitoring active clients, and managing multi-client scenarios.
get_own_connection_info() returns a WebsocketConnectionInfo object with details about your own connection (ID, address, connection time, TLS status, etc.).
get_ecw_connection_list() returns a list of all currently active WebSocket connections to the ECW.
drop_ecw_connection() disconnects a specific client by its connection info.
link = zl.ZahnerLinkExc(HOST, PORT)
try:
link.connect()
print("connected successfully")
except zl.ZahnerLinkException as e:
error_object: zl.ErrorObject = e.error
print(f"failed to connect, status: {error_object.get_error_code_enum()}, message: {error_object.get_message_formatted()}")
sys.exit()
# Get information about our own connection
own_info = link.get_own_connection_info()
print(f"\n--- Own Connection Info ---")
print(f"\tconnection_id: {own_info.connection_id}")
print(f"\tconnected_since: {own_info.connected_since}")
print(f"\tremote_addr: {own_info.remote_addr}:{own_info.remote_port}")
print(f"\ttarget: {own_info.target}")
print(f"\tuser_agent: {own_info.user_agent}")
print(f"\tusing_tls: {own_info.using_tls}")
# List all active connections to the ECW
connections = link.get_ecw_connection_list()
print(f"\n--- All Active Connections ({len(connections)}) ---")
for conn in connections:
marker = "(this client)" if conn.connection_id == own_info.connection_id else ""
print(f"\tid={conn.connection_id}{marker}")
print(f"\t addr={conn.remote_addr}:{conn.remote_port}, since={conn.connected_since}")
link.disconnect()
print("\nDisconnected")
connected successfully
--- Own Connection Info ---
connection_id: df7fc4b8-28f8-418a-9b7f-ddc9662a816b
connected_since: 2026-05-18T06:39:40.717253Z
remote_addr: ::ffff:10.10.255.28:53485
target: /default
user_agent: zahner_link_lib
using_tls: False
--- All Active Connections (2) ---
id=df7fc4b8-28f8-418a-9b7f-ddc9662a816b(this client)
addr=::ffff:10.10.255.28:53485, since=2026-05-18T06:39:40.717253Z
id=11494ec6-dd5e-428f-9907-3fdab84e3fce
addr=::ffff:10.10.255.28:61855, since=2026-05-18T06:39:06.970628Z
Disconnected
Dropping Another Client’s Connection
If multiple clients are connected to the same ECW, you can forcefully disconnect a specific client using drop_ecw_connection(). This is useful for administrative purposes - for example, to reclaim exclusive access or to clean up stale connections.
In the following example, we connect a second client, verify that it appears in the connection list, then drop it from the first client’s perspective.
# Connect the first client
link1 = zl.ZahnerLinkExc(HOST, PORT)
link1.connect()
print("Client 1 connected")
# Connect a second client
link2 = zl.ZahnerLinkExc(HOST, PORT)
link2.connect()
print("Client 2 connected")
# Get the second client's connection info
link2_info = link2.get_own_connection_info()
print(f"\nClient 2 connection_id: {link2_info.connection_id}")
# Verify both clients appear in the connection list
connections_before = link1.get_ecw_connection_list()
print(f"\nConnections before drop: {len(connections_before)}")
for conn in connections_before:
print(f"\tid={conn.connection_id}")
# Drop the second client's connection from the first client
error = link1.drop_ecw_connection(link2_info)
if not error:
print(f"\nSuccessfully dropped client 2")
else:
print(f"\nFailed to drop client 2: {error.get_message_formatted()}")
time.sleep(1) # Give the server a moment to process
# Verify the dropped connection is no longer in the list
connections_after = link1.get_ecw_connection_list()
print(f"\nConnections after drop: {len(connections_after)}")
for conn in connections_after:
print(f"\tid={conn.connection_id}")
remaining_ids = [c.connection_id for c in connections_after]
if link2_info.connection_id not in remaining_ids:
print(f"Confirmed: Client 2 is no longer connected")
else:
print(f"Unexpected: Client 2 is still in the connection list")
# Clean up
link1.disconnect()
try:
link2.disconnect()
except Exception:
pass # Expected - connection was already dropped server-side
print("\nBoth clients Disconnected")
Client 1 connected
Client 2 connected
Client 2 connection_id: a2f9fd81-2fc2-42d9-985c-c7e1e3bcfb47
Connections before drop: 3
id=a2f9fd81-2fc2-42d9-985c-c7e1e3bcfb47
id=3a09ba80-c9e4-4f79-8789-1a022ac80fea
id=11494ec6-dd5e-428f-9907-3fdab84e3fce
Successfully dropped client 2
Connections after drop: 2
id=3a09ba80-c9e4-4f79-8789-1a022ac80fea
id=11494ec6-dd5e-428f-9907-3fdab84e3fce
Confirmed: Client 2 is no longer connected
Both clients Disconnected
Cancelling Pending Operations
The cancel_pending_operations() method provides a thread-safe way to interrupt any blocking client-side call - such as do_job(), get_job_result_data(), or wait_for_job_to_finish() - causing them to return immediately with ErrorCodeEnum.STOPPED_MANUALLY.
Key characteristics:
Thread-safe: Can be called from any thread (e.g., a timer thread, a GUI button handler) while another thread is blocked on a library call.
Connection preserved: Unlike disconnect(), the WebSocket connection to the IM7 remains active after cancellation.
Device unaffected: Jobs already running on the IM7 continue normally - only the client-side wait is interrupted.
Re-synchronizable: After cancellation, you can call wait_for_job_to_finish() to wait for the device-side measurement to complete, then retrieve the full result data.
This is particularly useful in GUI applications where a user clicks “Cancel” to regain control without aborting the measurement, or in automation scripts that need to implement timeouts.
Cancelling a Running Job and Retrieving Results Later
The following example demonstrates the complete cancel-and-resync workflow:
A PogaJob (potentiostatic galvanic acquisition) is started in a background thread via do_job().
After a few seconds, the main thread calls cancel_pending_operations() to unblock the do_job() call.
The background thread receives the ErrorCodeEnum.STOPPED_MANUALLY error, but the IM7 continues its measurement.
The main thread then calls wait_for_job_to_finish() to wait until the device-side measurement actually completes.
Finally, the full measurement data is retrieved with get_job_result_data().
POGA_DURATION = 15 # seconds - long enough to cancel mid-measurement
CANCEL_AFTER = 3 # seconds before calling cancel_pending_operations()
link = zl.ZahnerLinkExc(HOST, PORT)
try:
link.connect()
print("connected successfully")
except zl.ZahnerLinkException as e:
error_object: zl.ErrorObject = e.error
print(f"failed to connect, status: {error_object.get_error_code_enum()}, message: {error_object.get_message_formatted()}")
sys.exit()
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=0.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
link.do_job(switch_on_job)
print("Potentiostat switched on")
# Create a long-running measurement job
poga_job = zl.meas.PogaJob(
bias=0,
duration=POGA_DURATION,
output_data_rate=10,
autorange=False,
current_range=0.1,
)
# Run do_job() in a background thread so we can cancel from the main thread
error_from_thread = [None]
def run_job_in_background():
try:
error_from_thread[0] = link.do_job(poga_job)
except zl.ZahnerLinkException as e:
error_from_thread[0] = e.error
thread = threading.Thread(target=run_job_in_background)
thread.start()
print(f"\nPogaJob started in background thread (duration={POGA_DURATION}s)")
# Wait a few seconds, then cancel from the main thread
time.sleep(CANCEL_AFTER)
print(f"\n[MAIN THREAD] Calling cancel_pending_operations() after {CANCEL_AFTER}s ...")
link.cancel_pending_operations()
# The background thread should now unblock
thread.join(timeout=10)
assert not thread.is_alive(), "Thread should have finished after cancel"
# Inspect the error - should be STOPPED_MANUALLY
print(f"\n--- Error returned to background thread ---")
print(f"\tError code: {error_from_thread[0].get_error_code_enum()}")
print(f"\tMessage: {error_from_thread[0].get_message_formatted()}")
# The IM7 is still running the measurement! Let's wait for it to finish.
print(f"\n[MAIN THREAD] Waiting for the IM7 to finish the measurement ...")
link.wait_for_job_to_finish(poga_job)
print("Measurement finished on the device")
# Now retrieve the full result data
data = link.get_job_result_data(poga_job)
print(f"\nData retrieved successfully!")
print(f"\tNumber of data points: {data.get_row_count()}")
link.do_job(switch_off_job)
print("\nPotentiostat switched off")
link.disconnect()
print("Disconnected")
connected successfully
Potentiostat switched on
PogaJob started in background thread (duration=15s)
[MAIN THREAD] Calling cancel_pending_operations() after 3s ...
--- Error returned to background thread ---
Error code: ErrorCodeEnum.STOPPED_MANUALLY
Message: manually stopped
[MAIN THREAD] Waiting for the IM7 to finish the measurement ...
Measurement finished on the device
Data retrieved successfully!
Number of data points: 150
Potentiostat switched off
Disconnected
Stopping the Device-Side Measurement with send_stop()
In the previous example, cancel_pending_operations() only unblocked the client - the IM7 continued measuring until the configured duration elapsed. If you want to actually abort the measurement on the device, use send_stop() after cancelling.
send_stop() takes a QueueStopModeEnum parameter:
QUEUE_STOP_MODE_FLUSH- Stops the currently running job and cancels all queued jobs.QUEUE_STOP_MODE_CONTINUE- Stops only the currently running job; queued jobs continue normally.
The pattern is:
cancel_pending_operations() - unblock the client
send_stop() - tell the IM7 to abort the measurement
wait_for_job_to_finish() - confirm the job has ended (returns quickly since the device stopped it)
get_job_result_data() - retrieve any partial data that was collected before the stop
POGA_DURATION_LONG = 60 # seconds - intentionally long, we will stop it early
CANCEL_AFTER = 3 # seconds before calling cancel_pending_operations()
link = zl.ZahnerLinkExc(HOST, PORT)
try:
link.connect()
print("connected successfully")
except zl.ZahnerLinkException as e:
error_object: zl.ErrorObject = e.error
print(f"failed to connect, status: {error_object.get_error_code_enum()}, message: {error_object.get_message_formatted()}")
sys.exit()
switch_on_job = zl.control.SwitchOnJob(
potentiostat="MAIN:1:POT",
coupling=zl.PotentiostatCoupling.POTENTIOSTATIC,
bias=0.0,
voltage_range_index=0,
compliance_range_index=0,
)
switch_off_job = zl.control.SwitchOffJob(potentiostat="MAIN:1:POT")
link.do_job(switch_on_job)
print("Potentiostat switched on")
# Create a long-running measurement job (60 seconds)
poga_job = zl.meas.PogaJob(
bias=0,
duration=POGA_DURATION_LONG,
output_data_rate=10,
autorange=False,
current_range=0.1,
)
# Run do_job() in a background thread
error_from_thread = [None]
def run_job_in_background():
try:
error_from_thread[0] = link.do_job(poga_job)
except zl.ZahnerLinkException as e:
error_from_thread[0] = e.error
thread = threading.Thread(target=run_job_in_background)
thread.start()
print(f"\nPogaJob started in background thread (duration={POGA_DURATION_LONG}s)")
# Wait a few seconds, then cancel the client-side wait
time.sleep(CANCEL_AFTER)
print(f"\n[MAIN THREAD] Calling cancel_pending_operations() after {CANCEL_AFTER}s ...")
link.cancel_pending_operations()
thread.join(timeout=10)
assert not thread.is_alive(), "Thread should have finished after cancel"
print(f"\tBackground thread unblocked with: {error_from_thread[0].get_error_code_enum()}")
# Now stop the device-side measurement
print(f"\n[MAIN THREAD] Calling send_stop(QUEUE_STOP_MODE_FLUSH) to abort the IM7 measurement ...")
link.send_stop(zl.QueueStopModeEnum.QUEUE_STOP_MODE_FLUSH)
# Wait for the job to finish (should return quickly since we stopped it)
print("[MAIN THREAD] Waiting for job to finish ...")
link.wait_for_job_to_finish(poga_job)
print("Job finished (stopped by user)")
# Retrieve partial data collected before the stop
data = link.get_job_result_data(poga_job)
print(f"\nPartial data retrieved!")
print(f"\tNumber of data points: {data.get_row_count()}")
print(f"\t(Expected ~{CANCEL_AFTER * 10} points at 10 samples/s for {CANCEL_AFTER}s of measurement)")
# Check the job status - should indicate it was stopped
print_job_status(poga_job)
link.do_job(switch_off_job)
print("\nPotentiostat switched off")
link.disconnect()
print("Disconnected")
connected successfully
Potentiostat switched on
PogaJob started in background thread (duration=60s)
[MAIN THREAD] Calling cancel_pending_operations() after 3s ...
Background thread unblocked with: ErrorCodeEnum.STOPPED_MANUALLY
[MAIN THREAD] Calling send_stop(QUEUE_STOP_MODE_FLUSH) to abort the IM7 measurement ...
[MAIN THREAD] Waiting for job to finish ...
Job finished (stopped by user)
Partial data retrieved!
Number of data points: 29
(Expected ~30 points at 10 samples/s for 3s of measurement)
job successfull: False
job status: JobStatusEnum.FAILED
job error message: manually stopped
JobInfo object id: ed429a5f-f1ac-4012-b61b-eb8d9ee5014c
JobInfo object status: JobStatusEnum.FAILED
JobInfo object status detail: JobStatusDetailEnum.STOPPED_BY_USER
JobInfo object error message: manually stopped
JobInfo started:
Potentiostat switched off
Disconnected