forked from saltastro/timDIMM
-
Notifications
You must be signed in to change notification settings - Fork 0
/
tcp_serial_redirect.py
executable file
·326 lines (285 loc) · 10.3 KB
/
tcp_serial_redirect.py
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
#!/usr/bin/env python
# (C) 2002-2009 Chris Liechti <[email protected]>
# redirect data from a TCP/IP connection to a serial port and vice versa
# requires Python 2.2 'cause socket.sendall is used
import sys
import os
import time
import threading
import socket
import codecs
import serial
try:
True
except NameError:
True = 1
False = 0
class Redirector:
def __init__(self, serial_instance, socket, ser_newline=None, net_newline=None, spy=False):
self.serial = serial_instance
self.socket = socket
self.ser_newline = ser_newline
self.net_newline = net_newline
self.spy = spy
self._write_lock = threading.Lock()
def shortcut(self):
"""connect the serial port to the TCP port by copying everything
from one side to the other"""
self.alive = True
self.thread_read = threading.Thread(target=self.reader)
self.thread_read.setDaemon(True)
self.thread_read.setName('serial->socket')
self.thread_read.start()
self.writer()
def reader(self):
"""loop forever and copy serial->socket"""
while self.alive:
try:
data = self.serial.read(1) # read one, blocking
n = self.serial.inWaiting() # look if there is more
if n:
data = data + self.serial.read(n) # and get as much as possible
if data:
# the spy shows what's on the serial port, so log it before converting newlines
if self.spy:
sys.stdout.write(codecs.escape_encode(data)[0])
sys.stdout.flush()
if self.ser_newline and self.net_newline:
# do the newline conversion
# XXX fails for CR+LF in input when it is cut in half at the begin or end of the string
data = net_newline.join(data.split(ser_newline))
# escape outgoing data when needed (Telnet IAC (0xff) character)
self._write_lock.acquire()
try:
self.socket.sendall(data) # send it over TCP
finally:
self._write_lock.release()
except socket.error, msg:
sys.stderr.write('ERROR: %s\n' % msg)
# probably got disconnected
break
self.alive = False
def write(self, data):
"""thread safe socket write with no data escaping. used to send telnet stuff"""
self._write_lock.acquire()
try:
self.socket.sendall(data)
finally:
self._write_lock.release()
def writer(self):
"""loop forever and copy socket->serial"""
while self.alive:
try:
data = self.socket.recv(1024)
if not data:
break
if self.ser_newline and self.net_newline:
# do the newline conversion
# XXX fails for CR+LF in input when it is cut in half at the begin or end of the string
data = ser_newline.join(data.split(net_newline))
self.serial.write(data) # get a bunch of bytes and send them
# the spy shows what's on the serial port, so log it after converting newlines
if self.spy:
sys.stdout.write(codecs.escape_encode(data)[0])
sys.stdout.flush()
except socket.error, msg:
sys.stderr.write('ERROR: %s\n' % msg)
# probably got disconnected
break
self.alive = False
self.thread_read.join()
def stop(self):
"""Stop copying"""
if self.alive:
self.alive = False
self.thread_read.join()
if __name__ == '__main__':
import optparse
parser = optparse.OptionParser(
usage = "%prog [options] [port [baudrate]]",
description = "Simple Serial to Network (TCP/IP) redirector.",
epilog = """\
NOTE: no security measures are implemented. Anyone can remotely connect
to this service over the network.
Only one connection at once is supported. When the connection is terminated
it waits for the next connect.
""")
parser.add_option("-q", "--quiet",
dest = "quiet",
action = "store_true",
help = "suppress non error messages",
default = False
)
parser.add_option("--spy",
dest = "spy",
action = "store_true",
help = "peek at the communication and print all data to the console",
default = False
)
group = optparse.OptionGroup(parser,
"Serial Port",
"Serial port settings"
)
parser.add_option_group(group)
group.add_option("-p", "--port",
dest = "port",
help = "port, a number (default 0) or a device name",
default = None
)
group.add_option("-b", "--baud",
dest = "baudrate",
action = "store",
type = 'int',
help = "set baud rate, default: %default",
default = 9600
)
group.add_option("", "--parity",
dest = "parity",
action = "store",
help = "set parity, one of [N, E, O], default=%default",
default = 'N'
)
group.add_option("--rtscts",
dest = "rtscts",
action = "store_true",
help = "enable RTS/CTS flow control (default off)",
default = False
)
group.add_option("--xonxoff",
dest = "xonxoff",
action = "store_true",
help = "enable software flow control (default off)",
default = False
)
group.add_option("--rts",
dest = "rts_state",
action = "store",
type = 'int',
help = "set initial RTS line state (possible values: 0, 1)",
default = None
)
group.add_option("--dtr",
dest = "dtr_state",
action = "store",
type = 'int',
help = "set initial DTR line state (possible values: 0, 1)",
default = None
)
group = optparse.OptionGroup(parser,
"Network settings",
"Network configuration."
)
parser.add_option_group(group)
group.add_option("-P", "--localport",
dest = "local_port",
action = "store",
type = 'int',
help = "local TCP port",
default = 7777
)
group = optparse.OptionGroup(parser,
"Newline Settings",
"Convert newlines between network and serial port. Conversion is normally disabled and can be enabled by --convert."
)
parser.add_option_group(group)
group.add_option("-c", "--convert",
dest = "convert",
action = "store_true",
help = "enable newline conversion (default off)",
default = False
)
group.add_option("--net-nl",
dest = "net_newline",
action = "store",
help = "type of newlines that are expected on the network (default: %default)",
default = "LF"
)
group.add_option("--ser-nl",
dest = "ser_newline",
action = "store",
help = "type of newlines that are expected on the serial port (default: %default)",
default = "CR+LF"
)
(options, args) = parser.parse_args()
# get port and baud rate from command line arguments or the option switches
port = options.port
baudrate = options.baudrate
if args:
if options.port is not None:
parser.error("no arguments are allowed, options only when --port is given")
port = args.pop(0)
if args:
try:
baudrate = int(args[0])
except ValueError:
parser.error("baud rate must be a number, not %r" % args[0])
args.pop(0)
if args:
parser.error("too many arguments")
else:
if port is None: port = 0
# check newline modes for network connection
mode = options.net_newline.upper()
if mode == 'CR':
net_newline = '\r'
elif mode == 'LF':
net_newline = '\n'
elif mode == 'CR+LF' or mode == 'CRLF':
net_newline = '\r\n'
else:
parser.error("Invalid value for --net-nl. Valid are 'CR', 'LF' and 'CR+LF'/'CRLF'.")
# check newline modes for serial connection
mode = options.ser_newline.upper()
if mode == 'CR':
ser_newline = '\r'
elif mode == 'LF':
ser_newline = '\n'
elif mode == 'CR+LF' or mode == 'CRLF':
ser_newline = '\r\n'
else:
parser.error("Invalid value for --ser-nl. Valid are 'CR', 'LF' and 'CR+LF'/'CRLF'.")
# connect to serial port
ser = serial.Serial()
ser.port = port
ser.baudrate = baudrate
ser.parity = options.parity
ser.rtscts = options.rtscts
ser.xonxoff = options.xonxoff
ser.timeout = 1 # required so that the reader thread can exit
if not options.quiet:
sys.stderr.write("--- TCP/IP to Serial redirector --- type Ctrl-C / BREAK to quit\n")
sys.stderr.write("--- %s %s,%s,%s,%s ---\n" % (ser.portstr, ser.baudrate, 8, ser.parity, 1))
try:
ser.open()
except serial.SerialException, e:
sys.stderr.write("Could not open serial port %s: %s\n" % (ser.portstr, e))
sys.exit(1)
if options.rts_state is not None:
ser.setRTS(options.rts_state)
if options.dtr_state is not None:
ser.setDTR(options.dtr_state)
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.bind( ('', options.local_port) )
srv.listen(1)
while True:
try:
sys.stderr.write("Waiting for connection on %s...\n" % options.local_port)
connection, addr = srv.accept()
sys.stderr.write('Connected by %s\n' % (addr,))
# enter network <-> serial loop
r = Redirector(
ser,
connection,
options.convert and ser_newline or None,
options.convert and net_newline or None,
options.spy,
)
r.shortcut()
if options.spy: sys.stdout.write('\n')
sys.stderr.write('Disconnected\n')
connection.close()
except KeyboardInterrupt:
break
except socket.error, msg:
sys.stderr.write('ERROR: %s\n' % msg)
sys.stderr.write('\n--- exit ---\n')