pytweezer.servers.device_client module

Generic device RPC-client factory.

Every device in CONFIG["Devices"] is served by a sipyco RPC server. Rather than hand-writing a client class per device, get_device() looks the device up by its config name, resolves the endpoint that serves it, and returns a transparent sipyco proxy:

cam = get_device("Rb HamCam")
cam.acquire()                 # remote method call, transparently proxied
cam.close_rpc()               # release the socket when done

Every device is addressed by its own name, whether it has a server to itself or shares one with other devices. A composite device (an entry with a "devices" sub-dict) runs several devices in one process so they can drive each other without RPC, but its sub-devices are named in config exactly like any other device and reached the same way — the caller never needs to know which process a device lives in, or that sipyco targets exist:

cam = get_device("Rb Feedback Cam")     # sub-device of the "Rb Feedback Rig" composite
dac = get_device("Rb Feedback DAC")     # ...same process, same port, different target
rig = get_device("Rb Feedback Rig")     # the composite itself -> its coordinator

target_name is resolved from the config and only needs passing to reach a target the config doesn’t name.

For driving two or more device servers concurrently (e.g. starting two MotMaster sequences in parallel), prefer pytweezer.parallel.run_parallel() — it runs blocking get_device() clients in threads with no async/await and works from the GUI. get_device_async() is the lower-level asyncio counterpart if you want to drive the servers with coroutines directly:

async def main():
    mm1 = await get_device_async("Rb MotMaster Server")
    mm2 = await get_device_async("CaF MotMaster Server")
    try:
        await asyncio.gather(
            mm1.start_motmaster_experiment(),
            mm2.start_motmaster_experiment(),
        )
    finally:
        await mm1.close_rpc()
        await mm2.close_rpc()

asyncio.run(main())

pytweezer.servers.device_client.get_device(name, host=None, port=None, target_name=<class 'sipyco.pc_rpc.AutoTarget'>, timeout=None)

Build a sipyco RPC client for the device named name in the config.

Parameters:

• name – Any device name: a top-level CONFIG["Devices"] key ("Rb HamCam"), a composite’s sub-device ("Rb Feedback Cam"), or a composite itself (which resolves to its coordinator).

• host – Optional overrides; default to the serving process’s host/port.

• port – Optional overrides; default to the serving process’s host/port.

• target_name – RPC target to bind. Defaults to AutoTarget, which selects the right target automatically — the server’s sole target for a plain device, or the one belonging to name on a composite. Override only to reach a target the config doesn’t name.

• timeout – Socket timeout in seconds (None = block indefinitely).

Returns:

A transparent proxy; remote methods are called as normal attributes. Call .close_rpc() when finished to release the socket.

Return type:

sipyco.pc_rpc.Client

async pytweezer.servers.device_client.get_device_async(name, host=None, port=None, target_name=<class 'sipyco.pc_rpc.AutoTarget'>)

Build an AsyncioClient for the device named name.

Same config lookup as get_device(), but every RPC method on the returned client is a coroutine. Use this (with asyncio.gather) to drive two independent device servers concurrently — e.g. starting two MotMaster sequences in parallel — since a blocking get_device() client waits for the full RPC round trip, including the remote call’s execution time, before the next line runs.

Call await client.close_rpc() when finished to release the socket.

pytweezer.servers.device_client.get_device_config(name)

Return the config entry for the device named name.

Matches leniently (whitespace-/case-insensitively), and finds composite sub-devices as well as top-level ones. Raises KeyError listing the known devices if nothing matches.

A sub-device’s entry carries only its own driver settings; host/port belong to the composite that serves it. Use get_device() (or device_server.resolve_address) rather than reading them off this dict.