Merge pull request #902 from BCDA-APS/628-acalcout-userArrayCalcs

add acalcout record and userArrayCalcs

CHANGES.rst

@@ -38,6 +38,7 @@ New Features
   * Measurement Computing USB CTR08 High-Speed Counter/Timer
   * Simulated process controller as positioner using EPICS swait record.
   * Simulated process controller as positioner using EPICS transform record.
+  * synApps userArrayCalcs and EPICS acalcout record.
 * Add subnet check for APSU beamlines.
 * Add template support for writing NeXus/HDF5 files.
 * New lineup2() plan can be used in console, notebooks, and queueserver.

apstools/synApps/__init__.py

@@ -7,6 +7,8 @@
 from ._common import EpicsRecordInputFields
 from ._common import EpicsRecordOutputFields
 from ._common import EpicsSynAppsRecordEnableMixin
+from .acalcout import AcalcoutRecord
+from .acalcout import UserArrayCalcDevice
 from .asyn import AsynRecord
 from .busy import BusyRecord
 from .calcout import CalcoutRecord

apstools/synApps/acalcout.py

@@ -0,0 +1,229 @@
+"""
+Ophyd support for the EPICS acalcout record
+
+https://epics-modules.github.io/calc/aCalcoutRecord.html
+
+Public Structures
+
+.. autosummary::
+
+    ~UserArrayCalcDevice
+    ~UserArrayCalcN
+    ~AcalcoutRecord
+    ~AcalcoutArrayRecordChannel
+    ~AcalcoutRecordChannel
+"""
+
+from collections import OrderedDict
+
+from ophyd import Component as Cpt
+from ophyd import Device
+from ophyd import DynamicDeviceComponent as DDC
+from ophyd import EpicsSignal
+from ophyd import EpicsSignalRO
+from ophyd import FormattedComponent as FC
+
+from .. import utils as APS_utils
+from ._common import EpicsRecordDeviceCommonAll
+from ._common import EpicsRecordFloatFields
+from ._common import EpicsSynAppsRecordEnableMixin
+
+CHANNEL_LETTERS_LIST = "A B C D E F G H I J K L".split()
+
+
+class AcalcoutRecordChannel(Device):
+    """
+    Float channel of a acalcout record: A-L.
+
+    .. index:: Ophyd Device; synApps AcalcoutRecordChannel
+
+    .. autosummary::
+
+        ~reset
+    """
+
+    input_value = FC(EpicsSignal, "{prefix}.{_ch_letter}", kind="config")
+    # last_value = FC(EpicsSignalRO, "{prefix}.L{_ch_letter}", kind="config")
+    input_pv = FC(EpicsSignal, "{prefix}.INP{_ch_letter}", kind="config")
+    input_pv_valid = FC(EpicsSignalRO, "{prefix}.IN{_ch_letter}V", kind="config")
+
+    read_attrs = [
+        "input_value",
+    ]
+    hints = {"fields": read_attrs}
+
+    def __init__(self, prefix, letter, **kwargs):
+        self._ch_letter = letter
+        super().__init__(prefix, **kwargs)
+
+    def reset(self):
+        """set all fields to default values"""
+        self.input_pv.put("")
+        self.input_value.put(0)
+
+
+class AcalcoutArrayRecordChannel(Device):
+    """
+    Array channel of a acalcout record: A-L.
+
+    .. index:: Ophyd Device; synApps AcalcoutArrayRecordChannel
+
+    .. autosummary::
+
+        ~reset
+    """
+
+    input_value = FC(EpicsSignal, "{prefix}.{_ch_letters}", kind="config")
+    input_pv = FC(EpicsSignal, "{prefix}.IN{_ch_letters}", kind="config")
+
+    read_attrs = [
+        "input_value",
+    ]
+    hints = {"fields": read_attrs}
+
+    def __init__(self, prefix, letter, **kwargs):
+        self._ch_letters = letter + letter
+        super().__init__(prefix, **kwargs)
+
+    def reset(self):
+        """set all fields to default values"""
+        self.input_pv.put("")
+        self.input_value.put([])
+
+
+def _channels(channel_list):
+    defn = OrderedDict()
+    for chan in channel_list:
+        defn[chan] = (AcalcoutRecordChannel, "", {"letter": chan})
+    for chan in channel_list:
+        defn[chan + chan] = (AcalcoutArrayRecordChannel, "", {"letter": chan})
+    return defn
+
+
+class AcalcoutRecord(EpicsRecordFloatFields, EpicsRecordDeviceCommonAll):
+    """
+    EPICS base aCalcout record support in ophyd
+
+    .. index:: Ophyd Device; synApps aCalcout
+
+    .. autosummary::
+
+        ~reset
+
+    :see: https://epics-modules.github.io/calc/aCalcoutRecord.html
+    """
+
+    units = Cpt(EpicsSignal, ".EGU", kind="config")
+    precision = Cpt(EpicsSignal, ".PREC", kind="config")
+
+    calculated_value = Cpt(EpicsSignal, ".VAL", kind="normal")
+    calculated_array = Cpt(EpicsSignalRO, ".AVAL", kind="normal")
+    calculation = Cpt(EpicsSignal, ".CALC", kind="config")
+
+    output_pv = Cpt(EpicsSignal, ".OUT", kind="config")
+    output_execute_option = Cpt(EpicsSignal, ".OOPT", kind="config")
+    output_execution_delay = Cpt(EpicsSignal, ".ODLY", kind="config")
+    output_data_option = Cpt(EpicsSignal, ".DOPT", kind="config")
+    output_calculation = Cpt(EpicsSignal, ".OCAL", kind="config")
+    output_value = Cpt(EpicsSignal, ".OVAL", kind="hinted")
+    output_array = Cpt(EpicsSignalRO, ".OAV", kind="hinted")
+    invalid_output_action = Cpt(EpicsSignal, ".IVOA", kind="config")
+    invalid_output_value = Cpt(EpicsSignal, ".IVOV", kind="config")
+    event_to_issue = Cpt(EpicsSignal, ".OEVT", kind="config")
+
+    output_pv_status = Cpt(EpicsSignal, ".OUTV", kind="config")
+    calculation_valid = Cpt(EpicsSignal, ".CLCV", kind="config")
+    output_calculation_valid = Cpt(EpicsSignal, ".OCLV", kind="config")
+    output_delay_active = Cpt(EpicsSignal, ".DLYA", kind="config")
+
+    array_elements_allocated = Cpt(EpicsSignalRO, ".NELM", kind="config")
+    array_elements_used = Cpt(EpicsSignal, ".NUSE", kind="config")
+    array_size_choice = Cpt(EpicsSignal, ".SIZE", kind="config", string=True)
+
+    channels = DDC(_channels(CHANNEL_LETTERS_LIST))
+
+    read_attrs = APS_utils.itemizer("channels.%s", CHANNEL_LETTERS_LIST)
+    hints = {"fields": read_attrs}
+
+    @property
+    def value(self):
+        return self.calculated_value.get()
+
+    def reset(self):
+        """set all fields to default values"""
+        pvname = self.description.pvname.split(".")[0]
+        self.scanning_rate.put("Passive")
+        self.description.put(pvname)
+        self.units.put("")
+        self.precision.put("5")
+
+        self.calculation.put("0")
+        self.calculated_value.put(0)
+        self.output_calculation.put("")
+        self.output_value.put(0)
+
+        self.forward_link.put("")
+        self.output_pv.put("")
+        self.invalid_output_action.put(0)
+        self.invalid_output_value.put(0)
+
+        self.output_execution_delay.put(0)
+        self.output_execute_option.put(0)
+        self.output_data_option.put(0)
+
+        self.array_elements_used.put(self.array_elements_allocated.get())
+        self.array_size_choice.put("NELM")
+
+        for letter in self.channels.read_attrs:
+            channel = getattr(self.channels, letter)
+            if isinstance(channel, (AcalcoutRecordChannel, AcalcoutArrayRecordChannel)):
+                channel.reset()
+        self.hints = {"fields": ["channels.%s" % c for c in CHANNEL_LETTERS_LIST]}
+        self.read_attrs = ["channels.%s" % c for c in CHANNEL_LETTERS_LIST]
+
+
+class UserArrayCalcN(EpicsSynAppsRecordEnableMixin, AcalcoutRecord):
+    """Single instance of the userCalcoutN database."""
+
+
+class UserArrayCalcDevice(Device):
+    """
+    EPICS synApps XXX IOC setup of user aCalcouts: ``$(P):userArrayCalc$(N)``
+
+    .. index:: Ophyd Device; synApps UserArrayCalcDevice
+
+    .. autosummary::
+
+        ~reset
+
+    """
+
+    enable = Cpt(EpicsSignal, "userArrayCalcEnable", kind="omitted")
+    acalcout1 = Cpt(UserArrayCalcN, "userArrayCalc1")
+    acalcout2 = Cpt(UserArrayCalcN, "userArrayCalc2")
+    acalcout3 = Cpt(UserArrayCalcN, "userArrayCalc3")
+    acalcout4 = Cpt(UserArrayCalcN, "userArrayCalc4")
+    acalcout5 = Cpt(UserArrayCalcN, "userArrayCalc5")
+    acalcout6 = Cpt(UserArrayCalcN, "userArrayCalc6")
+    acalcout7 = Cpt(UserArrayCalcN, "userArrayCalc7")
+    acalcout8 = Cpt(UserArrayCalcN, "userArrayCalc8")
+    acalcout9 = Cpt(UserArrayCalcN, "userArrayCalc9")
+    acalcout10 = Cpt(UserArrayCalcN, "userArrayCalc10")
+
+    def reset(self):
+        """set all fields to default values"""
+        for i in range(10):
+            getattr(self, f"acalcout{i+1}").reset()
+        self.read_attrs = ["acalcout%d" % (c + 1) for c in range(10)]
+        self.read_attrs.insert(0, "enable")
+
+
+# -----------------------------------------------------------------------------
+# :author:    Pete R. Jemian
+# :email:     jemian@anl.gov
+# :copyright: (c) 2017-2024, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Argonne National Laboratory Open Source License.
+#
+# The full license is in the file LICENSE.txt, distributed with this software.
+# -----------------------------------------------------------------------------

apstools/synApps/tests/test_acalcout.py

@@ -0,0 +1,69 @@
+"""Test the aCalcout support."""
+
+import math
+import time
+
+import pytest
+
+from ...tests import IOC_GP
+from ...tests import common_attribute_quantities_test
+from ...tests import timed_pause
+from .. import AcalcoutRecord
+from .. import UserArrayCalcDevice
+
+TEST_PV = f"{IOC_GP}userArrayCalc10"
+
+
+def test_connected():
+    acalcout = AcalcoutRecord(TEST_PV, name="acalcout")
+    timed_pause(0.25)
+    if not acalcout.connected:
+        for nm in acalcout.component_names:
+            assert getattr(acalcout, nm).connected, f"{nm}"
+
+
+@pytest.mark.parametrize(
+    "device, pv, connect, attr, expected",
+    [
+        [AcalcoutRecord, TEST_PV, False, "read_attrs", 12],
+        [AcalcoutRecord, TEST_PV, False, "configuration_attrs", 127],
+        [AcalcoutRecord, TEST_PV, True, "read()", 4],
+        [AcalcoutRecord, TEST_PV, True, "summary()", 266],
+        [UserArrayCalcDevice, IOC_GP, False, "read_attrs", 130],
+        [UserArrayCalcDevice, IOC_GP, False, "configuration_attrs", 1290],
+        [UserArrayCalcDevice, IOC_GP, True, "read()", 40],
+        [UserArrayCalcDevice, IOC_GP, True, "summary()", 2505],
+    ],
+)
+def test_attribute_quantities(device, pv, connect, attr, expected):
+    """Verify the quantities of the different attributes."""
+    common_attribute_quantities_test(device, pv, connect, attr, expected)
+
+
+def test_calcout_reset():
+    user = UserArrayCalcDevice(IOC_GP, name="user")
+    user.wait_for_connection(timeout=10)
+    user.enable.put("Enable")
+
+    acalcout = user.acalcout10
+    assert isinstance(acalcout, AcalcoutRecord)
+    acalcout.enable.put("E")  # Note: only "E"
+
+    # Change a few things from the default so reset can be tested.
+    timed_pause()
+    acalcout.description.put("sum the AA array")
+    acalcout.channels.AA.input_value.put([0, 0, 5, 0, 0])
+    acalcout.array_elements_used.put(5)
+    acalcout.calculation.put("sum(AA)")
+    timed_pause()
+    assert math.isclose(5, acalcout.calculated_value.get(), abs_tol=0.1)
+
+    # Reset and test for defaults.
+    acalcout.reset()
+    timed_pause()
+    assert acalcout.description.get() == acalcout.prefix
+    assert acalcout.calculation.get() == "0"
+    assert acalcout.array_elements_used.get() == acalcout.array_elements_allocated.get()
+    v1 = acalcout.calculated_value.get()
+    time.sleep(0.2)
+    assert v1 == acalcout.calculated_value.get()

docs/source/api/synApps/_acalcout.rst

@@ -0,0 +1,9 @@
+
+EPICS base aCalcout record
+-----------------------------
+
+The ``aCalcout`` record is part of EPICS base:
+https://epics-modules.github.io/calc/aCalcoutRecord.html
+
+.. automodule:: apstools.synApps.acalcout
+    :members: