Added Ethernet client server example code.

krejov100 · krejov100 · commit 349894e7fd6b · 2019-07-23T10:16:02.000-07:00
diff --git a/wrappers/python/examples/ethernet_client_server/AlwaysRunningServer.bash b/wrappers/python/examples/ethernet_client_server/AlwaysRunningServer.bash
@@ -0,0 +1,6 @@
+#!/bin/bash
+# this statement checks if there is an instance of the EtherSenseServer running
+if [[ ! `ps -eaf | grep "python EtherSenseServer.py" | grep -v grep` ]]; then
+# if not, EtherSenseServer is started with the PYTHONPATH set due to cron not passing Env 
+    PYTHONPATH=$HOME/.local/lib/python2.7/site-packages python EtherSenseServer.py
+fi
diff --git a/wrappers/python/examples/ethernet_client_server/EtherSenseClient.py b/wrappers/python/examples/ethernet_client_server/EtherSenseClient.py
@@ -0,0 +1,117 @@
+#!/usr/bin/python
+import pyrealsense2 as rs
+import sys, getopt
+import asyncore
+import numpy as np
+import pickle
+import socket
+import struct
+import cv2
+
+
+print('Number of arguments:', len(sys.argv), 'arguments.')
+print('Argument List:', str(sys.argv))
+mc_ip_address = '224.0.0.1'
+local_ip_address = '192.168.0.1'
+port = 1024
+chunk_size = 4096
+
+def main(argv):
+    multi_cast_message(mc_ip_address, port, 'EtherSensePing')
+        
+
+#UDP client for each camera server 
+class ImageClient(asyncore.dispatcher):
+    def __init__(self, server, source):   
+        asyncore.dispatcher.__init__(self, server)
+        self.address = server.getsockname()[0]
+        self.port = source[1]
+        self.buffer = bytearray()
+        self.windowName = self.port
+        # open cv window which is unique to the port 
+        cv2.namedWindow("window"+str(self.windowName))
+        self.remainingBytes = 0
+        self.frame_id = 0
+       
+    def handle_read(self):
+        if self.remainingBytes == 0:
+            # get the expected frame size
+            self.frame_length = struct.unpack('<I', self.recv(4))[0]
+            # get the timestamp of the current frame
+            self.timestamp = struct.unpack('<d', self.recv(8))
+            self.remainingBytes = self.frame_length
+        
+        # request the frame data until the frame is completely in buffer
+        data = self.recv(self.remainingBytes)
+        self.buffer += data
+        self.remainingBytes -= len(data)
+        # once the frame is fully recived, process/display it
+        if len(self.buffer) == self.frame_length:
+            self.handle_frame()
+
+    def handle_frame(self):
+        # convert the frame from string to numerical data
+        imdata = pickle.loads(self.buffer)
+        bigDepth = cv2.resize(imdata, (0,0), fx=2, fy=2, interpolation=cv2.INTER_NEAREST) 
+        cv2.putText(bigDepth, str(self.timestamp), (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (65536), 2, cv2.LINE_AA)
+        cv2.imshow("window"+str(self.windowName), bigDepth)
+        cv2.waitKey(1)
+        self.buffer = bytearray()
+        self.frame_id += 1
+    def readable(self):
+        return True
+
+    
+class EtherSenseClient(asyncore.dispatcher):
+    def __init__(self):
+        asyncore.dispatcher.__init__(self)
+        self.server_address = ('', 1024)
+        # create a socket for TCP connection between the client and server
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.settimeout(5)
+        
+        self.bind(self.server_address) 	
+        self.listen(10)
+
+    def writable(self): 
+        return False # don't want write notifies
+
+    def readable(self):
+        return True
+        
+    def handle_connect(self):
+        print("connection recvied")
+
+    def handle_accept(self):
+        pair = self.accept()
+        #print(self.recv(10))
+        if pair is not None:
+            sock, addr = pair
+            print ('Incoming connection from %s' % repr(addr))
+            # when a connection is attempted, delegate image receival to the ImageClient 
+            handler = ImageClient(sock, addr)
+
+def multi_cast_message(ip_address, port, message):
+    # send the multicast message
+    multicast_group = (ip_address, port)
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    connections = {}
+    try:
+        # Send data to the multicast group
+        print('sending "%s"' % message + str(multicast_group))
+        sent = sock.sendto(message.encode(), multicast_group)
+   
+        # defer waiting for a response using Asyncore
+        client = EtherSenseClient()
+        asyncore.loop()
+
+        # Look for responses from all recipients
+        
+    except socket.timeout:
+        print('timed out, no more responses')
+    finally:
+        print(sys.stderr, 'closing socket')
+        sock.close()
+
+if __name__ == '__main__':
+    main(sys.argv[1:])
diff --git a/wrappers/python/examples/ethernet_client_server/EtherSenseServer.py b/wrappers/python/examples/ethernet_client_server/EtherSenseServer.py
@@ -0,0 +1,140 @@
+#!/usr/bin/python
+import pyrealsense2 as rs
+import sys, getopt
+import asyncore
+import numpy as np
+import pickle
+import socket
+import struct
+import cv2
+
+
+print('Number of arguments:', len(sys.argv), 'arguments.')
+print('Argument List:', str(sys.argv))
+mc_ip_address = '224.0.0.1'
+port = 1024
+chunk_size = 4096
+#rs.log_to_console(rs.log_severity.debug)
+
+def getDepthAndTimestamp(pipeline, depth_filter):
+    frames = pipeline.wait_for_frames()
+    # take owner ship of the frame for further processing
+    frames.keep()
+    depth = frames.get_depth_frame()
+    if depth:
+	depth2 = depth_filter.process(depth)
+	# take owner ship of the frame for further processing
+	depth2.keep()
+	# represent the frame as a numpy array
+        depthData = depth2.as_frame().get_data()        
+	depthMat = np.asanyarray(depthData)
+	ts = frames.get_timestamp()
+        return depthMat, ts
+    else:
+        return None, None
+def openPipeline():
+    cfg = rs.config()
+    cfg.enable_stream(rs.stream.depth, 640, 480, rs.format.z16, 30)
+    pipeline = rs.pipeline()
+    pipeline_profile = pipeline.start(cfg)
+    sensor = pipeline_profile.get_device().first_depth_sensor()
+    return pipeline
+
+class DevNullHandler(asyncore.dispatcher_with_send):
+
+    def handle_read(self):
+        print(self.recv(1024))
+
+    def handle_close(self):
+        self.close()
+           
+		
+class EtherSenseServer(asyncore.dispatcher):
+    def __init__(self, address):
+        asyncore.dispatcher.__init__(self)
+        print("Launching Realsense Camera Server")
+	try:
+            self.pipeline = openPipeline()
+        except:
+            print("Unexpected error: ", sys.exc_info()[1])
+            sys.exit(1)
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        print('sending acknowledgement to', address)
+        
+	# reduce the resolution of the depth image using post processing
+        self.decimate_filter = rs.decimation_filter()
+        self.decimate_filter.set_option(rs.option.filter_magnitude, 2)
+        self.frame_data = ''
+        self.connect((address[0], 1024))
+        self.packet_id = 0        
+
+    def handle_connect(self):
+        print("connection received")
+
+    def writable(self):
+        return True
+
+    def update_frame(self):
+	depth, timestamp = getDepthAndTimestamp(self.pipeline, self.decimate_filter)
+        if depth is not None:
+	    # convert the depth image to a string for broadcast
+            data = pickle.dumps(depth)
+	    # capture the lenght of the data portion of the message	
+            length = struct.pack('<I', len(data))
+	    # include the current timestamp for the frame
+            ts = struct.pack('<d', timestamp)
+	    # for the message for transmission
+            self.frame_data = ''.join([length, ts, data])
+
+    def handle_write(self):
+	# first time the handle_write is called
+        if not hasattr(self, 'frame_data'):
+            self.update_frame()
+	# the frame has been sent in it entirety so get the latest frame
+        if len(self.frame_data) == 0:
+	    self.update_frame()
+        else:
+	    # send the remainder of the frame_data until there is no data remaining for transmition
+            remaining_size = self.send(self.frame_data)
+            self.frame_data = self.frame_data[remaining_size:]
+	
+
+    def handle_close(self):
+        self.close()
+            
+
+class MulticastServer(asyncore.dispatcher):
+    def __init__(self, host = mc_ip_address, port=1024):
+        asyncore.dispatcher.__init__(self)
+        server_address = ('', port)
+        self.create_socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.bind(server_address) 	
+
+    def handle_read(self):
+        data, addr = self.socket.recvfrom(42)
+        print('Recived Multicast message %s bytes from %s' % (data, addr))
+	# Once the server recives the multicast signal, open the frame server
+        EtherSenseServer(addr)
+        print(sys.stderr, data)
+
+    def writable(self): 
+        return False # don't want write notifies
+
+    def handle_close(self):
+        self.close()
+
+    def handle_accept(self):
+        channel, addr = self.accept()
+        print('received %s bytes from %s' % (data, addr))
+
+
+def main(argv):
+    # initalise the multicast receiver 
+    server = MulticastServer()
+    # hand over excicution flow to asyncore
+    asyncore.loop()
+   
+if __name__ == '__main__':
+    main(sys.argv[1:])
+
diff --git a/wrappers/python/examples/ethernet_client_server/MultiCameraEthernet.png b/wrappers/python/examples/ethernet_client_server/MultiCameraEthernet.png
diff --git a/wrappers/python/examples/ethernet_client_server/README.md b/wrappers/python/examples/ethernet_client_server/README.md
@@ -0,0 +1,80 @@
+# EtherSense
+Ethernet client and server for RealSense using python's Asyncore.
+
+## Prerequisites
+Installation and Setup of Server:
+These steps assume a fresh install of Ubuntu 18.04 on an UpBoard but has also been tested on an Intel NUC.
+
+```
+sudo apt-get update; sudo apt-get upgrade; 
+
+sudo apt-get install python
+
+sudo apt-get install python-pip  
+
+sudo apt-get install git
+```
+
+Clone the repo then run:
+
+```
+sudo python setup.py
+```
+
+This will first install the pip dependencies, followed by the creation of cronjobs in the /etc/crontab file that maintains an instance of the Server running whenever the device is powered. 
+
+## Overview
+Mulicast broadcast is used to establish connections to servers that are present on the network. 
+Once a server receives a request for connection from a client, Asyncore is used to establish a TCP connection for each server. 
+Frames are collected from the camera using librealsense pipeline. It is then resized and send in smaller chucks as to conform with TCP.
+
+### UpBoard PoE 
+Below shows use of a PoE switch and PoE breakout devices(avalible from online retailers) powering each dedicated UpBoard: 
+This configuration should allow for a number of RealSense cameras to be connected over distances greater then 30m 
+![Example Image](https://github.com/krejov100/EtherSense/blob/master/UpBoardSwitch.JPG)
+The 5 RealSense cameras are connected to each UpBoard using the provided USB3 cables.
+
+### Client Window
+Below shows the result of having connected to five cameras over the local network: 
+![Example Image](https://github.com/krejov100/EtherSense/blob/master/MultiCameraEthernet.png)
+The window titles indicate the port which the frames are being received over.
+
+## Error Logging
+Errors are piped to a log file stored in /tmp/error.log as part of the command that is setup in /etc/crontab
+
+## NOTES
+
+### Power Considerations
+The UpBoards require a 5v 4Amp power supply. When using PoE breakout adaptors I have found some stability issues, for example the device kernel can crash when the HDMI port is connected. As such I recommend running the UpBoard as a headless server when using PoE. 
+
+### Network bandwidth
+It is currently very easy to saturate the bandwidth of the Ethernet connection I have tested 5 servers connected to the same client without issue beyond limited framerate:
+
+cfg.enable_stream(rs.stream.depth, 640, 480, rs.format.z16, 30)
+
+self.decimate_filter.set_option(rs.option.filter_magnitude, 2)
+
+There are a number of strategies that can be used to increase this bandwidth but are left to the user for brevity and the specific tradeoff for your application, these include:
+
+Transmitting frames using UDP and allowing for frame drop, this requires implementation of packet ordering.
+
+Reducing the depth channel to 8bit.
+
+Reducing the resolution further.
+
+The addition of compression, either frame wise or better still temporal.
+
+Local recording of the depth data into a buffer, with asynchronous frame transfer.
+ 
+## TroubleShooting Tips
+
+I first of all suggest installing and configuring openssh-server on each of the UpBoards allowing remote connection from the client machine.
+
+Check that the UpBoards are avalible on the local network using "nmap -sP 192.168.2.*"
+
+Check that the server is running on the UpBoard using "ps -eaf | grep "python EtherSenseServer.py"
+
+Finally check the log file at /tmp/error.log
+
+There might still be some conditions where the Server is running but not in a state to transmit, help in narrowing these cases would be much appreciated. 
+
diff --git a/wrappers/python/examples/ethernet_client_server/UpBoardSwitch.JPG b/wrappers/python/examples/ethernet_client_server/UpBoardSwitch.JPG
diff --git a/wrappers/python/examples/ethernet_client_server/setup.py b/wrappers/python/examples/ethernet_client_server/setup.py
@@ -0,0 +1,34 @@
+# install the dependancies from pip
+try:
+    from pip import main as pipmain
+except:
+    from pip._internal import main as pipmain
+pipmain(['install', 'numpy'])
+pipmain(['install', 'python-crontab'])
+pipmain(['install', 'opencv-python'])
+pipmain(['install', 'pyrealsense2'])
+pipmain(['install', 'cron_descriptor'])
+
+# using python-crontab, setup a job that is ran on the minute to check if the server is running. 
+# https://pypi.org/project/python-crontab/
+from crontab import CronTab
+#using the system crontab file as sudo is required for librealsense aquisition
+system_cron = CronTab(tabfile='/etc/crontab', user=False)
+#requires the shell enviroment as ./AlwaysRunningServer.bash 
+system_cron.env['SHELL'] = '/bin/bash'
+job = system_cron.new(command="cd /home/$(ls /home/)/EtherSense; ./AlwaysRunningServer.bash >> /tmp/error.log 2>&1", user='root')
+job.every_reboot()
+i = 0
+#this while loop means that the are entrys in the cron file for each 5 sec interval
+while i < 60:
+    #the cron job is cd to the EtherSense dir then run AlwaysRunningServer.bash, logging to /temp/error.log
+    #have to use this $(ls /home/) to find home dir, assuming no other user spaces.
+    job = system_cron.new(command="sleep %i; cd /home/$(ls /home/)/EtherSense; ./AlwaysRunningServer.bash >> /tmp/error.log 2>&1"%i, user='root')
+
+    # this line sets the frequance of server checking, stars for all time slices means every minute
+    job.setall('* * * * *')
+    system_cron.write()
+    i += 5
+print('cron job set to run ' + job.description())
+
+
diff --git a/wrappers/python/examples/readme.md b/wrappers/python/examples/readme.md
@@ -15,6 +15,7 @@ These Examples demonstrate how to use the python wrapper of the SDK.
 8. [T265 Basic](./t265_example.py) - Demonstrates how to retrieve pose data from a T265
 9. [T265 Coordinates](./t265_rpy.py) - This example shows how to change coordinate systems of a T265 pose
 10. [T265 Stereo](./t265_stereo.py) - This example shows how to use T265 intrinsics and extrinsics in OpenCV to asynchronously compute depth maps from T265 fisheye images on the host.
+11. [Realsense over Ethernet](./ethernet_client_server/README.md) - This example shows how to stream depth data from RealSense depth cameras over ethernet.
 
 ## Pointcloud Visualization
 
