Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Newer
Older
100755 215 lines (165 sloc) 7.459 kB
f5f47e8 @lukec and add the actual server
lukec authored
1 #!/usr/bin/env python
2
3 """
4 serialserver multiplexes the Arduino serial port to concurrent incoming TCP
5 socket connections. this allows multiple clients to use the connected arduino
6 at once.
7
8 socket server listens on port 9994 and keeps open all incoming TCP connections
9
10 messages sent to the open socket are passed to the serial port and responses
11 are written to the client socket
12
13 the serial port is also polled periodically and any other messages are broadcast
14 to all open connections
15
16 future improvements:
17 - convert print's to logging.*
18 - subclass SocketServer.TCPServer constructor to eliminate global SERIAL_DAEMON
19 - switch from Lock protected lists to python's synchronized Queue class
20 - better granularity of existing locks
21 """
22
23 from __future__ import with_statement
24
25 import serial
26 import SocketServer, threading, socket, re, os, logging
27 import traceback
28
29 SERVER_HOST_PORT = 'localhost', 9994
30
31 SERIAL_PORT, LOG_FILENAME = {
32 'nt': ('COM12', 'serialserver.log'),
33 }.get(os.name, ('/dev/ttyUSB0', '/var/log/vhs-serialserver.log'))
34
35 logging.basicConfig(filename=LOG_FILENAME, level=logging.DEBUG)
36
37 # socket read timeout in seconds
38 TIMEOUT = 0.01
39 # serial port timeout in seconds
40 SERIAL_TIMEOUT = 0.05
41
42 global SERIAL_DAEMON
43
44 class TCPHandler(SocketServer.BaseRequestHandler):
45 def setup(self):
46 global SERIAL_DAEMON
47 assert(SERIAL_DAEMON)
48
49 self.seriald = SERIAL_DAEMON
50
51 (host, port) = self.client_address
52 self.client_id = port
53 self.seriald.register_client(self.client_id)
54
55 self.request.settimeout(TIMEOUT)
56
57 def handle(self):
58 logging.info('Opened TCP client connection %s' % str(self.client_address))
59
60 while True:
61 try:
62 self.data = self.request.recv(1024)
63
64 if self.data:
65 message = self.data.strip()
66 if not message: continue
67
68 logging.debug('received message "%s" length %d' % (message, len(message)))
69 # send to serial port
70 self.seriald.outgoing((self.client_id, message))
71 else:
72 # stream closed
73 self.request.close()
74 logging.info('Closed TCP client connection #%d' % self.client_id)
75 break
76 except socket.timeout:
77 # check if there are any incoming messages for this client
78 messages = self.seriald.incoming(self.client_id, pop=True)
79 for (id, msg) in messages:
80 logging.debug('sending message %s to %d' % (msg.strip(), self.client_id))
81 self.request.send(msg)
82
83 def finish(self):
84 self.seriald.unregister_client(self.client_id)
85
86 class TCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
87 pass
88
89 class SerialDaemon(object):
90 def __init__(self, ser):
91 assert(ser)
92
93 self.ser = ser # serial port
94 self.incoming_serial = [] # message queue from the serial port
95 self.outgoing_serial = [] # message queue writing to serial port
96 self.clients = set() # track a global list of clients for broadcast messages
97
98 self.client_lock = threading.Lock() # clients mutex
99 self.message_lock = threading.Lock() # message queues mutex
100
101 def register_client(self, client_id):
102 with self.client_lock:
103 self.clients.add(client_id)
104
105 def unregister_client(self, client_id):
106 logging.debug('unregistering client %s' % client_id)
107 with self.client_lock:
108 self.clients.remove(client_id)
109
110 # remove any messages related to this client
111 self.incoming(client_id, pop=True)
112 with self.message_lock:
113 self.outgoing_serial = filter(lambda (id, msg): id != client_id,
114 self.outgoing_serial)
115
116 def update(self):
117 """Called from main loop"""
118
119 valid_message = lambda msg: msg and not msg.startswith('#')
120
121 def broadcast(message):
122 for client in self.clients:
123 self.incoming_serial.append((client, msg))
124
125 # read and broadcast any new global messages
126 msg = self.ser.readline()
127 if valid_message(msg):
128 with self.message_lock: broadcast(msg)
129
130 # send client requests and accumulate responses
131 with self.message_lock:
132 while len(self.outgoing_serial):
133 (id, msg) = self.outgoing_serial.pop(0)
134 logging.debug('writing "%s" to serial port' % msg)
135 self.ser.write(msg + '\n')
136
137 # wait till we get a response to our query
138 if valid_message(msg):
139 attempts = 100
140 command = msg.split()[0]
141
142 while attempts:
143 response = self.ser.readline()
144 if valid_message(response):
145 logging.debug('response "%s"' % response.strip())
146 logging.debug(' command "%s"' % command)
147 if response and command == response.split()[0]:
148 logging.debug('matched response %s' % response.strip())
149 self.incoming_serial.append((id, response))
150 break
151 else:
152 broadcast(msg)
153
154 attempts -= 1
155 else:
156 # send out a timeout message
157 logging.warning('timeout on message "%s"' % msg)
158 self.incoming_serial.append((id, '!timeout\r\n'))
159
160 def outgoing(self, (client_id, message)):
161 """Send a message to the serial port"""
162
163 with self.message_lock:
164 self.outgoing_serial.append((client_id, message))
165
166 def incoming(self, client_id, pop=True):
167 """
168 Get incoming messages from the serial port
169
170 client_id -- matches only incoming messages associated with the given
171 client id
172 pop -- remove matching messages from the incoming queue
173 """
174
175 messages = list()
176 popped_queue = list()
177
178 with self.message_lock:
179 for msg in self.incoming_serial:
180 if msg[0] == client_id:
181 messages.append(msg)
182 else:
183 popped_queue.append(msg)
184 if pop:
185 self.incoming_serial = popped_queue
186
187 return messages
188
189 if __name__ == '__main__':
190 ser = serial.Serial(SERIAL_PORT, 9600, timeout=SERIAL_TIMEOUT)
191
192 try:
193 global SERIAL_DAEMON
194 SERIAL_DAEMON = SerialDaemon(ser)
195
196 # start up server thread
197 server = TCPServer(SERVER_HOST_PORT, TCPHandler)
198 server_thread = threading.Thread(target=server.serve_forever)
199 server_thread.setDaemon(True)
200 server_thread.start()
201
202 logging.info('Serial server initialized')
203 logging.info(' -- listening on serial port %s and %s\n' % (
204 SERIAL_PORT, '%s:%d' % (SERVER_HOST_PORT)))
205 while True:
206 # poll serial port
207 SERIAL_DAEMON.update()
208
209 except Exception, e:
210 logging.critical('serialserver exception:', e)
211 logging.critical(traceback.format_exc())
212 finally:
213 ser.close()
214
Something went wrong with that request. Please try again.