-
Notifications
You must be signed in to change notification settings - Fork 0
/
tcp_serial_redirect.py
145 lines (132 loc) · 4.97 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
#!/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
# connect to serial port
ser = serial.Serial()
ser.port = "/dev/ttyUSB0"
ser.baudrate = 9600
ser.parity = 'N'
ser.rtscts = False
ser.xonxoff = False
ser.timeout = 1
ser.open()
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#srv.bind( ('', 50000) )
srv.bind( ('', 50000) )
srv.listen(1)
while True:
try:
sys.stderr.write("Waiting for connection on %s...\n" % 50000)
connection, addr = srv.accept()
sys.stderr.write('Connected by %s\n' % (addr,))
# enter network <-> serial loop
r = Redirector(
ser,
connection,
None,
None,
True,
)
r.shortcut()
if True: 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')