from __future__ import annotations
import copy
import inspect
import re
import time
import warnings
from abc import ABC, abstractmethod
from functools import reduce
from itertools import chain
from typing import Any, List, Union, cast
import numpy as np
from bluesky.protocols import Reading
from event_model import DataKey
from frappy.client import CacheItem
from frappy.datatypes import (
ArrayOf,
BLOBType,
BoolType,
CommandType,
DataType,
EnumType,
FloatRange,
IntRange,
ScaledInteger,
StringType,
StructOf,
TupleOf,
)
from ophyd_async.core._utils import StrictEnum
from typing_extensions import NotRequired
SCALAR_DATATYPES = (
IntRange,
BoolType,
FloatRange,
ScaledInteger,
StringType,
EnumType,
)
[docs]
class SECoPDataKey(DataKey):
"""
A DataKey that is used to describe the SECoP Datatype.
"""
dtype_str: NotRequired[str] # type: ignore
"""
The array-protocol typestring of the data-type object.
"""
dtype_descr: NotRequired[list] # type: ignore
"""
String representation of the numpy structured array dtype.
"""
SECOP_datainfo: str
"""
String representation of the original secop datatype.
"""
SECoP_dtype: NotRequired[str] # type: ignore
"""
The SECoP datatype of the data.
"""
[docs]
class NestedRaggedArray(Exception):
"""The Datatype contains nested ragged arrays"""
[docs]
def deep_get(dictionary, keys, default=None) -> dict:
def get_val(obj, key, default):
if isinstance(obj, dict):
return obj.get(key, default)
if isinstance(obj, list):
return obj[key]
if isinstance(obj, tuple):
return obj[key]
return default
return reduce(lambda d, key: get_val(d, key, default), keys, dictionary)
[docs]
class Path:
def __init__(self, parameter_name: str, module_name: str) -> None:
self._accessible_name = parameter_name
self._module_name = module_name
self._last_named_param: int | None = None
self._dev_path: List[Union[str, int]] = []
# Path is extended
[docs]
def append(self, elem: Union[str, int]) -> Path:
new_path = copy.deepcopy(self)
if isinstance(elem, str):
new_path._last_named_param = len(new_path._dev_path)
new_path._dev_path.append(elem)
return new_path
[docs]
def get_param_path(self):
return {"module": self._module_name, "parameter": self._accessible_name}
[docs]
def get_path_tuple(self):
return (self._module_name, self._accessible_name)
# TODO typing of return value
[docs]
def get_memberinfo_path(self):
# inserting K after every Nth number
# using itertool.chain()
# insert element
k = "members"
# insert after every other element
n = 1
# using itertool.chain()
# inserting K after every Nth number
return list(
chain(
*[
(
[k] + self._dev_path[i : i + n] # type: ignore
if len(self._dev_path[i : i + n]) == n
else self._dev_path[i : i + n]
)
for i in range(0, len(self._dev_path), n)
]
)
)
[docs]
def get_signal_name(self):
# top level: signal name == Parameter name
if self._dev_path == []:
return self._accessible_name
sig_postfix = self._dev_path[self._last_named_param :]
if self._last_named_param is None:
sig_postfix = [self._accessible_name] + sig_postfix # type: ignore
delim = "-"
return delim.join(map(str, sig_postfix))
[docs]
def get_param_desc_path(self):
return [self._module_name, "parameters", self._accessible_name]
[docs]
def get_cmd_desc_path(self):
return [self._module_name, "commands", self._accessible_name]
[docs]
def get_leaf(self):
if self._dev_path == []:
return None
return self._dev_path[-1]
[docs]
def insert_val(self, dic: dict, new_val):
if self._dev_path == []:
return dic
d = dic
for key in self._dev_path[:-1]:
if key in d:
d = d[key]
else:
# wrong path
raise Exception(
"path is incorrect " + str(key) + " is not in dict: " + str(dic)
)
# insert new value
if self._dev_path[-1] in d:
d[self._dev_path[-1]] = new_val
else:
# wrong path
raise Exception(
"path is incorrect " + str(key) + " is not in dict: " + str(dic)
)
return dic
[docs]
class DtypeNP(ABC):
secop_dtype: DataType
name: str | None
array_element: bool = False
max_depth: int = 0
[docs]
@abstractmethod
def make_numpy_dtype(self) -> tuple:
"""Create Numpy Compatible structured Datatype"""
[docs]
@abstractmethod
def make_concrete_numpy_dtype(self, value) -> tuple:
"""Create Numpy Compatible structured Datatype from a concrete data value"""
[docs]
@abstractmethod
def make_numpy_compatible_list(self, value) -> Any:
"""make a make a list that is ready for numpy array import"""
[docs]
@abstractmethod
def make_secop_compatible_object(self, value) -> Any:
"""make a make an SECoP Compatible Object"""
[docs]
def dt_factory(
secop_dt: DataType, name: str = "", array_element: bool = False
) -> DtypeNP:
dt_class = secop_dt.__class__
dt_converters = {
StructOf: StructNP,
TupleOf: TupleNP,
ArrayOf: ArrayNP,
IntRange: IntNP,
FloatRange: FloatNP,
ScaledInteger: ScaledIntNP,
BLOBType: BLOBNP,
BoolType: BoolNP,
EnumType: EnumNP,
StringType: StringNP,
}
return dt_converters[dt_class](secop_dt, name, array_element) # type: ignore
STR_LEN_DEFAULT = 100
[docs]
class BLOBNP(DtypeNP):
def __init__(
self, blob_dt: BLOBType, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: BLOBType = blob_dt
self.array_element = array_element
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "U" + str(self.secop_dtype.maxbytes))
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return (self.name, "U" + str(self.secop_dtype.maxbytes))
[docs]
def make_numpy_compatible_list(self, value: str):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return value
[docs]
class BoolNP(DtypeNP):
def __init__(
self, bool_dt: BoolType, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: BoolType = bool_dt
self.array_element = array_element
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "<b1")
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return self.make_numpy_dtype()
[docs]
def make_numpy_compatible_list(self, value: bool):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return bool(value)
[docs]
class EnumNP(DtypeNP):
def __init__(
self, enum_dt: EnumType, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: EnumType = enum_dt
self.array_element = array_element
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "<i8")
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return self.make_numpy_dtype()
[docs]
def make_numpy_compatible_list(self, value: int):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return int(value)
[docs]
class FloatNP(DtypeNP):
def __init__(
self, float_dt: FloatRange, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: FloatRange = float_dt
self.array_element = array_element
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "<f8")
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return self.make_numpy_dtype()
[docs]
def make_numpy_compatible_list(self, value: float):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return float(value)
[docs]
class IntNP(DtypeNP):
def __init__(
self, int_dt: IntRange, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: IntRange = int_dt
self.array_element = array_element
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "<i8")
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return self.make_numpy_dtype()
[docs]
def make_numpy_compatible_list(self, value: int):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return int(value)
[docs]
class ScaledIntNP(DtypeNP):
def __init__(
self, scaled_int_dt: ScaledInteger, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: ScaledInteger = scaled_int_dt
self.array_element = array_element
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "<i8")
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return self.make_numpy_dtype()
[docs]
def make_numpy_compatible_list(self, value: int):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return int(value)
[docs]
class StringNP(DtypeNP):
def __init__(
self, string_dt: StringType, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: StringType = string_dt
self.array_element = array_element
if string_dt.maxchars == 1 << 64:
# warnings.warn(
# "maxchars was not set, default max char lenght is set to: "
# + str(STR_LEN_DEFAULT)
# )
self.strlen = STR_LEN_DEFAULT
else:
self.strlen = string_dt.maxchars
[docs]
def make_numpy_dtype(self) -> tuple:
return (self.name, "<U" + str(self.strlen))
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
return (self.name, "<U" + str(self.strlen))
[docs]
def make_numpy_compatible_list(self, value: str):
return value
[docs]
def make_secop_compatible_object(self, value) -> Any:
return str(value)
[docs]
class StructNP(DtypeNP):
def __init__(
self, struct_dt: StructOf, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype: StructOf = struct_dt
self.array_element = array_element
self.members: dict[str, DtypeNP] = {
name: dt_factory(member, name, self.array_element)
for (name, member) in struct_dt.members.items()
}
max_depth = [member.max_depth for member in self.members.values()]
self.max_depth = 1 + max(max_depth)
[docs]
def make_numpy_dtype(self) -> tuple:
dt_list = []
for member in self.members.values():
dt_list.append(member.make_numpy_dtype())
return (self.name, dt_list)
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
dt_list = []
for key, member in self.members.items():
member_val = value[key]
dt_list.append(member.make_concrete_numpy_dtype(member_val))
return (self.name, dt_list)
[docs]
def make_numpy_compatible_list(self, value: dict):
return tuple(
[
self.members[name].make_numpy_compatible_list(value[name])
for name in self.members.keys()
]
)
[docs]
def make_secop_compatible_object(self, value) -> Any:
return {
member: np_dtype.make_secop_compatible_object(value=value[member])
for (member, np_dtype) in self.members.items()
}
[docs]
class TupleNP(DtypeNP):
def __init__(
self, tuple_dt: TupleOf, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype = tuple_dt
self.array_element = array_element
self.members: list[DtypeNP] = [
dt_factory(
secop_dt=member, name="f" + str(idx), array_element=self.array_element
)
for idx, member in enumerate(tuple_dt.members)
]
max_depth = [member.max_depth for member in self.members]
self.max_depth = 1 + max(max_depth)
[docs]
def make_numpy_dtype(self) -> tuple:
dt_list = []
for member in self.members:
dt_list.append(member.make_numpy_dtype())
return (self.name, dt_list)
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
dt_list = []
for index, member in enumerate(self.members):
member_val = value[index]
dt_list.append(member.make_concrete_numpy_dtype(member_val))
return (self.name, dt_list)
[docs]
def make_numpy_compatible_list(self, value: tuple):
return tuple(
[
self.members[idx].make_numpy_compatible_list(value[idx])
for idx, member in enumerate(self.members)
]
)
[docs]
def make_secop_compatible_object(self, value) -> Any:
return tuple(
[
np_dtype.make_secop_compatible_object(value=value["f" + str(idx)])
for idx, np_dtype in enumerate(self.members)
]
)
[docs]
class ArrayNP(DtypeNP):
def __init__(
self, array_dt: ArrayOf, name: str = "", array_element: bool = False
) -> None:
self.name: str = name
self.secop_dtype = array_dt
self.maxlen = array_dt.maxlen
self.minlen = array_dt.minlen
self.array_element = array_element
self.ragged: bool = self.minlen != self.maxlen
self.shape = [self.maxlen]
self.members: DtypeNP = dt_factory(array_dt.members, array_element=True)
self.root_type: DtypeNP
self.max_depth = self.members.max_depth
if isinstance(self.members, ArrayNP):
self.shape.extend(self.members.shape)
self.root_type = self.members.root_type
self.members.shape = []
if self.members.ragged:
pass
# raise NestedRaggedArray(
# "ragged arrays with more than a single dimension are not supported"
# )int
else:
self.root_type = self.members
if self.array_element and self.ragged:
warnings.warn(
"ragged arrays inside of arrays of copmposite datatypes (struct/tuple)"
"are not supported"
)
[docs]
def get_root_np_str(self) -> str:
dtype_list = self.root_type.make_numpy_dtype()
return dtype_list[1]
[docs]
def make_numpy_dtype(self) -> tuple:
if self.shape == []:
return self.members.make_numpy_dtype()
else:
return (self.name, list(self.members.make_numpy_dtype()).pop(), self.shape)
[docs]
def make_concrete_numpy_dtype(self, value) -> tuple:
if self.ragged is False:
return (
self.name,
list(self.members.make_concrete_numpy_dtype(value)).pop(),
self.shape,
)
else:
if value == []:
member_np = self.members.make_concrete_numpy_dtype(None)
val_shape = [0]
elif value is None:
member_np = self.members.make_concrete_numpy_dtype(None)
val_shape = []
else:
member_np = self.members.make_concrete_numpy_dtype(value[0])
val_shape = [len(value)]
if isinstance(self.members, ArrayNP):
val_shape = val_shape + member_np[2]
return (self.name, member_np[1], val_shape)
[docs]
def make_numpy_compatible_list(self, value: list):
return [self.members.make_numpy_compatible_list(elem) for elem in value]
[docs]
def make_secop_compatible_object(self, value: np.ndarray) -> Any:
return [
self.members.make_secop_compatible_object(value[idx])
for idx in range(0, self.maxlen)
]
[docs]
class SECoPdtype:
def __init__(self, datatype: DataType) -> None:
self.raw_dtype: DataType = datatype
# Describe Fields ------------------------
self.dtype: str
# The array-protocol typestring of the data-type object.
self.dtype_str: str
# String representation of the numpy structured array dtype
self.dtype_descr: list
# string representation of the original secop datatype
self.secop_dtype_str = str(datatype.export_datatype())
# Shape of Data
self.shape = []
self.np_datatype: Any
# Describe Fields ------------------------
self.numpy_dtype: np.dtype
self.dtype_tree: DtypeNP
self._is_composite: bool = False
self._is_array: bool = False
self._is_enum: bool = False
self.dtype_tree = dt_factory(datatype)
self.max_depth: int = self.dtype_tree.max_depth
if isinstance(self.dtype_tree, EnumNP):
self._is_enum = True
if isinstance(self.dtype_tree, ArrayNP):
self.shape = self.dtype_tree.shape
self._is_array = True
self._is_composite = (
True
if isinstance(self.dtype_tree.root_type, (StructNP, TupleNP))
else False
)
self._is_enum = self._is_enum or isinstance(
self.dtype_tree.root_type, EnumNP
)
if isinstance(self.dtype_tree, (TupleNP, StructNP)):
self._is_composite = True
# Composite Datatypes & Arrays of COmposite Datatypes
if self._is_composite:
dt = self.dtype_tree.make_numpy_dtype()
# Top level elements are not named and shape is
# already covered by the shape var
dt = dt[1]
self.numpy_dtype = np.dtype(dt)
# all composite Dtypes are transported as numpy arrays
self.dtype = "array"
self.dtype_str = self.numpy_dtype.str
self.dtype_descr = self.numpy_dtype.descr
self.np_datatype = np.ndarray
# Scalar atomic Datatypes and arrays of atomic dataypes
else:
if self._is_array:
# root secop datatype that is contained in the array
self.dtype = "array"
self.np_datatype = np.ndarray
# Primitive datatypes
else:
self.np_datatype = SECOP2DTYPE[datatype.__class__][0]
self.dtype = SECOP2DTYPE[datatype.__class__][1]
[docs]
def get_datakey(self) -> dict:
describe_dict: dict = {}
# Composite Datatypes & Arrays of COmposite Datatypes
if self._is_composite:
describe_dict["dtype_str"] = self.dtype_str
# describe_dict["dtype_numpy"] = self.dtype_descr
describe_dict["dtype_descr"] = self.dtype_descr
if isinstance(self.dtype_tree, ArrayNP):
describe_dict["dtype_numpy"] = self.dtype_tree.get_root_np_str()
describe_dict["dtype"] = self.dtype
describe_dict["shape"] = self.shape
describe_dict["SECOP_datainfo"] = self.secop_dtype_str
return describe_dict
def _secop2numpy_array(self, value) -> np.ndarray:
np_list = self.dtype_tree.make_numpy_compatible_list(value)
return np.array(np_list, dtype=self.numpy_dtype)
[docs]
def secop2val(self, reading_val) -> Any:
if self._is_composite:
return self._secop2numpy_array(reading_val)
elif self._is_enum:
if isinstance(self.dtype_tree, EnumNP):
exp_val = self.dtype_tree.secop_dtype(reading_val).name
return exp_val
elif isinstance(self.dtype_tree, ArrayNP) and isinstance(
self.dtype_tree.root_type, EnumNP
):
def map_enum(val):
return list(
map(
lambda x: map_enum(x) if isinstance(x, tuple) else x.name,
val,
)
)
exp_val = map_enum(reading_val)
return exp_val
else:
raise Exception("enum type is not correctly identified in dtype tree")
else:
return reading_val
[docs]
def val2secop(self, input_val) -> Any:
# TODO check input_Val for conformity with datatype
# TODO check if it is already in SECoP Format
if self._is_composite and isinstance(input_val, np.ndarray):
return self.dtype_tree.make_secop_compatible_object(input_val)
else:
return self.raw_dtype.validate(input_val)
[docs]
def update_dtype(self, input_val):
if self._is_composite:
# Composite Datatypes & Arrays of Composite Datatypes
dt = self.dtype_tree.make_concrete_numpy_dtype(input_val)
# Top level elements are not named and shape is
# already covered by the shape var
inner_dt = dt[1]
if isinstance(self.raw_dtype, ArrayOf):
self.shape = dt[2]
self.numpy_dtype = np.dtype(inner_dt)
self.dtype_str = self.numpy_dtype.str
self.dtype_descr = self.numpy_dtype.descr
return
if isinstance(self.raw_dtype, ArrayOf):
dt = self.dtype_tree.make_concrete_numpy_dtype(input_val)
self.shape = dt[2]
[docs]
def secop_enum_name_to_python(member_name: str) -> str:
"""Convert SECoP enum member name to Python identifier.
Examples:
'Low Energy' -> 'LOW_ENERGY'
'high-power' -> 'HIGH_POWER'
'Mode 1' -> 'MODE_1'
:param member_name: Original SECoP enum member name
:return: Python-compatible identifier in UPPER_CASE
"""
# Replace spaces and hyphens with underscores, remove other special chars
cleaned = re.sub(r"[\s-]+", "_", member_name)
cleaned = re.sub(r"[^a-zA-Z0-9_]", "", cleaned)
# Convert to uppercase
cleaned = cleaned.upper()
# Ensure it doesn't start with a digit
if cleaned and cleaned[0].isdigit():
cleaned = "_" + cleaned
return cleaned
# Maps a SECoP command argument/result datatype to the plain python type used
# for both the runtime bluesky-plan-method signature and the generated
# Command[[ArgT], ResT] annotation (as opposed to SECoPdtype.np_datatype, which
# is the numpy-facing type used for Parameter/Property SignalR annotations).
COMMAND_DTYPE_MAPPING: dict[type[DataType], type] = {
StructOf: dict[str, Any],
ArrayOf: list[Any],
TupleOf: tuple[Any],
BLOBType: str,
BoolType: bool,
EnumType: StrictEnum,
FloatRange: float,
IntRange: int,
ScaledInteger: int,
StringType: str,
}
[docs]
def command_dtype_to_python_type(datatype: DataType) -> type:
"""Map a SECoP command argument/result datatype to a plain python type."""
return COMMAND_DTYPE_MAPPING[datatype.__class__]
def _dynamic_enum_class(enum_dt: EnumType, context_name: str) -> type[StrictEnum]:
"""Dynamically build a StrictEnum subclass with the real SECoP member
names/values, for a command signature annotation built at pure-runtime
introspection (no codegen involved, so no generated source file to name
a concrete class in) -- gives `build_command_signature()` a real,
member-carrying enum instead of the generic `StrictEnum` base class,
using Python's functional Enum API rather than generated source text.
"""
members = enum_dt.export_datatype().get("members", {})
class_name = f"{context_name.capitalize()}Enum"
# mypy doesn't recognize the functional Enum API on a custom StrEnum
# subclass with its own metaclass; this is exercised at runtime in
# tests/test_classgen.py.
enum_cls = StrictEnum( # type: ignore[call-arg]
class_name, {secop_enum_name_to_python(name): name for name in members}
)
return cast("type[StrictEnum]", enum_cls)
def _reused_annotated_enum(candidate: Any) -> type[StrictEnum] | None:
"""If `candidate` (a parameter/return annotation captured from a real
`Command[[ArgT], ResT]` class annotation, before init_command_from_introspection
would otherwise discard it) is already a genuine, concrete StrictEnum
subclass -- not the bare base class, not `inspect.Parameter.empty`/
`inspect.Signature.empty` -- return it so it can be reused verbatim
instead of building a fresh, differently-named class via
`_dynamic_enum_class`."""
if (
isinstance(candidate, type)
and issubclass(candidate, StrictEnum)
and candidate is not StrictEnum
):
return cast("type[StrictEnum]", candidate)
return None
[docs]
def python_type_to_str(python_type: Any) -> str:
"""Render a python type object as it should appear in generated source,
e.g. int -> "int", dict[str, Any] -> "dict[str, Any]"."""
if python_type in (None, type(None)):
return "None"
if isinstance(python_type, type):
return python_type.__name__
return str(python_type).replace("typing.", "")
[docs]
def command_dtype_to_annotation_str(datatype: DataType) -> str:
"""Render a SECoP command argument/result datatype as a python type string,
e.g. for use in a generated Command[[ArgT], ResT] annotation."""
return python_type_to_str(command_dtype_to_python_type(datatype))
[docs]
def build_command_signature(
cmd_datatype: CommandType,
annotated_signature: inspect.Signature | None = None,
) -> inspect.Signature:
"""Build the call signature for a SECoP command's argument(s)/result.
A StructOf argument is unraveled into one KEYWORD_ONLY parameter per
member (optional members default to None, matching frappy's own "None
means missing" convention for structs); any other argument datatype
becomes a single POSITIONAL_OR_KEYWORD 'arg' parameter; no argument means
no argument parameters. A trailing 'wait_for_idle: bool = False' is
always appended (KEYWORD_ONLY if the preceding param is, to avoid mixing
keyword-only and positional-or-keyword params). Used by
SECoPCommandBackend to build both its real call signature and to
validate/convert calls via Signature.bind().
`annotated_signature`, if given, is the signature captured from a real
`Command[[ArgT], ResT]` class annotation (before this function's caller
would otherwise discard it) -- when the bare (non-struct) argument or
result is an Enum and `annotated_signature` already carries a concrete
generated enum class there, that class is reused instead of building a
fresh, differently-named one via `_dynamic_enum_class`. Struct members
are unaffected: `annotated_signature` has a different shape there (a
single flat argument, not a per-member decomposition), and struct
members never get named enum classes at codegen time anyway.
"""
params: list[inspect.Parameter] = []
arg_dt = cmd_datatype.argument
annotated_params = (
list(annotated_signature.parameters.values()) if annotated_signature else []
)
if isinstance(arg_dt, StructOf):
for member_name, member_dtype in arg_dt.members.items():
annotation = (
_dynamic_enum_class(member_dtype, member_name)
if isinstance(member_dtype, EnumType)
else command_dtype_to_python_type(member_dtype)
)
params.append(
inspect.Parameter(
member_name,
inspect.Parameter.KEYWORD_ONLY,
annotation=annotation,
default=(
None
if member_name in arg_dt.optional
else inspect.Parameter.empty
),
)
)
elif arg_dt is not None:
if isinstance(arg_dt, EnumType):
reused = (
_reused_annotated_enum(annotated_params[0].annotation)
if annotated_params
else None
)
annotation = reused or _dynamic_enum_class(arg_dt, "arg")
else:
annotation = command_dtype_to_python_type(arg_dt)
params.append(
inspect.Parameter(
"arg",
inspect.Parameter.POSITIONAL_OR_KEYWORD,
annotation=annotation,
)
)
wait_for_idle_kind = (
inspect.Parameter.KEYWORD_ONLY
if params and params[-1].kind is inspect.Parameter.KEYWORD_ONLY
else inspect.Parameter.POSITIONAL_OR_KEYWORD
)
params.append(
inspect.Parameter(
"wait_for_idle", wait_for_idle_kind, annotation=bool, default=False
)
)
res_dt = cmd_datatype.result
return_annotation: type | None
if isinstance(res_dt, EnumType):
reused = (
_reused_annotated_enum(annotated_signature.return_annotation)
if annotated_signature
else None
)
return_annotation = reused or _dynamic_enum_class(res_dt, "result")
else:
return_annotation = (
command_dtype_to_python_type(res_dt) if res_dt is not None else None
)
return inspect.Signature(params, return_annotation=return_annotation)
[docs]
class SECoPReading:
def __init__(
self,
secop_dt: SECoPdtype,
entry: CacheItem | None = None,
) -> None:
self.secop_dt: SECoPdtype = secop_dt
if entry is None:
self.timestamp: float = time.time()
self.value = None
self.readerror = None
return
if entry.readerror is not None:
raise entry.readerror
# exported_val = secop_dt.raw_dtype.export_value(entry.value)
self.secop_dt.update_dtype(entry.value)
self.value = secop_dt.secop2val(entry.value)
self.secop_val = entry.value
self.timestamp = entry.timestamp
[docs]
def get_reading(self) -> Reading:
return {"value": self.value, "timestamp": self.timestamp}
[docs]
def get_value(self):
return self.value
[docs]
def get_secop_value(self):
return self.secop_val
[docs]
def set_reading(self, value) -> None:
self.value = value
self.secop_val = self.secop_dt.val2secop(value)
self.timestamp = time.time()
SECOP2DTYPE = {
FloatRange: (float, "number"),
IntRange: (int, "integer"),
ScaledInteger: (int, "integer"),
BoolType: (bool, "boolean"),
EnumType: (StrictEnum, "string"),
StringType: (str, "string"),
BLOBType: (str, "string"),
}
