bluesky / ophyd-async

Hardware abstraction for bluesky written using asyncio
https://blueskyproject.io/ophyd-async
BSD 3-Clause "New" or "Revised" License
8 stars 23 forks source link

Switch from `TypedDict` to row-wise numpy structured data for tables #442

Closed coretl closed 2 days ago

coretl commented 2 months ago

_Originally posted by @coretl in https://github.com/bluesky/ophyd-async/pull/430#discussion_r1669932540_

PVA enforces a strict datatype for each of the columns of the table which allows us to use a numpy structured data type for each row, which we can check against. This has the advantage that it is stored row-wise, so will always have columns the same length. It does mean doing a transpose from PVA column-wise to numpy row-wise, but this is done by numpy in C, so is not a big concern.

To do this:

Enums are expected to be difficult, so you might need to change them to SubsetEnum instead, or even drop to str

abbiemery commented 1 month ago

Background:

coretl commented 4 weeks ago

numpy structured data lost us all type hints, so decided that BaseModel was a better approach.

What we came up with:

from pydantic import BaseModel, ValidationError, model_validator
from typing_extensions import Self

class Table(BaseModel):
    @model_validator(mode="after")
    def check_rows_same_length(self) -> Self:
        lengths = [len(field) for field in self.__pydantic_fields_set__]
        assert all_lengths_match
        return self

    def __add__(self, other: Self) -> Self: ...

class SeqTable(Table):
    repeats: pnd.Np1DArrayUint16
    trigger: Sequence[SeqTrigger]
    position: pnd.Np1DArrayInt32
    time1: pnd.Np1DArrayUint32
    outa1: pnd.Np1DArrayBool
    outb1: pnd.Np1DArrayBool
    outc1: pnd.Np1DArrayBool
    outd1: pnd.Np1DArrayBool
    oute1: pnd.Np1DArrayBool
    outf1: pnd.Np1DArrayBool
    time2: pnd.Np1DArrayUint32
    outa2: pnd.Np1DArrayBool
    outb2: pnd.Np1DArrayBool
    outc2: pnd.Np1DArrayBool
    outd2: pnd.Np1DArrayBool
    oute2: pnd.Np1DArrayBool
    outf2: pnd.Np1DArrayBool

    @classmethod
    def row(
        cls,
        repeats: int = 1,
        trigger: SeqTrigger = SeqTrigger.IMMEDIATE,
        position: int = 0,
        time1: int = 0,
        outa1: bool = False,
        outb1: bool = False,
        outc1: bool = False,
        outd1: bool = False,
        oute1: bool = False,
        outf1: bool = False,
        time2: int = 0,
        outa2: bool = False,
        outb2: bool = False,
        outc2: bool = False,
        outd2: bool = False,
        oute2: bool = False,
        outf2: bool = False,
    ) -> Self:
        arrays = {k: [v] for k, v in locals()}
        return cls(**arrays)

def my_plan():
    table = (
        # Wait for pre-delay then open shutter
        SeqTable.row(
            time1=in_micros(pre_delay),
            time2=in_micros(shutter_time),
            outa2=True,
        ) +
        # Keeping shutter open, do N triggers
        SeqTable.row(
            repeats=number_of_frames,
            time1=in_micros(exposure),
            outa1=True,
            outb1=True,
            time2=in_micros(deadtime),
            outa2=True,
        ) +
        # Add the shutter close
        SeqTable.row(time2=in_micros(shutter_time)),
    )    
coretl commented 3 weeks ago

Decided: