Skip to content

Commit

Permalink
Switch from using apischema to pydantic
Browse files Browse the repository at this point in the history
  • Loading branch information
callumforrester committed Feb 13, 2023
1 parent 842aa86 commit fd6769a
Show file tree
Hide file tree
Showing 2 changed files with 20 additions and 28 deletions.
4 changes: 2 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@ dependencies = [
"ophyd",
"nslsii",
"pyepics",
"apischema",
"pydantic",
"stomp.py",
"scanspec<=0.5.5",
"scanspec",
"PyYAML",
"click",
]
Expand Down
44 changes: 18 additions & 26 deletions src/blueapi/service/model.py
Original file line number Diff line number Diff line change
@@ -1,29 +1,27 @@
from dataclasses import dataclass
from typing import Iterable, List

from apischema import settings
from bluesky.protocols import HasName
from pydantic import BaseModel, Field

from blueapi.core import BLUESKY_PROTOCOLS, Device, Plan

_UNKNOWN_NAME = "UNKNOWN"

settings.camel_case = True


@dataclass
class DeviceModel:
class DeviceModel(BaseModel):
"""
Representation of a device
"""

name: str
protocols: List[str]
name: str = Field(description="Name of the device")
protocols: List[str] = Field(
description="Protocols that a device conforms to, indicating its capabilities"
)

@classmethod
def from_device(cls, device: Device) -> "DeviceModel":
name = device.name if isinstance(device, HasName) else _UNKNOWN_NAME
return cls(name, list(_protocol_names(device)))
return cls(name=name, protocols=list(_protocol_names(device)))


def _protocol_names(device: Device) -> Iterable[str]:
Expand All @@ -32,59 +30,53 @@ def _protocol_names(device: Device) -> Iterable[str]:
yield protocol.__name__


@dataclass
class DeviceRequest:
class DeviceRequest(BaseModel):
"""
A query for devices
"""

...


@dataclass
class DeviceResponse:
class DeviceResponse(BaseModel):
"""
Response to a query for devices
"""

devices: List[DeviceModel]
devices: List[DeviceModel] = Field(description="Devices available to use in plans")


@dataclass
class PlanModel:
class PlanModel(BaseModel):
"""
Representation of a plan
"""

name: str
name: str = Field(description="Name of the plan")

@classmethod
def from_plan(cls, plan: Plan) -> "PlanModel":
return cls(plan.name)
return cls(name=plan.name)


@dataclass
class PlanRequest:
class PlanRequest(BaseModel):
"""
A query for plans
"""

...


@dataclass
class PlanResponse:
class PlanResponse(BaseModel):
"""
Response to a query for plans
"""

plans: List[PlanModel]
plans: List[PlanModel] = Field(description="Plans available to use by a worker")


@dataclass
class TaskResponse:
class TaskResponse(BaseModel):
"""
Acknowledgement that a task has started, includes its ID
"""

task_name: str
task_name: str = Field(description="Unique identifier for the task")

0 comments on commit fd6769a

Please sign in to comment.