Source code for event_model.documents.run_start

from typing import Any, Dict, List, Union

from typing_extensions import Annotated, Literal, NotRequired, TypedDict

from event_model.documents.generate.type_wrapper import Field, add_extra_schema

from .generate.type_wrapper import DataType


[docs] class Hints(TypedDict): """Start-level hints""" dimensions: NotRequired[ Annotated[ List[List[Union[List[str], str]]], Field( description="The independent axes of the experiment. " "Ordered slow to fast", ), ] ]
class Calculation(TypedDict): args: NotRequired[List] callable: Annotated[ str, Field(description="callable function to perform calculation") ] kwargs: NotRequired[ Annotated[Dict[str, Any], Field(description="kwargs for calcalation callable")] ]
[docs] class Projection(TypedDict): """Where to get the data from""" calculation: NotRequired[ Annotated[ Calculation, Field( description="required fields if type is calculated", title="calculation properties", ), ] ] config_index: NotRequired[int] config_device: NotRequired[str] field: NotRequired[str] location: NotRequired[ Annotated[ Literal["start", "event", "configuration"], Field( description="start comes from metadata fields in the start document, " "event comes from event, configuration comes from configuration " "fields in the event_descriptor document" ), ] ] stream: NotRequired[str] type: NotRequired[ Annotated[ Literal["linked", "calculated", "static"], Field( description="linked: a value linked from the data set, " "calculated: a value that requires calculation, " "static: a value defined here in the projection ", ), ] ] value: NotRequired[ Annotated[ Any, Field( description="value explicitely defined in the projection " "when type==static." ), ] ]
RUN_START_EXTRA_SCHEMA = { "$defs": { "DataType": { "patternProperties": {"^([^./]+)$": {"$ref": "#/$defs/DataType"}}, "additionalProperties": False, }, "Projection": { "allOf": [ { "if": { "allOf": [ {"properties": {"location": {"enum": ["configuration"]}}}, {"properties": {"type": {"enum": ["linked"]}}}, ] }, "then": { "required": [ "type", "location", "config_index", "config_device", "field", "stream", ] }, }, { "if": { "allOf": [ {"properties": {"location": {"enum": ["event"]}}}, {"properties": {"type": {"enum": ["linked"]}}}, ] }, "then": {"required": ["type", "location", "field", "stream"]}, }, { "if": { "allOf": [ {"properties": {"location": {"enum": ["event"]}}}, {"properties": {"type": {"enum": ["calculated"]}}}, ] }, "then": {"required": ["type", "field", "stream", "calculation"]}, }, { "if": {"properties": {"type": {"enum": ["static"]}}}, "then": {"required": ["type", "value"]}, }, ], }, }, "properties": { "hints": { "additionalProperties": False, "patternProperties": {"^([^.]+)$": {"$ref": "#/$defs/DataType"}}, }, }, "patternProperties": {"^([^./]+)$": {"$ref": "#/$defs/DataType"}}, "additionalProperties": False, }
[docs] class Projections(TypedDict): """Describe how to interperet this run as the given projection""" configuration: Annotated[ Dict[str, Any], Field(description="Static information about projection") ] name: NotRequired[Annotated[str, Field(description="The name of the projection")]] projection: Annotated[Dict[Any, Projection], Field(description="")] version: Annotated[ str, Field( description="The version of the projection spec. Can specify the version " "of an external specification.", ), ]
[docs] @add_extra_schema(RUN_START_EXTRA_SCHEMA) class RunStart(TypedDict): """ Document created at the start of run. Provides a seach target and later documents link to it """ data_groups: NotRequired[ Annotated[ List[str], Field( description="An optional list of data access groups that have meaning " "to some external system. Examples might include facility, beamline, " "end stations, proposal, safety form.", ), ] ] data_session: NotRequired[ Annotated[ str, Field( description="An optional field for grouping runs. The meaning is " "not mandated, but this is a data management grouping and not a " "scientific grouping. It is intended to group runs in a visit or " "set of trials.", ), ] ] data_type: NotRequired[Annotated[DataType, Field(description="")]] group: NotRequired[ Annotated[str, Field(description="Unix group to associate this data with")] ] hints: NotRequired[Annotated[Hints, Field(description="Start-level hints")]] owner: NotRequired[ Annotated[str, Field(description="Unix owner to associate this data with")] ] project: NotRequired[ Annotated[str, Field(description="Name of project that this run is part of")] ] projections: NotRequired[Annotated[List[Projections], Field(description="")]] sample: NotRequired[ Annotated[ Union[Dict[str, Any], str], Field( description="Information about the sample, may be a UID to " "another collection" ), ] ] scan_id: NotRequired[ Annotated[int, Field(description="Scan ID number, not globally unique")] ] time: Annotated[float, Field(description="Time the run started. Unix epoch time")] uid: Annotated[str, Field(description="Globally unique ID for this run")]