Programmatic Usage¶

Overview¶

The same operations exposed by the CLI are available as Python methods on LuciHubClient, imported from anabrid.lucihub.cli.client. The client is a synchronous wrapper around the REST API; it uses httpx internally and supports use as a context manager. It pairs naturally with pybrid-computing for direct manipulation of device objects (modules, integrators, multipliers, routing) and with pyredacc for converting SymPy ODEs into analang source.

Note that the actual compiler (redacc) is currently not publicly available. Contact anabrid for more information.

Installation¶

The Python client is installed alongside lucihub-cli (see Getting Started) — there is no separate package. Bring pybrid-computing and (optionally) pyredacc as direct dependencies of your project when you need their APIs.

Construction¶

import os
from anabrid.lucihub.cli.client import LuciHubClient

hub = LuciHubClient(
    base_url=os.getenv("LUCIHUB_URL", "https://redac.anabrid.com"),
    timeout=30.0,
    api_key=os.environ["LUCIHUB_API_KEY"],
)

Use the client as a context manager (with LuciHubClient(...) as hub: ...) or call hub.close() explicitly. The HTTP connection pool is otherwise kept open for the lifetime of the object.

API surface¶

Since the lucihub-cli and lucihub-common wheels are source packages, you are free to point your IDE or your code agent at your installed packages and explore the code for further information. The functions documented below are mostly thin wrappers around the corresponding LUCIHUB API calls.

System / users / devices¶

Task submission¶

Task lifecycle and retrieval¶

Errors¶

Integration with pybrid-computing¶

For a deeper understanding of pybrid-computing (commonly referred to simply as pybrid), please refer to its own documentation at https://anabrid.github.io/pybrid-computing.

pybrid-computing normally needs a direct connection to a device to download its specification and exchange protobuf messages. Lucihub bridges that gap: it caches each device's entity blob and serves it over the REST API. The Python client fetches the blob, the helper build_computer_from_entity (in anabrid.lucihub.cli.entity_tree) deserialises it into a live pybrid Computer / LUCIStack object, and you configure that object locally. The configured object is then re-serialised into a protobuf pb.File envelope and submitted to lucihub as the module of a run task.

from anabrid.lucihub.cli.entity_tree import build_computer_from_entity
import pybrid.base.proto.main_pb2 as pb
from pybrid.base.proto.versioning import ProtoVersioning

entity_bytes = hub.get_device_entity("lucistack")
lucistack = build_computer_from_entity(entity_bytes, device_type="LUCIDAC")

# configure pybrid Computer object (blocks, routing, ADC channels)
cluster = lucistack.entities[0].clusters[0]
cluster.m0block.elements[0].computation.ic = -1.0
cluster.route(m_out=1, u_out=0, c_factor=-1.0, m_in=0)
# ...

module = lucistack.get_serializer()().serialize(lucistack)
file_msg = pb.File(version=ProtoVersioning.current(), module=module)
module_bytes = file_msg.SerializeToString()

run_task_id = hub.submit_run_task(
    device_name="lucistack",
    module=module_bytes,
    sample_rate=100_000,
    op_time=0.2,
)

See examples/run_with_pybrid.py and examples/run_with_lucipy.py for end-to-end scripts. The integration currently supports single-circuit mode only; session-based pybrid workflows are not yet bridged.

TODO: the spec (specs/00_overview.md §Example) names a hub.get_pybrid_device(device_name) convenience method that would return a configured pybrid Computer directly. No such method exists on LuciHubClient in the current code; clients call get_device_entity + build_computer_from_entity themselves. Resolve before publishing — either add the convenience method or update the spec.

SymPy → analang via pyredacc¶

The compile endpoint accepts ODEs in analang format only. For workflows that start from SymPy, use pyredacc locally to translate the SymPy expression into analang text, then submit the text as ode_source.

import sympy as sp
from pyredacc import ODEAdapter, SympyODE

t = sp.Symbol("t")
X = sp.Function("X")
sympy_ode = SympyODE(
    odes=sp.Eq(X(t).diff(t, 2) + X(t), 0),
    initial_conditions={X(t): 1.0},
    probes=[X(t), X(t).diff(t)],
    name="harmonic",
)
ode_source = ODEAdapter().translate_sympy(sympy_ode)

End-to-end workflow¶

import os
import numpy as np
from anabrid.lucihub.cli.client import LuciHubClient
from anabrid.lucihub.models.enums import TaskStateEnum

hub = LuciHubClient(api_key=os.environ["LUCIHUB_API_KEY"])

DEVICE = "lucistack"
device = next(d for d in hub.get_devices() if d.name == DEVICE)
entity_bytes = hub.get_device_entity(DEVICE)

# 1. compile
compile_id = hub.submit_compile_task(
    ode_source=open("harmonic.ana").read(),
    backend=device.type,
    module=entity_bytes,
)
status = hub.wait_for_task_completion(compile_id)
if status.state != TaskStateEnum.SUCCEEDED:
    raise RuntimeError(f"compile failed: {status.error}")
module_bytes = hub.get_task_result(compile_id)   # raw APB

# 2. run
run_id = hub.submit_run_task(
    device_name=DEVICE,
    module=module_bytes,
    sample_rate=100_000,
    op_time=0.2,
)
status = hub.wait_for_task_completion(run_id)
if status.state != TaskStateEnum.SUCCEEDED:
    raise RuntimeError(f"run failed: {status.error}")

runs = hub.get_task_result(run_id)               # [run][channel][sample]
result = np.array(runs[0])

For a SymPy-driven variant, replace open("harmonic.ana").read() with the output of ODEAdapter().translate_sympy(...) shown above. The full set of working examples lives in examples/: run_with_analang.py, run_with_sympy.py, run_with_lucipy.py, and run_with_pybrid.py.