forked from gijzelaerr/python-snap7
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathread_multi.py
More file actions
68 lines (53 loc) · 2.01 KB
/
read_multi.py
File metadata and controls
68 lines (53 loc) · 2.01 KB
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
"""
Example ussage of the read_multi_vars function
This was tested against a S7-319 CPU
"""
import ctypes
import snap7
from snap7.common import check_error
from snap7.types import S7DataItem, S7AreaDB, S7WLByte
client = snap7.client.Client()
client.connect('10.100.5.2', 0, 2)
data_items = (S7DataItem * 3)()
data_items[0].Area = ctypes.c_int32(S7AreaDB)
data_items[0].WordLen = ctypes.c_int32(S7WLByte)
data_items[0].Result = ctypes.c_int32(0)
data_items[0].DBNumber = ctypes.c_int32(200)
data_items[0].Start = ctypes.c_int32(16)
data_items[0].Amount = ctypes.c_int32(4) # reading a REAL, 4 bytes
data_items[1].Area = ctypes.c_int32(S7AreaDB)
data_items[1].WordLen = ctypes.c_int32(S7WLByte)
data_items[1].Result = ctypes.c_int32(0)
data_items[1].DBNumber = ctypes.c_int32(200)
data_items[1].Start = ctypes.c_int32(12)
data_items[1].Amount = ctypes.c_int32(4) # reading a REAL, 4 bytes
data_items[2].Area = ctypes.c_int32(S7AreaDB)
data_items[2].WordLen = ctypes.c_int32(S7WLByte)
data_items[2].Result = ctypes.c_int32(0)
data_items[2].DBNumber = ctypes.c_int32(200)
data_items[2].Start = ctypes.c_int32(2)
data_items[2].Amount = ctypes.c_int32(2) # reading an INT, 2 bytes
# create buffers to receive the data
# use the Amount attribute on each item to size the buffer
for di in data_items:
# create the buffer
buffer = ctypes.create_string_buffer(di.Amount)
# cast the pointer to the buffer to the required type
pBuffer = ctypes.cast(ctypes.pointer(buffer),
ctypes.POINTER(ctypes.c_uint8))
di.pData = pBuffer
result, data_items = client.read_multi_vars(data_items)
for di in data_items:
check_error(di.Result)
result_values = []
# function to cast bytes to match data_types[] above
byte_to_value = [util.get_real, util.get_real, util.get_int]
# unpack and test the result of each read
for i in range(0, len(data_items)):
btv = byte_to_value[i]
di = data_items[i]
value = btv(di.pData, 0)
result_values.append(value)
print(result_values)
client.disconnect()
client.destroy()