1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
from epics import ca, dbr
import time
import debugtime
try:
from collections import OrderedDict
except:
from ordereddict import OrderedDict
dt = debugtime.debugtime()
def add(x):
print(x)
dt.add(x)
add('test of fast connection to many PVs')
pvnames = []
results = OrderedDict()
MAX_PVS = 12500
for line in open('fastconn_pvlist.txt','r').readlines():
if not line.startswith('#'):
pvnames.append(line.strip())
if MAX_PVS is not None:
pvnames = pvnames[:MAX_PVS]
add('Read PV list: Will connect to %i PVs' % len(pvnames))
libca = ca.initialize_libca()
for name in pvnames:
chid = ca.create_channel(name, connect=False, auto_cb=False)
results[name] = {'chid': chid}
time.sleep(0.001)
add("created PVs with ca_create_channel")
for name in pvnames:
ca.connect_channel(results[name]['chid'])
time.sleep(0.001)
add("connected to PVs with connect_channel")
ca.pend_event(1.e-2)
for name in pvnames:
chid = results[name]['chid']
val = ca.get(chid, wait=False)
results[name]['value'] = val
add("did ca.get(wait=False)")
ca.poll(2.e-3, 1.0)
add("ca.poll() complete")
for name in pvnames:
results[name]['value'] = ca.get_complete(results[name]['chid'])
add("ca.get_complete() for all PVs")
f = open('fastconn_pvdata.sav', 'w')
for name, val in results.items():
f.write("%s %s\n" % (name.strip(), val['value']))
f.close()
add("wrote PV values to disk")
dt.show()
time.sleep(0.01)
ca.poll()
|
