Update app.py

app.py

@@ -1,84 +1,363 @@
-# Import necessary modules
-from flask import Flask, render_template, Response, request, jsonify, redirect, url_for
-from aiortc import RTCPeerConnection, RTCSessionDescription
-import cv2
-import json
-import uuid
-import asyncio
-import logging
+from ultralytics import YOLO
 import time
+import numpy as np
+import mediapipe as mp
 
-# Create a Flask app instance
-app = Flask(__name__, static_url_path='/static')
+import uvicorn
+from socketio import ASGIApp
 
-# Set to keep track of RTCPeerConnection instances
-pcs = set()
+import cv2
+from flask import Flask, render_template, request, Response, session, redirect, url_for
+from flask import Flask, render_template
 
-# Function to generate video frames from the camera
-def generate_frames():
-    camera = cv2.VideoCapture(0)
-    while True:
-        start_time = time.time()
-        success, frame = camera.read()
-        if not success:
-            break
-        else:
-            ret, buffer = cv2.imencode('.jpg', frame)
-            frame = buffer.tobytes()
-            # Concatenate frame and yield for streaming
-            yield (b'--frame\r\n'
-                   b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n') 
-            elapsed_time = time.time() - start_time
-            logging.debug(f"Frame generation time: {elapsed_time} seconds")
-
-# Route to render the HTML template
-@app.route('/')
-def index():
-    return render_template('index.html')
-    # return redirect(url_for('video_feed')) #to render live stream directly
+from flask_socketio import SocketIO
+from flask_socketio import emit
+from flask_cors import CORS
+
+from flask_socketio import SocketIO
+import yt_dlp as youtube_dl
+import uvicorn
 
-# Asynchronous function to handle offer exchange
-async def offer_async():
-    params = await request.json
-    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])
+model_object_detection = YOLO("bisindov2.pt")
 
-    # Create an RTCPeerConnection instance
-    pc = RTCPeerConnection()
 
-    # Generate a unique ID for the RTCPeerConnection
-    pc_id = "PeerConnection(%s)" % uuid.uuid4()
-    pc_id = pc_id[:8]
 
-    # Create a data channel named "chat"
-    # pc.createDataChannel("chat")
 
-    # Create and set the local description
-    await pc.createOffer(offer)
-    await pc.setLocalDescription(offer)
+app = Flask(__name__)
 
-    # Prepare the response data with local SDP and type
-    response_data = {"sdp": pc.localDescription.sdp, "type": pc.localDescription.type}
+socketio = SocketIO(app, cors_allowed_origins="*")
 
-    return jsonify(response_data)
+CORS(app)
 
-# Wrapper function for running the asynchronous offer function
-def offer():
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
+app.secret_key = 'flask-sockets-builds'
+
+
+@socketio.on('image')
+def handle_image(image_data):
+    # Process the received image data
+    # Here, you can save the image, perform analysis, etc.
+    # For demonstration purposes, let's just print the length of the image data
+    print("Received image data length:", len(image_data))
+
+    # Send a response back to the client
+    # You can send any data you want back to the client
+    response_data = {'status': 'success'}
+    socketio.emit('response_back', response_data)
     
-    future = asyncio.run_coroutine_threadsafe(offer_async(), loop)
-    return future.result()
+######################################################
+classes_translation = {
+    "all": "الكل",
+    "A": "أ",
+    "B": "ب",
+    "C": "ج",
+    "D": "د",
+    "F": "ف",
+    "H": "هـ",
+    "I": "أنا",
+    "J": "جيم",
+    "L": "إل",
+    "M": "إم",
+    "O": "أو",
+    "R": "ر",
+    "T": "ت",
+    "U": "يو",
+    "V": "في",
+    "W": "دبليو",
+    "Z": "زد",
+    "additional": "إضافي",
+    "alcohol": "مدرسة",
+    "allergy": "حساسية",
+    "bacon": "لحم المقدد",
+    "bag": "حقيبة",
+    "barbecue": "شواء",
+    "bill": "فاتورة",
+    "biscuit": "بسكويت",
+    "bitter": "مر",
+    "bread": "خبز",
+    "burger": "برغر",
+    "bye": "وداعاً",
+    "cheese": "جبن",
+    "chicken": "دجاج",
+    "coke": "كوكاكولا",
+    "cold": "بارد",
+    "cost": "تكلفة",
+    "coupon": "كوبون",
+    "cup": "كوب",
+    "dessert": "حلوى",
+    "drink": "شراب",
+    "drive": "قيادة",
+    "eat": "تناول الطعام",
+    "eggs": "بيض",
+    "enjoy": "استمتع",
+    "fork": "شوكة",
+    "french fries": "بطاطس مقلية",
+    "fresh": "طازج",
+    "hello": "مرحبا",
+    "hot": "ساخن",
+    "icecream": "آيس كريم",
+    "ingredients": "مكونات",
+    "juicy": "عصيري",
+    "ketchup": "كاتشب",
+    "lactose": "لاكتوز",
+    "lettuce": "خس",
+    "lid": "غطاء",
+    "manager": "مدير",
+    "menu": "قائمة الطعام",
+    "milk": "حليب",
+    "mustard": "خردل",
+    "napkin": "منديل",
+    "no": "لا",
+    "order": "طلب",
+    "pepper": "فلفل",
+    "pickle": "مخلل",
+    "pizza": "بيتزا",
+    "please": "من فضلك",
+    "ready": "جاهز",
+    "refill": "إعادة ملء",
+    "repeat": "كرر",
+    "safe": "آمن",
+    "salt": "ملح",
+    "sandwich": "ساندويتش",
+    "sauce": "صلصة",
+    "small": "صغير",
+    "soda": "صودا",
+    "sorry": "آسف",
+    "spicy": "حار",
+    "spoon": "ملعقة",
+    "straw": "قش",
+    "sugar": "سكر",
+    "sweet": "حلو",
+    "tissues": "مناديل",
+    "total": "مجموع",
+    "urgent": "عاجل",
+    "vegetables": "خضروات",
+    "warm": "دافئ",
+    "water": "ماء",
+    "what": "ماذا",
+    "yoghurt": "زبادي",
+    "your": "لك",
+    "ILoveYou":"أحبك",
+    "Halo":"مرحبًا"
+    }
+######################################################
+class VideoStreaming(object):
+    def __init__(self):
+        super(VideoStreaming, self).__init__()
+        print ("===== Video Streaming =====")
+        self._preview = False
+        self._flipH = False
+        self._detect = False
+        self._model = False
+        self._mediaPipe = False
+        self._confidence = 75.0
+        self.mp_hands = mp.solutions.hands
+        self.hands = self.mp_hands.Hands()
+
+    @property
+    def confidence(self):
+        return self._confidence
+
+    @confidence.setter
+    def confidence(self, value):
+        self._confidence = int(value)
+
+    @property
+    def preview(self):
+        return self._preview
+
+    @preview.setter
+    def preview(self, value):
+        self._preview = bool(value)
+
+    @property
+    def flipH(self):
+        return self._flipH
+
+    @flipH.setter
+    def flipH(self, value):
+        self._flipH = bool(value)
+
+    @property
+    def detect(self):
+        return self._detect
+
+    @detect.setter
+    def detect(self, value):
+        self._detect = bool(value)
+
+    @property
+    def mediaPipe(self):
+        return self._mediaPipe
+
+    @mediaPipe.setter
+    def mediaPipe(self, value):
+        self._mediaPipe = bool(value)
+
+    def show(self, url):
+        print(url)
+        self._preview = False
+        self._flipH = False
+        self._detect = False
+        self._mediaPipe = False
+
+        self._confidence = 75.0
+        ydl_opts = {
+            "quiet": True,
+            "no_warnings": True,
+            "format": "best",
+            "forceurl": True,
+        }
+
+        if url == '0':
+            cap = cv2.VideoCapture(0)
+        else:
+            
+            ydl = youtube_dl.YoutubeDL(ydl_opts)
+
+            info = ydl.extract_info(url, download=False)
+            url = info["url"]
 
-# Route to handle the offer request
-@app.route('/offer', methods=['POST'])
-def offer_route():
-    return offer()
+            cap = cv2.VideoCapture(url)
+
+        while True:
+            if self._preview:
+                if stop_flag:
+                    print("Process Stopped")
+                    return
+
+                grabbed, frame = cap.read()
+                if not grabbed:
+                    break
+                if self.flipH:
+                    frame = cv2.flip(frame, 1)
+
+                if self.detect:
+                    frame_yolo = frame.copy()
+                    results_yolo = model_object_detection.predict(frame_yolo, conf=self._confidence / 100)
+
+                    frame_yolo, labels = results_yolo[0].plot()
+                    list_labels = []
+                    # labels_confidences
+
+                    for label in labels:
+                        confidence = label.split(" ")[-1]
+                        label_name = " ".join(label.split(" ")[:-1])
+                        # Translate the label if it exists in the translation dictionary
+                        translated_label = classes_translation.get(label_name, label_name)
+                        list_labels.append(translated_label)
+                        list_labels.append(confidence)
+                        socketio.emit('label', list_labels)
+
+                if self.mediaPipe:
+                    # Convert the image to RGB for processing with MediaPipe
+                    image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+                    results = self.hands.process(image)
+                    
+                    if results.multi_hand_landmarks:
+                        for hand_landmarks in results.multi_hand_landmarks:
+                            mp.solutions.drawing_utils.draw_landmarks(
+                                frame,
+                                hand_landmarks,
+                                self.mp_hands.HAND_CONNECTIONS,
+                                landmark_drawing_spec=mp.solutions.drawing_utils.DrawingSpec(color=(255, 0, 0), thickness=4, circle_radius=3),
+                                connection_drawing_spec=mp.solutions.drawing_utils.DrawingSpec(color=(255, 255, 255), thickness=2, circle_radius=2), 
+                            )
+
+                frame = cv2.imencode(".jpg", frame)[1].tobytes()
+                yield ( 
+                    b'--frame\r\n'
+                    b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n'
+                )
+            else:
+                snap = np.zeros((
+                    1000,
+                    1000
+                ), np.uint8)
+                label = "Streaming Off"
+                H, W = snap.shape
+                font = cv2.FONT_HERSHEY_PLAIN
+                color = (255, 255, 255)
+                cv2.putText(snap, label, (W//2 - 100, H//2),
+                            font, 2, color, 2)
+                frame = cv2.imencode(".jpg", snap)[1].tobytes()
+                yield (b'--frame\r\n'
+                       b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
+
+
+# check_settings()
+VIDEO = VideoStreaming()
+
+
+@app.route('/', methods=['GET', 'POST'])
+def homepage():
+    return render_template('hompage.html')
+
+
+@app.route('/index', methods=['GET', 'POST'])
+def index():
+    print("index")
+    global stop_flag
+    stop_flag = False
+    if request.method == 'POST':
+        print("Index post request")
+        url = request.form['url']
+        print("index: ", url)
+        session['url'] = url
+        return redirect(url_for('index'))
+    return render_template('index.html')
 
-# Route to stream video frames
 @app.route('/video_feed')
 def video_feed():
-    return Response(generate_frames(), mimetype='multipart/x-mixed-replace; boundary=frame')
+    url = session.get('url', None)
+    print("video feed: ", url)
+    if url is None:
+        return redirect(url_for('homepage'))
+    print("video feed: ", url)
+    return Response(VIDEO.show(url), mimetype='multipart/x-mixed-replace; boundary=frame')
+
+# * Button requests
+@app.route("/request_preview_switch")
+def request_preview_switch():
+    VIDEO.preview = not VIDEO.preview
+    print("*"*10, VIDEO.preview)
+    return "nothing"
+
+@app.route("/request_flipH_switch")
+def request_flipH_switch():
+    VIDEO.flipH = not VIDEO.flipH
+    print("*"*10, VIDEO.flipH)
+    return "nothing"
+
+@app.route("/request_run_model_switch")
+def request_run_model_switch():
+    VIDEO.detect = not VIDEO.detect
+    print("*"*10, VIDEO.detect)
+    return "nothing"
+
+@app.route("/request_mediapipe_switch")
+def request_mediapipe_switch():
+    VIDEO.mediaPipe = not VIDEO.mediaPipe
+    print("*"*10, VIDEO.mediaPipe)
+    return "nothing"
+
+@app.route('/update_slider_value', methods=['POST'])
+def update_slider_value():
+    slider_value = request.form['sliderValue']
+    VIDEO.confidence = slider_value
+    return 'OK'
+
+@app.route('/stop_process')
+def stop_process():
+    print("Process stop Request")
+    global stop_flag
+    stop_flag = True
+    return 'Process Stop Request'
+
+@socketio.on('connect')
+def test_connect():
+    print('Connected')
+    #emit('message', data, broadcast=True)
+
 
-# Run the Flask app
-if __name__ == "__main__":
-    app.run(debug=True, host='0.0.0.0')
+if __name__ == '__main__':
+    socketio.run(app, host="0.0.0.0", allow_unsafe_werkzeug=True,port=7860)