forked from labjack/LabJackPython
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathModbus.py
More file actions
324 lines (256 loc) · 9.68 KB
/
Copy pathModbus.py
File metadata and controls
324 lines (256 loc) · 9.68 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
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
# File: Modbus.py
# Author: LabJack Corp.
# Created: 05.05.2008
from __future__ import with_statement
import datetime
import threading
from struct import pack, unpack
AES_CHANNEL = 64000
IP_PART1_CHANNEL = 64008
IP_PART2_CHANNEL = 64009
PORT_CHANNEL = 64010
HEARTBEAT_CHANNEL = 64016
DEBUG_CHANNEL = 64017
DEVICE_TYPE_CHANNEL = 65000
SERIAL_NUMBER_CHANNEL = 65001
READ_PACKET = 3
WRITE_PACKET = 6
HEADER_LENGTH = 9
BYTES_PER_REGISTER = 2
GLOBAL_TRANSACTION_ID_LOCK = threading.Lock()
MAX_TRANS_ID = 64760
def _calcBaseTransId():
t = datetime.datetime.now()
d = "%s%s%s%s" % (t.hour, t.minute, t.second, t.microsecond)
d = int(d) % MAX_TRANS_ID
return d
BASE_TRANS_ID = _calcBaseTransId()
CURRENT_TRANS_IDS = set()
def _buildHeaderBytes(length = 6, unitId = None):
with GLOBAL_TRANSACTION_ID_LOCK:
global BASE_TRANS_ID, CURRENT_TRANS_IDS
if unitId is None:
basicHeader = (BASE_TRANS_ID, 0, length, 0x00)
else:
basicHeader = (BASE_TRANS_ID, 0, length, unitId)
CURRENT_TRANS_IDS.add(BASE_TRANS_ID)
BASE_TRANS_ID = ( BASE_TRANS_ID + 1 ) % MAX_TRANS_ID
return pack('>HHHB', *basicHeader)
def _checkTransId(transId):
with GLOBAL_TRANSACTION_ID_LOCK:
global CURRENT_TRANS_IDS
if transId in CURRENT_TRANS_IDS:
CURRENT_TRANS_IDS.remove(transId)
else:
raise ModbusException("Got an unexpected transaction ID. Id = %s, Set = %s" % (transId, CURRENT_TRANS_IDS))
def readHoldingRegistersRequest(addr, numReg = None, unitId = None):
if numReg is None:
numReg = calcNumberOfRegisters(addr)
packet = _buildHeaderBytes(unitId = unitId) + pack('>BHH', 0x03, addr, numReg)
return packet
def readHoldingRegistersResponse(packet, payloadFormat=None):
# Example: Device type is 9
# [0, 0, 5, 255, 3, 2, 9]
# H H H c c c payload
# 0 1 2 3 4 5 6+
HEADER_LENGTH = 9
header = unpack('>HHHBBB', packet[:HEADER_LENGTH])
#print "header", [ c for c in header ]
#print "header", header
# Check that protocol ID is 0
if header[1] != 0:
raise ModbusException("Got an unexpected protocol ID: %s (expected 0). Please make sure that you have the latest firmware. UE9s need a Comm Firmware of 1.50 or greater.\n\nThe packet you received: %s" % (header[1], repr(packet)))
# Check for valid Trans ID
_checkTransId(header[0])
#Check for exception
if header[4] == 0x83:
raise ModbusException("Error reading register: A Modbus error %s was raised.\n\nThe packet you received: %s" % (header[5], repr(packet)))
#Check for proper command
if header[4] != 0x03:
raise ModbusException("Not a read holding registers packet.\n\nGot: %s" % repr(packet))
#Check for proper length
payloadLength = header[5]
if (payloadLength + HEADER_LENGTH) != len(packet):
#print "packet length is", len(packet)
#print "payload and header is", payloadLength + HEADER_LENGTH
raise ModbusException("Packet length not valid. Expected %s, Got %s\n\nThe packet you received: %s" % (payloadLength + HEADER_LENGTH, len(packet), repr(packet)))
if payloadFormat is None:
payloadFormat = '>' + 'H' * (payloadLength//2)
# When we write '>s', we mean a variable-length string.
# We just didn't know the length when we wrote it.
if payloadFormat == '>s':
payloadFormat = '>' + 's' * payloadLength
payload = unpack(payloadFormat, packet[HEADER_LENGTH:])
if len(payload) == 1:
return payload[0]
else:
return list(payload)
def readInputRegistersRequest(addr, numReg = None):
if numReg is None:
numReg = calcNumberOfRegisters(addr)
packet = _buildHeaderBytes() + pack('>BHH', 0x04, addr, numReg)
return packet
def readInputRegistersResponse(packet, payloadFormat=None):
# Example: Device type is 9
# [0, 0, 5, 255, 3, 2, 9]
# H H H c c c payload
# 0 1 2 3 4 5 6+
HEADER_LENGTH = 9
header = unpack('>HHHBBB', packet[:HEADER_LENGTH])
# Check for valid Trans ID
_checkTransId(header[0])
#Check for exception
if header[4] == 0x83:
raise ModbusException(header[5])
#Check for proper command
if header[4] != 0x04:
raise ModbusException("Not a read holding registers packet.")
#Check for proper length
payloadLength = header[5]
if (payloadLength + HEADER_LENGTH) != len(packet):
#print "packet length is", len(packet)
#print "payload and header is", payloadLength + HEADER_LENGTH
raise ModbusException("Packet length not valid.")
if payloadFormat is None:
payloadFormat = '>' + 'H' * (payloadLength//2)
# When we write '>s', we mean a variable-length string.
# We just didn't know the length when we wrote it.
if payloadFormat == '>s':
payloadFormat = '>' + 's' * payloadLength
payload = unpack(payloadFormat, packet[HEADER_LENGTH:])
return payload
def writeRegisterRequest(addr, value, unitId = None):
if not isinstance(value, int):
raise TypeError("Value written must be an integer.")
packet = _buildHeaderBytes(unitId = unitId) + pack('>BHH', 0x06, addr, value)
return packet
def writeRegistersRequest(startAddr, values, unitId = None):
numReg = len(values)
for v in values:
if not isinstance(v, int):
raise TypeError("Value written must be an integer.")
if unitId is None:
unitId = 0xff
header = _buildHeaderBytes(length = 7+(numReg*2), unitId = unitId)
header += pack('>BHHB', *(16, startAddr, numReg, numReg*2) )
format = '>' + 'H' * numReg
packet = header + pack(format, *values)
return packet
def writeRegisterRequestValue(data):
"""Return the value to be written in a writeRegisterRequest Packet."""
packet = unpack('>H', data[10:])
return packet[0]
class ModbusException(Exception):
def __init__(self, exceptCode):
self.exceptCode = exceptCode
def __str__(self):
return repr(self.exceptCode)
def calcNumberOfRegisters(addr, numReg = None):
return calcNumberOfRegistersAndFormat(addr, numReg)[0]
def calcFormat(addr, numReg = None):
return calcNumberOfRegistersAndFormat(addr, numReg)[1]
def calcNumberOfRegistersAndFormat(addr, numReg = None):
# TODO add special cases for channels above
if addr < 1000:
# Analog Inputs
minNumReg = 2
format = 'f'
elif addr >= 5000 and addr < 6000:
# DAC Values
minNumReg = 2
format = 'f'
elif addr >= 7000 and addr < 8000:
# Timers / Counters
minNumReg = 2
format = 'I'
elif addr in range(64008,64018) or addr == 65001:
# Serial Number
minNumReg = 2
format = 'I'
elif addr in range(10000,10010):
# VBatt/Temp/RH/Light/Pressure
minNumReg = 2
format = 'f'
elif addr in range(12000,13000):
# RXLQI/TXLQI/VBatt/Temp/Light/Motion/Sound/RH/Pressure
minNumReg = 2
format = 'f'
elif addr in range(50100, 50103):
# Check-in interval
minNumReg = 2
format = 'I'
elif addr in range(57002, 57010):
# TX/RX Bridge stuff
minNumReg = 2
format = 'I'
elif addr in range(57050, 57056):
# VUSB/VJack/VST
minNumReg = 2
format = 'f'
elif addr == 59990:
# Rapid mode
minNumReg = 1
format = 'H'
elif addr == 59200:
# NumberOfKnownDevices
minNumReg = 2
format = 'I'
else:
minNumReg = 1
format = 'H'
if numReg:
if (numReg%minNumReg) == 0:
return (numReg, '>' + ( format * (numReg//minNumReg) ))
else:
raise ModbusException("For address %s, the number of registers must be divisible by %s" % (addr, minNumReg))
else:
return ( minNumReg, '>'+format)
def getStartingAddress(packet):
"""Get the address of a modbus request"""
return ((ord(packet[8]) << 8) + ord(packet[9]))
def getRequestType(packet):
"""Get the request type of a modbus request."""
return ord(packet[7])
def getTransactionId(packet):
"""Pulls out the transaction id of the packet"""
if isinstance(packet, list):
return unpack(">H", pack("BB", *packet[:2]) )[0]
else:
return unpack(">H", packet[:2])[0]
def getProtocolId(packet):
"""Pulls out the transaction id of the packet"""
if isinstance(packet, list):
return unpack(">H", pack("BB", *packet[2:4]) )[0]
else:
return unpack(">H", packet[2:4])[0]
def parseIntoPackets(packet):
while True:
if isinstance(packet, list):
firstLength = packet[5]+6
else:
firstLength = ord(packet[5])+6
if len(packet) == firstLength:
yield packet
raise StopIteration
else:
yield packet[:firstLength]
packet = packet[firstLength:]
def parseSpontaneousDataPacket(packet):
if isinstance(packet, list):
localId = packet[6]
packet = pack("B"*len(packet), *packet)
else:
localId = ord(packet[6])
transId = unpack(">H", packet[0:2])[0]
report = unpack(">HBBfHH"+"f"*8, packet[9:53])
results = dict()
results['unitId'] = localId
results['transId'] = transId
results['Rxlqi'] = report[1]
results['Txlqi'] = report[2]
results['Battery'] = report[3]
results['Temperature'] = report[6]
results['Light'] = report[7]
results['Bump'] = report[4]
results['Sound'] = report[11]
return results