diff --git a/src/ophyd_async/epics/_backend/_aioca.py b/src/ophyd_async/epics/_backend/_aioca.py index 89bf6d256b..394cdde199 100644 --- a/src/ophyd_async/epics/_backend/_aioca.py +++ b/src/ophyd_async/epics/_backend/_aioca.py @@ -2,7 +2,8 @@ import sys from dataclasses import dataclass from enum import Enum -from typing import Any, Dict, Optional, Type, Union +from math import isnan, nan +from typing import Any, Dict, List, Optional, Type, Union import numpy as np from aioca import ( @@ -29,7 +30,7 @@ ) from ophyd_async.core.utils import DEFAULT_TIMEOUT, NotConnected -from .common import get_supported_values +from .common import LimitPair, Limits, common_meta, get_supported_values dbr_to_dtype: Dict[Dbr, Dtype] = { dbr.DBR_STRING: "string", @@ -41,6 +42,64 @@ } +def _data_key_from_augmented_value( + value: AugmentedValue, + *, + choices: Optional[List[str]] = None, + dtype: Optional[str] = None, +) -> DataKey: + """Use the return value of get with FORMAT_CTRL to construct a DataKey + describing the signal. See docstring of AugmentedValue for expected + value fields by DBR type. + + Args: + value (AugmentedValue): Description of the the return type of a DB record + choices: Optional list of enum choices to pass as metadata in the datakey + dtype: Optional override dtype when AugmentedValue is ambiguous, e.g. booleans + + Returns: + DataKey: A rich DataKey describing the DB record + """ + source = f"ca://{value.name}" + assert value.ok, f"Error reading {source}: {value}" + + scalar = value.element_count == 1 + dtype = dtype or dbr_to_dtype[value.datatype] + + d = DataKey( + source=source, + dtype=dtype if scalar else "array", + # strictly value.element_count >= len(value) + shape=[] if scalar else [len(value)], + ) + for key in common_meta: + attr = getattr(value, key, nan) + if isinstance(attr, str) or not isnan(attr): + d[key] = attr + + if choices is not None: + d["choices"] = choices + + if limits := _limits_from_augmented_value(value): + d["limits"] = limits + + return d + + +def _limits_from_augmented_value(value: AugmentedValue) -> Limits: + def get_limits(limit: str) -> LimitPair: + low = getattr(value, f"lower_{limit}_limit", None) + high = getattr(value, f"upper_{limit}_limit", None) + return LimitPair(low=low, high=high) + + return Limits( + alarm=get_limits("alarm"), + control=get_limits("ctrl"), + display=get_limits("disp"), + warning=get_limits("warning"), + ) + + @dataclass class CaConverter: read_dbr: Optional[Dbr] @@ -62,8 +121,8 @@ def reading(self, value: AugmentedValue): "alarm_severity": -1 if value.severity > 2 else value.severity, } - def get_datakey(self, source: str, value: AugmentedValue) -> DataKey: - return {"source": source, "dtype": dbr_to_dtype[value.datatype], "shape": []} + def get_datakey(self, value: AugmentedValue) -> DataKey: + return _data_key_from_augmented_value(value) class CaLongStrConverter(CaConverter): @@ -77,15 +136,17 @@ def write_value(self, value: str): class CaArrayConverter(CaConverter): - def get_datakey(self, source: str, value: AugmentedValue) -> DataKey: - return {"source": source, "dtype": "array", "shape": [len(value)]} - def value(self, value: AugmentedValue): return np.array(value, copy=False) @dataclass class CaEnumConverter(CaConverter): + """To prevent issues when a signal is restarted and returns with different enum + values or orders, we put treat an Enum signal as a string, and cache the + choices on this class. + """ + choices: dict[str, str] def write_value(self, value: Union[Enum, str]): @@ -97,13 +158,18 @@ def write_value(self, value: Union[Enum, str]): def value(self, value: AugmentedValue): return self.choices[value] - def get_datakey(self, source: str, value: AugmentedValue) -> DataKey: - return { - "source": source, - "dtype": "string", - "shape": [], - "choices": list(self.choices), - } + def get_datakey(self, value: AugmentedValue) -> DataKey: + # Sometimes DBR_TYPE returns as String, must pass choices still + return _data_key_from_augmented_value(value, choices=list(self.choices.keys())) + + +@dataclass +class CaBoolConverter(CaConverter): + def value(self, value: AugmentedValue) -> bool: + return bool(value) + + def get_datakey(self, value: AugmentedValue) -> DataKey: + return _data_key_from_augmented_value(value, dtype="bool") class DisconnectedCaConverter(CaConverter): @@ -145,7 +211,7 @@ def make_converter( ) if pv_choices_len != 2: raise TypeError(f"{pv} has {pv_choices_len} choices, can't map to bool") - return CaConverter(dbr.DBR_SHORT, dbr.DBR_SHORT) + return CaBoolConverter(dbr.DBR_SHORT, dbr.DBR_SHORT) elif pv_dbr == dbr.DBR_ENUM: # This is an Enum pv_choices = get_unique( @@ -233,7 +299,7 @@ async def _caget(self, format: Format) -> AugmentedValue: async def get_datakey(self, source: str) -> DataKey: value = await self._caget(FORMAT_CTRL) - return self.converter.get_datakey(source, value) + return self.converter.get_datakey(value) async def get_reading(self) -> Reading: value = await self._caget(FORMAT_TIME) diff --git a/src/ophyd_async/epics/_backend/_p4p.py b/src/ophyd_async/epics/_backend/_p4p.py index d6e2b49b1c..84c8be68ae 100644 --- a/src/ophyd_async/epics/_backend/_p4p.py +++ b/src/ophyd_async/epics/_backend/_p4p.py @@ -4,6 +4,7 @@ import time from dataclasses import dataclass from enum import Enum +from math import isnan, nan from typing import Any, Dict, List, Optional, Sequence, Type, Union from bluesky.protocols import DataKey, Dtype, Reading @@ -20,7 +21,7 @@ ) from ophyd_async.core.utils import DEFAULT_TIMEOUT, NotConnected -from .common import get_supported_values +from .common import LimitPair, Limits, common_meta, get_supported_values # https://mdavidsaver.github.io/p4p/values.html specifier_to_dtype: Dict[str, Dtype] = { @@ -39,6 +40,67 @@ } +def _data_key_from_value( + source: str, + value: Value, + *, + shape: Optional[list[int]] = None, + choices: Optional[list[str]] = None, + dtype: Optional[str] = None, +) -> DataKey: + """ + Args: + value (Value): Description of the the return type of a DB record + shape: Optional override shape when len(shape) > 1 + choices: Optional list of enum choices to pass as metadata in the datakey + dtype: Optional override dtype when AugmentedValue is ambiguous, e.g. booleans + + Returns: + DataKey: A rich DataKey describing the DB record + """ + shape = shape or [] + dtype = dtype or specifier_to_dtype[value.type().aspy("value")] + display_data = getattr(value, "display", None) + + d = DataKey( + source=source, + dtype=dtype, + shape=shape, + ) + if display_data is not None: + for key in common_meta: + attr = getattr(display_data, key, nan) + if isinstance(attr, str) or not isnan(attr): + d[key] = attr + + if choices is not None: + d["choices"] = choices + + if limits := _limits_from_value(value): + d["limits"] = limits + + return d + + +def _limits_from_value(value: Value) -> Limits: + def get_limits( + substucture_name: str, low_name: str = "limitLow", high_name: str = "limitHigh" + ) -> LimitPair: + substructure = getattr(value, substucture_name, None) + low = getattr(substructure, low_name, nan) + high = getattr(substructure, high_name, nan) + return LimitPair( + low=None if isnan(low) else low, high=None if isnan(high) else high + ) + + return Limits( + alarm=get_limits("valueAlarm", "lowAlarmLimit", "highAlarmLimit"), + control=get_limits("control"), + display=get_limits("display"), + warning=get_limits("valueAlarm", "lowWarningLimit", "highWarningLimit"), + ) + + class PvaConverter: def write_value(self, value): return value @@ -56,8 +118,7 @@ def reading(self, value): } def get_datakey(self, source: str, value) -> DataKey: - dtype = specifier_to_dtype[value.type().aspy("value")] - return {"source": source, "dtype": dtype, "shape": []} + return _data_key_from_value(source, value) def metadata_fields(self) -> List[str]: """ @@ -74,7 +135,9 @@ def value_fields(self) -> List[str]: class PvaArrayConverter(PvaConverter): def get_datakey(self, source: str, value) -> DataKey: - return {"source": source, "dtype": "array", "shape": [len(value["value"])]} + return _data_key_from_value( + source, value, dtype="array", shape=[len(value["value"])] + ) class PvaNDArrayConverter(PvaConverter): @@ -98,7 +161,7 @@ def value(self, value): def get_datakey(self, source: str, value) -> DataKey: dims = self._get_dimensions(value) - return {"source": source, "dtype": "array", "shape": dims} + return _data_key_from_value(source, value, dtype="array", shape=dims) def write_value(self, value): # No clear use-case for writing directly to an NDArray, and some @@ -109,6 +172,11 @@ def write_value(self, value): @dataclass class PvaEnumConverter(PvaConverter): + """To prevent issues when a signal is restarted and returns with different enum + values or orders, we put treat an Enum signal as a string, and cache the + choices on this class. + """ + def __init__(self, choices: dict[str, str]): self.choices = tuple(choices.values()) @@ -122,20 +190,17 @@ def value(self, value): return self.choices[value["value"]["index"]] def get_datakey(self, source: str, value) -> DataKey: - return { - "source": source, - "dtype": "string", - "shape": [], - "choices": list(self.choices), - } + return _data_key_from_value( + source, value, choices=list(self.choices), dtype="string" + ) -class PvaEnumBoolConverter(PvaConverter): +class PvaEmumBoolConverter(PvaConverter): def value(self, value): - return value["value"]["index"] + return bool(value["value"]["index"]) def get_datakey(self, source: str, value) -> DataKey: - return {"source": source, "dtype": "integer", "shape": []} + return _data_key_from_value(source, value, dtype="bool") class PvaTableConverter(PvaConverter): @@ -144,7 +209,7 @@ def value(self, value): def get_datakey(self, source: str, value) -> DataKey: # This is wrong, but defer until we know how to actually describe a table - return {"source": source, "dtype": "object", "shape": []} # type: ignore + return _data_key_from_value(source, value, dtype="object") class PvaDictConverter(PvaConverter): @@ -213,7 +278,7 @@ def make_converter(datatype: Optional[Type], values: Dict[str, Any]) -> PvaConve ) if pv_choices_len != 2: raise TypeError(f"{pv} has {pv_choices_len} choices, can't map to bool") - return PvaEnumBoolConverter() + return PvaEmumBoolConverter() elif "NTEnum" in typeid: # This is an Enum pv_choices = get_unique( diff --git a/src/ophyd_async/epics/_backend/common.py b/src/ophyd_async/epics/_backend/common.py index bb89dfa756..e5a7453823 100644 --- a/src/ophyd_async/epics/_backend/common.py +++ b/src/ophyd_async/epics/_backend/common.py @@ -1,5 +1,28 @@ from enum import Enum -from typing import Dict, Optional, Tuple, Type +from typing import Dict, Optional, Tuple, Type, TypedDict + +common_meta = { + "units", + "precision", +} + + +class LimitPair(TypedDict): + high: float | None + low: float | None + + def __bool__(self) -> bool: + return self.low is None and self.high is None + + +class Limits(TypedDict): + alarm: LimitPair + control: LimitPair + display: LimitPair + warning: LimitPair + + def __bool__(self) -> bool: + return any(self.alarm, self.control, self.display, self.warning) def get_supported_values( diff --git a/tests/epics/test_records.db b/tests/epics/test_records.db index fad91d8f09..7bbb46f51c 100644 --- a/tests/epics/test_records.db +++ b/tests/epics/test_records.db @@ -11,6 +11,10 @@ record(bo, "$(P)bool_unnamed") { } record(longout, "$(P)int") { + field(LLSV, "MAJOR") # LOLO is alarm + field(LSV, "MINOR") # LOW is warning + field(HSV, "MINOR") # HIGH is warning + field(HHSV, "MAJOR") # HIHI is alarm field(HOPR, "100") field(HIHI, "98") field(HIGH, "96") @@ -23,6 +27,30 @@ record(longout, "$(P)int") { field(PINI, "YES") } +record(longout, "$(P)partialint") { + field(LLSV, "MAJOR") # LOLO is alarm + field(HHSV, "MAJOR") # HIHI is alarm + field(HOPR, "100") + field(HIHI, "98") + field(DRVH, "90") + field(DRVL, "10") + field(LOLO, "2") + field(LOPR, "0") + field(VAL, "42") + field(PINI, "YES") +} + +record(longout, "$(P)lessint") { + field(HSV, "MINOR") # LOW is warning + field(LSV, "MINOR") # HIGH is warning + field(HOPR, "100") + field(HIGH, "98") + field(LOW, "2") + field(LOPR, "0") + field(VAL, "42") + field(PINI, "YES") +} + record(ao, "$(P)float") { field(PREC, "1") field(EGU, "mm") diff --git a/tests/epics/test_signals.py b/tests/epics/test_signals.py index 7e80545da1..764c2c3644 100644 --- a/tests/epics/test_signals.py +++ b/tests/epics/test_signals.py @@ -10,27 +10,18 @@ from enum import Enum from pathlib import Path from types import GenericAlias -from typing import ( - Any, - Callable, - Dict, - Literal, - Optional, - Sequence, - Tuple, - Type, - TypedDict, -) +from typing import Any, Dict, Literal, Optional, Sequence, Tuple, Type, TypedDict from unittest.mock import ANY import numpy as np import numpy.typing as npt import pytest from aioca import CANothing, purge_channel_caches -from bluesky.protocols import Reading +from bluesky.protocols import DataKey, Reading from ophyd_async.core import SignalBackend, T, load_from_yaml, save_to_yaml from ophyd_async.core.utils import NotConnected +from ophyd_async.epics._backend.common import LimitPair, Limits from ophyd_async.epics.signal._epics_transport import EpicsTransport from ophyd_async.epics.signal.signal import ( _make_backend, @@ -164,7 +155,7 @@ def _is_numpy_subclass(t): async def assert_monitor_then_put( ioc: IOC, suffix: str, - descriptor: dict, + datakey: dict, initial_value: T, put_value: T, datatype: Optional[Type[T]] = None, @@ -174,9 +165,9 @@ async def assert_monitor_then_put( # Make a monitor queue that will monitor for updates q = MonitorQueue(backend) try: - # Check descriptor - pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:{suffix}" - assert dict(source=pv_name, **descriptor) == await backend.get_datakey(pv_name) + # Check datakey + source = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:{suffix}" + assert dict(source=source, **datakey) == await backend.get_datakey(source) # Check initial value await q.assert_updates( pytest.approx(initial_value), @@ -211,24 +202,53 @@ class MyEnum(str, Enum): c = "Ccc" -def integer_d(value): - return {"dtype": "integer", "shape": []} - - -def number_d(value): - return {"dtype": "number", "shape": []} - - -def string_d(value): - return {"dtype": "string", "shape": []} - - -def enum_d(value): - return {"dtype": "string", "shape": [], "choices": ["Aaa", "Bbb", "Ccc"]} - - -def waveform_d(value): - return {"dtype": "array", "shape": [len(value)]} +_metadata: Dict[str, Dict[str, Dict[str, Any]]] = { + "ca": { + "bool": {"units": ANY, "limits": ANY}, + "integer": {"units": ANY, "limits": ANY}, + "number": {"units": ANY, "limits": ANY, "precision": ANY}, + "enum": {"limits": ANY}, + "string": {"limits": ANY}, + }, + "pva": { + "bool": {"limits": ANY}, + "integer": {"units": ANY, "precision": ANY, "limits": ANY}, + "number": {"units": ANY, "precision": ANY, "limits": ANY}, + "enum": {"limits": ANY}, + "string": {"units": ANY, "precision": ANY, "limits": ANY}, + }, +} + + +def datakey(protocol: str, suffix: str, value=None) -> DataKey: + def get_internal_dtype(suffix: str) -> str: + # uint32, [u]int64 backed by DBR_DOUBLE, have precision + if "float" in suffix or "uint32" in suffix or "int64" in suffix: + return "number" + if "int" in suffix: + return "integer" + if "bool" in suffix: + return "bool" + if "enum" in suffix: + return "enum" + return "string" + + def get_dtype(suffix: str) -> str: + if suffix.endswith("a"): + return "array" + if "enum" in suffix: + return "string" + return get_internal_dtype(suffix) + + dtype = get_dtype(suffix) + + d = {"dtype": dtype, "shape": [len(value)] if dtype == "array" else []} + if get_internal_dtype(suffix) == "enum": + d["choices"] = [e.value for e in type(value)] + + d.update(_metadata[protocol].get(get_internal_dtype(suffix), {})) + + return d ls1 = "a string that is just longer than forty characters" @@ -236,21 +256,19 @@ def waveform_d(value): @pytest.mark.parametrize( - "datatype, suffix, initial_value, put_value, descriptor, supported_backends", + "datatype, suffix, initial_value, put_value, supported_backends", [ # python builtin scalars - (int, "int", 42, 43, integer_d, {"ca", "pva"}), - (float, "float", 3.141, 43.5, number_d, {"ca", "pva"}), - (str, "str", "hello", "goodbye", string_d, {"ca", "pva"}), - (MyEnum, "enum", MyEnum.b, MyEnum.c, enum_d, {"ca", "pva"}), - (str, "enum", "Bbb", "Ccc", enum_d, {"ca", "pva"}), + (int, "int", 42, 43, {"ca", "pva"}), + (float, "float", 3.141, 43.5, {"ca", "pva"}), + (str, "str", "hello", "goodbye", {"ca", "pva"}), + (MyEnum, "enum", MyEnum.b, MyEnum.c, {"ca", "pva"}), # numpy arrays of numpy types ( npt.NDArray[np.int8], "int8a", [-128, 127], [-8, 3, 44], - waveform_d, {"pva"}, ), ( @@ -258,7 +276,6 @@ def waveform_d(value): "uint8a", [0, 255], [218], - waveform_d, {"ca", "pva"}, ), ( @@ -266,7 +283,6 @@ def waveform_d(value): "int16a", [-32768, 32767], [-855], - waveform_d, {"ca", "pva"}, ), ( @@ -274,7 +290,6 @@ def waveform_d(value): "uint16a", [0, 65535], [5666], - waveform_d, {"pva"}, ), ( @@ -282,7 +297,6 @@ def waveform_d(value): "int32a", [-2147483648, 2147483647], [-2], - waveform_d, {"ca", "pva"}, ), ( @@ -290,7 +304,6 @@ def waveform_d(value): "uint32a", [0, 4294967295], [1022233], - waveform_d, {"pva"}, ), ( @@ -298,7 +311,6 @@ def waveform_d(value): "int64a", [-2147483649, 2147483648], [-3], - waveform_d, {"pva"}, ), ( @@ -306,7 +318,6 @@ def waveform_d(value): "uint64a", [0, 4294967297], [995444], - waveform_d, {"pva"}, ), ( @@ -314,7 +325,6 @@ def waveform_d(value): "float32a", [0.000002, -123.123], [1.0], - waveform_d, {"ca", "pva"}, ), ( @@ -322,7 +332,6 @@ def waveform_d(value): "float64a", [0.1, -12345678.123], [0.2], - waveform_d, {"ca", "pva"}, ), ( @@ -330,7 +339,6 @@ def waveform_d(value): "stra", ["five", "six", "seven"], ["nine", "ten"], - waveform_d, {"pva"}, ), ( @@ -338,12 +346,11 @@ def waveform_d(value): "stra", ["five", "six", "seven"], ["nine", "ten"], - waveform_d, {"ca"}, ), # Can't do long strings until https://github.com/epics-base/pva2pva/issues/17 - # (str, "longstr", ls1, ls2, string_d), - # (str, "longstr2.VAL$", ls1, ls2, string_d), + # (str, "longstr", ls1, ls2), + # (str, "longstr2.VAL$", ls1, ls2), ], ) async def test_backend_get_put_monitor( @@ -352,7 +359,6 @@ async def test_backend_get_put_monitor( suffix: str, initial_value: T, put_value: T, - descriptor: Callable[[Any], dict], tmp_path, supported_backends: set[str], ): @@ -366,7 +372,7 @@ async def test_backend_get_put_monitor( await assert_monitor_then_put( ioc, suffix, - descriptor(initial_value), + datakey(ioc.protocol, suffix, initial_value), initial_value, put_value, datatype, @@ -375,7 +381,7 @@ async def test_backend_get_put_monitor( await assert_monitor_then_put( ioc, suffix, - descriptor(put_value), + datakey(ioc.protocol, suffix, put_value), put_value, initial_value, datatype=None, @@ -388,17 +394,39 @@ async def test_backend_get_put_monitor( @pytest.mark.parametrize("suffix", ["bool", "bool_unnamed"]) -async def test_bool_conversion_of_enum(ioc: IOC, suffix: str) -> None: +async def test_bool_conversion_of_enum(ioc: IOC, suffix: str, tmp_path) -> None: + """Booleans are converted to Short Enumerations with values 0,1 as database does + not support boolean natively. + The flow of test_backend_get_put_monitor Gets a value with a dtype of None: we + cannot tell the difference between an enum with 2 members and a boolean, so + cannot get a DataKey that does not mutate form. + This test otherwise performs the same. + """ + # With the given datatype, check we have the correct initial value and putting + # works await assert_monitor_then_put( ioc, - suffix=suffix, - descriptor=integer_d(True), - initial_value=True, - put_value=False, - datatype=bool, - check_type=False, + suffix, + datakey(ioc.protocol, suffix), + True, + False, + bool, + ) + # With datatype guessed from CA/PVA, check we can set it back to the initial value + await assert_monitor_then_put( + ioc, + suffix, + datakey(ioc.protocol, suffix, True), + False, + True, + bool, ) + yaml_path = tmp_path / "test.yaml" + save_to_yaml([{"test": False}], yaml_path) + loaded = load_from_yaml(yaml_path) + assert np.all(loaded[0]["test"] is False) + async def test_error_raised_on_disconnected_PV(ioc: IOC) -> None: if ioc.protocol == "pva": @@ -534,17 +562,15 @@ async def test_pva_table(ioc: IOC) -> None: enum=[MyEnum.c, MyEnum.b], ) # TODO: what should this be for a variable length table? - descriptor = {"dtype": "object", "shape": []} + datakey = {"dtype": "object", "shape": [], "source": "test-source"} # Make and connect the backend for t, i, p in [(MyTable, initial, put), (None, put, initial)]: backend = await ioc.make_backend(t, "table") # Make a monitor queue that will monitor for updates q = MonitorQueue(backend) try: - # Check descriptor - dict(source="test-source", **descriptor) == await backend.get_datakey( - "test-source" - ) + # Check datakey + datakey == await backend.get_datakey("test-source") # Check initial value await q.assert_updates(approx_table(i)) # Put to new value and check that @@ -577,7 +603,7 @@ async def test_pvi_structure(ioc: IOC) -> None: } try: - # Check descriptor + # Check datakey with pytest.raises(NotImplementedError): await backend.get_datakey("") # Check initial value @@ -609,6 +635,7 @@ async def test_pva_ntdarray(ioc: IOC): "source": "test-source", "dtype": "array", "shape": [2, 3], + "limits": ANY, } == await backend.get_datakey("test-source") # Check initial value await q.assert_updates(pytest.approx(i)) @@ -694,3 +721,99 @@ async def test_str_returns_enum(ioc: IOC): assert val == MyEnum.b assert val == "Bbb" assert val is not MyEnum.b + + +async def test_signal_returns_units_and_precision(ioc: IOC): + await ioc.make_backend(float, "float") + pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:float" + + sig = epics_signal_rw(float, pv_name) + await sig.connect() + datakey = (await sig.describe())[""] + assert datakey["units"] == "mm" + assert datakey["precision"] == 1 + + +async def test_signal_not_return_none_units_and_precision(ioc: IOC): + await ioc.make_backend(str, "str") + pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:str" + + sig = epics_signal_rw(str, pv_name) + await sig.connect() + datakey = (await sig.describe())[""] + assert not hasattr(datakey, "units") + assert not hasattr(datakey, "precision") + + +async def test_signal_returns_limits(ioc: IOC): + await ioc.make_backend(int, "int") + pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:int" + + expected_limits = Limits( + # LOW, HIGH + warning=LimitPair(low=5.0, high=96.0), + # DRVL, DRVH + control=LimitPair(low=10.0, high=90.0), + # LOPR, HOPR + display=LimitPair(low=0.0, high=100.0), + # LOLO, HIHI + alarm=LimitPair(low=2.0, high=98.0), + ) + + sig = epics_signal_rw(int, pv_name) + await sig.connect() + limits = (await sig.describe())[""]["limits"] + assert limits == expected_limits + + +async def test_signal_returns_partial_limits(ioc: IOC): + await ioc.make_backend(int, "partialint") + pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:partialint" + not_set = 0 if ioc.protocol == "ca" else None + + expected_limits = Limits( + # LOLO, HIHI + alarm=LimitPair(low=2.0, high=98.0), + # DRVL, DRVH + control=LimitPair(low=10.0, high=90.0), + # LOPR, HOPR + display=LimitPair(low=0.0, high=100.0), + # HSV, LSV not set. + warning=LimitPair(low=not_set, high=not_set), + ) + + sig = epics_signal_rw(int, pv_name) + await sig.connect() + limits = (await sig.describe())[""]["limits"] + assert limits == expected_limits + + +async def test_signal_returns_warning_and_partial_limits(ioc: IOC): + await ioc.make_backend(int, "lessint") + pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:lessint" + not_set = 0 if ioc.protocol == "ca" else None + + expected_limits = Limits( + # HSV, LSV not set + alarm=LimitPair(low=not_set, high=not_set), + # control = display if DRVL, DRVH not set + control=LimitPair(low=0.0, high=100.0), + # LOPR, HOPR + display=LimitPair(low=0.0, high=100.0), + # LOW, HIGH + warning=LimitPair(low=2.0, high=98.0), + ) + + sig = epics_signal_rw(int, pv_name) + await sig.connect() + limits = (await sig.describe())[""]["limits"] + assert limits == expected_limits + + +async def test_signal_not_return_no_limits(ioc: IOC): + await ioc.make_backend(MyEnum, "enum") + pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:enum" + sig = epics_signal_rw(MyEnum, pv_name) + await sig.connect() + datakey = (await sig.describe())[""] + assert not hasattr(datakey, "limits")