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
|
#!/usr/bin/env python
#module for serial IO for POSIX compatible systems, like Linux
#see serial.py
#
#(C) 2001 Chris Liechti <cliechti@gmx.net>
# this is distributed under a free software license, see license.txt
#
#parts based on code from Grant B. Edwards <grante@visi.com>:
# ftp://ftp.visi.com/users/grante/python/PosixSerial.py
# references: http://www.easysw.com/~mike/serial/serial.html
import sys, os, fcntl, termios, struct, string, select
VERSION = string.split("$Revision: 1.1.1.1 $")[1] #extract CVS version
PARITY_NONE, PARITY_EVEN, PARITY_ODD = range(3)
STOPBITS_ONE, STOPBITS_TWO = (1, 2)
FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS = (5,6,7,8)
if (sys.hexversion < 0x020100f0):
import TERMIOS
else:
TERMIOS = termios
if (sys.hexversion < 0x020200f0):
import FCNTL
else:
FCNTL = fcntl
#try to detect the os so that a device can be selected...
plat = string.lower(sys.platform)
if plat[:5] == 'linux': #Linux (confirmed)
def device(port):
return '/dev/ttyS%d' % port
elif plat == 'openbsd3': #BSD (confirmed)
def device(port):
return '/dev/ttyp%d' % port
elif plat[:3] == 'bsd' or \
plat[:6] == 'netbsd' or \
plat[:7] == 'freebsd' or \
plat[:7] == 'openbsd' or \
plat[:6] == 'darwin': #BSD (confirmed for freebsd4: cuaa%d)
def device(port):
return '/dev/cuaa%d' % port
elif plat[:4] == 'irix': #IRIX® (not tested)
def device(port):
return '/dev/ttyf%d' % port
elif plat[:2] == 'hp': #HP-UX (not tested)
def device(port):
return '/dev/tty%dp0' % (port+1)
elif plat[:5] == 'sunos': #Solaris®/SunOS® (not tested)
def device(port):
return '/dev/tty%c' % (ord('a')+port)
elif plat[:3] == 'dgux': #Digital UNIX® (not tested)
def device(port):
return '/dev/tty0%d' % (port+1)
else:
#platform detection has failed...
info = "sys.platform = %r\nos.name = %r\nserialposix.py version = %s" % (sys.platform, os.name, VERSION)
print """send this information to the author of this module:
%s
also add the device name of the serial port and where the
counting starts for the first serial port.
e.g. 'first serial port: /dev/ttyS0'
and with a bit luck you can get this module running...
"""
raise Exception, "this module does not run on this platform, sorry."
#whats up with "aix", "beos", "sco", ....
#they should work, just need to know the device names.
#"cygwin" has a POSIX emulation but does not seem to have a /dev/ttyxx structure?
# construct dictionaries for baud rate lookups
baudEnumToInt = {}
baudIntToEnum = {}
for rate in (0,50,75,110,134,150,200,300,600,1200,1800,2400,4800,9600,
19200,38400,57600,115200,230400,460800,500000,576000,921600,
1000000,1152000,1500000,2000000,2500000,3000000,3500000,4000000
):
try:
i = eval('TERMIOS.B'+str(rate))
baudEnumToInt[i]=rate
baudIntToEnum[rate] = i
except:
pass
if hasattr(TERMIOS, 'TIOCMGET'):
TIOCMGET = TERMIOS.TIOCMGET
TIOCMBIS = TERMIOS.TIOCMBIS
TIOCMBIC = TERMIOS.TIOCMBIC
TIOCMSET = TERMIOS.TIOCMSET
TIOCM_LE = TERMIOS.TIOCM_LE
TIOCM_DTR = TERMIOS.TIOCM_DTR
TIOCM_RTS = TERMIOS.TIOCM_RTS
TIOCM_ST = TERMIOS.TIOCM_ST
TIOCM_SR = TERMIOS.TIOCM_SR
TIOCM_CTS = TERMIOS.TIOCM_CTS
TIOCM_CAR = TERMIOS.TIOCM_CAR
TIOCM_RNG = TERMIOS.TIOCM_RNG
TIOCM_DSR = TERMIOS.TIOCM_DSR
TIOCM_CD = TERMIOS.TIOCM_CD
TIOCM_RI = TERMIOS.TIOCM_RI
TIOCM_OUT1 = TERMIOS.TIOCM_OUT1
TIOCM_OUT2 = TERMIOS.TIOCM_OUT2
else: #workaround for older python versions
TIOCMGET = 0x5415
TIOCMBIS = 0x5416
TIOCMBIC = 0x5417
TIOCMSET = 0x5418
TIOCM_LE = 0x001
TIOCM_DTR = 0x002
TIOCM_RTS = 0x004
TIOCM_ST = 0x008
TIOCM_SR = 0x010
TIOCM_CTS = 0x020
TIOCM_CAR = 0x040
TIOCM_RNG = 0x080
TIOCM_DSR = 0x100
TIOCM_CD = TIOCM_CAR
TIOCM_RI = TIOCM_RNG
TIOCM_OUT1 = 0x2000
TIOCM_OUT2 = 0x4000
TIOCM_zero_str = struct.pack('I', 0)
TIOCM_RTS_str = struct.pack('I', TIOCM_RTS)
TIOCM_DTR_str = struct.pack('I', TIOCM_DTR)
portNotOpenError = ValueError('port not open')
class Serial:
def __init__(self,
port, #number of device, numbering starts at
#zero. if everything fails, the user
#can specify a device string, note
#that this isn't portable anymore
baudrate=9600, #baudrate
bytesize=EIGHTBITS, #number of databits
parity=PARITY_NONE, #enable parity checking
stopbits=STOPBITS_ONE, #number of stopbits
timeout=None, #set a timeout value, None for waiting forever
xonxoff=0, #enable software flow control
rtscts=0, #enable RTS/CTS flow control
):
self.fd = None
self.timeout = timeout
vmin = vtime = 0 #timeout is done via select
#open
if type(port) == type(''): #strings are taken directly
self.portstr = port
else:
self.portstr = device(port) #numbers are transformed to a os dependant string
self.fd = os.open(self.portstr, os.O_RDWR|os.O_NOCTTY|os.O_NONBLOCK)
fcntl.fcntl(self.fd, FCNTL.F_SETFL, 0) #set blocking
self.__tcgetattr() #read current settings
#set up raw mode / no echo / binary
self.cflag = self.cflag | (TERMIOS.CLOCAL|TERMIOS.CREAD)
self.lflag = self.lflag & ~(TERMIOS.ICANON|TERMIOS.ECHO|TERMIOS.ECHOE|TERMIOS.ECHOK|TERMIOS.ECHONL|
TERMIOS.ECHOCTL|TERMIOS.ECHOPRT|TERMIOS.ECHOKE|TERMIOS.ISIG|TERMIOS.IEXTEN)
self.oflag = self.oflag & ~(TERMIOS.OPOST)
if hasattr(TERMIOS, 'IUCLC'):
self.iflag = self.iflag & ~(TERMIOS.INLCR|TERMIOS.IGNCR|TERMIOS.ICRNL|TERMIOS.IUCLC|TERMIOS.IGNBRK)
else:
self.iflag = self.iflag & ~(TERMIOS.INLCR|TERMIOS.IGNCR|TERMIOS.ICRNL|TERMIOS.IGNBRK)
#setup baudrate
try:
self.ispeed = self.ospeed = baudIntToEnum[baudrate]
except:
raise ValueError,'invalid baud rate: %s' % baudrate
#setup char len
self.cflag = self.cflag & ~TERMIOS.CSIZE
if bytesize == 8:
self.cflag = self.cflag | TERMIOS.CS8
elif bytesize == 7:
self.cflag = self.cflag | TERMIOS.CS7
elif bytesize == 6:
self.cflag = self.cflag | TERMIOS.CS6
elif bytesize == 5:
self.cflag = self.cflag | TERMIOS.CS5
else:
raise ValueError,'invalid char len: '+str(clen)
#setup stopbits
if stopbits == STOPBITS_ONE:
self.cflag = self.cflag & ~(TERMIOS.CSTOPB)
elif stopbits == STOPBITS_TWO:
self.cflag = self.cflag | (TERMIOS.CSTOPB)
else:
raise ValueError,'invalid stopit specification:'+str(stopbits)
#setup parity
self.iflag = self.iflag & ~(TERMIOS.INPCK|TERMIOS.ISTRIP)
if parity == PARITY_NONE:
self.cflag = self.cflag & ~(TERMIOS.PARENB|TERMIOS.PARODD)
elif parity == PARITY_EVEN:
self.cflag = self.cflag & ~(TERMIOS.PARODD)
self.cflag = self.cflag | (TERMIOS.PARENB)
elif parity == PARITY_ODD:
self.cflag = self.cflag | (TERMIOS.PARENB|TERMIOS.PARODD)
else:
raise ValueError,'invalid parity: '+str(par)
#setup flow control
#xonxoff
if hasattr(TERMIOS, 'IXANY'):
if xonxoff:
self.iflag = self.iflag | (TERMIOS.IXON|TERMIOS.IXOFF|TERMIOS.IXANY)
else:
self.iflag = self.iflag & ~(TERMIOS.IXON|TERMIOS.IXOFF|TERMIOS.IXANY)
else:
if xonxoff:
self.iflag = self.iflag | (TERMIOS.IXON|TERMIOS.IXOFF)
else:
self.iflag = self.iflag & ~(TERMIOS.IXON|TERMIOS.IXOFF)
#rtscts
if hasattr(TERMIOS, 'CRTSCTS'):
if rtscts:
self.cflag = self.cflag | (TERMIOS.CRTSCTS)
else:
self.cflag = self.cflag & ~(TERMIOS.CRTSCTS)
#buffer
#vmin "minimal number of characters to be read. = for non blocking"
if vmin<0 or vmin>255:
raise ValueError,'invalid vmin: '+str(vmin)
self.cc[TERMIOS.VMIN] = vmin
#vtime
if vtime<0 or vtime>255:
raise ValueError,'invalid vtime: '+str(vtime)
self.cc[TERMIOS.VTIME] = vtime
#activate settings
self.__tcsetattr()
def __tcsetattr(self):
termios.tcsetattr(self.fd, TERMIOS.TCSANOW, [self.iflag,self.oflag,self.cflag,self.lflag,self.ispeed,self.ospeed,self.cc])
def __tcgetattr(self):
self.iflag,self.oflag,self.cflag,self.lflag,self.ispeed,self.ospeed,self.cc = termios.tcgetattr(self.fd)
def close(self):
if self.fd:
os.close(self.fd)
self.fd = None
def inWaiting(self):
s = fcntl.ioctl(self.fd, TERMIOS.FIONREAD, TIOCM_zero_str)
return struct.unpack('I',s)[0]
def write(self, data):
if not self.fd: raise portNotOpenError
t = len(data)
d = data
while t>0:
n = os.write(self.fd, d)
d = d[n:]
t = t - n
def read(self, size=1):
if not self.fd: raise portNotOpenError
read = ''
imp = None
if size > 0:
while len(read) < size:
if self.timeout:
#print "\tread(): size",size, "have", len(read) #debug
ready,_,_ = select.select([self.fd],[],[], self.timeout)
if not ready:
break #timeout
for fd in ready:
if fd is self.fd:
inp = os.read(self.fd, size-len(read))
else:
inp = os.read(self.fd, size-len(read))
if not inp and self.timeout: break #early abort on timeout
read = read + inp
return read
def flushInput(self):
if not self.fd:
raise portNotOpenError
termios.tcflush(self.fd, TERMIOS.TCIFLUSH)
def flushOutput(self):
if not self.fd:
raise portNotOpenError
termios.tcflush(self.fd, TERMIOS.TCOFLUSH)
def sendBreak(self):
if not self.fd:
raise portNotOpenError
termios.tcsendbreak(self.fd, 0)
def drainOutput(self):
if not self.fd: raise portNotOpenError
termios.tcdrain(self.fd)
def nonblocking(self):
if not self.fd:
raise portNotOpenError
fcntl.fcntl(self.fd, FCNTL.F_SETFL, FCNTL.O_NONBLOCK)
def getDSR(self):
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_DSR
def getCD(self):
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_CD
def getRI(self):
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_RI
def getCTS(self):
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_CTS
def setDTR(self,on=1):
if not self.fd: raise portNotOpenError
if on:
fcntl.ioctl(self.fd, TIOCMBIS, TIOCM_DTR_str)
else:
fcntl.ioctl(self.fd, TIOCMBIC, TIOCM_DTR_str)
def setRTS(self,on=1):
if not self.fd: raise portNotOpenError
if on:
fcntl.ioctl(self.fd, TIOCMBIS, TIOCM_RTS_str)
else:
fcntl.ioctl(self.fd, TIOCMBIC, TIOCM_RTS_str)
if __name__ == '__main__':
s = Serial(0,
baudrate=19200, #baudrate
bytesize=EIGHTBITS, #number of databits
parity=PARITY_EVEN, #enable parity checking
stopbits=STOPBITS_ONE, #number of stopbits
timeout=3, #set a timeout value, None for waiting forever
xonxoff=0, #enable software flow control
rtscts=0, #enable RTS/CTS flow control
)
s.setRTS(1)
s.setDTR(1)
s.flushInput()
s.flushOutput()
s.write('hello')
print repr(s.read(5))
print s.inWaiting()
del s
|