Error Handling

In most software, an unhandled exception means a bad user experience. In lab automation, it can mean a plate left in an incubator overnight, a gripper holding a tube, or reagents dispensed twice. The goal of error handling here is not just to recover gracefully; it is to leave every instrument in a safe, known state before the script can safely continue or exits.

Every error handler should ask: is the instrument in a safe state? before re-raising or continuing.

On this page you write error handlers yourself. When you're ready to have the workflow engine retry, pause, and resume on your behalf, see Error handling in Automate — same idea, moved from reactive code to declared policy.

Basic structure

Wrap your method steps with try / except / finally. The finally block runs regardless of success or failure, so use it for cleanup that must always happen.

import logging
from unitelabs.sdk import AsyncApiClient

async def run_method():
    async with AsyncApiClient() as client:
        incubator = await client.get_service_by_name("Incubator")
        gripper   = await client.get_service_by_name("Gripper")

        try:
            await incubator.door_controller.open()
            await gripper.arm_controller.move_plate(source="A1", destination="B1")
            await incubator.door_controller.close()

        except Exception as e:
            logging.exception("Step failed — attempting safe state")
            await incubator.door_controller.close()  # always close the door
            raise

        finally:
            await gripper.arm_controller.home()  # runs whether or not an exception occurred

Use logging.exception rather than print; it captures the full traceback and writes it to your log file.

Keep finally blocks simple. If safe-state actions can themselves raise, wrap them in their own try / except to prevent masking the original error.

Common failure patterns

Device timeout

The instrument did not respond in time. Retry once with a short delay, then raise:

for attempt in range(2):
    try:
        result = await reader.measurement.read_absorbance(well="A1")
        break
    except TimeoutError:
        if attempt == 1:
            logging.error("Reader timed out after 2 attempts")
            raise
        logging.warning("Reader timeout on attempt %d — retrying", attempt + 1)
        await asyncio.sleep(5)

Unexpected device state

Never silently continue when the instrument reports a state your script did not expect. Raise immediately with enough context to diagnose:

status = await incubator.heating.get_status()
if not status.door_closed:
    raise RuntimeError(
        f"Incubator door is open before step — expected closed. "
        f"Check that no plates were left inside."
    )

Silence here turns a detectable misconfiguration into an unexplained result.

Partial completion

If the method fails halfway through, write whatever results you have before re-raising. Data from completed samples is valuable even when the run did not finish:

results = []
try:
    for sample in samples:
        measurement = await reader.measurement.read_absorbance(well=sample["position"])
        results.append({"id": sample["id"], "absorbance": measurement.value})
except Exception:
    logging.error("Run aborted — saving %d partial results", len(results))
    with open("results_partial.json", "w") as f:
        json.dump(results, f)
    raise

Writing results incrementally during the loop makes this automatic. Your method accumulates partial data as it runs rather than buffering everything until the end.

Connection loss

Catch connection errors separately from device-logic errors, since the recovery is different:

try:
    await thermocycler.temperature_controller.set_target_temperature(target_temperature=37)
except ConnectionError as e:
    logging.error("Lost connection to thermocycler: %s", e)
    # Connectors handle reconnection to the instrument automatically
    # Automatic retries with a delay or exponential backoff may be sufficient
    raise
except Exception:
    logging.exception("Thermocycler command failed")
    raise

Marking labware as errored

After a failed tip pickup or transport step you often cannot trust what is physically in a spot — a tube may have been dropped, a tip half-mounted. Since labware 0.29.0, you can record that uncertainty on the labware itself instead of guessing later. Spot.mark_errored() (available on TipSpot and TubeSpot) clears whatever the spot held and replaces it with an Unknown placeholder, and spot.errored reports the state:

from unitelabs.labware import Unknown

spot = tip_rack["A1"]

try:
    await hamilton.pipettes.pick_up_tips(channels=[0], spots=[spot])
except Exception:
    spot.mark_errored()   # contents are now Unknown
    raise

spot.errored             # True
isinstance(spot.get(), Unknown)   # True

mark_errored() works whether the spot was occupied or empty; the previous occupant is detached. Clearing the spot with spot.remove() discards the Unknown placeholder and restores the empty, non-errored state. Marking spots as errored keeps the deck model honest so downstream steps — and anyone reloading the saved deck — know not to rely on those positions.

Hamilton STAR firmware errors

Errors from a Hamilton STAR are raised as HamiltonError or one of its subclasses (HardwareError, ProcessError, CommunicationError, InvalidCommand, HamiltonMultiError). Main codes the SDK doesn't classify, such as 0, raise a plain HamiltonError. The trace attributes are available from unitelabs-liquid-handling 0.39.0, and this API may change in a future release. Each error has:

trace_code can be set while trace_message is None, when the trace code has no known description.

The main code is often too generic to act on. A blocked X-drive and a blocked Y-drive both arrive as HardwareError with code 2, and the same fault can arrive with different main codes. The trace code tells you which component failed, so match on trace_code when your recovery depends on the exact fault:

from unitelabs.liquid_handling.hamilton import HamiltonError

try:
    await hamilton.core96.dispense(...)
except HamiltonError as error:
    if error.module == "H0" and error.trace_code == 51:
        ...  # Dispensing drive not initialized
    raise

If the trace code has a description, it is appended to message as (<module> trace <code>: <description>). For the error above, error.message reads Hardware error (H0 trace 51: Dispensing-Drive: Not initialized) and str(error) reads HardwareError[2]: Hardware error (H0 trace 51: Dispensing-Drive: Not initialized). Without a trace code or a known description, the message is unchanged. Checks that look for part of the message keep working. Checks that compare the whole message, such as error.message == "Hardware error", no longer match once a trace is present. Use error.code or error.trace_code for those.

A command that fails on several channels at once raises a HamiltonMultiError with code 99. Its errors list holds one HamiltonError per failed module, each with its own trace_code and trace_message, and errors_by_module maps each module name to its error. If only one channel fails, you get a single error instead of a HamiltonMultiError. Every HamiltonError has errors_by_module, so catch HamiltonError and loop over it to handle both cases:

import logging

from unitelabs.liquid_handling.hamilton import HamiltonError

try:
    await hamilton.pipettes.pick_up_tips(channels=[0, 1, 2, 3], spots=spots)
except HamiltonError as error:
    for channel_error in error.errors_by_module.values():
        logging.error("%s: trace %s", channel_error.module, channel_error.trace_code)
    raise

Logging for traceability

Set up logging once at the top of your script, before any instrument connections:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s — %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("run.log"),
    ],
)

Log before and after every state-changing operation, not for debugging, but for audit and post-run analysis:

logging.info("Opening incubator door")
await incubator.door.open()
logging.info("Incubator door open")

logging.info("Moving plate: %s → %s", source_well, dest_well)
await gripper.arm.move_plate(source=source_well, destination=dest_well)
logging.info("Plate move complete")

The run.log file gives you a timestamped record of every physical action your script took. Many logs are automatically written and stored on the platform (API logs), others are available but not surfaced unless logging is explicitly set up (inbuilt SDK logs).

Safe-state checklist

Before finishing an error handler, check:

1. Connections: are open connections closed or returned to their pool?
2. Held labware: is the gripper parked? Is a pipette tip ejected?
3. Instrument doors: are incubator and centrifuge doors closed?
4. Partial results: have you written whatever data you collected so far?
5. Notification: if the method was running unattended, does someone know it failed?
6. finally safety: does your finally block avoid raising its own exception?

Next steps

• Operate: define deck positions your error handlers can reference
• Basic Pipetting: tip ejection and liquid-handling-specific error patterns
• Automate → Error handling: declare retry policy and checkpoint-based resumption instead of writing it yourself