Upload New File

video_server.py

+import time
+import io
+import sys
+import math
+from picamera import PiCamera, Color
+from flask import Flask, render_template, Response
+from queue import Queue
+
+
+class VideoStream:
+    
+    def __init__(self, queue):
+        self.queue = queue
+        self.app = Flask(__name__)
+        
+        @self.app.route('/')
+        def index():
+            #Redner webpage
+            return render_template('index.html')
+
+        @self.app.route('/video_feed')
+        def video_feed():
+            #Display camera on webpage
+            return Response(self.gen_frames(), mimetype='multipart/x-mixed-replace; boundary=frame')
+        
+        #Run app
+        self.app.run(host='0.0.0.0', port=5000, debug=True)
+        
+    def formatDistance(self, dataDict, camera):
+        #Get data from dictionary
+        distance = dataDict.get('distance')
+        warningThreshold = dataDict.get('warningThreshold')
+        alertThreshold = dataDict.get('alertThreshold')
+        midThreshold = dataDict.get('midThreshold')
+        
+        if distance <= warningThreshold:
+            colour = 'red'
+        elif distance <= midThreshold:
+            colour = 'orange'
+        else:
+            colour = 'green'
+            
+        camera.annotate_background = Color(colour)
+        
+        #Display distance and optionally set warning message based on distance
+        warning = "" if distance > warningThreshold else "\nWATCH OUT!!!"
+        message = f"Distance: {distance}cm{warning}"
+        camera.annotate_text = message
+        
+    def gen_frames(self):
+        with PiCamera() as camera:
+            #Initialise camera
+            camera.stop_preview()
+            camera.resolution = (1024, 768)
+            camera.rotation = 90
+            camera.framerate = 16
+            camera.annotate_text_size = 64
+            camera.annotate_foreground = Color('white')
+            stream = io.BytesIO()
+            time.sleep(2)
+            text = None
+            for frame in camera.capture_continuous(stream, 'jpeg', use_video_port=True):
+                try:
+                    #Check for termination of queue, if not then read from the queue and put into dictionary
+                    value = self.queue.get(block=False)
+                    if isinstance(value, str) and value == 'terminate':
+                        sys.exit()
+                    else:
+                        dataDict = value
+                    #Display distance on camera
+                    self.formatDistance(dataDict, camera)
+                except:
+                    print("Error reading dictionary")
+                    pass
+                
+                stream.seek(0)
+                #Return the frame of the camera
+                yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + stream.read() + b'\r\n'
+                stream.seek(0)
+                stream.truncate()
+            
+
+def runVideo(queue):
+    vs = VideoStream(queue)
+    
\ No newline at end of file