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.

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 plestylib.device.base_device_sync import BaseDeviceSyncModel
from plestylib.traffic.serial import SerialTrafficManager
from plestylib.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

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(): generate a textual API overview.

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 plestylib.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())

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.