Added tenacity to worker and new recquirements. Improved reliability with Exception handlers

modules/server/master_server.py

@@ -169,7 +169,8 @@ class WorkerWebSocketHandler(tornado.websocket.WebSocketHandler):
             logging.debug("Message received from worker:" + message)
         else:  
             if 'transcription' in json_msg.keys(): #Receive the file path to process
-                logging.debug("Response send by worker : %s" % json.dumps({'transcript':json_msg['transcription']}))
+                print(json_msg['transcription'])
+                logging.debug("Response send by worker : %s" % json.dumps({'transcript':json_msg['transcription'].encode('utf-8')}))
                 self.client_handler.receive_response(json.dumps({'transcript':json_msg['transcription']}))
                 self.client_handler = None
                 self.application.available_workers.add(self)
@@ -198,7 +199,9 @@ def main():
     app = Application()
     app.listen(int(SERVER_PORT))
     logging.info('Starting up server listening on port %s' % SERVER_PORT)
-    tornado.ioloop.IOLoop.instance().start()
-    
+    try:
+        tornado.ioloop.IOLoop.instance().start()
+    except KeyboardInterrupt:
+        logging.info("Server close by user.")
 if __name__ == '__main__':
     main()
\ No newline at end of file

modules/server/requirements.txt

 ws4py
-configparser
\ No newline at end of file
+configparser
+tornado
\ No newline at end of file

modules/worker_offline/Dockerfile

@@ -42,7 +42,7 @@ WORKDIR $BASE_DIR
 
 # Install tornado
 COPY requirements.txt .
-RUN pip install -r requirements.txt
+RUN pip2 install -r requirements.txt
 
 # Deploy our offline server
 COPY . .

modules/worker_offline/requirements.txt

 ws4py
-configparser
\ No newline at end of file
+configparser
+tenacity 
\ No newline at end of file

modules/worker_offline/signal_to_noise_scipy.py

+def signaltonoise(data, axis=0):
+    """Calculates the signal-to-noise ratio, as the ratio of the mean over
+    standard deviation along the given axis.
+    Parameters
+    ----------
+    data : sequence
+        Input data
+    axis : {0, int}, optional
+        Axis along which to compute. If None, the computation is performed
+        on a flat version of the array.
+    """
+    data = ma.array(data, copy=False)
+    m = data.mean(axis)
+    sd = data.std(axis, ddof=0)
+return m/sd

modules/worker_offline/worker.cfg

 [server_params]
-server_ip : 10.31.2.168
+server_ip : 10.31.4.100
 server_port : 8888
 server_target : /worker/ws/speech
 

modules/worker_offline/worker_offline.py

@@ -13,6 +13,7 @@ import json
 import subprocess
 import configparser
 import re
+import tenacity
 
 from ws4py.client.threadedclient import WebSocketClient
 
@@ -26,6 +27,10 @@ DECODER_COMMAND = worker_settings.get('worker_params', 'decoder_command')
 TEMP_FILE_PATH = worker_settings.get('worker_params', 'temp_file_location')
 PREPROCESSING = True if worker_settings.get('worker_params', 'preprocessing') == 'true' else False
 
+class NoRouteException(Exception):
+    pass
+class ConnexionRefusedException(Exception):
+    pass
 
 class WorkerWebSocket(WebSocketClient):
     def __init__(self, uri):
@@ -89,7 +94,30 @@ class WorkerWebSocket(WebSocketClient):
     def finish_request(self):
         pass
     
-    
+@tenacity.retry(
+        wait=tenacity.wait.wait_fixed(2),
+        stop=tenacity.stop.stop_after_delay(45),
+        retry=tenacity.retry_if_exception(ConnexionRefusedException)
+    )
+def connect_to_server(ws):
+    try:
+        logging.info("Attempting to connect to server at %s:%s" % (SERVER_IP, SERVER_PORT))
+        ws.connect()
+        logging.info("Worker succefully connected to server at %s:%s" % (SERVER_IP, SERVER_PORT))
+        ws.run_forever()
+    except KeyboardInterrupt:
+        logging.info("Worker interrupted by user")
+        ws.close()
+    except Exception, e:
+        if "[Errno 113]" in str(e):
+            logging.info("Failed to connect")
+            raise NoRouteException
+        if "[Errno 111]" in str(e):
+            logging.info("Failed to connect")
+            raise ConnexionRefusedException
+        logging.debug(e)
+    logging.info("Worker stopped")
+
 def main():
     parser = argparse.ArgumentParser(description='Worker for linstt-dispatch')
     parser.add_argument('-u', '--uri', default="ws://"+SERVER_IP+":"+SERVER_PORT+SERVER_TARGET, dest="uri", help="Server<-->worker websocket URI")
@@ -103,12 +131,8 @@ def main():
     logging.info('Starting up worker')
     ws = WorkerWebSocket(args.uri)
     try:
-        ws.connect()
-        logging.info("Worker succefully connected to server at %s:%s" % (SERVER_IP, SERVER_PORT))
-        ws.run_forever()
-    except KeyboardInterrupt:
-        ws.close()
-    
-    
+        connect_to_server(ws)
+    except Exception:
+        logging.error("Worker did not manage to connect to server at %s:%s" % (SERVER_IP, SERVER_PORT))
 if __name__ == '__main__':
     main()
\ No newline at end of file