Spaces:
Sleeping
Sleeping
| import cv2 | |
| import numpy as np | |
| from utils.imageEmbedding.index import get_image_embedding | |
| from utils.imageToText.index import extract_text | |
| import requests | |
| def get_video_embedding(video_url): | |
| try: | |
| cap = cv2.VideoCapture(video_url) | |
| fps = cap.get(cv2.CAP_PROP_FPS) | |
| interval = int(fps) # Capture a frame every second | |
| frame_count = 0 | |
| video_embeddings = [] | |
| while(cap.isOpened()): | |
| ret, frame = cap.read() | |
| if ret: | |
| if frame_count % interval == 0: | |
| # Convert frame to binary format | |
| ret, buffer = cv2.imencode('.jpg', frame) | |
| if not ret: | |
| continue | |
| # Convert frame binary data to bytes | |
| frame_bytes = buffer.tobytes() | |
| # Call the route to get image embedding and extracted text | |
| response = requests.post('https://imageprocessing-backend.hf.space/extract_image_text_and_embedding_binary_data', data=frame_bytes) | |
| if response.status_code != 200: | |
| print(f"Failed to process image: {frame_bytes}") | |
| continue | |
| result = response.json() | |
| image_embedding = result.get("image_embedding") | |
| extracted_text = result.get("extracted_text") | |
| video_embeddings.append({"image_embedding": image_embedding, "extracted_text": extracted_text}) | |
| frame_count += 1 | |
| else: | |
| break | |
| cap.release() | |
| return video_embeddings | |
| except Exception as e: | |
| print(e) | |
| # Example usage: | |
| # video_url = "https://utfs.io/f/ef6c037f-fa61-471a-8956-562bc2d62531-fzxs1i.mp4" | |
| # video_embeddings = get_video_embedding(video_url) | |
| # print("Video Embeddings:", video_embeddings) |