Skip to content

Commit

Permalink
fix schema generation
Browse files Browse the repository at this point in the history
This is in one commit so it's easy to diff vs the last sorted schema.

Minimal changes from the last one. Funny little trick from
https://skaaptjop.medium.com/how-i-use-pydantic-unrequired-fields-so-that-the-schema-works-0010d8758072
to make it express optional things in the same way as before.
  • Loading branch information
sfoster1 committed Dec 3, 2024
1 parent e3a6b90 commit 74606e6
Show file tree
Hide file tree
Showing 36 changed files with 1,193 additions and 1,418 deletions.
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
"""Command models to initialize an Absorbance Reader."""
from __future__ import annotations
from typing import List, Optional, Literal, TYPE_CHECKING
from typing import List, Optional, Literal, TYPE_CHECKING, Any
from typing_extensions import Type

from pydantic import BaseModel, Field
from pydantic.json_schema import SkipJsonSchema

from opentrons.drivers.types import ABSMeasurementMode
from opentrons.protocol_engine.types import ABSMeasureMode
Expand All @@ -20,6 +21,10 @@
InitializeCommandType = Literal["absorbanceReader/initialize"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class InitializeParams(BaseModel):
"""Input parameters to initialize an absorbance reading."""

Expand All @@ -28,8 +33,10 @@ class InitializeParams(BaseModel):
..., description="Initialize single or multi measurement mode."
)
sampleWavelengths: List[int] = Field(..., description="Sample wavelengths in nm.")
referenceWavelength: Optional[int] = Field(
None, description="Optional reference wavelength in nm."
referenceWavelength: int | SkipJsonSchema[None] = Field(
None,
description="Optional reference wavelength in nm.",
json_schema_extra=_remove_default,
)


Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
"""Command models to read absorbance."""
from __future__ import annotations
from datetime import datetime
from typing import Optional, Dict, TYPE_CHECKING, List
from typing_extensions import Literal, Type
from typing import Optional, Dict, TYPE_CHECKING, List, Any

from typing_extensions import Literal, Type
from pydantic import BaseModel, Field
from pydantic.json_schema import SkipJsonSchema

from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
from ...errors import CannotPerformModuleAction, StorageLimitReachedError
Expand All @@ -22,16 +23,21 @@
from opentrons.protocol_engine.execution import EquipmentHandler


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


ReadAbsorbanceCommandType = Literal["absorbanceReader/read"]


class ReadAbsorbanceParams(BaseModel):
"""Input parameters for an absorbance reading."""

moduleId: str = Field(..., description="Unique ID of the Absorbance Reader.")
fileName: Optional[str] = Field(
fileName: str | SkipJsonSchema[None] = Field(
None,
description="Optional file name to use when storing the results of a measurement.",
json_schema_extra=_remove_default,
)


Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
"""Models and implementation for the calibrateGripper command."""

from enum import Enum
from typing import Optional, Type
from typing import Optional, Type, Any
from typing_extensions import Literal

from pydantic import BaseModel, Field
from pydantic.json_schema import SkipJsonSchema

from opentrons.types import Point
from opentrons.hardware_control import HardwareControlAPI
Expand All @@ -22,6 +23,10 @@
CalibrateGripperCommandType = Literal["calibration/calibrateGripper"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class CalibrateGripperParamsJaw(Enum): # noqa: D101
FRONT = "front"
REAR = "rear"
Expand All @@ -39,7 +44,7 @@ class CalibrateGripperParams(BaseModel):
),
)

otherJawOffset: Optional[Vec3f] = Field(
otherJawOffset: Vec3f | SkipJsonSchema[None] = Field(
None,
description=(
"If an offset for the other probe is already found, then specifying it here"
Expand All @@ -48,6 +53,7 @@ class CalibrateGripperParams(BaseModel):
" If this param is not specified then the command will only find and return"
" the offset for the specified probe."
),
json_schema_extra=_remove_default,
)


Expand All @@ -62,11 +68,12 @@ class CalibrateGripperResult(BaseModel):
),
)

savedCalibration: Optional[GripperCalibrationOffset] = Field(
savedCalibration: GripperCalibrationOffset | SkipJsonSchema[None] = Field(
None,
description=(
"Gripper calibration result data, when `otherJawOffset` is provided."
),
json_schema_extra=_remove_default,
)


Expand Down
13 changes: 11 additions & 2 deletions api/src/opentrons/protocol_engine/commands/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,12 @@
List,
Type,
Union,
Any,
Dict,
)

from pydantic import BaseModel, Field
from pydantic.json_schema import SkipJsonSchema

from opentrons.hardware_control import HardwareControlAPI
from opentrons.protocol_engine.state.update_types import StateUpdate
Expand Down Expand Up @@ -61,6 +64,10 @@ class CommandIntent(str, enum.Enum):
FIXIT = "fixit"


def _pop_default(s: Dict[str, Any]) -> None:
s.pop("default")


class BaseCommandCreate(
BaseModel,
# These type parameters need to be invariant because our fields are mutable.
Expand All @@ -80,7 +87,7 @@ class BaseCommandCreate(
),
)
params: _ParamsT = Field(..., description="Command execution data payload")
intent: Optional[CommandIntent] = Field(
intent: CommandIntent | SkipJsonSchema[None] = Field(
None,
description=(
"The reason the command was added. If not specified or `protocol`,"
Expand All @@ -93,14 +100,16 @@ class BaseCommandCreate(
"Use setup commands for activities like pre-run calibration checks"
" and module setup, like pre-heating."
),
json_schema_extra=_pop_default,
)
key: Optional[str] = Field(
key: str | SkipJsonSchema[None] = Field(
None,
description=(
"A key value, unique in this run, that can be used to track"
" the same logical command across multiple runs of the same protocol."
" If a value is not provided, one will be generated."
),
json_schema_extra=_pop_default,
)


Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
"""Configure for volume command request, result, and implementation models."""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional, Type, Any

from pydantic import BaseModel, Field
from typing import TYPE_CHECKING, Optional, Type
from pydantic.json_schema import SkipJsonSchema
from typing_extensions import Literal

from .pipetting_common import PipetteIdMixin
Expand All @@ -16,6 +18,10 @@
ConfigureForVolumeCommandType = Literal["configureForVolume"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class ConfigureForVolumeParams(PipetteIdMixin):
"""Parameters required to configure volume for a specific pipette."""

Expand All @@ -25,12 +31,13 @@ class ConfigureForVolumeParams(PipetteIdMixin):
"than a pipette-specific maximum volume.",
ge=0,
)
tipOverlapNotAfterVersion: Optional[str] = Field(
tipOverlapNotAfterVersion: str | SkipJsonSchema[None] = Field(
None,
description="A version of tip overlap data to not exceed. The highest-versioned "
"tip overlap data that does not exceed this version will be used. Versions are "
"expressed as vN where N is an integer, counting up from v0. If None, the current "
"highest version will be used.",
json_schema_extra=_remove_default,
)


Expand Down
10 changes: 8 additions & 2 deletions api/src/opentrons/protocol_engine/commands/dispense.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
"""Dispense command request, result, and implementation models."""

from __future__ import annotations
from typing import TYPE_CHECKING, Optional, Type, Union
from typing import TYPE_CHECKING, Optional, Type, Union, Any
from typing_extensions import Literal


from pydantic import Field
from pydantic.json_schema import SkipJsonSchema

from ..state.update_types import StateUpdate, CLEAR
from .pipetting_common import (
Expand Down Expand Up @@ -39,14 +40,19 @@
DispenseCommandType = Literal["dispense"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class DispenseParams(
PipetteIdMixin, DispenseVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin
):
"""Payload required to dispense to a specific well."""

pushOut: Optional[float] = Field(
pushOut: float | SkipJsonSchema[None] = Field(
None,
description="push the plunger a small amount farther than necessary for accurate low-volume dispensing",
json_schema_extra=_remove_default,
)


Expand Down
10 changes: 8 additions & 2 deletions api/src/opentrons/protocol_engine/commands/dispense_in_place.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
"""Dispense-in-place command request, result, and implementation models."""

from __future__ import annotations
from typing import TYPE_CHECKING, Optional, Type, Union
from typing import TYPE_CHECKING, Optional, Type, Union, Any
from typing_extensions import Literal
from pydantic import Field
from pydantic.json_schema import SkipJsonSchema

from .pipetting_common import (
PipetteIdMixin,
Expand Down Expand Up @@ -32,12 +33,17 @@
DispenseInPlaceCommandType = Literal["dispenseInPlace"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class DispenseInPlaceParams(PipetteIdMixin, DispenseVolumeMixin, FlowRateMixin):
"""Payload required to dispense in place."""

pushOut: Optional[float] = Field(
pushOut: float | SkipJsonSchema[None] = Field(
None,
description="push the plunger a small amount farther than necessary for accurate low-volume dispensing",
json_schema_extra=_remove_default,
)


Expand Down
14 changes: 11 additions & 3 deletions api/src/opentrons/protocol_engine/commands/drop_tip.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
"""Drop tip command request, result, and implementation models."""

from __future__ import annotations
from typing import TYPE_CHECKING, Optional, Type, Any

from pydantic import Field
from typing import TYPE_CHECKING, Optional, Type
from pydantic.json_schema import SkipJsonSchema

from typing_extensions import Literal

from opentrons.protocol_engine.errors.exceptions import TipAttachedError
Expand Down Expand Up @@ -37,6 +39,10 @@
DropTipCommandType = Literal["dropTip"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class DropTipParams(PipetteIdMixin):
"""Payload required to drop a tip in a specific well."""

Expand All @@ -46,15 +52,16 @@ class DropTipParams(PipetteIdMixin):
default_factory=DropTipWellLocation,
description="Relative well location at which to drop the tip.",
)
homeAfter: Optional[bool] = Field(
homeAfter: bool | SkipJsonSchema[None] = Field(
None,
description=(
"Whether to home this pipette's plunger after dropping the tip."
" You should normally leave this unspecified to let the robot choose"
" a safe default depending on its hardware."
),
json_schema_extra=_remove_default,
)
alternateDropLocation: Optional[bool] = Field(
alternateDropLocation: bool | SkipJsonSchema[None] = Field(
False,
description=(
"Whether to alternate location where tip is dropped within the labware."
Expand All @@ -63,6 +70,7 @@ class DropTipParams(PipetteIdMixin):
" labware well."
" If False, the tip will be dropped at the top center of the well."
),
json_schema_extra=_remove_default,
)


Expand Down
12 changes: 10 additions & 2 deletions api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
"""Drop tip in place command request, result, and implementation models."""
from __future__ import annotations

from typing import TYPE_CHECKING, Optional, Type, Any

from pydantic import Field, BaseModel
from typing import TYPE_CHECKING, Optional, Type
from pydantic.json_schema import SkipJsonSchema
from typing_extensions import Literal

from .command import (
Expand All @@ -24,16 +27,21 @@
DropTipInPlaceCommandType = Literal["dropTipInPlace"]


def _remove_default(s: dict[str, Any]) -> None:
s.pop("default")


class DropTipInPlaceParams(PipetteIdMixin):
"""Payload required to drop a tip in place."""

homeAfter: Optional[bool] = Field(
homeAfter: bool | SkipJsonSchema[None] = Field(
None,
description=(
"Whether to home this pipette's plunger after dropping the tip."
" You should normally leave this unspecified to let the robot choose"
" a safe default depending on its hardware."
),
json_schema_extra=_remove_default,
)


Expand Down
Original file line number Diff line number Diff line change
@@ -1,24 +1,17 @@
"""Generate a JSON schema against which all create commands statically validate."""

import json
import pydantic
import argparse
import sys
from opentrons.protocol_engine.commands.command_unions import CommandCreate


class CreateCommandUnion(pydantic.RootModel[CommandCreate]):
"""Model that validates a union of all CommandCreate models."""

root: CommandCreate
from opentrons.protocol_engine.commands.command_unions import CommandCreateAdapter


def generate_command_schema(version: str) -> str:
"""Generate a JSON Schema that all valid create commands can validate against."""
raw_json_schema = CreateCommandUnion.schema_json()
schema_as_dict = json.loads(raw_json_schema)
schema_as_dict = CommandCreateAdapter.json_schema(mode="validation")
schema_as_dict["$id"] = f"opentronsCommandSchemaV{version}"
schema_as_dict["$schema"] = "http://json-schema.org/draft-07/schema#"
return json.dumps(schema_as_dict, indent=2)
return json.dumps(schema_as_dict, indent=2, sort_keys=True)


if __name__ == "__main__":
Expand Down
Loading

0 comments on commit 74606e6

Please sign in to comment.