Consumer Code with Multiple Camera

abendigeri · abendigeri · commit 9b8d233145eb · 2019-07-20T18:06:31.000+05:30
diff --git a/deployment/raspbi/consumer.py b/deployment/raspbi/consumer.py
@@ -0,0 +1,67 @@
+import datetime
+from flask import Flask, Response, render_template
+from kafka import KafkaConsumer
+
+# Fire up the Kafka Consumer
+topic = "testpico"
+brokers = ["35.189.130.4:9092"]
+
+consumer = KafkaConsumer(
+    topic, 
+    bootstrap_servers=brokers,
+    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
+
+
+# Set the consumer in a Flask App
+app = Flask(__name__)
+
+@app.route('/')
+def index():
+    return render_template('index.html')
+
+@app.route('/camera_1', methods=['GET'])
+def camera_1():
+    """
+    This is the heart of our video display. Notice we set the mimetype to 
+    multipart/x-mixed-replace. This tells Flask to replace any old images with 
+    new values streaming through the pipeline.
+    """
+    return Response(
+        get_video_stream(id), 
+        mimetype='multipart/x-mixed-replace; boundary=frame')
+
+@app.route('/camera_2', methods=['GET'])
+def camera_2():
+    """
+    This is the heart of our video display. Notice we set the mimetype to 
+    multipart/x-mixed-replace. This tells Flask to replace any old images with 
+    new values streaming through the pipeline.
+    """
+    return Response(
+        get_video_stream(id), 
+        mimetype='multipart/x-mixed-replace; boundary=frame')
+
+
+@app.route('/camera_3', methods=['GET'])
+def camera_3():
+    """
+    This is the heart of our video display. Notice we set the mimetype to 
+    multipart/x-mixed-replace. This tells Flask to replace any old images with 
+    new values streaming through the pipeline.
+    """
+    return Response(
+        get_video_stream(id), 
+        mimetype='multipart/x-mixed-replace; boundary=frame')
+
+def get_video_stream(id):
+    """
+    Here is where we recieve streamed images from the Kafka Server and convert 
+    them to a Flask-readable format.
+    """
+    for msg in consumer:
+        if str(msg.value['camera_id']) == str(id):
+            yield (b'--frame\r\n'
+                   b'Content-Type: image/jpg\r\n\r\n' + base64.b64decode(msg.value['image_bytes']) + b'\r\n\r\n')
+
+if __name__ == "__main__":
+    app.run(host='0.0.0.0', debug=True)
diff --git a/deployment/raspbi/templates/index.html b/deployment/raspbi/templates/index.html
@@ -0,0 +1,15 @@
+<html>
+  <head>
+    <title>Video Streaming Demonstration</title>
+  </head>
+  <body>
+    <h1>Video Streaming Demonstration</h1>
+    <img src="{{ url_for('camera_1') }}">
+    
+    <img src="{{ url_for('camera_2') }}">
+      
+    <img src="{{ url_for('camera_3') }}">
+      
+      
+  </body>
+</html>
diff --git a/kafka/consumer.py b/kafka/consumer.py
@@ -4,10 +4,12 @@
 
 # Fire up the Kafka Consumer
 topic = "testpico"
+brokers = ["35.189.130.4:9092"]
 
 consumer = KafkaConsumer(
     topic, 
-    bootstrap_servers=['localhost:9093'])
+    bootstrap_servers=brokers,
+    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
 
 
 # Set the consumer in a Flask App
@@ -17,25 +19,49 @@
 def index():
     return render_template('index.html')
 
-@app.route('/video_feed', methods=['GET'])
-def video_feed():
+@app.route('/camera_1', methods=['GET'])
+def camera_1():
     """
     This is the heart of our video display. Notice we set the mimetype to 
     multipart/x-mixed-replace. This tells Flask to replace any old images with 
     new values streaming through the pipeline.
     """
     return Response(
-        get_video_stream(), 
+        get_video_stream(id), 
         mimetype='multipart/x-mixed-replace; boundary=frame')
 
-def get_video_stream():
+@app.route('/camera_2', methods=['GET'])
+def camera_2():
+    """
+    This is the heart of our video display. Notice we set the mimetype to 
+    multipart/x-mixed-replace. This tells Flask to replace any old images with 
+    new values streaming through the pipeline.
+    """
+    return Response(
+        get_video_stream(id), 
+        mimetype='multipart/x-mixed-replace; boundary=frame')
+
+
+@app.route('/camera_3', methods=['GET'])
+def camera_3():
+    """
+    This is the heart of our video display. Notice we set the mimetype to 
+    multipart/x-mixed-replace. This tells Flask to replace any old images with 
+    new values streaming through the pipeline.
+    """
+    return Response(
+        get_video_stream(id), 
+        mimetype='multipart/x-mixed-replace; boundary=frame')
+
+def get_video_stream(id):
     """
     Here is where we recieve streamed images from the Kafka Server and convert 
     them to a Flask-readable format.
     """
     for msg in consumer:
-        yield (b'--frame\r\n'
-               b'Content-Type: image/jpg\r\n\r\n' + msg.value + b'\r\n\r\n')
+        if str(msg.value['camera_id']) == str(id):
+            yield (b'--frame\r\n'
+                   b'Content-Type: image/jpg\r\n\r\n' + base64.b64decode(msg.value['image_bytes']) + b'\r\n\r\n')
 
 if __name__ == "__main__":
     app.run(host='0.0.0.0', debug=True)
diff --git a/kafka/templates/index.html b/kafka/templates/index.html
@@ -4,6 +4,12 @@
   </head>
   <body>
     <h1>Video Streaming Demonstration</h1>
-    <img src="{{ url_for('video_feed') }}">
+    <img src="{{ url_for('camera_1') }}">
+    
+    <img src="{{ url_for('camera_2') }}">
+      
+    <img src="{{ url_for('camera_3') }}">
+      
+      
   </body>
 </html>
