Implement new DataType fields and Attribute.description in EpicsBackend

Author: evvaaaa

src/fastcs/backends/epics/ioc.py

@@ -172,22 +172,43 @@ async def async_record_set(value: T):
 
 
 def _get_input_record(pv: str, attribute: AttrR) -> RecordWrapper:
+    attribute_fields = {}
+    if attribute.description is not None:
+        attribute_fields.update({"DESC": attribute.description})
+
     if attr_is_enum(attribute):
         assert attribute.allowed_values is not None and all(
             isinstance(v, str) for v in attribute.allowed_values
         )
         state_keys = dict(zip(MBB_STATE_FIELDS, attribute.allowed_values, strict=False))
-        return builder.mbbIn(pv, **state_keys)
+        return builder.mbbIn(pv, **state_keys, **attribute_fields)
 
     match attribute.datatype:
         case Bool(znam, onam):
-            return builder.boolIn(pv, ZNAM=znam, ONAM=onam)
-        case Int():
-            return builder.longIn(pv)
-        case Float(prec):
-            return builder.aIn(pv, PREC=prec)
+            return builder.boolIn(pv, ZNAM=znam, ONAM=onam, **attribute_fields)
+        case Int(units, min, max, min_alarm, max_alarm):
+            return builder.longIn(
+                pv,
+                EGU=units,
+                DRVL=min,
+                DRVH=max,
+                LOPR=min_alarm,
+                HOPR=max_alarm,
+                **attribute_fields,
+            )
+        case Float(prec, units, min, max, min_alarm, max_alarm):
+            return builder.aIn(
+                pv,
+                PREC=prec,
+                EGU=units,
+                DRVL=min,
+                DRVH=max,
+                LOPR=min_alarm,
+                HOPR=max_alarm,
+                **attribute_fields,
+            )
         case String():
-            return builder.longStringIn(pv)
+            return builder.longStringIn(pv, **attribute_fields)
         case _:
             raise FastCSException(
                 f"Unsupported type {type(attribute.datatype)}: {attribute.datatype}"
@@ -224,12 +245,21 @@ async def async_write_display(value: T):
 
 
 def _get_output_record(pv: str, attribute: AttrW, on_update: Callable) -> Any:
+    attribute_fields = {}
+    if attribute.description is not None:
+        attribute_fields.update({"DESC": attribute.description})
     if attr_is_enum(attribute):
         assert attribute.allowed_values is not None and all(
             isinstance(v, str) for v in attribute.allowed_values
         )
         state_keys = dict(zip(MBB_STATE_FIELDS, attribute.allowed_values, strict=False))
-        return builder.mbbOut(pv, always_update=True, on_update=on_update, **state_keys)
+        return builder.mbbOut(
+            pv,
+            always_update=True,
+            on_update=on_update,
+            **state_keys,
+            **attribute_fields,
+        )
 
     match attribute.datatype:
         case Bool(znam, onam):
@@ -240,12 +270,35 @@ def _get_output_record(pv: str, attribute: AttrW, on_update: Callable) -> Any:
                 always_update=True,
                 on_update=on_update,
             )
-        case Int():
-            return builder.longOut(pv, always_update=True, on_update=on_update)
-        case Float(prec):
-            return builder.aOut(pv, always_update=True, on_update=on_update, PREC=prec)
+        case Int(units, min, max, min_alarm, max_alarm):
+            return builder.longOut(
+                pv,
+                always_update=True,
+                on_update=on_update,
+                EGU=units,
+                DRVL=min,
+                DRVH=max,
+                LOPR=min_alarm,
+                HOPR=max_alarm,
+                **attribute_fields,
+            )
+        case Float(prec, units, min, max, min_alarm, max_alarm):
+            return builder.aOut(
+                pv,
+                always_update=True,
+                on_update=on_update,
+                PREC=prec,
+                EGU=units,
+                DRVL=min,
+                DRVH=max,
+                LOPR=min_alarm,
+                HOPR=max_alarm,
+                **attribute_fields,
+            )
         case String():
-            return builder.longStringOut(pv, always_update=True, on_update=on_update)
+            return builder.longStringOut(
+                pv, always_update=True, on_update=on_update, **attribute_fields
+            )
         case _:
             raise FastCSException(
                 f"Unsupported type {type(attribute.datatype)}: {attribute.datatype}"

tests/backends/epics/test_ioc.py

@@ -211,6 +211,15 @@ def test_get_output_record_raises(mocker: MockerFixture):
         _get_output_record("PV", mocker.MagicMock(), on_update=mocker.MagicMock())
 
 
+DEFAULT_SCALAR_FIELD_ARGS = {
+    "EGU": None,
+    "DRVL": None,
+    "DRVH": None,
+    "LOPR": None,
+    "HOPR": None,
+}
+
+
 def test_ioc(mocker: MockerFixture, mapping: Mapping):
     builder = mocker.patch("fastcs.backends.epics.ioc.builder")
     add_pvi_info = mocker.patch("fastcs.backends.epics.ioc._add_pvi_info")
@@ -222,15 +231,26 @@ def test_ioc(mocker: MockerFixture, mapping: Mapping):
 
     # Check records are created
     builder.boolIn.assert_called_once_with(f"{DEVICE}:ReadBool", ZNAM="OFF", ONAM="ON")
-    builder.longIn.assert_any_call(f"{DEVICE}:ReadInt")
-    builder.aIn.assert_called_once_with(f"{DEVICE}:ReadWriteFloat_RBV", PREC=2)
+    builder.longIn.assert_any_call(f"{DEVICE}:ReadInt", **DEFAULT_SCALAR_FIELD_ARGS)
+    builder.aIn.assert_called_once_with(
+        f"{DEVICE}:ReadWriteFloat_RBV", PREC=2, **DEFAULT_SCALAR_FIELD_ARGS
+    )
     builder.aOut.assert_any_call(
-        f"{DEVICE}:ReadWriteFloat", always_update=True, on_update=mocker.ANY, PREC=2
+        f"{DEVICE}:ReadWriteFloat",
+        always_update=True,
+        on_update=mocker.ANY,
+        PREC=2,
+        **DEFAULT_SCALAR_FIELD_ARGS,
+    )
+    builder.longIn.assert_any_call(f"{DEVICE}:BigEnum", **DEFAULT_SCALAR_FIELD_ARGS)
+    builder.longIn.assert_any_call(
+        f"{DEVICE}:ReadWriteInt_RBV", **DEFAULT_SCALAR_FIELD_ARGS
     )
-    builder.longIn.assert_any_call(f"{DEVICE}:BigEnum")
-    builder.longIn.assert_any_call(f"{DEVICE}:ReadWriteInt_RBV")
     builder.longOut.assert_called_with(
-        f"{DEVICE}:ReadWriteInt", always_update=True, on_update=mocker.ANY
+        f"{DEVICE}:ReadWriteInt",
+        always_update=True,
+        on_update=mocker.ANY,
+        **DEFAULT_SCALAR_FIELD_ARGS,
     )
     builder.mbbIn.assert_called_once_with(
         f"{DEVICE}:StringEnum_RBV", ZRST="red", ONST="green", TWST="blue"
@@ -390,9 +410,10 @@ def test_long_pv_names_discarded(mocker: MockerFixture):
         f"{DEVICE}:{short_pv_name}",
         always_update=True,
         on_update=mocker.ANY,
+        **DEFAULT_SCALAR_FIELD_ARGS,
     )
     builder.longIn.assert_called_once_with(
-        f"{DEVICE}:{short_pv_name}_RBV",
+        f"{DEVICE}:{short_pv_name}_RBV", **DEFAULT_SCALAR_FIELD_ARGS
     )
 
     long_pv_name = long_attr_name.title().replace("_", "")