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 textwrap import dedent
from caproto.server import PVGroup, ioc_arg_parser, pvproperty, run
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, doc='An integer')
B = pvproperty(value=2.0, doc='A float')
C = pvproperty(value=[1, 2, 3], doc='An array of integers')
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)
|
An IOC with three uncoupled read/writable PVs |
$ 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.
|
This function monitors the terminal it was run in for keystrokes. |
|
#!/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'
def __init__(self, *args, max_value, **kwargs):
super().__init__(*args, **kwargs)
self.max_value = max_value
# PV: {prefix}random - defaults to dtype of int
@pvproperty
async def random(self, instance):
print('Picking a random value from 1 to', self.max_value)
return random.randint(1, self.max_value)
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, max_value=5)
# PV: {prefix}group2-random
group2 = SubGroup(MySubGroup, prefix='group2-', max_value=20)
# 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
Records¶
See Records.
|
#!/usr/bin/env python3
import caproto
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, record='ai')
# And an analog output (ao) record:
B = pvproperty(value=2.0, record='ao',
precision=3)
# and an ai with all the bells and whistles
C = pvproperty(value=0.0,
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('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,
precision=3,
field_spec=B.field_spec)
# D will also be reported as an 'ao' record like B, as it uses the same
# field specification.
E = pvproperty(value='this is a test',
record='stringin',
dtype=caproto.ChannelType.STRING,
)
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.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([64], 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([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=1, metadata=None)
In [21]: randint.read()
Out[21]: ReadNotifyResponse(data=array([44], 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([99], 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)
Mini-Beamline¶
Simulate your own mini beamline with this IOC.
This example is quite large. |
|
|
A collection of detectors coupled to motors and an oscillating beam current |
|
A pinhole simulation device. |
|
An edge simulation device. |
|
A slit simulation device. |
|
More…¶
Take a look around the ioc_examples subpackage for more examples not covered here.
|
An IOC with some enums. |
|
An IOC with three uncoupled read/writable PVs |
|
Simulates an exponential decay to a set point. |
|
Helpers¶
caproto offers several “helper” subgroups (SubGroup)
that are of general use, and could be considered part of the “caproto server
standard library”, so to speak.
Autosave¶
|
|
|
Save to a primary filename, while rotating out old copies. |
Status / Statistics¶
|
An IocStats-like tool for caproto IOCs. |
|
|
|
An IocStats-like tool for caproto IOCs. |
|
A helper which quickly allows for tracing memory usage and allocations on a caproto server instance. |