synApps sseq record and userStringSeq databases

CHANGES.rst

@@ -49,6 +49,11 @@ Moved ``command_list_as_table()`` from `utils` into ``plans/command_list``.
 
 ``utils/``: Reorganized ``utils.py`` and ``_utils/`` into ``utils/`` subpackage.
 
+New Features and/or Enhancements
+---------------------------------------------
+
+* Add support for synApps ``sseq`` record and ``userStringSeq`` databases.
+
 Maintenance
 ---------------
 

apstools/synApps/__init__.py

@@ -30,6 +30,10 @@
 from .transform import TransformRecord
 from .transform import UserTransformsDevice
 
+from .sseq import EditStringSequence
+from .sseq import SseqRecord
+from .sseq import UserStringSequenceDevice
+
 # -----------------------------------------------------------------------------
 # :author:    Pete R. Jemian
 # :email:     [email protected]

apstools/synApps/sseq.py

@@ -0,0 +1,182 @@
+"""
+Ophyd support for the EPICS sseq (string sequence) record
+
+
+Public Structures
+
+.. autosummary::
+
+    ~EditStringSequence
+    ~SseqRecord
+    ~UserStringSequenceDevice
+"""
+
+# -----------------------------------------------------------------------------
+# :author:    Pete R. Jemian
+# :email:     [email protected]
+# :copyright: (c) 2017-2022, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Creative Commons Attribution 4.0 International Public License.
+#
+# The full license is in the file LICENSE.txt, distributed with this software.
+# -----------------------------------------------------------------------------
+
+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 ._common import EpicsRecordDeviceCommonAll
+
+
+STEP_LIST = [f"step{i+1}" for i in range(10)]  # step1, step2, step10
+
+
+class sseqRecordStep(Device):
+    """
+    Step of a synApps sseq record: 1..10  (note: for 10, the PVs use "A")
+
+    .. index:: Ophyd Device; synApps sseqRecordStep
+
+    .. autosummary::
+
+        ~reset
+    """
+
+    input_pv = FC(EpicsSignal, "{self.prefix}.DOL{self._step}", kind="config")
+    input_pv_valid = FC(EpicsSignalRO, "{self.prefix}.DOL{self._step}V", kind="config")
+    delay = FC(EpicsSignal, "{self.prefix}.DLY{self._step}", kind="config")
+    string_value = FC(
+        EpicsSignal, "{self.prefix}.STR{self._step}", string=True, kind="hinted"
+    )
+    numeric_value = FC(EpicsSignal, "{self.prefix}.DO{self._step}", kind="hinted")
+    output_pv = FC(EpicsSignal, "{self.prefix}.LNK{self._step}", kind="config")
+    output_pv_valid = FC(EpicsSignalRO, "{self.prefix}.LNK{self._step}V", kind="config")
+
+    waiting_completion = FC(
+        EpicsSignalRO, "{self.prefix}.WTG{self._step}", kind="config"
+    )
+    wait_completion = FC(EpicsSignal, "{self.prefix}.WAIT{self._step}", kind="config")
+    wait_error = FC(EpicsSignalRO, "{self.prefix}.WERR{self._step}", kind="config")
+
+    def __init__(self, prefix, step, **kwargs):
+        names = "_123456789A"  # step #10 (1-based) is called "A"
+        self._step = names[step]
+        super().__init__(prefix, **kwargs)
+
+    def reset(self):
+        """set all fields to default values"""
+        self.input_pv.put("")
+        self.delay.put(0)
+        self.numeric_value.put(0)  # EPICS will set string_value from this
+        self.output_pv.put("")
+        self.wait_completion.put("NoWait")
+
+
+def _steps(step_list):
+    defn = OrderedDict()
+    for step in step_list:
+        step_number = int(step[4:])
+        defn[step] = (sseqRecordStep, "", {"step": step_number})
+    return defn
+
+
+class SseqRecord(EpicsRecordDeviceCommonAll):
+    """
+    EPICS sseq record support in ophyd
+
+    .. index:: Ophyd Device; synApps SseqRecord
+
+    .. autosummary::
+
+        ~abort
+        ~reset
+
+    :see: https://htmlpreview.github.io/?https://raw.githubusercontent.com/epics-modules/calc/R3-6-1/documentation/TransformRecord.html#Fields
+    """
+
+    enable = Cpt(EpicsSignal, "Enable", kind="config")
+    precision = Cpt(EpicsSignal, ".PREC", kind="config")
+    busy = Cpt(EpicsSignalRO, ".PREC", kind="config")
+    _abort = Cpt(EpicsSignal, ".ABORT", kind="omitted")
+
+    selection_link = Cpt(EpicsSignal, ".SELL", kind="config")
+    selection_mask = Cpt(EpicsSignal, ".SELM", kind="config")
+    selection_number = Cpt(EpicsSignal, ".SELN", kind="config")
+
+    steps = DDC(_steps(STEP_LIST))
+
+    def abort(self):
+        """
+        .ABORT is a push button.  Send a 1 to the PV to "push" it.
+
+        Push this button without a timeout from the .put() method.
+        """
+        self._abort.put(1, use_complete=False, force=True)
+
+    def reset(self):
+        """set all fields to default values"""
+        self.scanning_rate.put("Passive")
+        self.description.put(self.description.pvname.split(".")[0])
+        self.forward_link.put("")
+        self.precision.put(5)
+        self.selection_link.put("")
+        self.selection_mask.put("All")
+        self.selection_number.put(1)
+        for ch in self.steps.component_names:
+            step = getattr(self.steps, ch)
+            if isinstance(step, sseqRecordStep):
+                step.reset()
+        self.hints["fields"] = ["steps_%s" % c for c in self.steps.component_names]
+        self.read_attrs = ["steps.%s" % c for c in STEP_LIST]
+
+
+class UserStringSequenceDevice(Device):
+    """
+    synApps XXX IOC setup of userStringSeqs: ``$(P):userStringSeq$(N)``
+
+    Note: This will connect more than 1,000 EpicsSignal objects!
+
+    .. index:: Ophyd Device; synApps UserStringSequenceDevice
+
+    .. autosummary::
+
+        ~reset
+    """
+
+    enable = Cpt(EpicsSignal, "userStringSeqEnable", kind="config")
+    sseq1 = Cpt(SseqRecord, "userStringSeq1")
+    sseq2 = Cpt(SseqRecord, "userStringSeq2")
+    sseq3 = Cpt(SseqRecord, "userStringSeq3")
+    sseq4 = Cpt(SseqRecord, "userStringSeq4")
+    sseq5 = Cpt(SseqRecord, "userStringSeq5")
+    sseq6 = Cpt(SseqRecord, "userStringSeq6")
+    sseq7 = Cpt(SseqRecord, "userStringSeq7")
+    sseq8 = Cpt(SseqRecord, "userStringSeq8")
+    sseq9 = Cpt(SseqRecord, "userStringSeq9")
+    sseq10 = Cpt(SseqRecord, "userStringSeq10")
+
+    def reset(self):  # lgtm [py/similar-function]
+        """set all fields to default values"""
+        for c in self.component_names:
+            if not c.startswith("sseq"):
+                continue
+            getattr(self, c).reset()
+        self.read_attrs = self.component_names
+
+
+class EditStringSequence(Device):
+    """
+    Assistance to quickly re-arrange steps in an sseq record configuration.
+
+    See the editSseq_more GUI screen for assistance.
+    """
+    record_name = Cpt(EpicsSignal, "ES:recordName", kind="config")
+    command = Cpt(EpicsSignal, "ES:command", kind="config")
+    message_acknowledge = Cpt(EpicsSignal, "ES:OperAck", kind="config")
+    message = Cpt(EpicsSignalRO, "ES:message", kind="normal")
+    alert = Cpt(EpicsSignalRO, "ES:Alert", kind="normal")
+    debug = Cpt(EpicsSignal, "ES:Debug", kind="config")

apstools/synApps/tests/test_sseq.py

@@ -0,0 +1,46 @@
+from ophyd import EpicsSignal
+
+from ..sseq import SseqRecord
+from ..sseq import sseqRecordStep
+from ..sseq import UserStringSequenceDevice
+
+
+IOC = "gp:"
+TEST_SSEQ_PV = f"{IOC}userStringSeq10"
+
+
+def test_sseq_read():
+    sseq = SseqRecord(TEST_SSEQ_PV, name="sseq")
+    assert sseq is not None
+    sseq.wait_for_connection()
+
+    r = sseq.read()
+    assert len(r) == 20
+
+
+def test_sseq_reset():
+    user = UserStringSequenceDevice(IOC, name="user")
+    user.wait_for_connection()
+    user.enable.put("Enable")
+
+    sseq = user.sseq10
+    assert isinstance(sseq, SseqRecord)
+    sseq.enable.put("E")  # Note: only "E"
+
+    step = sseq.steps.step1
+    assert isinstance(step, sseqRecordStep)
+
+    step.reset()
+    assert step.input_pv.get() == ""
+
+    uptime = EpicsSignal(f"{IOC}UPTIME", name="uptime")
+    uptime.wait_for_connection()
+
+    step.input_pv.put(uptime.pvname)
+    assert step.string_value.get() == f"{0:.5f}"
+    sseq.process_record.put(1)
+    assert step.string_value.get() <= uptime.get()
+
+    user.reset()
+    assert step.input_pv.get() == ""
+    assert step.string_value.get() == f"{0:.5f}"

apstools/synApps/transform.py

@@ -122,7 +122,7 @@ def reset(self):
             channel = getattr(self.channels, letter)
             if isinstance(channel, transformRecordChannel):
                 channel.reset()
-        self.hints = {"fields": ["channels.%s" % c for c in CHANNEL_LETTERS_LIST]}
+        self.hints["fields"] = ["channels.%s" % c for c in CHANNEL_LETTERS_LIST]
         self.read_attrs = ["channels.%s" % c for c in CHANNEL_LETTERS_LIST]
 
 

docs/source/api/synApps/_sseq.rst

@@ -0,0 +1,9 @@
+
+synApps sseq record
+-------------------------
+
+The ``sseq`` (String Sequence) record is part of the ``calc`` module:
+    :see: https://htmlpreview.github.io/?https://raw.githubusercontent.com/epics-modules/calc/R3-6-1/documentation/sseqRecord.html
+
+.. automodule:: apstools.synApps.sseq
+    :members: