everything does serialize and deserialize to json! Resources Introduction — PyLabRobot documentation
the middleware could ask either, but the client has to tell the server at some point
I’m more of saying that source of truth isn’t a data file but is in code - so to keep up to date, you have to run code. The save function also doesn’t have schema-level guarantees, so any system depending on it that isn’t PLR might get out of date because there is no schema associated.
(that might actually be a useful pr, plus associated github action)
1 Like
Had a call with Rick, here is the idea right now:
- The resource backend is actually separate from the specific robotic backend. We’re not gonna figure that out right now, just focus on a specific robotic backend
- We have two options: do arcane python bullshit to generate an API by introspecting the kwargs of something like the STAR backend OR just build a 1-to-1 translation with pydantic.
I’m gonna play with the arcane stuff, but worst comes to worst, we just go deep on AI generating the backend.
ok, some testing
It doesn’t work very well to do the type introspection. For whatever reason, pydantic doesn’t really like recursing down the type tree. It is almost always the ops bit of the code, like in ops: List[SingleChannelAspiration],, because these reference a whole bunch of resource classes:
@dataclass(frozen=True)
class SingleChannelAspiration:
resource: Container
offset: Coordinate
tip: Tip
volume: float
flow_rate: Optional[float]
liquid_height: Optional[float]
blow_out_air_volume: Optional[float]
liquids: List[Tuple[Optional[Liquid], float]]
That said, it seems like we could just define the dataclass as a pydantic data class -
from pydantic.dataclasses import dataclass
@dataclass # Just add this decorator
class SingleChannelAspiration:
volume: float
location: str
Or inherit the BaseModel:
from pydantic import BaseModel
class SingleChannelAspiration(BaseModel):
volume: float
location: str
# ... other fields
Here is some test code that works just fine:
"""FastAPI prototype for STAR backend aspirate command."""
from fastapi import FastAPI
from aspirate_models import AspirateRequest
app = FastAPI()
@app.post("/aspirate")
async def aspirate_endpoint(req: AspirateRequest):
"""Test endpoint for aspirate with BaseModel ops."""
# In a real implementation, you would convert the BaseModel ops
# back to SingleChannelAspiration objects and call backend.aspirate()
return {
"status": "success",
"ops_count": len(req.ops),
"use_channels": req.use_channels
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
"""Pydantic models for STAR backend aspirate command."""
from typing import List, Optional
from pydantic import BaseModel
from enum import IntEnum
class LLDMode(IntEnum):
OFF = 0
GAMMA = 1
dP = 2
class SingleChannelAspirationModel(BaseModel):
"""BaseModel version of SingleChannelAspiration for testing."""
resource_name: str
offset_x: float
offset_y: float
offset_z: float
tip_id: str
volume: float
flow_rate: Optional[float] = None
liquid_height: Optional[float] = None
blow_out_air_volume: Optional[float] = None
class AspirateRequest(BaseModel):
"""Request model for aspirate endpoint."""
ops: List[SingleChannelAspirationModel]
use_channels: List[int]
jet: Optional[List[bool]] = None
blow_out: Optional[List[bool]] = None
lld_search_height: Optional[List[float]] = None
clot_detection_height: Optional[List[float]] = None
pull_out_distance_transport_air: Optional[List[float]] = None
second_section_height: Optional[List[float]] = None
second_section_ratio: Optional[List[float]] = None
minimum_height: Optional[List[float]] = None
immersion_depth: Optional[List[float]] = None
immersion_depth_direction: Optional[List[int]] = None
surface_following_distance: Optional[List[float]] = None
transport_air_volume: Optional[List[float]] = None
pre_wetting_volume: Optional[List[float]] = None
lld_mode: Optional[List[LLDMode]] = None
gamma_lld_sensitivity: Optional[List[int]] = None
dp_lld_sensitivity: Optional[List[int]] = None
aspirate_position_above_z_touch_off: Optional[List[float]] = None
detection_height_difference_for_dual_lld: Optional[List[float]] = None
swap_speed: Optional[List[float]] = None
settling_time: Optional[List[float]] = None
mix_volume: Optional[List[float]] = None
mix_cycles: Optional[List[int]] = None
mix_position_from_liquid_surface: Optional[List[float]] = None
limit_curve_index: Optional[List[int]] = None
This one basically shows how if the classes inherit BaseModel they can generate downstream.
That said, I actually think I am against this, because you lose some of the rich documentation of the schema that is actually really useful. For example:
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum
class LLDMode(str, Enum):
"""Liquid level detection mode"""
OFF = "off"
GAMMA = "gamma"
PRESSURE = "pressure"
DUAL = "dual"
class SingleChannelAspiration(BaseModel):
"""Single channel aspiration operation"""
volume: float = Field(description="Volume to aspirate in microliters", gt=0, example=100.0)
location: str = Field(description="Well location (e.g., 'A1', 'B2')", example="A1")
flow_rate: Optional[float] = Field(None, description="Flow rate in μL/s", gt=0, example=50.0)
class AspirateRequest(BaseModel):
"""
Request model for aspirating liquid from specified channels.
This operation performs liquid level detection (LLD) and aspirates
the specified volume from each channel. Supports pre-wetting, mixing,
and various LLD modes.
"""
ops: List[SingleChannelAspiration] = Field(
description="List of aspiration operations to perform"
)
use_channels: List[int] = Field(
description="Channel indices to use (0-based)",
example=[0, 1]
)
jet: Optional[List[bool]] = Field(
None,
description="Whether to search for a jet liquid class (only used on dispense)",
example=[False, False]
)
lld_mode: Optional[List[LLDMode]] = Field(
None,
description="Liquid level detection mode for each channel",
example=["gamma", "gamma"]
)
immersion_depth: Optional[List[float]] = Field(
None,
description="Z distance to move after detecting liquid (mm). Positive values move into liquid.",
example=[2.0, 2.0]
)
transport_air_volume: Optional[List[float]] = Field(
None,
description="Volume of air to aspirate after the liquid (μL)",
example=[5.0, 5.0]
)
mix_volume: Optional[List[float]] = Field(
None,
description="Volume to aspirate for mixing (μL)",
example=[50.0, 50.0]
)
mix_cycles: Optional[List[int]] = Field(
None,
description="Number of mix cycles to perform before aspiration",
example=[3, 3]
)
class Config:
schema_extra = {
"example": {
"ops": [
{"volume": 100.0, "location": "A1", "flow_rate": 50.0},
{"volume": 150.0, "location": "B2", "flow_rate": 50.0}
],
"use_channels": [0, 1],
"jet": [False, False],
"lld_mode": ["gamma", "gamma"],
"immersion_depth": [2.0, 2.0],
"transport_air_volume": [5.0, 5.0],
"mix_volume": [50.0, 50.0],
"mix_cycles": [3, 3]
}
}
@app.post(
"/aspirate",
summary="Aspirate liquid from channels",
description="Performs aspiration operation with optional LLD, pre-wetting, and mixing",
response_description="Aspiration status and results",
tags=["liquid-handling"]
)
async def aspirate_endpoint(req: AspirateRequest):
"""
Aspirate liquid from the specified channels.
This endpoint supports various advanced features:
- **Liquid Level Detection (LLD)**: Gamma, pressure, or dual mode
- **Pre-wetting**: Prime tips before aspiration
- **Mixing**: Mix liquid before aspirating
- **Clot detection**: Detect clots during aspiration
The operation will:
1. Move to the specified locations
2. Perform LLD if enabled
3. Mix if configured
4. Aspirate the specified volumes
5. Aspirate transport air if configured
"""
# Convert Pydantic model to dict, excluding unset values
params = req.dict(exclude_unset=True)
# Call the underlying function
await your_backend.aspirate(**params)
return {
"status": "success",
"message": f"Aspirated from {len(req.ops)} positions using channels {req.use_channels}"
}
Here, you get descriptions of all the fields, descriptions of the endpoint overall, and example inputs. So my thoughts are right now:
- Rewrite all the functional inputs as endpoints, with rich schema validation + documentation
- Hook that up to the actual functions
Thoughts @Rick ? It has the problem that updates are very annoying, but has the niceness of rich documentation and a fully annotated schema.
yes, on these methods (the PLR universal ones) it won’t work well. I mainly suggest this approach for the whole range of firmware commands that you said you wanted to expose (I think we have around 150 star firmware commands implemented)
for these particular objects, why not use the dataclasses defined in lh standard.py?
what is “this”? not having the attributes as Field?
It is duplicating a lot of information. We have all these parameters + types + docs already in the star backend file, and having one source of truth is ideal. If it’s gonna be in PLR (which makes the most sense to me), ideally I do not become responsible for maintaining two sources of truth.
If the client is gonna use a modified version of the STARBackend, they get access to docs+types through that client. And the server in that case does not require documentation today (until someone is writing a new client to it in the future…)
The firmware commands use classes from the resource models, so they don’t work just the same.
Mainly because I’m much more interested in implementing the entire STAR backend rather than a higher level standardization.
I think my bad there for communicating. Field is all good.
Hmmm, how about something like this?
from pydantic import BaseModel, Field
from typing import List, Optional
class AspirationParams(BaseModel):
"""Parameters for aspiration operations."""
jet: Optional[List[bool]] = Field(None, description="Enable jet mode per channel")
blow_out: Optional[List[bool]] = Field(None, description="Enable blow out per channel")
lld_search_height: Optional[List[float]] = Field(None, description="Liquid level detection search height (mm)")
clot_detection_height: Optional[List[float]] = Field(None, description="Height for clot detection (mm)")
pull_out_distance_transport_air: Optional[List[float]] = Field(None, description="Pull out distance for transport air (mm)")
second_section_height: Optional[List[float]] = Field(None, description="Height of second section (mm)")
second_section_ratio: Optional[List[float]] = Field(None, description="Ratio for second section aspiration")
minimum_height: Optional[List[float]] = Field(None, description="Minimum aspiration height (mm)")
immersion_depth: Optional[List[float]] = Field(None, description="Tip immersion depth (mm)")
immersion_depth_direction: Optional[List[int]] = Field(None, description="Direction of immersion")
surface_following_distance: Optional[List[float]] = Field(None, description="Distance to follow liquid surface (mm)")
transport_air_volume: Optional[List[float]] = Field(None, description="Transport air volume (µL)")
pre_wetting_volume: Optional[List[float]] = Field(None, description="Pre-wetting volume (µL)")
# etc, etc, etc
model_config = {"extra": "forbid"} # Pydantic v2 syntax
async def aspirate(
self,
ops: List[SingleChannelAspiration],
use_channels: List[int],
**kwargs # Accept all kwargs
) -> None:
"""
Aspirate liquid using specified channels.
Args:
ops: List of aspiration operations
use_channels: Channels to use
**kwargs: Additional parameters - see AspirationParams for options
"""
# Validate and parse kwargs using Pydantic
params = AspirationParams(**kwargs)
# Now you can access validated params with autocomplete:
# params.jet, params.lld_search_height, etc.
# Your existing implementation...
# You can convert back to dict if needed: params.model_dump(exclude_none=True)
Or something like that, where the kwargs are in a pydantic class and then you can just import them into the API, so there is one source of truth, and you can annotate them with Field. The doc-string we could make annotate the endpoint. The only “duplication” you get is on the per parameter documentation
This is probably correct and also feels kinda gross
most pure-firmware commands don’t, but I guess we want a more general fix than some-vs-others
the most useful methods of the star backend use the standard.py objects
the standard.py objects are almost identical to your SingleChannelAspirationModel. I think they could be the same
interesting proposal. I am not sure how I feel. @CamilloMoschner what do you think?
why do we need “pydantic”? I hate dependencies. why can’t we use data classes?
Pydantic allows for autogeneration of schemas with some level of validation. It’s pretty much entirely about schema generation, whereas data classes don’t do that as well.
Yes, but they likely need a little more to have schemas generate on top of em (either BaseModel inheritance or @dataclass decorator from pydantic).