The BaseDeviceSyncModel Contract

Every PLESTY device inherits from BaseDeviceSyncModel (from plesty.lib.device.base_device_sync). It defines a lifecycle, a parameter system, and a query/write interface.

Required methods

Your subclass must implement these six methods:

Method	Signature	What it must do
`connect`	`(self) -> None`	Open the connection to the instrument
`disconnect`	`(self) -> None`	Close the connection cleanly
`_write_`	`(self, key: str, value: str \| float \| int \| bool) -> bool`	Send a write command; return `True` on success
`_query_`	`(self, key: str) -> str`	Send a query command; return raw string response
`check_errors`	`(self) -> list[str]`	Query device error register; return `[]` if healthy
`check_operatability`	`(self) -> bool`	Return `True` if the device is ready to operate

Optional overrides

These are implemented in the base class but you can override them:

Method	Default behavior	Why override
`init(self, main=None)`	No-op	Create traffic manager and solver objects here
`identity(self) -> str`	Returns `""`	Return vendor/model string from `*IDN?` or equivalent
`query_param_range(self, key) -> tuple`	Returns `(None, None)`	If the device can report runtime min/max
`query_param_options(self, key) -> list`	Returns `[]`	If the device can report categorical options

What the base class provides

You do not need to implement:

write(key, value) — validates input, calls _write_, updates cache
query(key) — validates access, calls _query_, parses and caches response
state / get_state() — queries all registered parameters, returns dict
summary() — generates a human-readable API overview
Context manager (with device:) — calls init(), connect(), disconnect()
synchronize_param_from_device() — pulls all current values from hardware into the model
is_operatable property — delegates to check_operatability()
Async wrappers (AsyncWrapperSafe, AsyncDeviceThread) — wraps the sync device for async use
TCP/IP server (build_server) — exposes the device over ZMQ

Lifecycle

# Recommended: use context manager for guaranteed cleanup
with MyDevice("resource-id") as dev:
    dev.write("WAVELENGTH", 1064)
    power = dev.query("POWER")

# Manual:
dev = MyDevice("resource-id")
dev.init()
dev.connect()
try:
    dev.write("WAVELENGTH", 1064)
    power = dev.query("POWER")
finally:
    dev.disconnect()

Typical class structure

from plesty.lib.device.base_device_sync import BaseDeviceSyncModel
from plesty.lib.traffic.visa import VisaTrafficManager
from plesty.lib.solver.scpi import SCPISolver


class PowermeterDevice(BaseDeviceSyncModel):
    def __init__(self, address: str, sensor_type: str):
        super().__init__(id=address)
        self._address = address
        self._sensor_type = sensor_type

        # Register all parameters in __init__
        self.register_config("POWER", dtype=float, unit="watt", read_only=True,
                             command="MEAS:SCAL:POW")
        self.register_config("WAVELENGTH", dtype=int, unit="nm",
                             min_value=400, max_value=1700, command="SENS:CORR:WAV")

    def init(self, main=None) -> None:
        self.tm = VisaTrafficManager(self._address)
        self.solver = SCPISolver()

    def connect(self) -> None:
        self.tm.open()

    def disconnect(self) -> None:
        self.tm.close()

    def _write_(self, key: str, value: str | float | int | bool) -> bool:
        cfg = self.get_config(key)
        cmd = self.solver.get_write_cmd(cfg, value)
        return bool(self.tm.send_command(cmd))

    def _query_(self, key: str) -> str:
        cfg = self.get_config(key)
        cmd = self.solver.get_query_cmd(cfg)
        return self.tm.send_command(cmd)

    def check_errors(self) -> list[str]:
        response = self.tm.send_command("SYST:ERR?")
        if response.startswith("+0"):
            return []
        return [response]

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

    def identity(self) -> str:
        return self.tm.send_command("*IDN?")

Resource tracking for shared devices

If a device has lockable resources (e.g., a multi-channel DAQ), declare them in _resources:

def __init__(self, device_id: str):
    super().__init__(id=device_id)
    self._resources = ["Dev1/port0/line0", "Dev1/port0/line1"]

The TCP server reads _resources automatically and enforces exclusive access per client.