Constraints and Units
Constraints tell a client which values an endpoint accepts or returns, and units tell it what those values mean. Where you declare a constraint depends on when its range is known:
| The range is known… | Declare it with |
|---|---|
| When you write the code, and never changes | typing.Annotated on the signature |
| Once the connector has queried the device | self.add_constraint(...) in on_before_start |
| Only at call time, and can change per call | A validator plus a documented range |
For why every value should be constrained, see Designing Data Types.
Units
Define each unit once, in io/units.py, and import it wherever you need it. Don't redefine units per feature.
from unitelabs.cdk import sila
RPM = sila.constraints.Unit(
label="RPM",
components=[sila.constraints.UnitComponent(sila.constraints.SIUnit.SECOND, -1)],
factor=1 / 60,
)
Any numerical value users give to or receive from the device should be annotated with a Unit constraint to give it concrete meaning. Annotate on the type in the signature, and leave units out of identifiers and docstrings.
import typing
from ...io.units import RPM
@sila.ObservableProperty()
async def subscribe_speed(self) -> sila.Stream[typing.Annotated[float, RPM]]:
"""The current shaking speed."""
Use the unit the device reports. See Units for how to handle devices that use fractional units.
Fixed constraints
When the range is known while you write the code and never changes, wrap the type in typing.Annotated:
@sila.UnobservableCommand()
async def create_folder(self, path: typing.Annotated[pathlib.Path, sila.constraints.MinimalLength(1)]) -> None:
"""
Create a new folder.
Args:
Path: The absolute path of the folder to create.
"""
Constraints apply to return values too. A temperature that can never go below 0 Kelvin says so:
@sila.ObservableProperty()
async def subscribe_current_temperature(
self,
) -> sila.Stream[typing.Annotated[float, KELVIN, sila.constraints.MinimalInclusive(0.0)]]:
"""The current chamber temperature."""
For the full list of constraints per data type, see SiLA Data Types.
Constraints that depend on the device
When the range depends on the connected device, query it at startup and add the constraint in on_before_start:
class ShakerController(sila.Feature):
@typing.override
async def on_before_start(self) -> None:
max_speed = await self.protocol.get_max_speed()
self.add_constraint(
self.start,
parameter="target_speed",
constraints=[sila.constraints.MaximalInclusive(max_speed)],
)
@sila.UnobservableCommand()
async def start(self, target_speed: typing.Annotated[int, RPM]) -> None:
"""
Start shaking at the given speed until `Stop` is called.
Args:
TargetSpeed: The shaking speed.
"""
After startup, clients see start with the device's real maximum. This only works if the protocol is connected before the server starts, which create_app already handles. See Connector Architecture.
Limits that change between calls
Some limits can't be pinned down even at startup, because they depend on device state that changes between calls. For those:
- State the valid range in the parameter's description, so the client knows why a value may be rejected.
- Register a validator that checks the value and raises a
ValidationErrornaming the parameter.
@sila.UnobservableCommand()
async def run_method(self, speed: typing.Annotated[float, RPM]) -> None:
"""
Run the method at the given speed.
Args:
Speed: The target speed. Must not exceed the device's current maximum speed.
"""
await self.protocol.run_method(speed)
@run_method.validator
async def _check_speed(self, speed: float) -> None:
max_speed = await self.protocol.get_max_speed()
if speed > max_speed:
msg = f"Speed {speed} exceeds the current maximum of {max_speed}."
raise sila.errors.ValidationError(msg, "speed")
See Validating in the Implementation for how validators run.
Last updated