Add support for pydantic model registration with BlueAPI

[oliwenmandiamond]

When testing with Blueapi, I have a pydantic model loaded and ready to give to my plan / device client side.

>>> region1
VGScientaRegion[LensMode, PassEnergy](name='New_Region', enabled=True, slices=1, iterations=1, excitation_energy_source=<SelectedSource.SOURCE1: 'source1'>, lens_mode=<LensMode.ANGULAR45: 'Angular45'>, pass_energy=<PassEnergy.E10: '10'>, acquisition_mode=<AcquisitionMode.SWEPT: 'Swept'>, low_energy=149.0, centre_energy=150.0, high_energy=151.0, acquire_time=1.0, energy_step=0.2, energy_mode=<EnergyMode.KINETIC: 'Kinetic'>, id='_bMnQkCkaEfG3xKiZ1VupOQ', total_steps=15.0, total_time=15.0, min_x=1, sensor_max_size_x=1000, min_y=101, sensor_max_size_y=800, detector_mode=<DetectorMode.ADC: 'ADC'>)
>>> plans.set_absolute(devs.ew4000, region1)

When trying to set my device with these settings

>>> plans.set_absolute(devs.ew4000, region1)

I can see these logs:

2026-03-26 15:05:25,816  WARNING blueapi.worker.task_worker Invalid correlation id detected: None
2026-03-26 15:05:25,817     INFO blueapi.service.main 10.36.0.26:49436 POST /tasks 201
2026-03-26 15:05:25,830     INFO blueapi.worker.task_worker Submitting: task_id='3ea512cf-4a5c-4349-9bcd-39a6ec357e3f' task=Task(name='set_absolute', params={'movable': 'ew4000', 'value': {'name': 'New_Region', 'enabled': True, 'slices': 1, 'iterations': 1, 'excitation_energy_source': 'source1', 'lens_mode': 'Angular45', 'pass_energy': '10', 'acquisition_mode': 'Swept', 'low_energy': 149.0, 'centre_energy': 150.0, 'high_energy': 151.0, 'acquire_time': 1.0, 'energy_step': 0.2, 'energy_mode': 'Kinetic', 'id': '_bMnQkCkaEfG3xKiZ1VupOQ', 'total_steps': 15.0, 'total_time': 15.0, 'min_x': 1, 'sensor_max_size_x': 1000, 'min_y': 101, 'sensor_max_size_y': 800, 'detector_mode': 'ADC'}}, metadata={'user': 'xol73553', 'instrument_session': 'cm12345-1'}) request_id=None is_complete=False is_pending=True errors=[] outcome=None
2026-03-26 15:05:25,831     INFO blueapi.worker.task_worker Got new task: task_id='3ea512cf-4a5c-4349-9bcd-39a6ec357e3f' task=Task(name='set_absolute', params={'movable': 'ew4000', 'value': {'name': 'New_Region', 'enabled': True, 'slices': 1, 'iterations': 1, 'excitation_energy_source': 'source1', 'lens_mode': 'Angular45', 'pass_energy': '10', 'acquisition_mode': 'Swept', 'low_energy': 149.0, 'centre_energy': 150.0, 'high_energy': 151.0, 'acquire_time': 1.0, 'energy_step': 0.2, 'energy_mode': 'Kinetic', 'id': '_bMnQkCkaEfG3xKiZ1VupOQ', 'total_steps': 15.0, 'total_time': 15.0, 'min_x': 1, 'sensor_max_size_x': 1000, 'min_y': 101, 'sensor_max_size_y': 800, 'detector_mode': 'ADC'}}, metadata={'user': 'xol73553', 'instrument_session': 'cm12345-1'}) request_id=None is_complete=False is_pending=True errors=[] outcome=None
2026-03-26 15:05:25,831     INFO blueapi.worker.task Asked to run plan set_absolute with {'movable': 'ew4000', 'value': {'name': 'New_Region', 'enabled': True, 'slices': 1, 'iterations': 1, 'excitation_energy_source': 'source1', 'lens_mode': 'Angular45', 'pass_energy': '10', 'acquisition_mode': 'Swept', 'low_energy': 149.0, 'centre_energy': 150.0, 'high_energy': 151.0, 'acquire_time': 1.0, 'energy_step': 0.2, 'energy_mode': 'Kinetic', 'id': '_bMnQkCkaEfG3xKiZ1VupOQ', 'total_steps': 15.0, 'total_time': 15.0, 'min_x': 1, 'sensor_max_size_x': 1000, 'min_y': 101, 'sensor_max_size_y': 800, 'detector_mode': 'ADC'}} and metadata {'user': 'xol73553', 'instrument_session': 'cm12345-1'} for all runs
2026-03-26 15:05:25,831     INFO bluesky Executing plan <generator object set_absolute at 0x7fbbf3c32240>
2026-03-26 15:05:25,832     INFO bluesky.RE.state Change state on <bluesky.run_engine.RunEngine object at 0x7fbbf993af50> from 'idle' -> 'running'
2026-03-26 15:05:25,836  WARNING opentelemetry.attributes Invalid type dict in attribute 'msg.args' value sequence. Expected one of ['bool', 'str', 'bytes', 'int', 'float'] or None
2026-03-26 15:05:25,836     INFO bluesky.RE.state Change state on <bluesky.run_engine.RunEngine object at 0x7fbbf993af50> from 'running' -> 'idle'
2026-03-26 15:05:25,837     INFO bluesky Cleaned up from plan <generator object set_absolute at 0x7fbbf3c32240>
2026-03-26 15:05:25,837     INFO blueapi.worker.task_worker Task ran successfully - returned: <AsyncStatus, device: ew4000, task: <coroutine object AsyncStatusBase.__init__.<locals>.wait_with_error_message at 0x7fbbf3c30a40>, errored: AttributeError("'dict' object has no attribute 'excitation_energy_source'")>
2026-03-26 15:05:25,838     INFO blueapi.worker.task_worker Awaiting task
2026-03-26 15:05:25,838     INFO blueapi.service.main 10.36.0.26:49436 PUT /worker/task 200With key bit being
2026-03-26 15:05:25,836  WARNING opentelemetry.attributes Invalid type dict in attribute 'msg.args' value sequence. 

With the key parts being

Expected one of ['bool', 'str', 'bytes', 'int', 'float'] or None

ew4000, task: <coroutine object AsyncStatusBase.__init__.<locals>.wait_with_error_message at 0x7fbbf3c30a40>, errored: AttributeError("'dict' object has no attribute 'excitation_energy_source'")>

So my device expects the pydantic model

    @AsyncStatus.wrap
    async def set(self, region: TAbstractBaseRegion) -> None:

So the issue is that Blueapi does the following

Python object -> JSON -> Python dic

So my device set now fails because it expects the model and not the python dict.

When creating devices in dodal, we were told to make it as strongly typed as possible but now were needing to regress as we either have to make the device handle a dict or create a custom set plan for this device to convert the dict into a pydantic model before we give it to the device.

It would be good if we could add Blueapi support to be able to register pydantic models in the blueapi service values.yaml. Then we retain the type information during serialisation, something like "__type__": "dodal.my.pydantic.model.Implementation" for example so that then on server side when executing the plan, it can attempt to look up registered models and convert back into the pydantic model so that we don't need to write wrapper plans for every pydantic object.

[oliwenmandiamond]

I've done some more testing with this to try and replicate. The real issue is because the plan is using type T. I created this test to confirm.

from typing import Generic, TypeVar

from pydantic import BaseModel

T = TypeVar("T")


class GenericPlanArgs(BaseModel):
    value1: int


class SpecialisedPlanArgs(GenericPlanArgs, Generic[T]):
    value2: T


def plan_with_model(val: SpecialisedPlanArgs) -> MsgGenerator:
    yield from ()
    assert isinstance(val, SpecialisedPlanArgs)


def plan_using_base_model(val: GenericPlanArgs) -> MsgGenerator:
    yield from ()
    assert isinstance(val, SpecialisedPlanArgs)


def plan_with_generic_model(val: T) -> MsgGenerator:
    yield from ()
    assert isinstance(val, SpecialisedPlanArgs)


def test_plan_args_are_converted_back_to_model(
    context: BlueskyContext,
) -> None:
    context.register_plan(plan_with_model)

    task = Task(name="plan_with_model", params={"val": {"value1": 1, "value2": "test"}})
    task.do_task(context)


def test_base_model_plan_args_are_converted_back_to_specialised_model(
    context: BlueskyContext,
) -> None:
    context.register_plan(plan_using_base_model)

    task = Task(
        name="plan_using_base_model", params={"val": {"value1": 1, "value2": "test"}}
    )
    task.do_task(context)


def test_generic_plan_args_are_converted_back_to_specialised_model(
    context: BlueskyContext,
) -> None:
    context.register_plan(plan_with_generic_model)

    task = Task(
        name="plan_with_generic_model", params={"val": {"value1": 1, "value2": "test"}}
    )
    task.do_task(context)

First test passes

Second test fails

FAILED tests/unit_tests/worker/test_task_worker.py::test_base_model_plan_args_are_converted_back_to_specialised_model - assert False
 +  where False = isinstance(GenericPlanArgs(value1=1), SpecialisedPlanArgs)

Third test fails

FAILED tests/unit_tests/worker/test_task_worker.py::test_generic_plan_args_are_converted_back_to_specialised_model - AssertionError: assert False
 +  where False = isinstance({'value1': 1, 'value2': 'test'}, SpecialisedPlanArgs)

So BlueAPI is looking at the plan types and then tying to convert the arguments to fit the plan types. These has two big drawbacks.

The first is that any device that use a movable plan must support either primitive types or a dictionary, otherwise it will fail with a model. E.g any device that implements Movable must use a dictionary rather a pydantic model, undoing the strong type checking. This is because the plan is using generics type T and this is erased so BlueAPI can't convert it using plan args.

The second is that if a plan is using class A as the argument and then I provide it class B that extends A, it will incorrectly convert it to type A and cause issues if the device is expecting the extra data of B.

This means that we either make every device that uses a model instead handle a dictionary, or we create many wrapper plans for every possible set for a device expecting a model which eliminates the use of all generic move plans if a device is not using primitive types.

This process needs to be reversed. Instead, the model that was given to BlueAPI should preserve the type data when serialised and then when preparing the arguments to execute the plan, it should attempt to convert the parameters back to the model using the type data provided and lookup some sort of model register with BlueAPI. It should not being trying to use the plan types to conform the data to because then we end up with all this unexpected and weird behaviour unless nothing else is provided.

[fajinyuan]

I am not sure I fully understand the description and comment of this ticket yet. But My question is if Diamond has chosen to use pydantic model in python software development, why this pydantic model cannot be passed through all the layers of the software stack? why there is a need to convert from model type to primitive types or dictionary and back to model again through different layers?
Should we decide pydantic model only be used in certain layers, would it be better to implement generic pydantic model type eraser before passing the arguments or parameters to other layers that does not support pydantic model? This way every layer of the software stack can be totally independent of one another.
Furthermore, does a user need to use pydantic model to create plan themselves?

[oliwenmandiamond]

It's because BlueAPI needs to serialise to send data across from client to server which is using generic Json.

BlueAPI currently does this

Client, constructs plan arguments via python (from Model) or via web (from Json)
Python Model -> Json to send to server -> python dict -> Check plan parameter type, attempt to convert back to Model -> Execute plan with converted arguments

The above loses data to give to plan because plan type is too generic if the data to send is a Model so doesn't convert to the correct type for plan / device to accept. So any generic plan such as move or set_relative does not work with Models. Must wrap the plan with more type hints so data is not lost (impractical with number of plans you would have to wrap for number of devices to work) or make the device accept a python dict (lose benefit of type checking in dodal). For it to work correctly without any work around plans or devices, we need to do something like below in BlueAPI so works as expected sending data across the different layers.

Python Model -> Json to send to server (add additional type information) -> python dict -> Check python dict for type, attempt to convert back to true type Model via registration with BlueAPI -> Check plan argument type, attempt to convert back to Model if needed -> Execute plan with converted arguments

[fajinyuan]

Thank for the clarification, so it is a serialisation/deserialisation issue as in Java object over the wire. The server have to deserilise the json from client back to correct object type before passing it to the same plan. If this is the case, serilisation must carry model class type in the Json to the server for server to correctly reconstruct the input model type parameter passing to the same plan.