- added daemon option

- added sys.exit() to examples

Author: ac133451

DataSocket/TCPDataSocket.py

@@ -22,7 +22,14 @@ def _get_socket():
 
 
 class TCPSendSocket(object):
- def __init__(self, tcp_port, tcp_ip='localhost', send_type=NUMPY, verbose=True, as_server=True, include_time=False):
+ def __init__(self,
+ tcp_port,
+ tcp_ip='localhost',
+ send_type=NUMPY,
+ verbose=True,
+ as_server=True,
+ include_time=False,
+ as_daemon=True):
  """
  A TCP socket class to send data to a specific port and address.
  :param tcp_port: TCP port to use.
@@ -36,6 +43,7 @@ def __init__(self, tcp_port, tcp_ip='localhost', send_type=NUMPY, verbose=True,
  :param as_server: Whether to run this socket as a server (default: True) or client. When run as a server, the
  socket supports multiple clients and sends each message to every connected client.
  :param include_time: Appends time.time() value when sending the data message.
+ :param as_daemon: runs the underlying threads as daemon.
  """
  self.send_type = send_type
  self.data_to_send = b'0'
@@ -48,8 +56,8 @@ def __init__(self, tcp_port, tcp_ip='localhost', send_type=NUMPY, verbose=True,
  self.as_server = as_server
  self.include_time = include_time
  self.connected_clients = []
- self._gather_connections_thread = Thread(target=self._gather_connections)
- self.sending_thread = Thread(target=self._run)
+ self._gather_connections_thread = Thread(target=self._gather_connections, daemon=as_daemon)
+ self.sending_thread = Thread(target=self._run, daemon=as_daemon)
 
  def send_data(self, data):
  """
@@ -242,7 +250,8 @@ def __init__(self,
  verbose=True,
  as_server=False,
  receive_as_raw=False,
- receive_buffer_size=4095):
+ receive_buffer_size=4095,
+ as_daemon=True):
  """
  Receiving TCP socket to be used with TCPSendSocket.
  :param tcp_port: TCP port to use.
@@ -257,6 +266,7 @@ def my_handler(received_data):
  whatever the SendSocket is configured to be.
  :param receive_as_raw: Whether or not the incoming data is just raw bytes or is a predefined format (JSON, NUMPY, HDF)
  :param receive_buffer_size: available buffer size in bytes when receiving messages
+ :param as_daemon: runs underlying threads as daemon.
  """
  self.receive_buffer_size = receive_buffer_size
  self.receive_as_raw = receive_as_raw
@@ -274,9 +284,9 @@ def pass_func(data):
  self._new_data = None
  self._new_data_lock = Lock()
  self.new_data_flag = Event()
- self.handler_thread = Thread(target=self._handler, daemon=True)
+ self.handler_thread = Thread(target=self._handler, daemon=as_daemon)
  self.socket = _get_socket()
- self.thread = Thread(target=self._run, daemon=True)
+ self.thread = Thread(target=self._run, daemon=as_daemon)
  self.port = int(tcp_port)
  self.ip = tcp_ip
  self.block_size = 0
@@ -289,8 +299,7 @@ def pass_func(data):
  def start(self, blocking=False):
  """
  Start the socket service.
- :param blocking: Will block the calling thread until a connection is established to at least one receiver.
- :return: Nothing
+ :param blocking: Will block the calling thread until a connection is established.
  """
  self.thread.start()
  if blocking:

examples/hdp_example.py

@@ -1,6 +1,7 @@
 from DataSocket import TCPSendSocket, TCPReceiveSocket, HDF
 import time
 from threading import Thread
+import sys
 
 
 number_of_messages = 5 # number of sample messages to send
@@ -50,3 +51,5 @@ def print_value(data):
 
  send_thread.join()
  rec_thread.join()
+
+ sys.exit()

examples/json_example.py

@@ -1,6 +1,7 @@
 from DataSocket import TCPSendSocket, TCPReceiveSocket, JSON
 import time
 from threading import Thread
+import sys
 
 
 number_of_messages = 5 # number of sample messages to send
@@ -53,3 +54,5 @@ def print_value(data):
  send_thread.join()
  rec_thread1.join()
  rec_thread2.join()
+
+ sys.exit()

examples/numpy_example.py

@@ -1,6 +1,8 @@
 from DataSocket import TCPSendSocket, TCPReceiveSocket, NUMPY
 import time
 from threading import Thread
+import sys
+import numpy as np
 
 
 number_of_messages = 5 # number of sample messages to send
@@ -13,7 +15,8 @@ def sending_function():
  send_socket.start(blocking=True)
 
  for i in range(number_of_messages):
- send_socket.send_data(i*10)
+ send_socket.send_data({'i': i*10,
+ 'data': np.random.random(4)})
  time.sleep(0.25)
 
  print("closing send socket.")
@@ -26,7 +29,7 @@ def receiving_function():
 
  # function to run when a new piece of data is received
  def print_value(data):
- print("value received: ", data['data'])
+ print('i=', data['i'], "data received: ", data['data'])
  num_messages_received[0] = 1 + num_messages_received[0]
 
  rec_socket = TCPReceiveSocket(tcp_port=port, handler_function=print_value)
@@ -50,3 +53,4 @@ def print_value(data):
 
  send_thread.join()
  rec_thread.join()
+ sys.exit()

examples/simulink_example.py

@@ -9,10 +9,13 @@
 receive_port = 4343
 ip = '0.0.0.0'
 
+start_time = time.time()
 
 # define function to print the echo back from matlab
 def print_data(data):
- print('length of returned array:', np.frombuffer(data, dtype='float32').shape[0])
+ global start_time
+ now = time.time()
+ print('length of returned array:', np.frombuffer(data, dtype='float32').shape[0], 'Time for full trip:', now - start_time)
 
 
 # create a send and receive socket
@@ -27,11 +30,13 @@ def print_data(data):
 
 
 def send_sig():
+ global start_time
  while not stop_flag.is_set():
  data = np.random.random((100, 100)) # create 100x100 array of random numbers
  data_as_bytes = data.astype('float32').flatten().tostring() # flatten it before sending
+ start_time = time.time()
  send_socket.send_data(data_as_bytes)
- time.sleep(1)
+ time.sleep(0.5)
 
 
 thread = threading.Thread(target=send_sig)