Metadata-Version: 2.4
Name: lab-sync-driver-sdk
Version: 0.4.0
Summary: Public driver-authoring SDK for the device engine: subclass Device, decorate commands, serve() over stdio
Author: Lab Sync
License-Expression: MIT
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering
Classifier: Typing :: Typed
Requires-Python: >=3.14
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: lab-sync-driver-protocol==0.4.0
Provides-Extra: lab
Requires-Dist: lab-device-sdk>=0.1.12; extra == "lab"
Requires-Dist: fastapi>=0.110; extra == "lab"
Requires-Dist: uvicorn>=0.30; extra == "lab"
Dynamic: license-file

# lab-sync-driver-sdk

Write a lab device driver as a decorated Python class and serve it as a
supervised child process. A driver subclasses `Device`, decorates async
methods with `@command` and the typed in/out decorators, and calls
`serve(device)` to speak the stdio wire protocol defined by
[`lab-sync-driver-protocol`](https://pypi.org/project/lab-sync-driver-protocol/). The SDK depends on
nothing but that package, so a driver ships with its own runtime and never
imports the supervisor that launches it.

```bash
pip install lab-sync-driver-sdk
```

## A driver

```python
import asyncio

from lab_sync_driver_sdk import Device, command, float_in, string_in, string_out, serve


class Greeter(Device):
    device_type_id = "acme-greeter"
    manufacturer = "Acme"
    model = "Greeter 1"

    @command("Greet", "Say hello", default_sim_sec=1.0)
    @string_in("Name", "who to greet", default="world", maps_to="name")
    @float_in("Delay", "seconds to wait first", default=0.0, min_value=0.0, maps_to="delay")
    @string_out("Greeting", "what was said")
    async def greet(self, name: str = "world", delay: float = 0.0, progress=None) -> dict[str, object]:
        await asyncio.sleep(delay)
        if progress is not None:
            await progress(1.0, "greeted")
        return {"Greeting": f"Hello, {name}!"}


if __name__ == "__main__":
    asyncio.run(serve(Greeter()))
```

The decorators produce the driver's `describe` reply: its commands, their
inputs and outputs, and any optional metadata the class declares (`category`,
physical dimensions, `default_nests`, `configurations`, telemetry channels).
A command returns a dict, and every key in it must be declared with an output
decorator. An undeclared key fails the command with `OutputValidationError`.
A command that takes a `progress` parameter is handed an async
`progress(fraction, message)` reporter.

A device may define `async def abort(self, command_name: str)`. The host calls
it after cancelling a command's task, with the name of the command that was
cancelled -- so a driver waiting on a run can tell that wait from a
non-exclusive status read cancelled beside it, and stop the hardware only for
the one that owns it. The name is `"*"` only when the host does not know the
execution being aborted.

Once `serve` starts, stdout carries protocol frames only. The host points
`sys.stdout` at stderr, so a stray `print` lands on the log channel rather
than corrupting a frame.

## Moving parts in the 3D views

A device whose 3D model moves declares its joints in `articulation` (each names
a node in the GLB, or one of its animation clips; see the protocol's
`articulation` field). Where they are reaches the views as `articulation`
telemetry frames, only while a view is open.

A part only commands move -- a lid, a door, a stage -- needs no state of the
driver's own. Declare it on the commands and the SDK reports it:

```python
from typing import ClassVar

from lab_sync_driver_sdk import UNKNOWN, Device, articulates, bool_in, command, when_param

class Sealer(Device):
    articulation: ClassVar[dict[str, object]] = {
        "joints": [{"id": "stage", "label": "Stage", "kind": "clip", "clip": "Stage_In"}],
    }

    @command("Seal", "Take the stage in, seal, bring it out")
    @articulates("stage", during=1.0, after=0.0)
    async def seal(self) -> dict[str, object]: ...

    @command("Load", "Bring the stage in, and out again unless told to hold it")
    @bool_in("Hold", maps_to="hold")
    @articulates("stage", during=1.0, after=when_param("hold", truthy=1.0, falsy=0.0))
    async def load(self, *, hold: bool = False) -> dict[str, object]: ...

    @command("Send Command", "A raw line that may move anything")
    @articulates("stage", after=UNKNOWN)
    async def send_command(self, line: str = "") -> dict[str, object]: ...
```

`during` is reported as the command starts and `after` when it returns (left
out, the joint keeps `during`). A command that raises or is aborted leaves every
joint it declares unknown, so the views keep the last pose rather than being
told a guess -- a refusal before anything moved included, since the SDK cannot
tell the two apart. `when_param` picks by one of the method's own keywords,
after defaults; a branch left out leaves the joint as it is.

What a command cannot declare -- a position read from the hardware, the state
seeded at connect -- the driver writes with `self.set_joint(id, value)` and
`self.forget_joint(id)`, and reads back with `self.joint_value(id)`. One rule
orders every write: it carries the start order of the command it is made in (or
a fresh one outside any command), and is dropped if a command that started
later has already written that joint. A status read asked for before a move
cannot overwrite where the move is taking the part, and a slow command's
`after` cannot undo a newer command's `during`.

A device that declares `articulation` and no `@telemetry("articulation")`
sampler of its own gets that channel from the host, which pushes the known pose
when a view opens and each change as it is written. A device with a sampler
keeps it, and its readings are filled in with the joints the store knows (the
sample wins for any joint it reads). The declarations apply when a command runs
through the host or `run_command`; a command method called directly from
Python -- one command calling another, or a test -- is a plain method call.

## Testing a driver

`lab_sync_driver_sdk.testing.run_command` runs one command in-process, applying the
same output validation the host applies and the same `@articulates` reporting
(read it back with `device.joint_value(...)`):

```python
from lab_sync_driver_sdk.testing import run_command

result = await run_command(Greeter(), "Greet", {"name": "lab"})
assert result == {"Greeting": "Hello, lab!"}
```

## Connection settings

`@settings()` plus the typed input decorators declare the form an operator fills
in. Put them on `connect` itself, with one parameter per setting — this is the
standard shape, and every driver in this repository uses it:

```python
@settings()
@string_in("Host", "device address", default="10.0.0.1", maps_to="host")
@int_in("Port", "device port", default=4321, maps_to="port")
async def connect(self, host: str = "10.0.0.1", port: int = 4321) -> None: ...
```

The host binds each parameter by `maps_to` (else the display name). The value
comes from the saved setting, else the declared default, else the signature's
own default, and is coerced to the declared type, so `connect` receives an `int`
even when the value was saved as text. A setting with no matching parameter, or
a required parameter with no setting, fails the connect with a message naming
it. Saved settings are keyed by display name, so adding `maps_to` to an existing
driver keeps its devices' saved values. A driver that reconnects by itself (on a
simulation switch, say) keeps the keyword arguments of its last connect and
replays them. See `lab_sync_driver_sdk.settings_binding`.

The older shape — an empty `@settings()` carrier such as `connection_settings`
holding the form, beside `connect(self, settings)` taking the saved dict keyed
by display name — is still served, for drivers written before binding existed,
but new and updated drivers should not use it.

## Hosting a lab-device-sdk driver

`lab_sync_driver_sdk.lab_host` serves a driver written against
[`lab-device-sdk`](https://pypi.org/project/lab-device-sdk/) over the same
protocol, unchanged. It needs the `lab` extra, which also brings in FastAPI
and uvicorn for drivers that serve HTTP routes of their own:

```bash
pip install "lab-sync-driver-sdk[lab]"
```

It hosts one device per process, so it has to be told which instance it is
running. A driver manifest declares that with the supervisor's `{device_id}`
token rather than a hardcoded id, which is what lets a site run several
devices of the same type:

```toml
[launch]
type = "command"
command = ["python", "-m", "lab_sync_driver_sdk.lab_host",
           "--device", "my_pkg.my_device:MyDevice",
           "--device-id", "{device_id}"]
```

## License

MIT
