app.py

app.py

@@ -0,0 +1,186 @@
+import gradio as gr
+import os
+import json
+from huggingface_hub import hf_hub_download, list_repo_files
+import logging
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# Cricket annotation categories
+ANNOTATION_CATEGORIES = {
+    "Bowler's Run Up": ["Fast", "Slow"],
+    "Delivery Type": ["Yorker", "Bouncer", "Length Ball", "Slower ball", "Googly", "Arm Ball", "Other"],
+    "Ball's trajectory": ["In Swing", "Out Swing", "Off spin", "Leg spin"],
+    "Shot Played": ["Cover Drive", "Straight Drive", "On Drive", "Pull", "Square Cut", "Defensive Block"],
+    "Outcome of the shot": ["four", "six", "wicket", "runs (1,2,3)"],
+    "Shot direction": ["long on", "long off", "cover", "point", "midwicket", "square leg", "third man", "fine leg"],
+    "Fielder's Action": ["Catch taken", "Catch dropped", "Misfield", "Run-out attempt", "Fielder fields"]
+}
+
+# HuggingFace repo info - replace with your actual repo
+HF_REPO_ID = "username/cricket-videos"  # Replace with your actual repo
+HF_REPO_TYPE = "dataset"  # or "model" depending on how you've stored the videos
+
+class VideoAnnotator:
+    def __init__(self):
+        self.video_files = []
+        self.current_video_idx = 0
+        self.annotations = {}
+        
+    def load_videos_from_hf(self):
+        try:
+            logger.info(f"Attempting to list files from HuggingFace repo: {HF_REPO_ID}")
+            self.video_files = [f for f in list_repo_files(HF_REPO_ID, repo_type=HF_REPO_TYPE) 
+                               if f.endswith(('.mp4', '.avi', '.mov'))]
+            logger.info(f"Found {len(self.video_files)} video files")
+            return len(self.video_files) > 0
+        except Exception as e:
+            logger.error(f"Error accessing HuggingFace repo: {e}")
+            return False
+    
+    def get_current_video(self):
+        if not self.video_files:
+            logger.warning("No video files available")
+            return None
+        
+        video_path = self.video_files[self.current_video_idx]
+        logger.info(f"Loading video: {video_path}")
+        
+        try:
+            local_path = hf_hub_download(
+                repo_id=HF_REPO_ID,
+                filename=video_path,
+                repo_type=HF_REPO_TYPE
+            )
+            logger.info(f"Video downloaded to: {local_path}")
+            return local_path
+        except Exception as e:
+            logger.error(f"Error downloading video: {e}")
+            return None
+    
+    def save_annotation(self, annotations_dict):
+        video_name = os.path.basename(self.video_files[self.current_video_idx])
+        annotation_file = f"{video_name}_annotations.json"
+        
+        logger.info(f"Saving annotations for {video_name} to {annotation_file}")
+        
+        try:
+            with open(annotation_file, 'w') as f:
+                json.dump(annotations_dict, f, indent=4)
+            logger.info("Annotations saved successfully")
+            return f"Annotations saved for {video_name}"
+        except Exception as e:
+            logger.error(f"Error saving annotations: {e}")
+            return f"Error saving: {str(e)}"
+    
+    def next_video(self, *current_annotations):
+        # Save current annotations before moving to next video
+        if self.video_files:
+            annotations_dict = {}
+            for i, category in enumerate(ANNOTATION_CATEGORIES.keys()):
+                if current_annotations[i]:
+                    annotations_dict[category] = current_annotations[i]
+            
+            if annotations_dict:
+                self.save_annotation(annotations_dict)
+        
+        # Move to next video
+        if self.current_video_idx < len(self.video_files) - 1:
+            self.current_video_idx += 1
+            logger.info(f"Moving to next video (index: {self.current_video_idx})")
+            return self.get_current_video(), *[None] * len(ANNOTATION_CATEGORIES)
+        else:
+            logger.info("Already at the last video")
+            return self.get_current_video(), *[None] * len(ANNOTATION_CATEGORIES)
+    
+    def prev_video(self, *current_annotations):
+        # Save current annotations before moving to previous video
+        if self.video_files:
+            annotations_dict = {}
+            for i, category in enumerate(ANNOTATION_CATEGORIES.keys()):
+                if current_annotations[i]:
+                    annotations_dict[category] = current_annotations[i]
+            
+            if annotations_dict:
+                self.save_annotation(annotations_dict)
+        
+        # Move to previous video
+        if self.current_video_idx > 0:
+            self.current_video_idx -= 1
+            logger.info(f"Moving to previous video (index: {self.current_video_idx})")
+            return self.get_current_video(), *[None] * len(ANNOTATION_CATEGORIES)
+        else:
+            logger.info("Already at the first video")
+            return self.get_current_video(), *[None] * len(ANNOTATION_CATEGORIES)
+
+def create_interface():
+    annotator = VideoAnnotator()
+    success = annotator.load_videos_from_hf()
+    
+    if not success:
+        logger.error("Failed to load videos. Using demo mode with sample video.")
+        # In real app, you might want to provide a sample video or show an error
+    
+    with gr.Blocks() as demo:
+        gr.Markdown("# Cricket Video Annotation Tool")
+        
+        with gr.Row():
+            video_player = gr.Video(label="Current Video")
+        
+        annotation_components = []
+        
+        with gr.Row():
+            with gr.Column():
+                for category, options in list(ANNOTATION_CATEGORIES.items())[:4]:
+                    radio = gr.Radio(
+                        choices=options,
+                        label=category,
+                        info=f"Select {category}"
+                    )
+                    annotation_components.append(radio)
+            
+            with gr.Column():
+                for category, options in list(ANNOTATION_CATEGORIES.items())[4:]:
+                    radio = gr.Radio(
+                        choices=options,
+                        label=category,
+                        info=f"Select {category}"
+                    )
+                    annotation_components.append(radio)
+        
+        with gr.Row():
+            prev_btn = gr.Button("Previous Video")
+            save_btn = gr.Button("Save Annotations", variant="primary")
+            next_btn = gr.Button("Next Video")
+            
+        # Initialize with first video
+        current_video = annotator.get_current_video()
+        if current_video:
+            video_player.value = current_video
+        
+        # Event handlers
+        save_btn.click(
+            fn=annotator.save_annotation,
+            inputs=[gr.Group(annotation_components)],
+            outputs=gr.Textbox(label="Status")
+        )
+        
+        next_btn.click(
+            fn=annotator.next_video,
+            inputs=annotation_components,
+            outputs=[video_player] + annotation_components
+        )
+        
+        prev_btn.click(
+            fn=annotator.prev_video,
+            inputs=annotation_components,
+            outputs=[video_player] + annotation_components
+        )
+        
+    return demo
+
+if __name__ == "__main__":
+    demo = create_interface()
+    demo.launch()