video prompting

Author: Mika

test_uralicnlp.py

@@ -154,12 +154,14 @@
 #uralicApi.import_dictionary_to_db("sms")
 
 #llm = get_llm("chatgpt", open_read(os.path.expanduser("~/.openaiapikey")).read().strip())
-#llm = get_llm("gemini", open_read(os.path.expanduser("~/.geminiapikey")).read().strip())
+llm = get_llm("gemini", open_read(os.path.expanduser("~/.geminiapikey")).read().strip())
 #llm = get_llm("mistral", open_read(os.path.expanduser("~/.mistralapikey")).read().strip())
 
 #llm = get_llm("perplexity", open_read(os.path.expanduser("~/.perplexityapikey")).read().strip())
 #llm = get_llm("claude", open_read(os.path.expanduser("~/.claudeapikey")).read().strip())
 
+print(llm.prompt_video("What is happening on this video?", "/Users/mikahama/Downloads/6830385-uhd_4096_2160_25fps.mp4"))
+
 #print(llm.prompt_image("What is this image about?", "/Users/mikahama/Desktop/teams.jpg"))
 
 #print(llm.prompt("I forgot where I put my hat..."))
@@ -203,7 +205,7 @@
 #t = TartuTranslator()
 #print(t.translate("Hello, how are you?", "eng", "fin"))
 
-llm = get_llm("chatgpt", open_read(os.path.expanduser("~/.openaiapikey")).read().strip(), model="omni-moderation-latest")
-print(llm.moderate("those faggots punched idiots and fucked each other."))
+#llm = get_llm("chatgpt", open_read(os.path.expanduser("~/.openaiapikey")).read().strip(), model="omni-moderation-latest")
+#print(llm.moderate("those faggots punched idiots and fucked each other."))
 
 

uralicNLP/llm.py

@@ -100,7 +100,10 @@ def _openai_format_prompt(self, prompt, extra_content):
 		if extra_content is None:
 			return prompt
 		else:
-			return [{"type": "text", "text": prompt}, extra_content]
+			if type(extra_content) == list:
+				return [{"type": "text", "text": prompt}] + extra_content
+			else:
+				return [{"type": "text", "text": prompt}, extra_content]
 
 	def _embed_cache(func):
 		def inner(*args, **kwargs):
@@ -117,19 +120,61 @@ def inner(*args, **kwargs):
 	def _prompt_cache(func):
 		def inner(*args, **kwargs):
 			self = args[0]
-			if self.cache and "_".join(args[1:]) in self._prompt_cache_dict:
-				return self._prompt_cache_dict["_".join(args[1:])]
+			if self.cache and "_".join([str(x) for x in args[1:]]) in self._prompt_cache_dict:
+				return self._prompt_cache_dict["_".join([str(x) for x in args[1:]])]
 			else:
 				r = func(*args, **kwargs)
 				if self.cache:
-					self._prompt_cache_dict["_".join(args[1:])] = r
+					self._prompt_cache_dict["_".join([str(x) for x in args[1:]])] = r
 				return r
 		return inner
 
 	def prompt_image(self, text, image):
 		prompt_image = self._convert_image(image)
 		return self._prompt_image_decorated(text, prompt_image)
 
+	def _prepare_video_frame(self, frame, size, i, b64):
+		height, width, channels = frame.shape
+		if height > width and height > size:
+			ratio = width/height
+			frame = cv2.resize(frame, (round(size*ratio), size))
+		elif width > height and width > size:
+			ratio = height/width
+			frame = cv2.resize(frame, (size, round(size*ratio)))
+		#cv2.imwrite("tmp/img" + str(i) + ".png", frame)	
+		_, buffer = cv2.imencode(".png", frame)
+		if b64:
+			return base64.b64encode(buffer).decode("utf-8")
+		else:
+			return buffer
+
+	@_prompt_cache
+	def prompt_video(self, text, video, size=1000, n_frames=5):
+		frames = self._process_video(video,size,n_frames,b64=True)
+		return self._prompt_video(text, frames)
+
+
+	def _process_video(self, video, size, n_frames, b64=True):
+		video = cv2.VideoCapture(video)
+		frames = []
+		while video.isOpened():
+			success, frame = video.read()
+			if not success:
+				break
+			frames.append(frame)
+			
+		video.release()
+		if (n_frames) >= len(frames):
+			intervals = 1
+		elif n_frames <= 0:
+			intervals = 1
+		else:
+			intervals = round(len(frames)/n_frames)
+		frames = frames[0::intervals]
+		frames = [self._prepare_video_frame(x, size, i, b64) for i, x in enumerate(frames)]
+		return frames
+
+
 	def _convert_image(self, image):
 		prompt_image = None
 		if type(image) == str:
@@ -170,6 +215,9 @@ def prompt(self, text):
 	def _prompt(self, text):
 		raise NotImplementedException("LLM does not support prompting")
 
+	def _prompt_video(self, text, frames):
+		raise NotImplementedException("LLM does not support prompting")
+
 	@_embed_cache
 	def embed(self, text):
 		return self._embed(text)
@@ -249,6 +297,10 @@ def set_system_prompt(self, text):
 	def _prompt_image(self, text, prompt_image):
 		return self._prompt(text, extra_content = {"type":"image_url", "image_url": {"url": prompt_image} })
 
+	def _prompt_video(self, text, frames):
+		extra_content = [{"type":"image_url", "image_url": {"url": "data:image/png;base64,"+ prompt_image} } for prompt_image in frames]
+		return self._prompt(text, extra_content=extra_content)
+
 	def _prompt(self, prompt, temperature=1, extra_content=None):
 		prompt = self._openai_format_prompt(prompt, extra_content)
 		chat_completion = self.client.chat.completions.create(
@@ -296,6 +348,11 @@ def __init__(self, api_key, model="gemini-1.5-flash", task_type="retrieval_docum
 		self.model_name = model
 		self.task_type = task_type
 
+	def _prompt_video(self, text, frames):
+		img_prompts = [{'mime_type':'image/png', 'data': img} for img in frames]
+		response = self.model.generate_content(img_prompts + [text])
+		return response.text
+
 	def set_system_prompt(self, text):
 		self.model = genai.GenerativeModel(self.model_name, system_instruction=text)
 
@@ -381,6 +438,10 @@ def _prompt_image(self, text, prompt_image):
 	def set_system_prompt(self, text):
 		self.system = text
 
+	def _prompt_video(self, text, frames):
+		extra_content = [{"type":"image", "source": {"data": prompt_image, "type":"base64", "media_type":"image/png" }} for prompt_image in frames]
+		return self._prompt(text, extra_content=extra_content)
+
 	def _prompt(self, prompt, temperature=1, extra_content=None):
 		prompt = self._openai_format_prompt(prompt, extra_content)
 		if self.system: