With the base class and schemas defined, it's time to write the actual device implementation. This connects the abstract API to real hardware communication.
Full example
Here's a complete device implementation:
from plesty.lib.device import BaseDeviceSyncModel
from plesty.lib.schema import ParameterSchema, OperationSchema
class PowerSupply(BaseDeviceSyncModel):
"""A simple power supply device."""
def __init__(self, main=None) -> None:
super().__init__(main=main)
self._connected = False
self._voltage = 0.0
self._current = 0.0
self._output = False
# Define parameter schemas
self._params = {
"voltage": ParameterSchema(
key="voltage",
unit="V",
dtype="float",
default=0.0,
min_value=0.0,
max_value=30.0,
description="Output voltage setpoint",
),
"current": ParameterSchema(
key="current",
unit="A",
dtype="float",
default=0.0,
min_value=0.0,
max_value=5.0,
description="Current limit",
),
"output": ParameterSchema(
key="output",
dtype="bool",
default=False,
description="Output enable",
),
}
# Define operation schemas
self._ops = {
"calibrate": OperationSchema(
key="calibrate",
description="Run calibration routine",
),
}
def connect(self) -> bool:
"""Connect to the power supply hardware."""
try:
# Hardware connection logic here
self._connected = True
return True
except ConnectionError:
return False
def disconnect(self) -> None:
"""Disconnect from the power supply."""
self._connected = False
def _write_(self, key: str, value: object) -> None:
"""Write a parameter value to the hardware."""
if key == "voltage":
self._voltage = float(value)
# Send to hardware: set_voltage(value)
elif key == "current":
self._current = float(value)
# Send to hardware: set_current(value)
elif key == "output":
self._output = bool(value)
# Send to hardware: set_output(value)
else:
raise ValueError(f"Unknown parameter: {key}")
def _query_(self, key: str) -> object:
"""Read a parameter value from the hardware."""
if key == "voltage":
return self._voltage
elif key == "current":
return self._current
elif key == "output":
return self._output
else:
raise ValueError(f"Unknown parameter: {key}")
def identity(self) -> str:
"""Return device identity."""
return "PowerSupply v1.0"
def check_errors(self) -> list[str]:
"""Check for hardware errors."""
errors = []
if self._voltage > 30.0:
errors.append("Voltage exceeds maximum")
return errors
def check_operatability(self) -> bool:
"""Check if the device is ready."""
return self._connected
Key patterns
State management
Store device state in instance variables. The _write_ and _query_ methods read and write these variables, which in a real device would communicate with hardware.
Error handling
Use check_errors() to report hardware issues. Return an empty list when everything is healthy — this is required by gate d1.
Type safety
Use type hints on all methods. The SDK checks return type annotations in gate 4.
Next steps
- Set up the TCP server to expose your device over the network
- Pass gate d1 — the device API pipeline