Update app.py

app.py

@@ -515,145 +515,145 @@ elif app_mode =='Run on Test Videos':
     
         prevTime = 0
         postfix = info["format"]["format_name"].split(",")[0]
-        
-        with tempfile.NamedTemporaryFile(suffix=f'.{postfix}',delete=False) as tfflie:
-            output_file = tfflie.name#'./output.mp4'
-            # width = int(vid.get(cv2.CAP_PROP_FRAME_WIDTH))
-            # height = int(vid.get(cv2.CAP_PROP_FRAME_HEIGHT))
-            fps_input = int(vid.get(cv2.CAP_PROP_FPS))
 
-            #codec = cv2.VideoWriter_fourcc(*FLAGS.output_format)
-            
-            # codec = cv2.VideoWriter_fourcc('m','p','4','v')
-            out = None
-            writer=None
-            weighted_avg_dict={}
-            
-            idx=0
-            
-            for _, row in current_test_df.iterrows():#enumerate(file_df.rolling(window=20, step=20,min_periods=1)):
-                # print(f'captured frame#{idx}')
-                if(vid.isOpened()):
-                    ret, frame = vid.read()
-                    
+
+        output_file = f"/tmp/output_{uuid.uuid4().hex}.{postfix}"
+
+        # height = int(vid.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        fps_input = int(vid.get(cv2.CAP_PROP_FPS))
+
+        #codec = cv2.VideoWriter_fourcc(*FLAGS.output_format)
+        
+        # codec = cv2.VideoWriter_fourcc('m','p','4','v')
+        out = None
+        writer=None
+        weighted_avg_dict={}
+        
+        idx=0
+        
+        for _, row in current_test_df.iterrows():#enumerate(file_df.rolling(window=20, step=20,min_periods=1)):
+            # print(f'captured frame#{idx}')
+            if(vid.isOpened()):
+                ret, frame = vid.read()
+                
+                
+                if len(window)<window_size:
+                    canvas=util.drawStickmodel(frame,eval(row['bodypose_circles']),eval(row['bodypose_sticks']),eval(row['handpose_edges']),eval(row['handpose_peaks']))
+                    canvas_with_plot=util.draw_bar_plot_below_image(canvas,{}, f'Prediction bar plot - Frame number {idx+1} [** no predictions]',canvas)
+                    canvas_with_plot=util.draw_bar_plot_below_image(canvas_with_plot,weighted_avg_dict, f'Weighted avg - Frame number {idx+1} [** no predictions]',canvas)
+                    canvas_with_plot=util.add_padding_to_bottom(canvas_with_plot,(255,255,255),100)# Adds padding at bottom
                     
-                    if len(window)<window_size:
-                        canvas=util.drawStickmodel(frame,eval(row['bodypose_circles']),eval(row['bodypose_sticks']),eval(row['handpose_edges']),eval(row['handpose_peaks']))
-                        canvas_with_plot=util.draw_bar_plot_below_image(canvas,{}, f'Prediction bar plot - Frame number {idx+1} [** no predictions]',canvas)
-                        canvas_with_plot=util.draw_bar_plot_below_image(canvas_with_plot,weighted_avg_dict, f'Weighted avg - Frame number {idx+1} [** no predictions]',canvas)
-                        canvas_with_plot=util.add_padding_to_bottom(canvas_with_plot,(255,255,255),100)# Adds padding at bottom
-                        
-                        if writer is None:                        
-                            input_framesize = canvas_with_plot.shape[:2]
-                            writer = Writer(output_file, input_fps, input_framesize, input_pix_fmt,
-                                            input_vcodec)
-
-                        # if out is None:
-                        #     out=cv2.VideoWriter(output_file, codec, fps_input, frame.shape[:2])
-                        
-
-                        writer(canvas_with_plot)
-                        # out.write(canvas)
-                        with runtime_progress.container():
-                            df1 = pd.DataFrame([[f'{idx+1}/{current_test_df.shape[0]}','<model will output after 20 frames>']], columns=['Frames Processed','Detected Class'])
-
-                            my_table = st.table(df1)
-                        window.append(frame)
-                        # kpi1_text.write(f"<h1 style='text-align: center; color: red;'>{idx+1}/{current_test_df.shape[0]}</h1>", unsafe_allow_html=True)
-                        # kpi2_text.write(f"<h1 style='text-align: center; color: red;'>--</h1>", unsafe_allow_html=True)
-                        with view.container():
-                            st.image(canvas_with_plot,channels = 'BGR',use_column_width=True)
-                    else:
-
-                        window[:-1] = window[1:]
-                        window[-1]=frame
-                        translation_model=get_translator_model()
-                        # testing_df[]
-                        
-                        encoded_translation = translation_model(X_test_filtered[idx-20].reshape(1,X_test_filtered[idx-20].shape[0],X_test_filtered[idx-20].shape[1]))
-                        encoded_translation=encoded_translation[0].cpu().detach().numpy()
-                        sorted_index=np.argsort(encoded_translation)[::-1]
-                        maxindex=np.argmax(encoded_translation)
-
-                        top_3_probs = encoded_translation.argsort()[-3:][::-1]  # Get indices of top 3 probabilities (descending order)
-                        top_3_categories = [expression_mapping[i] for i in top_3_probs]  # Convert indices to category names (assuming class_names list exists)
-                        top_3_values = encoded_translation[top_3_probs]  # Get corresponding probabilities
-                        # print(f'{idx} {encoded_translation[maxindex]:0.4f} {maxindex}-{expression_mapping[maxindex]} ')#{[(pi,encoded_translation[pi],expression_mapping[pi]) for pi in sorted_index]}
-                        for category, prob in zip(top_3_categories, top_3_values):
-                            if category not in frame_wise_outputs:
-                                frame_wise_outputs[category]=[]
-                            frame_wise_outputs[category].append(prob)
-                        current_prob={}
-
-                        for category, prob in zip(top_3_categories, top_3_values):
-                            current_prob[category]=prob
-
-                        for key in frame_wise_outputs:
-                            weighted_avg_dict[key]=weighted_average(frame_wise_outputs[key],[len(frame_wise_outputs[key]) for i in range(len(frame_wise_outputs[key]))])
-
-                        sorted_dict = dict(sorted(weighted_avg_dict.items(), key=lambda item: item[1], reverse=True))
-                        canvas=util.drawStickmodel(frame,eval(row['bodypose_circles']),eval(row['bodypose_sticks']),eval(row['handpose_edges']),eval(row['handpose_peaks']))
-                        canvas_with_plot=util.draw_bar_plot_below_image(canvas,current_prob, f'Prediction at frame window({idx-20+1}-{idx+1})',canvas)
-                        canvas_with_plot=util.draw_bar_plot_below_image(canvas_with_plot,weighted_avg_dict, f'Weighted avg till window {idx+1}',canvas)
-                        canvas_with_plot=util.add_padding_to_bottom(canvas_with_plot,(255,255,255),100)
-                        writer(canvas_with_plot)
-
-                        
-                        currTime = time.time()
-                        fps = 1 / (currTime - prevTime)
-                        prevTime = currTime
-                        # out.write(frame)
-                        # if record:
-                        #     #st.checkbox("Recording", value=True)
-                        #     out.write(frame)
-                        #Dashboard
-
-                        max_prob = float('-inf')  # Initialize with negative infinity
-                        max_key = None
-
-                        for exp, prob in weighted_avg_dict.items():
-                            if prob > max_prob:
-                                max_prob = prob
-                                max_key = exp
-                        with runtime_progress.container():
-                            df1 = pd.DataFrame([[f'{idx+1}/{current_test_df.shape[0]}',f'{max_key} ({max_prob*100:.2f}%)']], columns=['Frames Processed','Detected Class'])
-                            my_table = st.table(df1)
-                        # kpi1_text.write(f"<h1 style='text-align: center; color: red;'>{idx+1}/{current_test_df.shape[0]}</h1>", unsafe_allow_html=True)
-                        # kpi2_text.write(f"<h1 style='text-align: center; color: red;'>{max_key} ({max_prob*100:.2f}%)</h1>", unsafe_allow_html=True)
-                        # with placeholder.container():
-                        #     # st.write(weighted_avg_dict)
-                        #     # data = {
-                        #     #     "I": 0.7350964583456516,
-                        #     #     "Hello": 0.1078806109726429,
-                        #     #     "you": 0.11776176246348768,
-                        #     #     "you (plural)": 0.12685142129916568
-                        #     # }
-
-                        #     # Convert the dictionary to a Pandas DataFrame for easier plotting
-                        #     df = pd.DataFrame.from_dict(weighted_avg_dict, orient='index', columns=['Values'])
-
-                        #     # Create a bar chart with Streamlit
-                        #     st.bar_chart(df)
-                        # frame = cv2.resize(frame,(0,0),fx = 0.8 , fy = 0.8)
-                        # frame = image_resize(image = frame, width = 640)
-                        with view.container():
-                            st.image(canvas_with_plot,channels = 'BGR',use_column_width=True)
-
-                    idx=idx+1
+                    if writer is None:                        
+                        input_framesize = canvas_with_plot.shape[:2]
+                        writer = Writer(output_file, input_fps, input_framesize, input_pix_fmt,
+                                        input_vcodec)
+
+                    # if out is None:
+                    #     out=cv2.VideoWriter(output_file, codec, fps_input, frame.shape[:2])
                     
 
-            # st.text('Video Processed')
-            with view.container():
-                if writer is not None:  # ✅ safeguard
-                    writer.close()
-                    output_video = open(output_file,'rb')
-                    out_bytes = output_video.read()
-                    st.video(out_bytes)
-                    print(f'Output file - {output_file}')
+                    writer(canvas_with_plot)
+                    # out.write(canvas)
+                    with runtime_progress.container():
+                        df1 = pd.DataFrame([[f'{idx+1}/{current_test_df.shape[0]}','<model will output after 20 frames>']], columns=['Frames Processed','Detected Class'])
+
+                        my_table = st.table(df1)
+                    window.append(frame)
+                    # kpi1_text.write(f"<h1 style='text-align: center; color: red;'>{idx+1}/{current_test_df.shape[0]}</h1>", unsafe_allow_html=True)
+                    # kpi2_text.write(f"<h1 style='text-align: center; color: red;'>--</h1>", unsafe_allow_html=True)
+                    with view.container():
+                        st.image(canvas_with_plot,channels = 'BGR',use_column_width=True)
                 else:
-                    st.warning("No video was processed, writer is empty.")
-                # out.release()
+
+                    window[:-1] = window[1:]
+                    window[-1]=frame
+                    translation_model=get_translator_model()
+                    # testing_df[]
+                    
+                    encoded_translation = translation_model(X_test_filtered[idx-20].reshape(1,X_test_filtered[idx-20].shape[0],X_test_filtered[idx-20].shape[1]))
+                    encoded_translation=encoded_translation[0].cpu().detach().numpy()
+                    sorted_index=np.argsort(encoded_translation)[::-1]
+                    maxindex=np.argmax(encoded_translation)
+
+                    top_3_probs = encoded_translation.argsort()[-3:][::-1]  # Get indices of top 3 probabilities (descending order)
+                    top_3_categories = [expression_mapping[i] for i in top_3_probs]  # Convert indices to category names (assuming class_names list exists)
+                    top_3_values = encoded_translation[top_3_probs]  # Get corresponding probabilities
+                    # print(f'{idx} {encoded_translation[maxindex]:0.4f} {maxindex}-{expression_mapping[maxindex]} ')#{[(pi,encoded_translation[pi],expression_mapping[pi]) for pi in sorted_index]}
+                    for category, prob in zip(top_3_categories, top_3_values):
+                        if category not in frame_wise_outputs:
+                            frame_wise_outputs[category]=[]
+                        frame_wise_outputs[category].append(prob)
+                    current_prob={}
+
+                    for category, prob in zip(top_3_categories, top_3_values):
+                        current_prob[category]=prob
+
+                    for key in frame_wise_outputs:
+                        weighted_avg_dict[key]=weighted_average(frame_wise_outputs[key],[len(frame_wise_outputs[key]) for i in range(len(frame_wise_outputs[key]))])
+
+                    sorted_dict = dict(sorted(weighted_avg_dict.items(), key=lambda item: item[1], reverse=True))
+                    canvas=util.drawStickmodel(frame,eval(row['bodypose_circles']),eval(row['bodypose_sticks']),eval(row['handpose_edges']),eval(row['handpose_peaks']))
+                    canvas_with_plot=util.draw_bar_plot_below_image(canvas,current_prob, f'Prediction at frame window({idx-20+1}-{idx+1})',canvas)
+                    canvas_with_plot=util.draw_bar_plot_below_image(canvas_with_plot,weighted_avg_dict, f'Weighted avg till window {idx+1}',canvas)
+                    canvas_with_plot=util.add_padding_to_bottom(canvas_with_plot,(255,255,255),100)
+                    writer(canvas_with_plot)
+
+                    
+                    currTime = time.time()
+                    fps = 1 / (currTime - prevTime)
+                    prevTime = currTime
+                    # out.write(frame)
+                    # if record:
+                    #     #st.checkbox("Recording", value=True)
+                    #     out.write(frame)
+                    #Dashboard
+
+                    max_prob = float('-inf')  # Initialize with negative infinity
+                    max_key = None
+
+                    for exp, prob in weighted_avg_dict.items():
+                        if prob > max_prob:
+                            max_prob = prob
+                            max_key = exp
+                    with runtime_progress.container():
+                        df1 = pd.DataFrame([[f'{idx+1}/{current_test_df.shape[0]}',f'{max_key} ({max_prob*100:.2f}%)']], columns=['Frames Processed','Detected Class'])
+                        my_table = st.table(df1)
+                    # kpi1_text.write(f"<h1 style='text-align: center; color: red;'>{idx+1}/{current_test_df.shape[0]}</h1>", unsafe_allow_html=True)
+                    # kpi2_text.write(f"<h1 style='text-align: center; color: red;'>{max_key} ({max_prob*100:.2f}%)</h1>", unsafe_allow_html=True)
+                    # with placeholder.container():
+                    #     # st.write(weighted_avg_dict)
+                    #     # data = {
+                    #     #     "I": 0.7350964583456516,
+                    #     #     "Hello": 0.1078806109726429,
+                    #     #     "you": 0.11776176246348768,
+                    #     #     "you (plural)": 0.12685142129916568
+                    #     # }
+
+                    #     # Convert the dictionary to a Pandas DataFrame for easier plotting
+                    #     df = pd.DataFrame.from_dict(weighted_avg_dict, orient='index', columns=['Values'])
+
+                    #     # Create a bar chart with Streamlit
+                    #     st.bar_chart(df)
+                    # frame = cv2.resize(frame,(0,0),fx = 0.8 , fy = 0.8)
+                    # frame = image_resize(image = frame, width = 640)
+                    with view.container():
+                        st.image(canvas_with_plot,channels = 'BGR',use_column_width=True)
+
+                idx=idx+1
                 
+
+        # st.text('Video Processed')
+        with view.container():
+            if writer is not None:  # ✅ safeguard
+                writer.close()
+                output_video = open(output_file,'rb')
+                out_bytes = output_video.read()
+                st.video(out_bytes)
                 print(f'Output file - {output_file}')
-            cv2.destroyAllWindows()
-            vid.release()
+            else:
+                st.warning("No video was processed, writer is empty.")
+            # out.release()
+            
+            print(f'Output file - {output_file}')
+        cv2.destroyAllWindows()
+        vid.release()