Serve It

A device is never imported by the programs that use it: it runs as a device server that holds the instrument and listens on a port, and everything else is a client (see Networking). The scaffold's __main__.py is that server already:

uv run python -m plesty.powermeter --tcp-port 5555 --mock    # --mock: the module's simulator
uv run python -m plesty.powermeter --gui --mock              # its control panel, for trying it by hand

Talk to it as a client

A client needs nothing but the address. Anything the device exposes becomes a method on it:

from plesty.lib.service import build_client

with build_client("tcp://localhost:5555") as pm:
    print("identity:", pm.identity())
    print("operations:", ", ".join(sorted(pm.describe()["methods"])))

    reading = pm.measure_power(averaging=10)          # an operation: something it does
    print(f"power: {reading['power'] * 1e6:.3f} uW")

    pm.write("wavelength", 780.0)                     # a parameter: something it holds
    print("wavelength:", pm.query("wavelength"), "nm")

identity: THORLABS,PM100D,MOCK0001,1.0.0
operations: check_errors, check_operatability, clear, configure_and_measure_power, doc_model,
  get_data_path, get_sensor_info, identity, measure_current, measure_energy, measure_frequency,
  measure_power, measure_temperature, measure_voltage, query, reset, set_data_path,
  synchronize_param_from_device, write, zero_and_measure_power, zero_sensor
power: 1.471 uW
wavelength: 780.0 nm

The server's log shows the other side — what each call became on the wire:

That translation — SCPI, units, the order commands must come in — is the whole job of a device module, and what no client ever has to know.

Starting the server

from plesty.power_meter import PowermeterDevice
from plesty.lib.service import build_server, run_server


async def main() -> None:
    device = PowermeterDevice(
        address="USB0::0x1313::0x8078::P0000001::INSTR",
        sensor_type="S155C",
    )
    server = build_server(device, address="tcp://*:5555")
    await server.run()


run_server(main())

run_server takes the place of asyncio.run: it starts the loop on an event-loop type the network library (pyzmq) can use on every platform, which a bare asyncio.run will stop doing on Windows from Python 3.16.

The generated __main__.py in the scaffold already includes this pattern. Run it with:

python -m plesty.power_meter
# or
uv run python -m plesty.power_meter

build_server parameters

Connecting a client

From any machine that can reach the server:

from plesty.lib.service import build_client

client = build_client(address="tcp://192.168.1.10:5555", timeout=5000)

On connecting, the client asks the server to describe the device and binds every operation the device exposes as a method of its own, so it reads like the local device:

# Query a parameter
wavelength = client.query("WAVELENGTH")

# Write a parameter
client.write("WAVELENGTH", 1064)

# Get device identity string
idn = client.identity()

# Get a snapshot of all parameter values — an operation, so it is called
state = client.state()

# Call an exposed operation by its name…
result = client.measure_power_sequence(count=10, delay_ms=50)

# …or through the one call every method goes through
result = client.call("measure_power_sequence", count=10, delay_ms=50)

client.describe() returns the whole description — every parameter and operation, with their documentation and time budgets — which is what a tool that lists a device's operations reads.

Connection lifecycle

The client connects when it is built. Close it when you are done, so the server releases what this client held — or let a with block do it:

with build_client(address="tcp://host:5555") as client:
    data = client.query("POWER")

client.close() is the same by hand. Do not call client.disconnect() for this: disconnect is one of the device's own operations, forwarded to the server, and it closes the instrument.

Resource locking (shared devices)

When a device is shared by multiple clients, use the resources argument to request exclusive access to specific hardware channels:

# Experiment A locks two lines
client_a = build_client(
    address="tcp://daq-host:5555",
    resources=["Dev1/port0/line0", "Dev1/port0/line1"],
    client_id="experiment-A",
)

# Experiment B tries to lock the same line — will be rejected
client_b = build_client(
    address="tcp://daq-host:5555",
    resources=["Dev1/port0/line0"],  # conflict with client_a
    client_id="experiment-B",
)

The server enforces exclusive access. Client B receives a rejection until client A disconnects and releases its resources.

Running server at startup

For persistent instrument servers, write a __main__.py that the systemd service or Docker container launches:

# plesty/power_meter/__main__.py
import argparse
from plesty.power_meter import PowermeterDevice
from plesty.lib.service import build_server, run_server


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--address", default="USB0::0x1313::0x8078::P0000001::INSTR")
    parser.add_argument("--sensor", default="S155C")
    parser.add_argument("--port", default=5555, type=int)
    args = parser.parse_args()

    async def run() -> None:
        device = PowermeterDevice(args.address, sensor_type=args.sensor)
        server = build_server(device, address=f"tcp://*:{args.port}")
        await server.run()

    run_server(run())


if __name__ == "__main__":
    main()