Composite Device

Combine multiple devices into a single interface using CompositeDevice — with automatic reconnection and retry logic.

A CompositeDevice wraps multiple sub-devices into a single interface. This is useful when an experiment needs to control several instruments simultaneously.

Import

from plesty.lib.composite_device import CompositeDevice

Constructor

composite = CompositeDevice(
    devices={
        "power": power_supply,
        "laser": laser_device,
        "meter": power_meter,
    },
    addresses={
        "power": "tcp://localhost:5555",
        "laser": "tcp://localhost:5556",
        "meter": "tcp://localhost:5557",
    },
)

Parameters

Parameter Type Description
devices dict Mapping of sub-device name to connected device instance
addresses dict[str, str] Optional mapping of sub-device name to ZMQ address

Calling sub-device methods

Use the call() method to invoke a function on a sub-device:

result = composite.call(
    dev="power",
    func="_write_",
    retries=3,
    sleep=0.0,
    reconnect_wait_s=2.0,
    key="voltage",
    value=5.0,
)

Parameters

Parameter Type Description
dev str Sub-device name (key in the devices dict)
func str Method name to call
*args tuple Positional arguments for the method
retries int Number of retry attempts on failure (default: 3)
sleep float Seconds to sleep between retries (default: 0.0)
reconnect_wait_s float Seconds to wait before reconnecting (default: 2.0)
**kwargs dict Keyword arguments for the method

Reconnection

If a sub-device connection fails, use reconnect() to rebuild the client:

composite.reconnect("power")

This creates a new client connection to the sub-device's address.

Example

from plesty.lib.composite_device import CompositeDevice
from plesty.lib.service import build_client

# Create clients for each device
power_client = build_client(address="tcp://localhost:5555")
laser_client = build_client(address="tcp://localhost:5556")

# Build composite
composite = CompositeDevice(
    devices={
        "power": power_client,
        "laser": laser_client,
    },
)

# Set laser power
composite.call("laser", "_write_", key="power", value=50.0)

# Read voltage from power supply
voltage = composite.call("power", "_query_", key="voltage")

Next steps