Metadata-Version: 2.5
Name: benchmesh
Version: 0.1.0
Summary: Python client for BenchMesh: control and record lab instruments (power supplies, meters, loads, generators, scopes) from scripts and notebooks
Project-URL: Homepage, https://benchmesh.online
Project-URL: Documentation, https://benchmesh.online/docs/python
Project-URL: Repository, https://github.com/BenchMesh/benchmesh-python
Project-URL: Issues, https://github.com/BenchMesh/benchmesh-python/issues
Author-email: BenchMesh Team <support@benchmesh.online>
License-Expression: MIT
License-File: LICENSE
Keywords: automation,benchmesh,instruments,lab,multimeter,power-supply,scpi,test-equipment
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: System :: Hardware
Classifier: Typing :: Typed
Requires-Python: >=3.9
Provides-Extra: pandas
Requires-Dist: pandas>=1.3; extra == 'pandas'
Description-Content-Type: text/markdown

# benchmesh

Python client for [BenchMesh](https://benchmesh.online), one cockpit for lab instruments:
power supplies, multimeters, electronic loads, signal generators, oscilloscopes, LCR meters and
spectrum analyzers. Control them, read them and record them from scripts and Jupyter notebooks.

- No runtime dependencies (standard library only). Python 3.9 or newer.
- One API for every instrument of a class: the limits, ranges and modes come from the server.
- Live readings, recordings into pandas, automation flows, remote access with a pinned certificate.

## Install

```bash
pip install benchmesh              # the client
pip install "benchmesh[pandas]"    # plus pandas for DataFrames
```

The BenchMesh app (or `benchmesh-server`) must be running. The client talks to its API at
`http://127.0.0.1:57666` on the same computer.

## Quick start

```python
import benchmesh

bench = benchmesh.connect()  # $BENCHMESH_URL or http://127.0.0.1:57666

for inst in bench.instruments():
    print(inst.id, inst.classes, inst.model, "online" if inst.online else "offline")

psu = bench["tenma-72-2540"]  # the id from the Instruments drawer
psu.set("voltage", 5.0)  # setter, checked by the server against 0-60 V
psu.set("current", 0.5)
psu.set("output", "ON")  # switches the output ON at once
print(psu.read("VOUT"))  # latest polled value: 4.94 (float, volts)
print(psu.status())  # {"VOUT": 4.94, "IOUT": 0.012, "OUTPUT": "ON", ...}
print(psu.query("voltage"))  # asks the instrument now
psu.set("output", "OFF")
```

An instrument with several classes (the OWON SPM is a `PSU` and a `DMM`) needs the class:

```python
spm = bench["owon-spm3103"]
spm.set("voltage", 3.3, cls="PSU")
spm.read("MEAS", cls="DMM")
```

Multi-channel instruments take `ch=` (1-9, default 1): `scope.set("scale", 0.5, ch=2)`.

## Values

- Measured values (SI values on the server) become `float` in base units (volts, amperes, ohms).
  An instrument that reports something that is not a number (for example `OL` on an overloaded
  meter) gives that text back.
- Modes and switches stay strings (`"ON"`, `"VOLT"`); flags stay `bool`; missing values are `None`.
- Binary results (screenshots) are `bytes`.
- `raw=True` on `query`, `read` and `status` returns the server's JSON unchanged.
- Setter values are sent as given: floats with `repr`, integers as integers, strings unchanged.
  The server validates them against the definition (range, allowed values) and refuses invalid
  ones with `BadRequest`. `inst.operations()` lists every query and setter with its value spec.

## Examples

### Read a meter, adjust a load

The classic loop: hold 5 V on a source by raising or lowering the load current
(full script: [examples/hold_voltage.py](examples/hold_voltage.py)).

```python
import benchmesh

bench = benchmesh.connect()
load = bench["owon-oel1515"]
current = 0.0
load.set("mode", "CURR")
load.set("curr", current)
load.set("input", "ON")  # the load starts drawing current
try:
    for t, values in bench.watch({"v": ("owon-xdm1241", "DMM", "MEAS")}, interval=0.5, duration=60):
        v = values["v"]
        if v is None:
            continue  # meter offline
        current = min(current + 0.01, 1.0) if v > 5.0 else max(current - 0.01, 0.0)
        load.set("curr", round(current, 3))
finally:
    load.set("input", "OFF")
```

`bench.watch(signals)` polls the latest status (nothing extra is sent to the instruments) and
yields `(time_s, {label: value})` only when a reading was updated. A signal is
`(device_id, class, field)` or `(device_id, class, field, channel)`.

### Sweep into pandas

The server records on whole seconds; the script only steps the voltage
(full script: [examples/sweep_to_pandas.py](examples/sweep_to_pandas.py)).

```python
import time
import benchmesh

bench = benchmesh.connect()
psu = bench["tenma-72-2540"]
psu.set("voltage", 1.0)
psu.set("output", "ON")
try:
    with bench.record(
        "sweep", {"vout": (psu.id, "PSU", "VOUT"), "meter": ("owon-xdm1241", "DMM", "MEAS")}
    ) as rec:
        for v in (1.0, 2.0, 3.0, 4.0, 5.0):
            psu.set("voltage", v)
            time.sleep(2)
finally:
    psu.set("output", "OFF")

df = rec.to_pandas()  # index: time (UTC), one column per label
df.plot()
```

Older recordings: `bench.recordings()` (newest first), `rec.data()` for plain lists
`{label: (times_s, values)}`, `rec.export_csv()` for the CSV the app exports.

### Jupyter

```python
import benchmesh

bench = benchmesh.connect()
rec = bench.recordings()[0]
df = rec.to_pandas()
df.resample("10s").mean().plot(subplots=True)
```

### Remote access

On the bench computer open the BenchMesh app, Instruments - Remote access: set a password,
switch it on and note the address and the certificate fingerprint. On the other computer:

```python
bench = benchmesh.connect(
    "https://192.168.1.20:57667",
    password="the password",
    fingerprint="AB:CD:...:EF",  # as shown in the app; case and colons do not matter
)
```

The certificate is self-signed, so the client pins it: right after the TLS handshake it compares
the certificate's SHA-256 fingerprint with the one you give, and only then sends the password.
A different certificate raises `FingerprintMismatch`. Without a fingerprint the system's
certificate authorities must trust it; otherwise `TLSError` tells you which fingerprint the
server presented, so you can compare it with the app. An expired session signs in again by
itself. `bench.close()` (or `with benchmesh.connect(...) as bench:`) signs out.

`BENCHMESH_URL`, `BENCHMESH_PASSWORD` and `BENCHMESH_FINGERPRINT` can replace the arguments
(see [examples/remote_watch.py](examples/remote_watch.py)).

### Automation flows

```python
for f in bench.flows():
    print(f.id, f.name)

status = bench.run_flow("hold-5v", params={"target": 4.8}, arm=True, on_event=lambda e: print(e["msg"]))
print(status.state, status.vars)  # "finished" / "failed" / "aborted"
```

A flow that can switch outputs on needs `arm=True`, otherwise the server refuses it with
`Refused` (`error.body["energizes"]` lists the operations). `run_flow` waits until the run ends;
`timeout=` and Ctrl-C abort it (its cleanup still runs). A flow document (`dict`) can be run
directly instead of a saved id.

## API overview

| | |
|---|---|
| `benchmesh.connect(url=None, *, password=None, fingerprint=None, timeout=30.0)` | `Bench`; checks the server answers |
| `bench.instruments(cls=None)` / `bench.instrument(id)` / `bench[id]` | `Instrument` |
| `bench.watch(signals, interval=0.5, duration=None)` | iterator of `(time_s, {label: value})` |
| `bench.record(name, signals, interval_s=1, description=None)` | `Recording` (context manager stops it) |
| `bench.recordings()` / `bench.recording(id)` | `Recording` |
| `bench.flows()` / `bench.flow(id)` / `bench.run_flow(flow, params=None, *, arm=False, wait=True, timeout=None, on_event=None)` | `Flow`, flow document, `RunStatus` |
| `bench.run_status()` / `bench.abort_run()` / `bench.continue_run()` | current run |
| `bench.version()` / `bench.close()` | |
| `inst.id`, `.name`, `.model`, `.idn`, `.classes`, `.channels`, `.online`, `.parked`, `.health` | facts from the server |
| `inst.set(op, value, *, cls=None, ch=1)` | run a setter |
| `inst.query(op, *, cls=None, ch=1, raw=False)` | run a query on the instrument |
| `inst.read(field, *, cls=None, ch=1, raw=False)` / `inst.status(...)` | latest polled status |
| `inst.operations(cls=None)` / `inst.features(cls=None)` / `inst.refresh()` | description |
| `rec.data(start=None, end=None, max_points=None)` / `rec.to_pandas(...)` / `rec.export_csv()` | samples |
| `rec.add(label, signal)` / `rec.pause()` / `rec.resume()` / `rec.stop()` / `rec.delete()` / `rec.refresh()` | control |

`cls` is omitted when the instrument has one class. Class names are three letters (`PSU`, `DMM`,
`ELL`, `AWG`, `OSC`, `LCR`, `SAL`), any case.

### Errors

Every server error is a `BenchMeshError` with `.status`, `.detail` and `.body`:

| Exception | When |
|---|---|
| `ServerNotRunning` | nothing answers at the address: start the BenchMesh app |
| `BadRequest` (400, 422) | value outside the setter's range, invalid flow, ... |
| `AuthenticationError` (401, 403, sign-in) | wrong password, remote session not allowed |
| `NotFound` (404) | unknown instrument, class, operation, field, recording, flow |
| `Refused` (409) | parked instrument, a run is busy, flow needs `arm`, name taken |
| `InstrumentError` (502) | the instrument answered something unexpected |
| `InstrumentUnavailable` (503, 504) | not connected, parked or no answer in time |
| `TLSError`, `FingerprintMismatch` | remote certificate not trusted / not the pinned one |

Wrong arguments (class not on the instrument, channel 0, an empty name) raise `ValueError` or
`TypeError` before anything is sent.

## Safety

The API acts at once. `set("output", "ON")` switches a supply output on, `set("input", "ON")` a
load input; there is no second confirmation as in the app. Check wiring and limits before you
run a script, keep `try` / `finally` blocks that switch outputs off, and test new scripts with
the simulated bench (`benchmesh-server --simulate`). Automation flows are safer for unattended
runs: they need `arm`, confirm every setting and switch off what they switched on when they fail.

## Development

```bash
uv sync                                   # or: python -m venv .venv && pip install -e . pytest cryptography pandas
uv run pytest                             # unit tests (a fake server; no hardware)
uvx ruff check . && uvx ruff format --check .
```

Integration tests run against a real server and are skipped unless `BENCHMESH_INTEGRATION_URL`
is set; use a simulated server:

```bash
# in the BenchMesh repository
cargo build -p bm-server
BENCHMESH_ONLINE=off BENCHMESH_DATA_DIR=$(mktemp -d) ./target/debug/benchmesh-server \
    --config captures/bench-config.yaml --simulate captures/20261016-203514 --port 57690
# here
BENCHMESH_INTEGRATION_URL=http://127.0.0.1:57690 uv run pytest tests/integration
# also remote access over HTTPS (simulation only; switches it on at that port and off again)
BENCHMESH_INTEGRATION_URL=http://127.0.0.1:57690 BENCHMESH_INTEGRATION_REMOTE_PORT=57697 uv run pytest tests/integration
```

### Releasing

Bump `src/benchmesh/_version.py`, commit, tag `vX.Y.Z` and push the tag. The `release`
workflow builds and publishes to PyPI with trusted publishing. Once, on PyPI: add a trusted
publisher for project `benchmesh` with this repository, workflow `release.yml` and
environment `pypi` (Account settings - Publishing - Add a new pending publisher).

## Licence

MIT, see [LICENSE](LICENSE). BenchMesh: https://benchmesh.online
