Source code for event_model.documents.event_descriptor

# ruff: noqa
# generated by datamodel-codegen:
#   filename:  event_descriptor.json

from __future__ import annotations

from typing import Any, Literal, TypeAlias, TypedDict

from typing_extensions import NotRequired

DtypeNumpy: TypeAlias = str
"""
A numpy dtype e.g `<U9`, `<f16`
"""


DtypeNumpyItemItem: TypeAlias = str
"""
A numpy dtype e.g `<U9`, `<f16`
"""


DtypeNumpyItem: TypeAlias = tuple[str, DtypeNumpyItemItem]


DataType: TypeAlias = dict[str, "DataType"]


Dtype: TypeAlias = Literal["string", "number", "array", "boolean", "integer"]


class LimitsRange(TypedDict):
    high: float | None
    low: float | None


class PerObjectHint(TypedDict):
    """
    The 'interesting' data keys for this device.
    """

    NX_class: NotRequired[str]
    """
    The NeXus class definition for this device.
    """
    fields: NotRequired[list[str]]
    """
    The 'interesting' data keys for this device.
    """


class RdsRange(TypedDict):
    """
    RDS (Read different than set) parameters range.


    https://tango-controls.readthedocs.io/en/latest/development/device-api/attribute-alarms.html#the-read-different-than-set-rds-alarm
    """

    time_difference: float
    """
    ms since last update to fail after if set point and read point are not within `value_difference` of each other.
    """
    value_difference: float
    """
    Allowed difference in value between set point and read point after `time_difference`.
    """


class Limits(TypedDict):
    """
    Epics limits:
    https://docs.epics-controls.org/en/latest/getting-started/EPICS_Intro.html#channel-access
    """

    alarm: NotRequired[LimitsRange | None]
    """
    Alarm limits.
    """
    control: NotRequired[LimitsRange | None]
    """
    Control limits.
    """
    display: NotRequired[LimitsRange | None]
    """
    Display limits.
    """
    hysteresis: NotRequired[float | None]
    """
    Hysteresis.
    """
    rds: NotRequired[RdsRange | None]
    """
    RDS parameters.
    """
    warning: NotRequired[LimitsRange | None]
    """
    Warning limits.
    """




class DataKey(TypedDict):
    """
    Describes the objects in the data property of Event documents
    """

    choices: NotRequired[list[str]]
    """
    Choices of enum value.
    """
    dims: NotRequired[list[str]]
    """
    The names for dimensions of the data. Null or empty list if scalar data
    """
    dtype: Dtype
    """
    The type of the data in the event, given as a broad JSON schema type.
    """
    dtype_numpy: NotRequired[DtypeNumpy | list[DtypeNumpyItem]]
    """
    The type of the data in the event, given as a numpy dtype string (or, for structured dtypes, array).
    """
    external: NotRequired[str]
    """
    Where the data is stored if it is stored external to the events
    """
    limits: NotRequired[Limits]
    """
    Epics limits.
    """
    object_name: NotRequired[str]
    """
    The name of the object this key was pulled from.
    """
    precision: NotRequired[int | None]
    """
    Number of digits after decimal place if a floating point number
    """
    shape: list[int | None]
    """
    The shape of the data.  Empty list indicates scalar data. None indicates a dimension with unknown or variable length.
    """
    source: str
    """
    The source (ex piece of hardware) of the data.
    """
    units: NotRequired[str | None]
    """
    Engineering units of the value
    """



class Configuration(TypedDict):
    data: NotRequired[dict[str, Any]]
    """
    The actual measurement data
    """
    data_keys: NotRequired[dict[str, DataKey]]
    """
    This describes the data stored alongside it in this configuration object.
    """
    timestamps: NotRequired[dict[str, Any]]
    """
    The timestamps of the individual measurement data
    """


class EventDescriptor(TypedDict):
    """
    Document to describe the data captured in the associated event
    documents
    """

    configuration: NotRequired[dict[str, Configuration]]
    """
    Readings of configurational fields necessary for interpreting data in the Events.
    """
    data_keys: dict[str, DataKey]
    """
    This describes the data in the Event Documents.
    """
    hints: NotRequired[PerObjectHint]
    name: NotRequired[str]
    """
    A human-friendly name for this data stream, such as 'primary' or 'baseline'.
    """
    object_classes: NotRequired[dict[str, str]]
    """
    Maps a Device/Signal name to the runengine process' import path with __qualname__ of that object's type.
    """
    object_keys: NotRequired[dict[str, Any]]
    """
    Maps a Device/Signal name to the names of the entries it produces in data_keys.
    """
    run_start: str
    """
    Globally unique ID of this run's 'start' document.
    """
    time: float
    """
    Creation time of the document as unix epoch time.
    """
    uid: str
    """
    Globally unique ID for this event descriptor.
    """