Using the API from code

Habitat's REST API is plain JSON over HTTP, so you can drive it from any language. The examples below use Python with httpx, but nothing here is Habitat-specific — any HTTP client works.

Point at your device

Every call is relative to your Habitat device's base URL — the device on your local network, http://<device-name>.local:8000 (e.g. http://mistyforest.local:8000). Set it once:

import httpx

BASE_URL = "http://mistyforest.local:8000"

Check the device is up

r = httpx.get(f"{BASE_URL}/health", timeout=10.0)
r.raise_for_status()
print(r.json())          # device identity + driver status

Read hardware state

# List every pump and its live status
pumps = httpx.get(f"{BASE_URL}/pumps", timeout=10.0).json()
for pump in pumps:
    print(pump["addr"], pump["name"], "ready:", pump["ready"])

Run an operation

Fluidic operations are atomics (single primitives) and routines (compositions of atomics). You execute one by POSTing to its name with a params body. By default the call blocks until the hardware finishes and returns the result:

resp = httpx.post(
    f"{BASE_URL}/atomics/centris.motion.plunger_relative_ul",
    json={"params": {"addr": 0, "volume_ul": 50.0}},
    timeout=300.0,
)
resp.raise_for_status()
print(resp.json())       # job result

Poll for readiness between commands

When you issue several motion commands in a row, wait for the pump to report ready before the next one:

import time

def wait_until_ready(addr: int, timeout_s: float = 120.0) -> None:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        pump = httpx.get(f"{BASE_URL}/pumps/{addr}", timeout=10.0).json()
        if pump.get("ready"):
            return
        time.sleep(0.5)
    raise TimeoutError(f"pump {addr} not ready after {timeout_s}s")

Long-running work without blocking

Add the Prefer: respond-async header to get a 202 Accepted immediately with a Location you can poll:

resp = httpx.post(
    f"{BASE_URL}/routines/wash",
    json={"params": {"addr": 0}},
    headers={"Prefer": "respond-async"},
    timeout=10.0,
)
job_url = resp.headers["Location"]          # e.g. /jobs/abc123
status = httpx.get(f"{BASE_URL}{job_url}").json()

Error handling

Failures come back as application/problem+json with a detail field and, where relevant, structured context:

resp = httpx.post(f"{BASE_URL}/atomics/centris.motion.plunger_relative_ul",
                  json={"params": {"addr": 99, "volume_ul": 50.0}})
if resp.is_error:
    problem = resp.json()
    print(problem["title"], "-", problem.get("detail"))

A fuller worked example

A short end-to-end session — prime each reagent line, run a dosing sequence, then check the pump and recover if needed. It builds on exactly the calls above. Reagent names, volumes, and timings below are placeholders; substitute your own.

import time
import httpx

BASE_URL = "http://mistyforest.local:8000"


def run(path: str, params: dict | None = None) -> dict:
    """POST a routine/atomic and block until the hardware finishes."""
    r = httpx.post(f"{BASE_URL}{path}", json={"params": params or {}}, timeout=600.0)
    r.raise_for_status()
    return r.json()


def wait_ready(addr: int, timeout_s: float = 120.0) -> None:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        if httpx.get(f"{BASE_URL}/pumps/{addr}", timeout=10.0).json().get("ready"):
            return
        time.sleep(0.5)
    raise TimeoutError(f"pump {addr} not ready")


# 1. Prime each reagent source line so it's full of fresh fluid.
for reagent in ["reagent_a", "reagent_b", "pbs"]:
    run("/routines/prime_source", {"reagent": reagent})
    wait_ready(0)

# 2. Run a dosing sequence — three doses, spaced apart.
for dose in range(3):
    run("/routines/dispense", {"reagent": "reagent_a", "volume_ul": 50.0})
    wait_ready(0)
    if dose < 2:
        time.sleep(30 * 60)          # inter-dose interval

# 3. Check pump state; recover if anything faulted.
status = httpx.get(f"{BASE_URL}/pumps/0", timeout=10.0).json()
if not status.get("ready"):
    httpx.post(f"{BASE_URL}/pumps/0/reset", timeout=30.0)     # clear the fault
    httpx.post(f"{BASE_URL}/pumps/initialize-all", json={}, timeout=60.0)  # re-home

Browse the complete endpoint list on the API overview, or try requests live from Try it.