Base Device

BaseDeviceSyncModel is the core abstraction for building a synchronized device API in PlestyLib.

It combines:

  1. Device lifecycle (init, connect, disconnect)

  2. Parameter system (ConfigSystem)

  3. Function system (FunctionSystem)

  4. Standardized read/write/query workflow

  5. Resource ownership and context-manager cleanup

What It Provides

When your device class inherits from BaseDeviceSyncModel, you get:

  1. Parameter registration, validation, and typed parsing through ConfigSystem.

  2. Function registration and invocation metadata through FunctionSystem.

  3. Built-in high-level methods such as write, query, state, and summary.

  4. Context manager support (with device:) that calls init, connect, and disconnect.

  5. Resource tracking through ResourceRegistry to avoid duplicate ownership.

  6. Per-device resource management (register_resources, allocate_resource, …) for multi-client access control.

Typical Layering

A concrete device class usually composes three parts:

  1. Device class (inherits BaseDeviceSyncModel)

  2. Traffic manager (transport I/O)

  3. Command/operation solver (protocol translation)

Data flow:

  1. write("KEY", value) validates input against parameter metadata.

  2. Device _write_ builds protocol command and sends it via traffic manager.

  3. query("KEY") sends protocol query and converts response to target dtype.

Required Methods to Implement

Your subclass must implement these methods:

  1. connect(self)

  2. disconnect(self)

  3. _write_(self, key, value)

  4. _query_(self, key)

  5. check_errors(self)

  6. check_operatability(self)

Commonly overridden methods:

  1. init(self, main=None) to create traffic manager and solver objects.

  2. identity(self) for model/vendor identification command.

  3. query_param_range(self, key) if device can report runtime min/max.

  4. query_param_options(self, key) if categorical options are queryable.

Lifecycle

Recommended lifecycle:

  1. Instantiate device object.

  2. Enter context (with device:) or call init() and connect() manually.

  3. Perform query/write operations.

  4. Call disconnect() (automatic when using context manager).

Using context manager is recommended because cleanup is guaranteed:

with MyDevice("dev-id") as dev:
	print(dev.query("POWER"))

Minimal Implementation Example

from typing import Any

from plesty.lib.device.base_device_sync import BaseDeviceSyncModel
from plesty.lib.traffic.serial import SerialTrafficManager
from plesty.lib.solver.scpi import SCPISolver


class MyScpiSerialDevice(BaseDeviceSyncModel):
	def __init__(self, port: str):
		super().__init__(id=port)
		self.port = port

		# Register parameters for validation + parsing.
		self.register_config("POWER", dtype=float, read_only=True, command="MEAS:POW")
		self.register_config("WAVELENGTH", dtype=int, command="SENS:WAV")

	def init(self, main=None):
		self.traffic_manager = SerialTrafficManager(
			port=self.port,
			baudrate=9600,
			timeout=5,
			write_termination="\r",
			read_termination="\r\n",
		)
		self.cmd_solver = SCPISolver()

	def connect(self):
		return self.traffic_manager.open(parity="none", stopbits="one", bytesize=8)

	def disconnect(self):
		self.traffic_manager.close()

	def _write_(self, key: str, value: str | float | int | bool) -> bool:
		cfg = self.get_config(key)
		command = self.cmd_solver.get_write_cmd(cfg, value)
		return bool(self.traffic_manager.send_command(command))

	def _query_(self, key: str) -> str:
		cfg = self.get_config(key)
		command = self.cmd_solver.get_query_cmd(cfg)
		return self.traffic_manager.send_command(command)

	def check_errors(self) -> list[str]:
		# Replace with device-specific error query if available.
		return []

	def check_operatability(self) -> bool:
		return self.traffic_manager is not None and self.traffic_manager.is_open

Ready-Made Base Devices

Before subclassing BaseDeviceSyncModel directly, check whether one of the shipped intermediates already covers your transport. Each wires up a traffic manager and implements connect, disconnect, _write_, _query_ and check_operatability, so a module only declares its schemas and its device-specific operations.

Class

Location

For

BaseVisaScpiDevice

plesty.lib.device.base_visa_scpi_device

Any SCPI instrument reachable over VISA (USB-TMC, GPIB, serial-over-VISA).

BaseTCPScpiDevice

plesty.lib.device.base_tcp_scpi_device

A SCPI instrument that listens on a raw TCP socket rather than through VISA.

BaseAptDevice

plesty.lib.device.base_apt_device

Single-channel Thorlabs APT motion controllers — binary frames rather than text.

 
from plesty.lib.device.base_visa_scpi_device import BaseVisaScpiDevice


class MyPowerMeter(BaseVisaScpiDevice):
	def __init__(self, address: str):
		super().__init__(
			address,
			timeout=5000,
			param_schema="assets/param_schema.json",
			op_schema="assets/op_schema.json",
		)

Both SCPI bases take write_termination / read_termination (and an optional baud_rate for serial-backed resources), pair themselves with SCPISolver, and add query_range, identity, check_errors and reset on top of the standard device API.

BaseAptDevice is the odd one out because APT is not a text protocol. It takes a serial port, an AptAxis destination and a channel, and exposes the exchange as messages rather than strings: command(...) sends one frame, request(...) sends one and waits for its reply, hw_info() and get_status() read the controller. Motion is handled in the controller’s own count domain — home(), move_absolute_counts(), move_relative_counts(), stop(), get_position_counts(), get_velocity_params() / set_velocity_params() — and the subclass converts counts to physical units in its user-facing operations. Parameter keys are registered with register_apt_param() so the config system can read and write them like any other. Passing transport= a plesty.lib.sim.apt.AptMotorSimulator runs the whole class without a motor; see Simulation.

State and Synchronization Helpers

Useful built-in helpers:

  1. state / get_state(): query all registered parameters and return a dictionary.

  2. synchronize_param_from_device(keys=None, sync_constraints=False): pull current values from hardware into the model.

  3. summary(style=...): render a textual API overview (short, md, or google).

  4. doc_model(): return the structured documentation model (see below).

If your device supports querying dynamic constraints, implement:

  1. query_param_range for numeric parameters.

  2. query_param_options for categorical parameters.

Sync vs Async Use

BaseDeviceSyncModel is synchronous by design. To use it in async applications, wrap it with one of the provided wrappers.

  1. AsyncWrapperSafe: thread-offloaded calls protected by an async lock (simple, safe default).

  2. AsyncDeviceThread: dedicated worker thread with queued calls (better for high-frequency workflows).

Example with AsyncWrapperSafe:

import asyncio
from plesty.lib.device.async_wrapper import AsyncWrapperSafe


async def main():
	async with AsyncWrapperSafe(MyScpiSerialDevice("/dev/ttyUSB0")) as dev:
		power = await dev.query("POWER")
		print(power)


asyncio.run(main())

Multi-Client Resource Control

BaseDeviceSyncModel embeds resource tracking that supports exclusive access control when a device is shared by multiple clients (for example, over a TCP/IP service):

Member

Type

Purpose

_resources

list

Inventory of physical resources this device owns; passed to the TCP server for per-client access control

_resource_manager

ResourceManager | None

Embedded manager for direct (non-TCP) clients, created lazily by register_resources()

Four public methods delegate to the embedded ResourceManager (plesty.lib.service.resource_manager):

  1. register_resources(resources): declare the resources the device owns (flat list, set, or nested dict tree).

  2. allocate_resource(resources, client_id): request exclusive access; returns False if any requested resource is already held by another client.

  3. release_resource(client_id): free every resource held by that client.

  4. client_resources(client_id): canonical resource paths currently allocated to that client.

The default _user_permission_check always returns True (no restriction). Override it to implement real guard logic — query current allocations via client_resources or the embedded manager:

def _user_permission_check(self, user_resources: list, key: str) -> bool:
    if not user_resources or self._resource_manager is None:
        return True
    locked = self._resource_manager._locked_resources
    return any(r in locked for r in user_resources)

Both write and query pass a resources kwarg through to _user_permission_check when the caller supplies it, so access control integrates transparently with the standard operation interface.

Data Path

Bulk data never travels over the TCP channel: a device writes acquisitions to a shared disk and returns only the file path. Which directory it writes to is steerable by the client, so an experiment can route a whole run into its own session directory.

The lab’s shared disk — the Plesty share — is mounted under a different path on every machine (/plesty, /Volumes/plesty, Z:\xyz\plesty), and each bench may write only in its own folder of it (platform-a). PLESTY Bench tells every device it starts all three through the environment; a device module passes a constructor argument only to override one:

Argument

Environment

Purpose

data_path

DEVICE_DATA_PATH

Default directory, in effect until a client sets one

allowed_data_roots

DEVICE_DATA_ROOT

Where client paths are taken from and must stay under — the bench’s folder; defaults to [data_path]

—

PLESTY_DATA

The share as this machine mounts it; a routed folder is reported relative to it

 
class MyCamera(BaseDeviceSyncModel):
	def __init__(self):
		super().__init__(id="cam")   # the bench's DEVICE_DATA_* and PLESTY_DATA

	def acquire(self):
		target = Path(self.get_data_path()) / "frame-001.tiff"
		...
		return str(target)  # only the path goes back over TCP

Device code that writes acquisition files must read the directory from self.get_data_path() rather than from its own configuration — otherwise a client-set path has no effect.

set_data_path(path) takes a path relative to the bench’s folder, at any depth, resolves it (.. collapsed, symlinks followed), checks it stays inside, creates it if missing, and returns it relative to the Plesty share — the spelling every machine reads back by joining it to its own PLESTY_DATA. An absolute path is refused, since no other machine could read it:

# PLESTY_DATA=Z:\xyz\plesty, DEVICE_DATA_ROOT=Z:\xyz\plesty\platform-a
client.set_data_path("pol-pl/run-3/raw/cam")  # -> 'platform-a/pol-pl/run-3/raw/cam'
client.set_data_path("../platform-b")         # ValueError: outside the allowed roots
client.set_data_path("D:/data")               # ValueError: must be relative to the data root

Without PLESTY_DATA the folder comes back relative to the bench’s folder. get_data_path() is the absolute folder on this machine, for the device’s own writes.

Both methods are @expose_to_api, so the TCP server lists them in describe and the client binds them automatically — a device module inherits the behaviour with no code of its own. Constructing a device still creates no directories, keeping documentation tooling able to instantiate devices offline.

All clients of one device server share a single data path: the last set_data_path wins, matching the single-operator resource-lock model. A CompositeDevice fans the call out to every sub-device (set_data_path(path, subdirs=True) gives each one its own <path>/<name> directory).

Documentation Model (doc_model)

Every device exposes a structured, render-agnostic documentation model — the single source of truth for its user-facing documentation:

model = device.doc_model()
model.parameters         # {group_name: [ConfigParameter, ...]} from the parameter system
model.functions          # the device's own operations as unified FuncDoc objects
model.standard_methods   # the common device API (connect/write/query/reset/...)

functions unifies the two ways a device gains operations: schema-registered operations (source="schema") and @expose_to_api methods (source="custom"); framework plumbing is kept out and surfaced separately under standard_methods (source="standard"). Each FuncDoc also carries its kind — what the operation does to the world (read, motion, acquire, lifecycle, control, configure), declared with @expose_to_api(kind=..., position_key=..., step=...) or a "kind" block in schema_func.json and used by the field test to drive the operation (see test_helper, “Kinds”). Per-parameter and return descriptions are parsed from Google-style docstring Args:/Returns: sections — write those in your device methods and the generated documentation inherits them.

Renderers consume the model instead of re-deriving metadata: the SDK docs generator, summary(), and the server describe endpoint all build on it. Because constructing a device is side-effect-free (the per-device resource lock is only acquired on connect), documentation tooling can instantiate your device without touching hardware.

Best Practices

  1. Keep transport logic in traffic managers, not in device classes.

  2. Keep protocol formatting/parsing in solvers.

  3. Register all parameters with correct dtype/range/options for safe validation.

  4. Prefer with device: for guaranteed cleanup.

  5. Implement check_operatability using a real device health query when possible.

  6. Avoid mixing direct synchronous access and async wrapper access on the same device instance.