TCP Server and Remote Client

Once a device class is implemented, exposing it over TCP is a single function call.

Starting the server

import asyncio
from plesty.power_meter import PowermeterDevice
from plesty.lib.service import build_server


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


asyncio.run(main())

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)

The client exposes the exact same interface as the local device:

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

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

# Get all parameter names
params = client.get_config_list()

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

# Get a snapshot of all parameter values
state = client.state

# Call a registered operation
result = client.run_operation("measure_power_sequence", count=10, delay_ms=50)

Connection lifecycle

The client manages connection automatically. To explicitly control the connection:

client = build_client(address="tcp://host:5555")
client.connect()

try:
    data = client.query("POWER")
finally:
    client.disconnect()

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 asyncio
import argparse
from plesty.power_meter import PowermeterDevice
from plesty.lib.service import build_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()

    asyncio.run(run())


if __name__ == "__main__":
    main()