Add get_field and set_field methods to records

These allow access to all EPICS attributes on a record.

Also add documentation and tests

CHANGELOG.rst

@@ -10,7 +10,7 @@ Versioning <https://semver.org/spec/v2.0.0.html>`_.
 Unreleased_
 -----------
 
-Nothing yet
+- 'Add get_field and set_field methods to records <../../pull/140>'_
 
 4.4.0_ - 2023-07-06
 -------------------

docs/reference/api.rst

@@ -578,6 +578,27 @@ class which provides the methods documented below.
         Note that channel access puts to a Python soft IOC input record are
         completely ineffective, and this includes waveform records.
 
+    ..  method:: get_field(field)
+
+        This returns the named field from the record. An exception will be raised
+        if the field cannot be found.
+
+        Note that this function can only be used after the IOC has been initialized.
+        If you need to retrieve a field's value before that, access it directly via
+        an attribute e.g. ``my_record.EGU``. (This will not work after the IOC is
+        initialized)
+
+    ..  method:: set_field(field, value)
+
+        This sets the given field to the given value. The value will
+        always be converted to a Python String, which is then interpreted by
+        EPICS as a DBF_STRING type. Note that values can be no longer than 39 bytes.
+
+        Note that this function can only be used after the IOC has been initialized.
+        If you need to set a field's value before that, set it directly as an attribute
+        on the record e.g. ``my_record.EGU``. (This will not work after the IOC is
+        initialized)
+
 Working with OUT records
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -599,4 +620,25 @@ Working with OUT records
 
         Returns the value associated with the record.
 
+    ..  method:: get_field(field)
+
+        This returns the named field from the record. An exception will be raised
+        if the field cannot be found.
+
+        Note that this function can only be used after the IOC has been initialized.
+        If you need to retrieve a field's value before that, access it directly via
+        an attribute e.g. ``my_record.EGU``. (This will not work after the IOC is
+        initialized)
+
+    ..  method:: set_field(field, value)
+
+        This sets the given field to the given value. The value will
+        always be converted to a Python String, which is then interpreted by
+        EPICS as a DBF_STRING type. Note that values can be no longer than 39 bytes.
+
+        Note that this function can only be used after the IOC has been initialized.
+        If you need to set a field's value before that, set it directly as an attribute
+        on the record e.g. ``my_record.EGU``. (This will not work after the IOC is
+        initialized)
+
 .. _epics_device: https://github.com/Araneidae/epics_device

softioc/device.py

@@ -12,6 +12,7 @@
     signal_processing_complete,
     recGblResetAlarms,
     db_put_field,
+    db_get_field,
 )
 from .device_core import DeviceSupportCore, RecordLookup
 
@@ -83,6 +84,26 @@ def _read_value(self, record):
     def _write_value(self, record, value):
         record.write_val(value)
 
+    def get_field(self, field):
+        ''' Returns the given field value as a string.'''
+        assert hasattr(self, "_record"), \
+            'get_field may only be called after iocInit'
+
+        data = (c_char * 40)()
+        name = self._name + '.' + field
+        db_get_field(name, fields.DBF_STRING, addressof(data), 1)
+        return _string_at(data, 40)
+
+    def set_field(self, field, value):
+        '''Sets the given field to the given value. Value will be transported as
+        a DBF_STRING.'''
+        assert hasattr(self, "_record"), \
+            'set_field may only be called after iocInit'
+
+        data = (c_char * 40)()
+        data.value = str(value).encode() + b'\0'
+        name = self._name + '.' + field
+        db_put_field(name, fields.DBF_STRING, addressof(data), 1)
 
 class ProcessDeviceSupportIn(ProcessDeviceSupportCore):
     _link_ = 'INP'

softioc/extension.c

@@ -113,6 +113,25 @@ static PyObject *db_put_field(PyObject *self, PyObject *args)
     Py_RETURN_NONE;
 }
 
+static PyObject *db_get_field(PyObject *self, PyObject *args)
+{
+    const char *name;
+    short dbrType;
+    void *pbuffer;
+    long length;
+    if (!PyArg_ParseTuple(args, "shnl", &name, &dbrType, &pbuffer, &length))
+        return NULL;
+
+    long options = 0;
+    struct dbAddr dbAddr;
+    if (dbNameToAddr(name, &dbAddr))
+        return PyErr_Format(
+            PyExc_RuntimeError, "dbNameToAddr failed for %s", name);
+    if (dbGetField(&dbAddr, dbrType, pbuffer, &options, &length, NULL))
+        return PyErr_Format(
+            PyExc_RuntimeError, "dbGetField failed for %s", name);
+    Py_RETURN_NONE;
+}
 
 /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
 /* IOC PV put logging */
@@ -266,6 +285,8 @@ static struct PyMethodDef softioc_methods[] = {
      "Get offset, size and type for each record field"},
     {"db_put_field",  db_put_field, METH_VARARGS,
      "Put a database field to a value"},
+    {"db_get_field",  db_get_field, METH_VARARGS,
+     "Get a database field's value"},
     {"install_pv_logging",  install_pv_logging, METH_VARARGS,
      "Install caput logging to stdout"},
     {"signal_processing_complete",  signal_processing_complete, METH_VARARGS,

softioc/imports.py

@@ -23,6 +23,10 @@ def db_put_field(name, dbr_type, pbuffer, length):
     '''Put field where pbuffer is void* pointer. Returns RC'''
     return _extension.db_put_field(name, dbr_type, pbuffer, length)
 
+def db_get_field(name, dbr_type, pbuffer, length):
+    '''Get field where pbuffer is void* pointer. Returns Py_RETURN_NONE'''
+    return _extension.db_get_field(name, dbr_type, pbuffer, length)
+
 def install_pv_logging(acf_file):
     '''Install pv logging'''
     _extension.install_pv_logging(acf_file)

tests/test_records.py

@@ -34,6 +34,7 @@
     builder.WaveformIn,
 ]
 
+# TODO: This should never pass, but somehow does in CI...
 def test_records(tmp_path):
     # Ensure we definitely unload all records that may be hanging over from
     # previous tests, then create exactly one instance of expected records.
@@ -904,3 +905,203 @@ async def query_record(index):
             log(f"PARENT: Join completed with exitcode {process.exitcode}")
             if process.exitcode is None:
                 pytest.fail("Process did not terminate")
+
+class TestGetSetField:
+    """Tests related to get_field and set_field on records"""
+
+    test_result_rec = "TestResult"
+
+    def test_set_field_before_init_fails(self):
+        """Test that calling set_field before iocInit() raises an exception"""
+
+        ao = builder.aOut("testAOut")
+
+        with pytest.raises(AssertionError) as e:
+            ao.set_field("EGU", "Deg")
+
+        assert "set_field may only be called after iocInit" in str(e.value)
+
+    def test_get_field_before_init_fails(self):
+        """Test that calling get_field before iocInit() raises an exception"""
+
+        ao = builder.aOut("testAOut")
+
+        with pytest.raises(AssertionError) as e:
+            ao.get_field("EGU")
+
+        assert "get_field may only be called after iocInit" in str(e.value)
+
+    def get_set_test_func(self, device_name, conn):
+        """Run an IOC and do simple get_field/set_field calls"""
+
+        builder.SetDeviceName(device_name)
+
+        lo = builder.longOut("TestLongOut", EGU="unset", DRVH=12)
+
+        # Record to indicate success/failure
+        bi = builder.boolIn(self.test_result_rec, ZNAM="FAILED", ONAM="SUCCESS")
+
+        dispatcher = asyncio_dispatcher.AsyncioDispatcher()
+        builder.LoadDatabase()
+        softioc.iocInit(dispatcher)
+
+        conn.send("R")  # "Ready"
+
+        log("CHILD: Sent R over Connection to Parent")
+
+        # Set and then get the EGU field
+        egu = "TEST"
+        lo.set_field("EGU", egu)
+        log("CHILD: set_field successful")
+        readback_egu = lo.get_field("EGU")
+        log(f"CHILD: get_field returned {readback_egu}")
+        assert readback_egu == egu, \
+            f"EGU field was not {egu}, was {readback_egu}"
+
+        log("CHILD: assert passed")
+
+        # Test completed, report to listening camonitor
+        bi.set(True)
+
+        # Keep process alive while main thread works.
+        while (True):
+            if conn.poll(TIMEOUT):
+                val = conn.recv()
+                if val == "D":  # "Done"
+                    break
+
+        log("CHILD: Received exit command, child exiting")
+
+
+    @pytest.mark.asyncio
+    async def test_get_set(self):
+        """Test a simple set_field/get_field is successful"""
+        ctx = get_multiprocessing_context()
+        parent_conn, child_conn = ctx.Pipe()
+
+        device_name = create_random_prefix()
+
+        process = ctx.Process(
+            target=self.get_set_test_func,
+            args=(device_name, child_conn),
+        )
+
+        process.start()
+
+        log("PARENT: Child started, waiting for R command")
+
+        from aioca import camonitor
+
+        try:
+            # Wait for message that IOC has started
+            select_and_recv(parent_conn, "R")
+
+            log("PARENT: received R command")
+
+            queue = asyncio.Queue()
+            record = device_name + ":" + self.test_result_rec
+            monitor = camonitor(record, queue.put)
+
+            log(f"PARENT: monitoring {record}")
+            new_val = await asyncio.wait_for(queue.get(), TIMEOUT)
+            log(f"PARENT: new_val is {new_val}")
+            assert new_val == 1, \
+                f"Test failed, value was not 1(True), was {new_val}"
+
+
+        finally:
+            monitor.close()
+            # Clear the cache before stopping the IOC stops
+            # "channel disconnected" error messages
+            aioca_cleanup()
+
+            log("PARENT: Sending Done command to child")
+            parent_conn.send("D")  # "Done"
+            process.join(timeout=TIMEOUT)
+            log(f"PARENT: Join completed with exitcode {process.exitcode}")
+            if process.exitcode is None:
+                pytest.fail("Process did not terminate")
+
+    def get_set_too_long_value(self, device_name, conn):
+        """Run an IOC and deliberately call set_field with a too-long value"""
+
+        builder.SetDeviceName(device_name)
+
+        lo = builder.longOut("TestLongOut", EGU="unset", DRVH=12)
+
+        # Record to indicate success/failure
+        bi = builder.boolIn(self.test_result_rec, ZNAM="FAILED", ONAM="SUCCESS")
+
+        dispatcher = asyncio_dispatcher.AsyncioDispatcher()
+        builder.LoadDatabase()
+        softioc.iocInit(dispatcher)
+
+        conn.send("R")  # "Ready"
+
+        log("CHILD: Sent R over Connection to Parent")
+
+        # Set a too-long value and confirm it reports an error
+        try:
+            lo.set_field("EGU", "ThisStringIsFarTooLongToFitIntoTheEguField")
+        except ValueError as e:
+            # Expected error, report success to listening camonitor
+            assert "byte string too long" in e.args[0]
+            bi.set(True)
+
+        # Keep process alive while main thread works.
+        while (True):
+            if conn.poll(TIMEOUT):
+                val = conn.recv()
+                if val == "D":  # "Done"
+                    break
+
+        log("CHILD: Received exit command, child exiting")
+
+    @pytest.mark.asyncio
+    async def test_set_too_long_value(self):
+        """Test that set_field with a too-long value raises the expected
+        error"""
+        ctx = get_multiprocessing_context()
+        parent_conn, child_conn = ctx.Pipe()
+
+        device_name = create_random_prefix()
+
+        process = ctx.Process(
+            target=self.get_set_too_long_value,
+            args=(device_name, child_conn),
+        )
+
+        process.start()
+
+        log("PARENT: Child started, waiting for R command")
+
+        from aioca import camonitor
+
+        try:
+            # Wait for message that IOC has started
+            select_and_recv(parent_conn, "R")
+
+            log("PARENT: received R command")
+
+            queue = asyncio.Queue()
+            record = device_name + ":" + self.test_result_rec
+            monitor = camonitor(record, queue.put)
+
+            log(f"PARENT: monitoring {record}")
+            new_val = await asyncio.wait_for(queue.get(), TIMEOUT)
+            log(f"PARENT: new_val is {new_val}")
+            assert new_val == 1, \
+                f"Test failed, value was not 1(True), was {new_val}"
+
+        finally:
+            monitor.close()
+            # Clear the cache before stopping the IOC stops
+            # "channel disconnected" error messages
+            aioca_cleanup()
+
+            log("PARENT: Sending Done command to child")
+            parent_conn.send("D")  # "Done"
+            process.join(timeout=TIMEOUT)
+            log(f"PARENT: Join completed with exitcode {process.exitcode}")
+            if process.exitcode is None:
+                pytest.fail("Process did not terminate")