Update app.py

app.py

@@ -3,171 +3,84 @@ import time
 import cv2 # opencv2 package for python.
 import torch
 from pytube import YouTube
-from ultralyticsplus import YOLO, render_result
-#from imageai.Detection import ObjectDetection
 
 
-#obj_detect.setModelPath(r"C:/Datasets/yolo.h5")
-#obj_detect.loadModel()
+segmentor = pipeline("image-segmentation", model="facebook/detr-resnet-50-panoptic")
 
-#from torch import hub # Hub contains other models like FasterRCNN
-model = YOLO('ultralyticsplus/yolov8s')
-device = 'cuda' if torch.cuda.is_available() else 'cpu'
-URL = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" #URL to parse
-
-# set model parameters
-model.overrides['conf'] = 0.25  # NMS confidence threshold
-model.overrides['iou'] = 0.45  # NMS IoU threshold
-model.overrides['agnostic_nms'] = False  # NMS class-agnostic
-model.overrides['max_det'] = 1000  # maximum number of detections per image
-model.to(device)
-
-
-#play = pafy.new(_URL).streams[-1] #'-1' means read the lowest quality of video.
-
-#assert play is not None # we want to make sure their is a input to read.
-#stream = cv2.VideoCapture(play.url) #create a opencv video stream.
-#stream = cv2.VideoCapture(0) # 0 means read from local camera.
-#camera_ip = "rtsp://username:password@IP/port"
-#stream = cv2.VideoCapture(camera_ip)
-#class Capvid:
- 
-
-# load model
-
-# set image
-#image = 'https://github.com/ultralytics/yolov5/raw/master/data/images/zidane.jpg'
-
-# perform inference
-#def show(feed)
-#    return model.predict(feed)
-
-# observe results
-#print(results[0].boxes)
-#render = render_result(model=model, image=image, result=results[0])
-#render.show()
 
-"""
-The function below identifies the device which is availabe to make the prediction and uses it to load and infer the frame. Once it has results it will extract the labels and cordinates(Along with scores) for each object detected in the frame.
-"""
-def score_frame(frame):
-    #frame = [torch.tensor(frame)]
-    result = model(frame)
-    results = [torch.tensor(result)]
-    labels = results[0][:, -1].numpy()
-    cord = results[0][:, :-1].numpy()
-    return labels, cord
-
-"""
-The function below takes the results and the frame as input and plots boxes over all the objects which have a score higer than our threshold.
-"""
-def plot_boxes(results, frame):
-    labels, cord = results
-    n = len(labels)
-    x_shape, y_shape = frame.shape[1], frame.shape[0]
-    for i in range(n):
-        row = cord[i]
-        # If score is less than 0.2 we avoid making a prediction.
-        if row[4] < 0.2: 
-            continue
-        x1 = int(row[0]*x_shape)
-        y1 = int(row[1]*y_shape)
-        x2 = int(row[2]*x_shape)
-        y2 = int(row[3]*y_shape)
-        bgr = (0, 255, 0) # color of the box
-        classes = model.names # Get the name of label index
-        label_font = cv2.FONT_HERSHEY_SIMPLEX #Font for the label.
-        cv2.rectangle(frame, \
-                      (x1, y1), (x2, y2), \
-                       bgr, 2) #Plot the boxes
-        cv2.putText(frame,\
-                    classes[labels[i]], \
-                    (x1, y1), \
-                    label_font, 0.9, bgr, 2) #Put a label over box.
-        return frame
+device = 'cuda' if torch.cuda.is_available() else 'cpu'
+URL = "https://www.youtube.com/watch?v=6NBwbKMyzEE" #URL to parse
 
-"""
-The Function below oracestrates the entire operation and performs the real-time parsing for video stream.
-"""
-def vid_play(vid_cap):
-    stream = cv2.VideoCapture(vid_cap)
+def load(URL):
     
-    player = stream #Get your video stream.
-    assert player.isOpened() # Make sure that their is a stream. 
-    #Below code creates a new video writer object to write our
-    #output stream.
-    out_vid = ("vid_tmp.avi")
-    x_shape = int(player.get(cv2.CAP_PROP_FRAME_WIDTH))
-    y_shape = int(player.get(cv2.CAP_PROP_FRAME_HEIGHT))
-    four_cc = cv2.VideoWriter_fourcc(*"MJPG") #Using MJPEG codex
-    out = cv2.VideoWriter(out_vid, four_cc, 20, \
-                          (x_shape, y_shape)) 
-    ret, frame = player.read() # Read the first frame.
-    while True: # Run until stream is out of frames
-        start_time = time.time() # We would like to measure the FPS.
-        results = score_frame(frame) # Score the Frame
-        frame = plot_boxes(results, frame) # Plot the boxes.
-        end_time = time.time()
-        fps = 1/np.round(end_time - start_time, 3) #Measure the FPS.
-        print(f"Frames Per Second : {fps}")
-        out.write(frame) # Write the frame onto the output.
-        ret, frame = player.read() # Read next frame.
-        return out
-
-with gr.Blocks() as app:
-    stream = gr.State()
-    def load(URL):
-        yt = YouTube(URL)
-        vid_cap = yt.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc().last().download(filename="tmp.mp4")
-        process = cv2.VideoCapture(vid_cap)
-        frame_num = int(process.get(cv2.CAP_PROP_POS_FRAMES))
-        frame_count = int(process.get(cv2.CAP_PROP_FRAME_COUNT))
-        process.release()
-       
-        return vid_cap,frame_num,frame_count
-        
-    def vid_play2(cap,frame_num):
-        player = cv2.VideoCapture(cap)
+    yt = YouTube(URL)
+    vid_cap = yt.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc().last().download(filename="tmp.mp4")
+    global player
+    player = cv2.VideoCapture(vid_cap)
+    frame_num = int(player.get(cv2.CAP_PROP_POS_FRAMES))
+    frame_count = int(player.get(cv2.CAP_PROP_FRAME_COUNT))
+    frame_fps = (player.get(cv2.CAP_PROP_FPS))
+    tog = 0
+    return vid_cap,frame_num,frame_count,frame_fps,tog
+
+def fw_fn(cur,last):
+    next = cur+1
+    if next > last:
+        next = last
+    return next
+def bk_fn(cur):
+    next = cur-1
+    if next < 0:
+        next = 0
+    return next
+def tog_on():
+    return 1,gr.Markdown.update("""<center><h7>Status: Playing 😁</h7></center>""")
+def tog_off():
+    return 0,gr.Markdown.update("""<center><h7>Status: Stopped 💀</h7></center>""")
         
-        #player = stream #Get your video stream.
-        assert player.isOpened() # Make sure that their is a stream. 
-        #Below code creates a new video writer object to write our
-        #output stream.
-        #out_vid = ("vid_tmp.avi")
-        #x_shape = int(player.get(cv2.CAP_PROP_FRAME_WIDTH))
-        #y_shape = int(player.get(cv2.CAP_PROP_FRAME_HEIGHT))
-        #four_cc = cv2.VideoWriter_fourcc(*"MJPG") #Using MJPEG codex
-        #out = cv2.VideoWriter(out_vid, four_cc, 20,(x_shape, y_shape)) 
-        #stream.set(cv2.CAP_PROP_POS_FRAMES, int(frame_num))
-
-        ret, frame = player.read(int(frame_num))
-        results = model.predict(frame)
-
-        render = render_result(model=model, image=frame, result=results[0])
-        #out = render.show()
-        #start_time = time.time() # We would like to measure the FPS.
-        #results = score_frame(frame) # Score the Frame
-        #frame = plot_boxes(results, frame) # Plot the boxes.
-        #end_time = time.time()
-        #fps = 1/np.round(end_time - start_time, 3) #Measure the FPS.
-        #print(f"Frames Per Second : {fps}")
-        #out.write(frame) # Write the frame onto the output.
-        #ret, frame = player.read() # Read next frame.
-        return render
-   
-     
+def pl_fn(cap,cur,last,fps,pl_tog):
+    player.set(cv2.CAP_PROP_POS_FRAMES, cur)
+    ret, frame_bgr = player.read(cur)
+    frame = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
+    output = segmentor(frame)
+
+    if pl_tog ==1:
+        cur+=1
+    else:
+        cur = cur        
+    return output,cur
 
+with gr.Blocks() as app:
+    gr.Markdown("""<center><h1>Testing</h1><h4>stuff</h4><h4></h4></center>""")
+    play_state = gr.Markdown("""<right><h7></h7></right>""")
 
-    youtube_url = gr.Textbox(label="YouTube URL",value=f"{URL}")
-    load_button = gr.Button("Load Video")
     with gr.Row():
-        total_frames = gr.Number(interactive=False)
-        run_button = gr.Button()
-        cur_frame = gr.Number()
+        with gr.Column():
+            youtube_url = gr.Textbox(label="YouTube URL",value=f"{URL}")
+            load_button = gr.Button("Load Video")
+            output_win = gr.Video()
+        with gr.Column():
+            with gr.Row():
+                cur_frame = gr.Number(label="Current Frame")
+                fps_frames = gr.Number(label="Video FPS",interactive=False)
+                total_frames = gr.Number(label="Total Frames",interactive=False)
+            #run_button = gr.Button()
+            with gr.Row():
+                bk = gr.Button("<")
+                pl = gr.Button("Play")
+                st = gr.Button("Stop")
+                fw = gr.Button(">")
+            det_win = gr.Image(source="webcam", streaming=True)
     with gr.Row():
-        output_win = gr.Video()
-        det_win = gr.Image(source="webcam", streaming=True)
-    load_button.click(load,youtube_url,[output_win,cur_frame,total_frames])
-    run_button.click(vid_play2, [output_win,cur_frame], det_win) 
-
-app.launch(enable_queue=False)
+        pl_tog=gr.Number(visible=False)
+        ins_cnt=gr.Number(visible=False)   
+    pl.click(tog_on,None,[pl_tog,play_state],show_progress=False)
+    st.click(tog_off,None,[pl_tog,play_state],show_progress=False)
+    pl_tog.change(pl_fn,[output_win,cur_frame,total_frames,fps_frames,pl_tog],[det_win,cur_frame],show_progress=False)
+    cur_frame.change(pl_fn,[output_win,cur_frame,total_frames,fps_frames,pl_tog],[det_win,cur_frame],show_progress=False)
+    bk.click(bk_fn,[cur_frame],cur_frame,show_progress=False)
+    fw.click(fw_fn,[cur_frame,total_frames],cur_frame,show_progress=False)
+    load_button.click(load,youtube_url,[output_win,cur_frame,total_frames,fps_frames,pl_tog])
+    #run_button.click(vid_play, [output_win,cur_frame], det_win) 
+
+app.queue(concurrency_count=10).launch()