Input-Output Controllers (IOCs)

EPICS Input-Output Controllers (IOCs) expose an EPICS server. Behind the server, they may connect to a device driver, data-processing code, and/or an EPICS client for chaining with other servers.

In this section, we will review some of the example IOCs that are included with caproto, intended as demonstrations of what is possible.

Why Write an IOC Using Caproto?

Caproto makes it is easy to launch a protocol-compliant Channel Access server in just a couple lines of Python. This opens up some interesting possibilities:

  • In Python, it is easy to invoke standard web protocols. For example, writing an EPICS server around a device that speaks JSON may be easier with caproto than with any previously-existing tools.
  • Many scientists who rely on EPICS but may not understand the details of EPICS already know some Python for the data analysis work. Caproto may make it easier for scientists and controls engineers to collaborate.
  • As with its clients, caproto’s servers handle a human-friendly encapsulation of every message sent and received, which can be valuable for development, logging, and debugging.

Using the IOC Examples

They can be started like so:

$ python3 -m caproto.ioc_examples.simple
[I 16:51:09.751 server:93] Server starting up...
[I 16:51:09.752 server:109] Listening on 0.0.0.0:54966
[I 16:51:09.753 server:121] Server startup complete.

and stopped using Ctrl+C:

^C
[I 16:51:10.828 server:129] Server task cancelled. Must shut down.
[I 16:51:10.828 server:132] Server exiting....

Use --list-pvs to display which PVs they serve:

$ python3 -m caproto.ioc_examples.simple --list-pvs
[I 16:52:36.087 server:93] Server starting up...
[I 16:52:36.089 server:109] Listening on 0.0.0.0:62005
[I 16:52:36.089 server:121] Server startup complete.
[I 16:52:36.089 server:123] PVs available:
    simple:A
    simple:B
    simple:C

and use --prefix to conveniently customize the PV prefix:

$ python3 -m caproto.ioc_examples.simple --list-pvs --prefix my_custom_prefix:
[I 16:54:14.528 server:93] Server starting up...
[I 16:54:14.530 server:109] Listening on 0.0.0.0:55810
[I 16:54:14.530 server:121] Server startup complete.
[I 16:54:14.530 server:123] PVs available:
    my_custom_prefix:A
    my_custom_prefix:B
    my_custom_prefix:C

Type python3 -m caproto.ioc_examples.simple -h for more options.

Examples

Below, we will use caproto’s threading client to interact with caproto IOC.

In [1]: from caproto.threading.client import Context

In [2]: ctx = Context()  # a client Context used to explore the servers below

Of course, standard epics-base clients or other caproto clients may also be used.

Simple IOC

This IOC has two PVs that simply store a value.

#!/usr/bin/env python3
from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run
from textwrap import dedent


class SimpleIOC(PVGroup):
    """
    An IOC with three uncoupled read/writable PVs

    Scalar PVs
    ----------
    A (int)
    B (float)

    Vectors PVs
    -----------
    C (vector of int)
    """
    A = pvproperty(value=1)
    B = pvproperty(value=2.0)
    C = pvproperty(value=[1, 2, 3])


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='simple:',
        desc=dedent(SimpleIOC.__doc__))
    ioc = SimpleIOC(**ioc_options)
    run(ioc.pvdb, **run_options)
$ python3 -m caproto.ioc_examples.simple --list-pvs
[I 18:08:47.628 server:93] Server starting up...
[I 18:08:47.630 server:109] Listening on 0.0.0.0:56840
[I 18:08:47.630 server:121] Server startup complete.
[I 18:08:47.630 server:123] PVs available:
    simple:A
    simple:B
    simple:C

Using the threading client context we created above, we can read these values and write to them.

In [3]: a, b, c = ctx.get_pvs('simple:A', 'simple:B', 'simple:C')

In [4]: a.read()
Out[4]: ReadNotifyResponse(data=array([1], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=0, metadata=None)

In [5]: b.read()
Out[5]: ReadNotifyResponse(data=array([2.]), data_type=<ChannelType.DOUBLE: 6>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=1, metadata=None)

In [6]: c.read()
Out[6]: ReadNotifyResponse(data=array([1, 2, 3], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=3, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=2, metadata=None)

In [7]: b.write(5)
Out[7]: WriteNotifyResponse(data_type=<ChannelType.DOUBLE: 6>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=3)

In [8]: b.read()
Out[8]: ReadNotifyResponse(data=array([5.]), data_type=<ChannelType.DOUBLE: 6>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=4, metadata=None)

In [9]: c.write([4, 5, 6])
Out[9]: WriteNotifyResponse(data_type=<ChannelType.LONG: 5>, data_count=3, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=5)

In [10]: c.read()
Out[10]: ReadNotifyResponse(data=array([4, 5, 6], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=3, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=6, metadata=None)

Write to a File When a PV is Written To

#!/usr/bin/env python3
import os
import sys

from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run
import pathlib


temp_path = pathlib.Path('/tmp' if sys.platform != 'win32'
                         else os.environ.get('TEMP'))


class CustomWrite(PVGroup):
    """
    When a PV is written to, write the new value into a file as a string.
    """
    DIRECTORY = temp_path

    async def my_write(self, instance, value):
        # Compose the filename based on whichever PV this is.
        pv_name = instance.pvspec.attr  # 'A' or 'B', for this IOC
        with open(self.DIRECTORY / pv_name, 'w') as f:
            f.write(str(value))
        print(f'Wrote {value} to {self.DIRECTORY / pv_name}')
        return value

    A = pvproperty(put=my_write, value=0)
    B = pvproperty(put=my_write, value=0)


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='custom_write:',
        desc='Run an IOC with PVs that, when written to, update a file.')
    ioc = CustomWrite(**ioc_options)
    run(ioc.pvdb, **run_options)
$ python3 -m caproto.ioc_examples.custom_write --list-pvs
[I 18:12:07.282 server:93] Server starting up...
[I 18:12:07.284 server:109] Listening on 0.0.0.0:57539
[I 18:12:07.284 server:121] Server startup complete.
[I 18:12:07.284 server:123] PVs available:
    custom_write:A
    custom_write:B

On the machine where the server resides, we will see files update whenever any client writes.

In [11]: a, b = ctx.get_pvs('custom_write:A', 'custom_write:B')

In [12]: a.write(5)
Out[12]: WriteNotifyResponse(data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=0)

In [13]: print(open('/tmp/A').read())
5

In [14]: a.write(10)
Out[14]: WriteNotifyResponse(data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=1)

In [15]: print(open('/tmp/A').read())
10

It is easy to imagine extending this example to write a socket or a serial device rather than a file.

Random Walk

This example contains a PV random_walk:x that takes random steps at an update rate controlled by a second PV, random_walk:dt.

#!/usr/bin/env python3
import random
from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run


class RandomWalkIOC(PVGroup):
    dt = pvproperty(value=3.0)
    x = pvproperty(value=0.0)

    @x.startup
    async def x(self, instance, async_lib):
        'Periodically update the value'
        while True:
            # compute next value
            x = self.x.value + 2 * random.random() - 1

            # update the ChannelData instance and notify any subscribers
            await instance.write(value=x)

            # Let the async library wait for the next iteration
            await async_lib.library.sleep(self.dt.value)


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='random_walk:',
        desc='Run an IOC with a random-walking value.')
    ioc = RandomWalkIOC(**ioc_options)
    run(ioc.pvdb, **run_options)

Note

What is async_lib.library.sleep?

As caproto supports three different async libraries, we have an “async layer” class that gives compatible async versions of commonly-used synchronization primitives. The attribute async_lib.library would be either the Python module asyncio (default), trio, or curio, depending on how the server is run. It happens that all three of these modules have a sleep function at the top level, so async_lib.library.sleep accesses the appropriate sleep function for each library.

Why not use time.sleep?

The gist is that asyncio.sleep doesn’t hold up your entire thread / event loop, but gives back control to the event loop to run other tasks while sleeping. The function time.sleep, on the other hand, would cause noticeable delays and problems.

This is a fundamental consideration in concurrent programming generally, not specific to caproto. See for example this StackOverflow post for more information.

I/O Interrupt

This example listens for key presses.

#!/usr/bin/env python3
import termios
import fcntl
import sys
import os
import threading
import atexit

from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run


def start_io_interrupt_monitor(new_value_callback):
    '''
    This function monitors the terminal it was run in for keystrokes.
    On each keystroke, it calls new_value_callback with the given keystroke.

    This is used to simulate the concept of an I/O Interrupt-style signal from
    the EPICS world. Those signals depend on hardware to tell EPICS when new
    values are available to be read by way of interrupts - whereas we use
    callbacks here.
    '''

    # Thanks stackoverflow and Python 2 FAQ!
    if not sys.__stdin__.isatty():
        print('[IO Interrupt] stdin is not a TTY, exiting')
        return

    fd = sys.stdin.fileno()

    oldterm = termios.tcgetattr(fd)
    newattr = termios.tcgetattr(fd)
    newattr[3] &= ~termios.ICANON & ~termios.ECHO
    oldflags = fcntl.fcntl(fd, fcntl.F_GETFL)

    print('Started monitoring the keyboard outside of the async library')

    # When the process exits, be sure to reset the terminal settings
    @atexit.register
    def reset_terminal():
        print('Resetting the terminal settings')
        termios.tcsetattr(fd, termios.TCSAFLUSH, oldterm)
        fcntl.fcntl(fd, fcntl.F_SETFL, oldflags)
        print('Done')

    termios.tcsetattr(fd, termios.TCSANOW, newattr)
    fcntl.fcntl(fd, fcntl.F_SETFL, oldflags | os.O_NONBLOCK)

    # Loop forever, sending back keypresses to the callback
    while True:
        try:
            char = sys.stdin.read(1)
        except IOError:
            ...
        else:
            if char:
                print(f'New keypress: {char!r}')
                new_value_callback(char)


class IOInterruptIOC(PVGroup):
    keypress = pvproperty(value='', max_length=10)

    # NOTE the decorator used here:
    @keypress.startup
    async def keypress(self, instance, async_lib):
        # This method will be called when the server starts up.
        print('* keypress method called at server startup')
        queue = async_lib.ThreadsafeQueue()

        # Start a separate thread that monitors keyboard input, telling it to
        # put new values into our async-friendly queue
        thread = threading.Thread(target=start_io_interrupt_monitor,
                                  daemon=True,
                                  kwargs=dict(new_value_callback=queue.put))
        thread.start()

        # Loop and grab items from the queue one at a time
        while True:
            value = await queue.async_get()
            print(f'Saw new value on async side: {value!r}')

            # Propagate the keypress to the EPICS PV, triggering any monitors
            # along the way
            await self.keypress.write(str(value))


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='io:',
        desc='Run an IOC that updates via I/O interrupt on key-press events.')

    ioc = IOInterruptIOC(**ioc_options)
    run(ioc.pvdb, **run_options)
$ python -m caproto.ioc_examples.io_interrupt --list-pvs
[I 10:18:57.643 server:132] Server starting up...
[I 10:18:57.644 server:145] Listening on 0.0.0.0:54583
[I 10:18:57.646 server:218] Server startup complete.
[I 10:18:57.646 server:220] PVs available:
    io:keypress
* keypress method called at server startup
Started monitoring the keyboard outside of the async library

Typing causes updates to be sent to any client subscribed to io:keypress. If we monitoring using the commandline-client like so:

$ caproto-monitor io:keypress

and go back to the server and type some keys:

New keypress: 'a'
Saw new value on async side: 'a'
New keypress: 'b'
Saw new value on async side: 'b'
New keypress: 'c'
Saw new value on async side: 'c'
New keypress: 'd'
Saw new value on async side: 'd'

the client will receive the updates:

io:keypress                               2018-06-14 10:19:04 [b'd']
io:keypress                               2018-06-14 10:20:26 [b'a']
io:keypress                               2018-06-14 10:20:26 [b's']
io:keypress                               2018-06-14 10:20:26 [b'd']

Macros for PV names

#!/usr/bin/env python3
from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run


class MacroifiedNames(PVGroup):
    """

    """
    placeholder1 = pvproperty(value=0, name='{beamline}:{thing}.VAL')
    placeholder2 = pvproperty(value=0, name='{beamline}:{thing}.RBV')


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='macros:',
        desc='Run an IOC with PVs that have macro-ified names.',
        # Provide default values for macros.
        # Passing None as a default value makes a parameter required.
        macros=dict(beamline='my_beamline', thing='thing'))
    ioc = MacroifiedNames(**ioc_options)
    run(ioc.pvdb, **run_options)

The help string for this IOC contains two extra entries at the bottom:

$ python3 -m caproto.ioc_examples.macros -h
usage: macros.py [-h] [--prefix PREFIX] [-q | -v] [--list-pvs]
                [--async-lib {asyncio,curio,trio}]
                [--interfaces INTERFACES [INTERFACES ...]]
                [--beamline BEAMLINE] [--thing THING]

Run an IOC with PVs that have macro-ified names.

optional arguments:

<...snipped...>

--beamline BEAMLINE   Macro substitution, optional
--thing THING         Macro substitution, optional
$ python3 -m caproto.ioc_examples.macros --beamline XF31ID --thing detector --list-pvs
[I 18:44:39.528 server:93] Server starting up...
[I 18:44:39.530 server:109] Listening on 0.0.0.0:56365
[I 18:44:39.531 server:121] Server startup complete.
[I 18:44:39.531 server:123] PVs available:
    macros:XF31ID:detector.VAL
    macros:XF31ID:detector.RBV
In [16]: pv, = ctx.get_pvs('macros:XF31ID:detector.VAL')

In [17]: pv.read()
Out[17]: ReadNotifyResponse(data=array([0], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=0, metadata=None)

Observe that the command line arguments fill in the PV names.

Subgroups

The PVGroup is designed to be nested, which provides a nice path toward future capability: IOCs that are natively “V7”, encoding semantic structure for PV Access clients and decomposing into flat PVs for Channel Access clients.

#!/usr/bin/env python3
import logging
import random
from caproto.server import (pvproperty, PVGroup, SubGroup, ioc_arg_parser, run)


logger = logging.getLogger('caproto')


class RecordLike(PVGroup):
    'Example group, mirroring a V3 record'

    # stop : from being added to the prefix
    attr_separator = ''

    # PV #1: {prefix}.RTYP
    record_type = pvproperty(name='.RTYP', value='ai')
    # PV #2: {prefix}.VAL
    value = pvproperty(name='.VAL', value=1)
    # PV #3: {prefix}.DESC
    description = pvproperty(name='.DESC', value='Description')


class MySubGroup(PVGroup):
    'Example group of PVs, where the prefix is defined on instantiation'
    # PV: {prefix}random - defaults to dtype of int
    @pvproperty
    async def random(self, instance):
        return random.randint(1, 100)


class MyPVGroup(PVGroup):
    'Example group of PVs, a mix of pvproperties and subgroups'

    # PV: {prefix}random
    @pvproperty
    async def random(self, instance):
        logger.debug('read random from %s', type(self).__name__)
        return random.randint(1, 100)

    # PVs: {prefix}RECORD_LIKE1.RTYP, .VAL, and .DESC
    recordlike1 = SubGroup(RecordLike, prefix='RECORD_LIKE1')
    # PVs: {prefix}recordlike2.RTYP, .VAL, and .DESC
    recordlike2 = SubGroup(RecordLike)

    # PV: {prefix}group1:random
    group1 = SubGroup(MySubGroup)
    # PV: {prefix}group2-random
    group2 = SubGroup(MySubGroup, prefix='group2-')

    # PV: {prefix}group3_prefix:random
    @SubGroup(prefix='group3_prefix:')
    class group3(PVGroup):
        @pvproperty
        async def random(self, instance):
            logger.debug('read random from %s', type(self).__name__)
            return random.randint(1, 100)

    # PV: {prefix}group4:subgroup4:random
    # (TODO BUG) {prefix}subgroup4:random
    @SubGroup
    class group4(PVGroup):
        @SubGroup
        class subgroup4(PVGroup):
            @pvproperty
            async def random(self, instance):
                logger.debug('read random from %s', type(self).__name__)
                return random.randint(1, 100)


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='subgroups:',
        desc='Run an IOC with groups of groups of PVs.')
    ioc = MyPVGroup(**ioc_options)

    # here's what accessing a pvproperty descriptor looks like:
    print('random using the descriptor getter is:', ioc.random)

    # and for subgroups:
    print('subgroup4 is:', ioc.group4.subgroup4)
    print('subgroup4.random is:', ioc.group4.subgroup4.random)

    run(ioc.pvdb, **run_options)
$ python -m caproto.ioc_examples.subgroups --list-pvs
random using the descriptor getter is: <caproto.server.server.PvpropertyInteger object at 0x10928ffd0>
subgroup4 is: <__main__.MyPVGroup.group4.subgroup4 object at 0x10928fe10>
subgroup4.random is: <caproto.server.server.PvpropertyInteger object at 0x10928fef0>
[I 11:08:35.074 server:132] Server starting up...
[I 11:08:35.074 server:145] Listening on 0.0.0.0:63336
[I 11:08:35.076 server:218] Server startup complete.
[I 11:08:35.077 server:220] PVs available:
    subgroups:random
    subgroups:RECORD_LIKE1.RTYP
    subgroups:RECORD_LIKE1.VAL
    subgroups:RECORD_LIKE1.DESC
    subgroups:recordlike2.RTYP
    subgroups:recordlike2.VAL
    subgroups:recordlike2.DESC
    subgroups:group1:random
    subgroups:group2-random
    subgroups:group3_prefix:random
    subgroups:group4:subgroup4:random

Mocking Records

See Mock Records.

#!/usr/bin/env python3
from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run


class RecordMockingIOC(PVGroup):
    # Define three records, an analog input (ai) record:
    A = pvproperty(value=1.0, mock_record='ai')
    # And an analog output (ao) record:
    B = pvproperty(value=2.0, mock_record='ao',
                   precision=3)
    # and an ai with all the bells and whistles
    C = pvproperty(value=0.0,
                   mock_record='ai',
                   upper_alarm_limit=2.0,
                   lower_alarm_limit=-2.0,
                   upper_warning_limit=1.0,
                   lower_warning_limit=-1.0,
                   upper_ctrl_limit=3.0,
                   lower_ctrl_limit=-3.0,
                   units="mm",
                   precision=3,
                   doc='The C pvproperty')

    @B.putter
    async def B(self, instance, value):
        if value == 1:
            # Mocked record will pick up the alarm status simply by us raising
            # an exception in the putter:
            raise ValueError('Invalid value!')

    # It's also possible to modify some of the behavior of fields on a per-
    # record basis. The following function is called whenever A.RVAL is put to:
    @A.fields.current_raw_value.putter
    async def A(fields, instance, value):
        # However, somewhat confusingly, 'self' in this case is the fields
        # associated with 'A'. To access the IOC (i.e., the main PV group)
        # use the following:
        ioc = fields.parent.group
        print(f'A.RVAL: Writing values to A and B: {value}')
        await ioc.B.write(value)
        await ioc.A.write(value)

    # Similarly, you can refer to the fields by their usual PV name:
    @B.fields.RVAL.putter
    async def B(fields, instance, value):
        ioc = fields.parent.group
        print(f'B.RVAL: Writing modified values to A, B')
        await ioc.B.write(value + 10)
        await ioc.A.write(value - 10)

    # Now that the field specification has been set on B, it can be reused:
    D = pvproperty(value=2.0, mock_record='ao',
                   precision=3, field_spec=B.field_spec)


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='mock:',
        desc='Run an IOC that mocks an ai (analog input) record.')

    # Instantiate the IOC, assigning a prefix for the PV names.
    ioc = RecordMockingIOC(**ioc_options)
    print('PVs:', list(ioc.pvdb))

    # ... but what you don't see are all of the analog input record fields
    print('Fields of B:', list(ioc.B.fields.keys()))

    print('Custom field specifications of A:', RecordMockingIOC.A.fields)
    print('Custom field specifications of B:', RecordMockingIOC.B.fields)

    # Run IOC.
    run(ioc.pvdb, **run_options)
python -m caproto.ioc_examples.mocking_records --list-pvs
PVs: ['mock:A', 'mock:B']
Fields of B: ['ACKS', 'ACKT', 'ASG', 'DESC', 'DISA', 'DISP', 'DISS', 'DISV', 'DTYP', 'EVNT', 'FLNK', 'LCNT', 'NAME', 'NSEV', 'NSTA', 'PACT', 'PHAS', 'PINI', 'PRIO', 'PROC', 'PUTF', 'RPRO', 'SCAN', 'SDIS', 'SEVR', 'TPRO', 'TSE', 'TSEL', 'UDF', 'RTYP', 'STAT', 'RVAL', 'INIT', 'MLST', 'LALM', 'ALST', 'LBRK', 'ORAW', 'ROFF', 'SIMM', 'SVAL', 'HYST', 'HIGH', 'HSV', 'HIHI', 'HHSV', 'LOLO', 'LLSV', 'LOW', 'LSV', 'AOFF', 'ASLO', 'EGUF', 'EGUL', 'LINR', 'EOFF', 'ESLO', 'SMOO', 'ADEL', 'PREC', 'EGU', 'HOPR', 'LOPR', 'MDEL', 'INP', 'SIOL', 'SIML', 'SIMS']
[I 11:07:48.635 server:132] Server starting up...
[I 11:07:48.636 server:145] Listening on 0.0.0.0:49637
[I 11:07:48.638 server:218] Server startup complete.
[I 11:07:48.638 server:220] PVs available:
    mock:A
    mock:B

RPC Server from Python Function

This automatically generates a SubGroup. In the future, this could be used to spin up a PVA RPC service. As is, for Channel Access, this provides an RPC function for single-user access.

#!/usr/bin/env python3
from pprint import pprint
import random

from caproto.server import (pvproperty, PVGroup, pvfunction, ioc_arg_parser,
                            run)


class MyPVGroup(PVGroup):
    'Example group of PVs, where the prefix is defined on instantiation'
    # PV #1: {prefix}random - defaults to dtype of int
    @pvproperty
    async def fixed_random(self, instance):
        'Random integer between 1 and 100'
        return random.randint(1, 100)

    @pvfunction(default=[0])
    async def get_random(self,
                         low: int = 100,
                         high: int = 1000) -> int:
        'A configurable random number'
        return random.randint(low, high)


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='rpc:',
        desc='Run an IOC with an RPC function.')

    ioc = MyPVGroup(**ioc_options)

    # here's what accessing a pvproperty descriptor looks like:
    print(f'fixed_random using the descriptor getter is: {ioc.fixed_random}')
    print(f'get_random using the descriptor getter is: {ioc.get_random}')
    print('get_random is an autogenerated subgroup with PVs:')
    for pvspec in ioc.get_random.pvdb.items():
        print(f'\t{pvspec!r}')

    # here is the auto-generated pvdb:
    pprint(ioc.pvdb)
    run(ioc.pvdb, **run_options)
$ python -m caproto.ioc_examples.rpc_function --list-pvs

   fixed_random using the descriptor getter is: <caproto.server.server.PvpropertyInteger object at 0x1041672b0>
   get_random using the descriptor getter is: <caproto.server.server.get_random object at 0x1041675f8>
   get_random is an autogenerated subgroup with PVs:
       ('rpc:get_random:low', <caproto.server.server.PvpropertyInteger object at 0x104167358>)
       ('rpc:get_random:high', <caproto.server.server.PvpropertyInteger object at 0x104167588>)
       ('rpc:get_random:Status', <caproto.server.server.PvpropertyStringRO object at 0x104167320>)
       ('rpc:get_random:Retval', <caproto.server.server.PvpropertyIntegerRO object at 0x104167550>)
       ('rpc:get_random:Process', <caproto.server.server.PvpropertyInteger object at 0x1041672e8>)
   OrderedDict([('rpc:fixed_random',
               <caproto.server.server.PvpropertyInteger object at 0x1041672b0>),
               ('rpc:get_random:low',
               <caproto.server.server.PvpropertyInteger object at 0x104167358>),
               ('rpc:get_random:high',
               <caproto.server.server.PvpropertyInteger object at 0x104167588>),
               ('rpc:get_random:Status',
               <caproto.server.server.PvpropertyStringRO object at 0x104167320>),
               ('rpc:get_random:Retval',
               <caproto.server.server.PvpropertyIntegerRO object at 0x104167550>),
               ('rpc:get_random:Process',
               <caproto.server.server.PvpropertyInteger object at 0x1041672e8>)])

“Inline” Style Read and Write Customization

This example shows custom write and read behavior similar to what we have seen before, but implemented “inline” rather than siloed into a separate method. This may be useful, from a readability point of view, for implementing one-off behavior.

#!/usr/bin/env python3
from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run
import random


class InlineStyleIOC(PVGroup):
    "An IOC with a read-only PV and a read-write PV with inline customization"
    @pvproperty  # default dtype is int
    async def random_int(self, instance):
        return random.randint(1, 100)

    @pvproperty(dtype=str)
    async def random_str(self, instance):
        return random.choice('abc')

    @pvproperty(value='c')  # initializes value and, implicitly, dtype
    async def A(self, instance):
        print('reading A')
        return instance.value

    @A.putter
    async def A(self, instance, value):
        print('writing to A the value', value)
        return value


if __name__ == '__main__':
    ioc_options, run_options = ioc_arg_parser(
        default_prefix='inline_style:',
        desc='Run an IOC PVs having custom update behavior, defined "inline".')
    ioc = InlineStyleIOC(**ioc_options)
    run(ioc.pvdb, **run_options)
$ python3 -m caproto.ioc_examples.inline_style --list-pvs
[I 18:47:05.344 server:93] Server starting up...
[I 18:47:05.346 server:109] Listening on 0.0.0.0:64443
[I 18:47:05.346 server:121] Server startup complete.
[I 18:47:05.346 server:123] PVs available:
    inline_style:random_int
    inline_style:random_str
    inline_style:A
In [18]: randint, randstr = ctx.get_pvs('inline_style:random_int', 'inline_style:random_str')

In [19]: randint.read()
Out[19]: ReadNotifyResponse(data=array([36], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=0, metadata=None)

In [20]: randint.read()
Out[20]: ReadNotifyResponse(data=array([77], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=1, metadata=None)

In [21]: randint.read()
Out[21]: ReadNotifyResponse(data=array([91], dtype=int32), data_type=<ChannelType.LONG: 5>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=2, metadata=None)

In [22]: randstr.read()
Out[22]: ReadNotifyResponse(data=array([97], dtype=uint8), data_type=<ChannelType.CHAR: 4>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=3, metadata=None)

In [23]: randstr.read()
Out[23]: ReadNotifyResponse(data=array([97], dtype=uint8), data_type=<ChannelType.CHAR: 4>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=4, metadata=None)

In [24]: randstr.read()
Out[24]: ReadNotifyResponse(data=array([98], dtype=uint8), data_type=<ChannelType.CHAR: 4>, data_count=1, status=CAStatusCode(name='ECA_NORMAL', code=0, code_with_severity=1, severity=<CASeverity.SUCCESS: 1>, success=1, defunct=False, description='Normal successful completion'), ioid=5, metadata=None)

More…

Take a look around the ioc_examples subpackage for more examples not covered here.