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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
|
#!/usr/bin/env python
# M Newville <newville@cars.uchicago.edu>
# The University of Chicago, 2010
# Epics Open License
#
# Epics Database Records (DBR) Constants and Definitions
# most of the code here is copied from db_access.h
#
""" constants and declaration of data types for Epics database records
This is mostly copied from CA header files
"""
import ctypes
import functools
import os
import sys
import platform
HAS_NUMPY = False
try:
import numpy
HAS_NUMPY = True
except ImportError:
pass
PY64_WINDOWS = (os.name == 'nt' and platform.architecture()[0].startswith('64'))
IRON_PYTHON = platform.python_implementation() == 'IronPython'
PY_MAJOR, PY_MINOR = sys.version_info[:2]
# EPICS Constants
ECA_NORMAL = 1
ECA_TIMEOUT = 80
ECA_IODONE = 339
ECA_ISATTACHED = 424
ECA_BADCHID = 410
CS_CONN = 2
OP_CONN_UP = 6
OP_CONN_DOWN = 7
CS_NEVER_SEARCH = 4
#
# Note that DBR_XXX should be replaced with dbr.XXX
#
STRING = 0
INT = 1
SHORT = 1
FLOAT = 2
ENUM = 3
CHAR = 4
LONG = 5
DOUBLE = 6
TIME_STRING = 14
TIME_INT = 15
TIME_SHORT = 15
TIME_FLOAT = 16
TIME_ENUM = 17
TIME_CHAR = 18
TIME_LONG = 19
TIME_DOUBLE = 20
CTRL_STRING = 28
CTRL_INT = 29
CTRL_SHORT = 29
CTRL_FLOAT = 30
CTRL_ENUM = 31
CTRL_CHAR = 32
CTRL_LONG = 33
CTRL_DOUBLE = 34
MAX_STRING_SIZE = 40
MAX_UNITS_SIZE = 8
MAX_ENUM_STRING_SIZE = 26
MAX_ENUMS = 16
#EPICS2UNIX_EPOCH = 631173600.0 - time.timezone
EPICS2UNIX_EPOCH = 631152000.0
# create_subscription mask constants
DBE_VALUE = 1
DBE_LOG = 2
DBE_ALARM = 4
DBE_PROPERTY = 8
chid_t = ctypes.c_long
# Note that Windows needs to be told that chid is 8 bytes for 64-bit,
# except that Python2 is very weird -- using a 4byte chid for 64-bit,
# but needing a 1 byte padding!
if PY64_WINDOWS and PY_MAJOR > 2:
chid_t = ctypes.c_int64
short_t = ctypes.c_short
ushort_t = ctypes.c_ushort
int_t = ctypes.c_int
uint_t = ctypes.c_uint
long_t = ctypes.c_long
ulong_t = ctypes.c_ulong
float_t = ctypes.c_float
double_t = ctypes.c_double
byte_t = ctypes.c_byte
ubyte_t = ctypes.c_ubyte
string_t = ctypes.c_char * MAX_STRING_SIZE
char_t = ctypes.c_char
char_p = ctypes.c_char_p
void_p = ctypes.c_void_p
py_obj = ctypes.py_object
value_offset = None
# extended DBR types:
class TimeStamp(ctypes.Structure):
"emulate epics timestamp"
_fields_ = [('secs', uint_t), ('nsec', uint_t)]
_STAT_SEV = (('status', short_t), ('severity', short_t))
_STAT_SEV_TS = (('status', short_t), ('severity', short_t),
('stamp', TimeStamp))
_UNITS = ('units', char_t * MAX_UNITS_SIZE)
def make_unixtime(stamp):
"UNIX timestamp (seconds) from Epics TimeStamp structure"
return (EPICS2UNIX_EPOCH + stamp.secs + 1.e-6*int(1.e-3*stamp.nsec))
class time_string(ctypes.Structure):
"dbr time string"
_fields_ = list(_STAT_SEV_TS) + [('value', MAX_STRING_SIZE*char_t)]
class time_short(ctypes.Structure):
"dbr time short"
_fields_ = list(_STAT_SEV_TS) + [('RISC_pad', short_t),
('value', short_t)]
class time_float(ctypes.Structure):
"dbr time float"
_fields_ = list(_STAT_SEV_TS) + [('value', float_t)]
class time_enum(ctypes.Structure):
"dbr time enum"
_fields_ = list(_STAT_SEV_TS) + [('RISC_pad', short_t),
('value', ushort_t)]
class time_char(ctypes.Structure):
"dbr time char"
_fields_ = list(_STAT_SEV_TS) + [('RISC_pad0', short_t),
('RISC_pad1', byte_t),
('value', byte_t)]
class time_long(ctypes.Structure):
"dbr time long"
_fields_ = list(_STAT_SEV_TS) + [('value', int_t)]
class time_double(ctypes.Structure):
"dbr time double"
_fields_ = list(_STAT_SEV_TS) + [('RISC_pad', int_t),
('value', double_t)]
# DBR types with full control and graphical fields
# yes, this strange order is as in db_access.h!!!
ctrl_limits = ('upper_disp_limit', 'lower_disp_limit',
'upper_alarm_limit', 'upper_warning_limit',
'lower_warning_limit','lower_alarm_limit',
'upper_ctrl_limit', 'lower_ctrl_limit')
def _gen_ctrl_lims(t=short_t):
"create types for control limits"
return [(s, t) for s in ctrl_limits]
class ctrl_enum(ctypes.Structure):
"dbr ctrl enum"
_fields_ = list(_STAT_SEV)
_fields_.extend([ ('no_str', short_t),
('strs', (char_t * MAX_ENUM_STRING_SIZE) * MAX_ENUMS),
('value', ushort_t)])
class ctrl_short(ctypes.Structure):
"dbr ctrl short"
_fields_ = list(_STAT_SEV) + [_UNITS] + _gen_ctrl_lims(t=short_t)
_fields_.extend([('value', short_t )])
class ctrl_char(ctypes.Structure):
"dbr ctrl long"
_fields_ = list(_STAT_SEV) +[_UNITS] + _gen_ctrl_lims(t=byte_t)
_fields_.extend([('RISC_pad', byte_t), ('value', ubyte_t)])
class ctrl_long(ctypes.Structure):
"dbr ctrl long"
_fields_ = list(_STAT_SEV) +[_UNITS] + _gen_ctrl_lims(t=int_t)
_fields_.extend([('value', int_t)])
class ctrl_float(ctypes.Structure):
"dbr ctrl float"
_fields_ = list(_STAT_SEV)
_fields_.extend([('precision', short_t),
('RISC_pad', short_t)] + [_UNITS])
_fields_.extend( _gen_ctrl_lims(t=float_t) )
_fields_.extend([('value', float_t)])
class ctrl_double(ctypes.Structure):
"dbr ctrl double"
_fields_ = list(_STAT_SEV)
_fields_.extend([('precision', short_t),
('RISC_pad', short_t)] + [_UNITS])
_fields_.extend( _gen_ctrl_lims(t=double_t) )
_fields_.extend([('value', double_t)])
NP_Map = {}
if HAS_NUMPY:
NP_Map = {INT: numpy.int16,
FLOAT: numpy.float32,
ENUM: numpy.uint16,
CHAR: numpy.uint8,
LONG: numpy.int32,
DOUBLE: numpy.float64}
# map of Epics DBR types to ctypes types
Map = {STRING: string_t,
INT: short_t,
FLOAT: float_t,
ENUM: ushort_t,
CHAR: ubyte_t,
LONG: int_t,
DOUBLE: double_t,
TIME_STRING: time_string,
TIME_INT: time_short,
TIME_SHORT: time_short,
TIME_FLOAT: time_float,
TIME_ENUM: time_enum,
TIME_CHAR: time_char,
TIME_LONG: time_long,
TIME_DOUBLE: time_double,
# Note: there is no ctrl string in the C definition
CTRL_STRING: time_string,
CTRL_SHORT: ctrl_short,
CTRL_INT: ctrl_short,
CTRL_FLOAT: ctrl_float,
CTRL_ENUM: ctrl_enum,
CTRL_CHAR: ctrl_char,
CTRL_LONG: ctrl_long,
CTRL_DOUBLE: ctrl_double
}
def native_type(ftype):
"return native field type from TIME or CTRL variant"
if ftype == CTRL_STRING:
ftype = TIME_STRING
ntype = ftype
if ftype > CTRL_STRING:
ntype -= CTRL_STRING
elif ftype >= TIME_STRING:
ntype -= TIME_STRING
return ntype
def Name(ftype, reverse=False):
""" convert integer data type to dbr Name, or optionally reverse that
look up (that is, name to integer)"""
m = {STRING: 'STRING',
INT: 'INT',
FLOAT: 'FLOAT',
ENUM: 'ENUM',
CHAR: 'CHAR',
LONG: 'LONG',
DOUBLE: 'DOUBLE',
TIME_STRING: 'TIME_STRING',
TIME_SHORT: 'TIME_SHORT',
TIME_FLOAT: 'TIME_FLOAT',
TIME_ENUM: 'TIME_ENUM',
TIME_CHAR: 'TIME_CHAR',
TIME_LONG: 'TIME_LONG',
TIME_DOUBLE: 'TIME_DOUBLE',
CTRL_STRING: 'CTRL_STRING',
CTRL_SHORT: 'CTRL_SHORT',
CTRL_FLOAT: 'CTRL_FLOAT',
CTRL_ENUM: 'CTRL_ENUM',
CTRL_CHAR: 'CTRL_CHAR',
CTRL_LONG: 'CTRL_LONG',
CTRL_DOUBLE: 'CTRL_DOUBLE',
}
if reverse:
name = ftype.upper()
if name in list(m.values()):
for key, val in m.items():
if name == val: return key
return m.get(ftype, 'unknown')
def cast_args(args):
"""returns casted array contents
returns: [dbr_ctrl or dbr_time struct,
count * native_type structs]
If data is already of a native_type, the first
value in the list will be None.
"""
ftype = args.type
if ftype not in Map:
ftype = double_t
ntype = native_type(ftype)
if ftype != ntype:
native_start = args.raw_dbr + value_offset[ftype]
return [ctypes.cast(args.raw_dbr,
ctypes.POINTER(Map[ftype])).contents,
ctypes.cast(native_start,
ctypes.POINTER(args.count * Map[ntype])).contents
]
else:
return [None,
ctypes.cast(args.raw_dbr,
ctypes.POINTER(args.count * Map[ftype])).contents
]
if PY64_WINDOWS:
def make_callback(func, args):
""" make callback function"""
# note that ctypes.POINTER is needed for 64-bit Python on Windows
@functools.wraps(func)
def wrapped(arg, **kwargs):
# On 64-bit Windows, `arg.contents` seems to be equivalent to other
# platforms' `arg`
if hasattr(arg, 'contents'):
return func(arg.contents, **kwargs)
return func(arg, **kwargs)
return ctypes.CFUNCTYPE(None, ctypes.POINTER(args))(wrapped)
else:
def make_callback(func, args):
""" make callback function"""
return ctypes.CFUNCTYPE(None, args)(func)
class event_handler_args(ctypes.Structure):
"event handler arguments"
_fields_ = [('usr', ctypes.py_object),
('chid', chid_t),
('type', long_t),
('count', long_t),
('raw_dbr', void_p),
('status', int_t)]
class connection_args(ctypes.Structure):
"connection arguments"
_fields_ = [('chid', chid_t),
('op', long_t)]
class access_rights_handler_args(ctypes.Structure):
"access rights arguments"
_fields_ = [('chid', chid_t),
('access', ubyte_t)]
if PY64_WINDOWS and PY_MAJOR == 2:
# need to add padding on 64-bit Windows for Python2 -- yuck!
class event_handler_args(ctypes.Structure):
"event handler arguments"
_fields_ = [('usr', ctypes.py_object),
('chid', chid_t),
('_pad_', ctypes.c_int8),
('type', ctypes.c_int32),
('count', ctypes.c_int32),
('raw_dbr', void_p),
('status', ctypes.c_int32)]
class connection_args(ctypes.Structure):
"connection arguments"
_fields_ = [('chid', chid_t),
('_pad_',ctypes.c_int8),
('op', long_t)]
class access_rights_handler_args(ctypes.Structure):
"access rights arguments"
_fields_ = [('chid', chid_t),
('_pad_',ctypes.c_int8),
('access', ubyte_t)]
|
