blissclient 1.4.1

A python client library for the bliss REST API

blissclient

A python client for the BLISS REST API, the high-level client is fully typed ready for auto-completion in any modern IDE.

Getting Started

Set the BLISSAPI_URL

export BLISSAPI_URL=http://localhost:5000

Then:

from blissclient import BlissClient

client = BlissClient()

# or the blissapi url can be set directly when instantiating the client:
client = BlissClient("http://localhost:5000")

omega = client.hardware.get("omega")
print(omega.position)

future = omega.move(100)
# Wait for the call to temrinate, blocking
future.get()

Execute calls in the session:

import time
from blissclient import BlissClient, get_object

client = BlissClient()

test_session = client.session
future = test_session.call("ascan", get_object("omega"), 0, 10, 10, 0.1, get_object("diode"))

# Ask for the current future state
print(future.state)

# Block until terminated
result = future.get()

# The redis scan key, can be used with `blissdata``
print(result["key"])

get_object("<name>") are translated to the relevant beacon objects.

See the test suite for more examples.

Timeouts

The default timeout for httpx is 5 seconds. This can be a little short for some requests to the BLISS REST API. The timeout can be tuned either when instantiating the client:

client = BlissClient("http://localhost:5000", timeout=10)

or via the environement variable BLISSAPI_TIMEOUT:

export BLISSAPI_TIMEOUT=30

Events

Events are propagated via a websocket using socket.io. blissclient provides a helper function to create connect functions to instantiate the socket.io connection in either threaded or asyncio forms:

connect = self._client.create_connect()
connect()

connect = self._client.create_connect(async_client=True)
await connect()

This function should be run somewhere in the background to ensure event reception. After that objects will be set in evented mode to limit polling.

The client can then subscribe to hardware object events:

omega = client.hardware.get("omega")

def property_event(data: dict[str, any]):
    for key, value in data.items():
        # for example position
        logger.info(f"property event key=`{key}` value=`{value}`")

def online_event(online: bool):
    logger.info(f"onine event {online}")

def locked_event(reason: str):
    logger.info(f"locked event {reason}")

omega.subscribe("property", property_event)
omega.subscribe("online", online_event)
omega.subscribe("locked", locked_event)

Events also allow stdout to be captured from a session call:

connect = client.create_connect(async_client=True)
task = asyncio.create_task(connect())

future = test_session.call(
    "ascan",
    get_object("robx"),
    0,
    5,
    5,
    0.2,
    get_object("diode1"),
    in_terminal=True,
    emit_stdout=True,
)

response = future.get()
print(future.stdout)

Future API

The future returned by each call tries to emulate a celery future.

future = test_session.call(...)

# Any progress if the task supports it
future.progress

# Assume the called function returns a bliss `Scan` object
future = test_session.call(
    "ascan",
    ...
    has_scan_factory=True,
)

# Wait for the scan to start
progress = future.wait_scan()

# Now access the bliss `Scan` key that can be used with blissdata
print(progress["scan"]["key"])


# Ask for the current state (a single REST call)
future.state

# Blocking until terminated (polls the REST API in the background)
# default `monitor_interval` = 5s
future.get(monitor_interval=1)

# Kill the current task
future.kill()

# If socketio was connected and the call requested `emit_stdout=True`
print(future.stdout)

Accessing scan data via blissdata

blissclient will attempt to autoload scan info and provide access to scan data via blissdata. By default blissclient will try to infer the BLISSDATA_URL from the current BLISSAPI_URL, this assumes that redis is running on the same machine as the REST API. In situations where this is not the case you can manually set the blissdata url, where the default port at ESRF is 25002:

export BLISSDATA_URL="redis://myhost:25002"

Once set blissclient will try to resolve the scans:

# blissdata url can also be set when instantiating the client
client = BlissClient(blissdata_url="redis://myhost:25002")
...

future = test_session.call(
    "ascan",
    ...
)

scan = future.get()
scan.info
    {'type': 'ascan',
    'save': False,
    ...
    }

scan.stream('timer:epoch')[:]
    array([1.7732378e+09])

You can completely disable this behaviour by setting:

export BLISSCLIENT_AUTOLOAD_SCAN=false

or when you instantiate the client:

client = BlissClient(autoload_scans=False)