Metadata-Version: 2.4
Name: blissclient
Version: 1.4.0
Summary: A python client library for the bliss REST API
Author-email: ESRF <bcu@esrf.fr>
License-Expression: MIT
Project-URL: Homepage, https://gitlab.esrf.fr/bliss/blissclient
Project-URL: Bug Tracker, https://gitlab.esrf.fr/bliss/blissclient/issues
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: httpx
Requires-Dist: pydantic>=2
Requires-Dist: python-socketio[client]>=5.11
Requires-Dist: tblib
Provides-Extra: test
Requires-Dist: pytest>=7; extra == "test"
Requires-Dist: pint; extra == "test"
Requires-Dist: marshmallow<4; extra == "test"
Provides-Extra: dev
Requires-Dist: black>=24; extra == "dev"
Requires-Dist: flake8>=4; extra == "dev"
Requires-Dist: flake8-pyproject; extra == "dev"

# 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`

```bash
export BLISSAPI_URL=http://localhost:5000
```

Then:

```python
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:

```python
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.

# 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:

```python
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:

```python
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:

```python
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.

```python
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:

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

Once set blissclient will try to resolve the scans:

```python
# 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:

```bash
export BLISSCLIENT_AUTOLOAD_SCAN=false
```

or when you instantiate the client:

```python
client = BlissClient(autoload_scans=False)
```
