Driver API¶
What a driver package implements and how LabNet finds it. See Writing a driver for the walkthrough.
The physical device¶
protocol ¶
DeviceAvailability ¶
Bases: str, Enum
Standardized control-readiness reported by a physical device.
AvailabilityAwareDevice ¶
Bases: Protocol
Optional extension for devices that explicitly report availability.
availability ¶
availability() -> DeviceAvailability
Return whether control operations may currently be attempted.
PhysicalDevice ¶
Bases: Protocol
Runtime contract required by the Lab PC device manager.
With the exception of the immutable device_id used to build the
catalog, the device manager invokes this contract exclusively on the
device's dedicated worker thread. Hardware wrappers may therefore keep
thread-affine driver sessions without adding their own executors.
status ¶
status() -> str
Return a human-readable transport-neutral status string.
New physical drivers should additionally implement
:class:AvailabilityAwareDevice; status text is retained for display
and diagnostics rather than used as the primary health contract.
allowed_methods ¶
allowed_methods() -> Collection[str]
Return the physical methods that may be remotely invoked.
Site limits¶
limits ¶
Site limits: per-device limits a Lab PC's inventory sets, enforced on that Lab PC.
A driver may publish hints in metadata()["hints"] (its own hard ranges and
choices). An inventory can narrow them for one instrument on one bench::
from labnet.devices.core import site_limits
def create_devices():
return [
site_limits(
PhysicalSimulatedTunableLaser(device_id="lab01-laser"),
limits={"go_to_wavelength_nm.wavelength_nm": (770.0, 790.0)},
choices={"move_steps.velocity_mode": ["slow"]},
),
]
Keys are <method>.<argument>, exactly as hints and Python clients name them
(set_frequency.value, stream_data.channel). :func:site_limits returns
the same device object, so its type and contract are unchanged; the limits ride
on it as labnet_site_limits.
The Lab PC then:
- refuses to start if a key names no method or argument of the device's
contract, a range is inverted or not finite, a choice has the wrong type
(:func:
site_limitsitself, when the inventory runs), or a site limit is not inside the driver's own published limit or choices (when the device has connected and published its hints): site limits may only narrow; - refuses, before the driver is called, every remote call whose argument is
outside the effective range or choices (the driver's and the site's together),
so published hints are never merely advisory (:meth:
DeviceLimits.check); - publishes the effective values in
metadata()["hints"], withhints["sources"][key]="site"or"driver".
Limits apply to the physical method a call reaches. Two RPCs of one physical
method (read_value and stream_data both call read_value) share their
limits: a site choice for read_value.channel also limits stream_data's
channel, and both keys are published. A stream's own controls
(sample_rate_hz, sample_count) are not arguments and are not limited
here.
site_limits ¶
site_limits(device: D, *, limits: Mapping[str, Sequence[float]] | None = None, choices: Mapping[str, Sequence[Any]] | None = None) -> D
Give device site limits and return the same device.
limits maps "<method>.<argument>" of a number argument to
(min, max); choices maps one of a text or true/false argument to
the values allowed. Raises :class:DeviceLimitsError (a ValueError)
for anything that can't be a limit of this device's contract. Whether the
limits lie inside the driver's own is checked when the device connects.
The contract registry¶
registry ¶
Where LabNet code finds device contracts: by device type, service or RPC.
Driver packages register their contracts in the entry-point group
labnet.devices, one entry per device type::
[project.entry-points."labnet.devices"]
dummy_aom = "labnet_sim._generated.contracts:DUMMY_AOM_CONTRACT"
:func:get imports only the package that provides the requested type;
:func:all, :func:by_service and :func:rpc import every one. Two packages
declaring one type, or one that fails to import, become :func:problems, and
:func:get for that type raises saying so.
all shadows the builtin inside this module; callers write registry.all().
UnknownDeviceType ¶
UnknownDeviceType(device_type: object, reason: str = '')
Bases: KeyError
No contract is registered for a device type.
RegistryProblem
dataclass
¶
RegistryProblem(device_type: str, source: str, message: str)
A contract that could not be registered, and why.
DriverInfo
dataclass
¶
DriverInfo(package: str, version: str, fingerprint: str)
Which installed package provides a device type: what a Lab PC reports, and a Control Center compares.
get ¶
get(device_type: str) -> DeviceContract
The contract for device_type; :class:UnknownDeviceType (a KeyError) if none.
find ¶
find(device_type: str | None) -> DeviceContract | None
The contract for device_type, or None if there is none (or no type).
by_service ¶
by_service(service_full_name: str) -> DeviceContract | None
The contract serving <package>.<Name>Service, or None.
problems ¶
problems() -> tuple[RegistryProblem, ...]
Contracts that could not be registered: conflicting or failing packages.
driver_info ¶
driver_info(device_type: str) -> DriverInfo | None
The package, version and fingerprint behind device_type here, or None.
refresh ¶
refresh() -> None
Forget what was loaded: look at the installed packages again on the next lookup.
override ¶
override(contracts: Iterable[DeviceContract]) -> Iterator[None]
Use exactly contracts inside the block (for tests).