deploy

.gitattributes

@@ -32,3 +32,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+Opening.mp4 filter=lfs diff=lfs merge=lfs -text

Opening.mp4

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:856ea80ee310b8ef402bbdf53c426f6395e3bcf29adee5e0605b0736eea129a8
+size 1100428

app.py

@@ -0,0 +1,153 @@
+import os
+import re
+import gradio as gr
+import utils
+from utils import ChatCompletion  #my customly built helpers to hold commonly used data models
+from random import choice
+from time import sleep
+from uuid import uuid4
+chat = ChatCompletion()
+error_code = 0
+
+images = {"Narrator": "Narrator.jpg"}
+dubs = {"Narrator": "Josh"}
+males = ["Adam", "Antoni", "Arnold", "Sam"]
+females = ["Bella", "Domi", "Elli", "Rachel"]
+
+im_gen_msg = f"""Stable Diffusion is an AI art generation model similar to DALLE-2.
+Below is a list of prompts that can be used to generate images with Stable Diffusion:
+
+- pirate, concept art, deep focus, fantasy, intricate, highly detailed, digital painting, artstation, matte, sharp focus, illustration, art by magali villeneuve, chippy, ryan yee, rk post, clint cearley, daniel ljunggren, zoltan boros, gabor szikszai, howard lyon, steve argyle, winona nelson
+- a fantasy style portrait painting of rachel lane / alison brie hybrid in the style of francois boucher oil painting unreal 5 daz. rpg portrait, extremely detailed artgerm greg rutkowski alphonse mucha greg hildebrandt tim hildebrandt
+- athena, greek goddess, claudia black, art by artgerm and greg rutkowski and magali villeneuve, bronze greek armor, owl crown, d & d, fantasy, intricate, portrait, highly detailed, headshot, digital painting, trending on artstation, concept art, sharp focus, illustration
+- closeup portrait shot of a large strong female biomechanic woman in a scenic scifi environment, intricate, elegant, highly detailed, centered, digital painting, artstation, concept art, smooth, sharp focus, warframe, illustration, thomas kinkade, tomasz alen kopera, peter mohrbacher, donato giancola, leyendecker, boris vallejo
+
+Follow the structure of the example prompts."""
+
+
+start_msg = """Greetings, traveler from far-off worlds. I am Korkut, your humble guide through the boundless expanse of the universe. In this game, we shall embark on a journey like no other, delving deep into the mysteries of a sci-fi realm filled with endless wonders and unfathomable dangers.
+As the Game Master, it is my honor to lead you on this epic quest, where we shall explore uncharted territories, encounter strange creatures and civilizations, and uncover hidden treasures beyond imagination.
+But be warned, for the challenges that lie ahead are not for the faint of heart. Only the bravest and most cunning among us will survive. We shall be tested at every turn, and our fates will be decided by the choices we make.
+So, dear friend, I ask you to take up your sword, your laser gun, or whatever weapon you prefer, and prepare to face the unknown. But first, let us know each other. What is your name, brave adventurer?"""
+
+
+def chat_handler(hist):
+    global error_code
+    reply = chat.single_response(hist)
+
+    if reply == "" or reply == -1:
+        raise gr.Error("Chatbot response is empty. Please reload the page. If the issue repeats please contact me.")
+    sleep(20) #TODO REMOVE/REDUCE    
+
+    hist.append({"role": "assistant", "content": reply})
+    speaker = re.findall(r"\[([A-Za-z0-9\s]+)\]:", reply)
+    dialog = re.findall(r"\[[A-Za-z\s0-9]+\]:([^\[]*)", reply)
+    results = list(zip(speaker, dialog))
+    speaker_set = set(speaker).difference(set(images))
+    if(len(speaker_set)>=1):
+        hists = []
+        for speaker in speaker_set:
+            gender = chat.decide_gender(speaker)
+            if gender == "male":
+                dub = choice(males)
+                dubs[speaker] = dub
+            else:
+                dub = choice(females)
+                dubs[speaker] = dub
+
+            tmp = []
+            tmp.append({"role": "assistant", "content": reply})
+            tmp.append({"role": "user", "content": im_gen_msg + f" Generate a prompt for {speaker}'s appearance. If you don't have enough information then generate an arbitrary prompt. Keep the prompt shorter than 70 words."})
+            hists.append(tmp)
+            tmp = []
+
+        replies = chat.multi_response(hists)
+        if -1 in replies:
+            raise gr.Error("Image generation failed. Please reload the page. If the issue repeats please contact me.")
+        images.update(dict(zip(speaker_set, replies)))
+
+    for speaker in speaker_set:
+        reply = images[speaker]
+        img_path = chat.generate_image(reply, speaker)
+        images[speaker] = img_path
+
+    out_clip = []
+    text = ""
+    for result in results:
+        if result[1].strip()=="": continue
+        text += "**" + result[0] + ":** " + result[1].strip() + "\n"
+        dub = dubs[result[0]]
+        audio_name = utils.generate_audio(dub, result[1].strip())
+        if audio_name == -1:
+            error_code = -1
+            raise gr.Error("Audio generation failed.Please contact me.")
+        out_clip.append([result[0], audio_name, images[result[0]]])
+    out_path = "video" + str(uuid4()) + ".mp4"
+    try:
+        utils.generate_video(out_clip, out_path)
+    except Exception as e:
+        error_code = -1
+        print("Error:" + str(e))
+        raise gr.Error("Video generation failed. Please contact me.")
+    return out_path, text, hist
+
+
+def btn_1(text,history, hist):
+    if error_code == -1:
+        return {button1: gr.update(interactive=False), button2: gr.update(visible=False),textbox: gr.update(value="System error! Please contant me."), chatbot: gr.update(value=history), state: hist}
+    
+    user_name = utils.get_user_name(chat, text)
+    
+    if user_name == -1:
+        return {button1: gr.update(interactive=False), button2: gr.update(visible=False),textbox: gr.update(value="This name is not appropriate! Please restart the game and choose an appropriate name."), chatbot: gr.update(value=history), state: hist}
+    if user_name == -2:
+        return {button1: gr.update(interactive=False), button2: gr.update(visible=False),textbox: gr.update(value="This name is invalid! Please restart the game and choose a valid name."), chatbot: gr.update(value=history), state: hist}
+    
+    history += [(None,text)]
+    system_message = f"""You are a game master. You are going to create a DND game takes place in sci-fi setting. You use a mysterious, exciting and interesting language. The story you are telling must be compelling. Since I am the only one playing with you, you need to refer me as "you" when you are talking about my character's actions. You should use a lot dialogs in your story. Other characters can refer me as {user_name}. Dialogs should be formatted like 
+    "[Person1 Name]: Speech
+    [Person2 Name]: Speech"
+    Also when there is no dialog and you are narrating story the you should output like "[Narrator]: Story".
+    Overall an example output should look like the following 
+    "[Narrator]: You wake up in a deserted planet.
+    [Stranger]: Wake up! Who are you?
+    [Narrator]: What are you going to do?"
+    You mustn't generate my dialogs. Instead at the end of your message you should ask what I will do or say and your next message you should shape the story according to my response."""
+    hist = [{"role": "system", "content": system_message}, {"role": "user", "content": "Start"}]
+    try:
+        out_path, text, hist = chat_handler(hist)
+    except gr.Error as e:
+        return {button1: gr.update(interactive=False), button2: gr.update(visible=False),textbox: gr.update(value=str(e)), chatbot: gr.update(value=history), state: hist}
+    
+    history += [((out_path,None),None),(text,None)]
+    return {button1: gr.update(visible=False), button2: gr.update(visible=True), textbox: gr.update(label="What will you do?",value=""), chatbot: gr.update(value=history), state: hist}
+
+def btn_2(text, history, hist):
+    if error_code == -1:
+        return {button2: gr.update(interactive=False), textbox: gr.update(value="System error! Please contant me."), chatbot: gr.update(value=history), state:hist}
+
+    if not chat.safety_check(text):
+        return {button2: gr.update(interactive=False), textbox: gr.update(value="This message is inappropriate! Please restart the game and use an appropriate language."), chatbot: gr.update(value=history), state:hist}
+
+    hist.append({"role": "user", "content": text})
+
+    try:
+        out_path, text, hist = chat_handler(hist)
+    except gr.Error as e:
+        return {button2: gr.update(interactive=False), textbox: gr.update(value=str(e)), chatbot: gr.update(value=history), state: hist}
+    
+    history += [((out_path,None),None),(text,None)]
+    return {textbox: gr.update(value=""), chatbot: gr.update(value=history), state: hist}
+
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    state = gr.State([])
+    gr.Markdown("# KORKUT: A Spacetime Odyssey")
+    history = [(("./Opening.mp4",None),None),(start_msg, None)]
+    chatbot = gr.Chatbot(history, elem_id="chatbot", starts_with="bot").style(height=600)
+    textbox = gr.Textbox(lines=3, label="What is your name, brave adventurer?")
+    button1 = gr.Button(value="Submit")
+    button2 = gr.Button(value="Submit", visible=False)
+    button2.click(btn_2, [textbox, chatbot, state],[textbox, chatbot, state, button2])
+    button1.click(btn_1 ,[textbox, chatbot, state],[button1, button2, textbox, chatbot, state])
+demo.queue().launch(share=True)
+

requirements.txt

@@ -0,0 +1,8 @@
+openai
+dotenv
+httpx
+asyncio
+aiometer
+elevenlabs
+moviepy
+uuid

utils.py

@@ -0,0 +1,165 @@
+# pip install python-dotenv   first install this package to be able to call custom env variables from .env
+# don't forget to call the load_dotenv() function to initialize the getenv() method of os module 
+
+import openai
+import os
+from dotenv import load_dotenv
+import httpx
+import asyncio
+import aiometer
+from functools import partial
+from elevenlabs import generate, save, set_api_key
+from base64 import b64decode
+import re
+from moviepy.editor import ImageClip, AudioFileClip, CompositeVideoClip, concatenate_videoclips, concatenate_audioclips, TextClip, CompositeAudioClip
+from random import choice
+from uuid import uuid4
+
+TIMEOUT = 300
+RATE_LIMIT = 0.05 # x requests per minute
+
+# Load environment variables from the .env file
+load_dotenv()
+
+set_api_key(os.getenv("elevenlabs_api_key"))
+openai.api_key = os.getenv("openai_api_key")
+
+class ChatCompletion:
+    def __init__(self, temperature=0.8,):
+        self.model = "gpt-3.5-turbo",
+        self.temperature = temperature
+        self.total_tokens = 0
+
+
+    def single_response(self, hist):
+        response = openai.ChatCompletion.create(
+            model= "gpt-3.5-turbo",
+            messages=hist)
+        try:
+            print("Tokens used: " + str(response["usage"]["total_tokens"]))
+            self.total_tokens += response["usage"]["total_tokens"]
+        except:
+            print("Error: " + str(response["error"]["message"]))
+            return -1
+        return response["choices"][0]["message"]["content"]
+
+    async def _async_response(self,payload):
+        async with httpx.AsyncClient() as client:
+            return await client.post(
+                url="https://api.openai.com/v1/chat/completions",
+                json=payload,
+                headers={"content_type": "application/json", "Authorization": f"Bearer {openai.api_key}"},
+                timeout=TIMEOUT,
+            )
+
+    async def _request(self, hist):
+        response = await self._async_response({
+            "model": "gpt-3.5-turbo",
+            "messages": hist,
+        })
+        try:
+            print("Tokens used: " + str(response.json()["usage"]["total_tokens"]))
+            self.total_tokens += response.json()["usage"]["total_tokens"]
+            reply = response.json()["choices"][0]["message"]["content"]
+            return reply
+        except:
+            print("Error: " + str(response.json()["error"]["message"]))
+            return -1
+
+    async def _multi_response(self, hists):
+        return await aiometer.run_all(
+            [partial(self._request, hist) for hist in hists],
+            max_per_second = RATE_LIMIT
+        )
+
+    def multi_response(self, hists):
+        return asyncio.run(self._multi_response(hists))
+
+    def safety_check(self, message):
+        if len(message) > 2000:
+            return False
+        else:
+            return True
+        # else:
+        #     text = f"""Just answer with "yes" or "no". Is the following message appropriate in DND game context?
+            
+        #     {message}"""
+        #     hist = [{"role": "user", "content": text}]
+        #     response = self.single_response(hist).lower()
+        #     if(response=="no." or response=="no"):
+        #         return False
+        #     else:
+        #         return True
+    
+    def decide_gender(self, message):
+        return choice(["male","female"])
+        # text = f"""Only reply with "male" or "female". Select a gender for {message}. If unknown or both just arbitrarily select one gender."""
+        # hist = [{"role": "user", "content": text}]
+        # response = self.single_response(hist).lower()
+        # match = re.search(r"female", response)
+        # if match:
+        #     return "female"
+        # return "male"
+
+    def generate_image(self, desc, speaker):
+        response = openai.Image.create(
+            prompt=desc,
+            n=1,
+            size="256x256",
+            response_format = "b64_json"
+        )
+        image_b64 = response["data"][0]["b64_json"]
+        with open(f"{speaker}.png","wb") as img:
+            img.write(b64decode(image_b64))
+        return f"{speaker}.png"
+    
+def _str_check(message):
+    unwanted = re.findall(r"[^A-Za-z\s0-9]", message)
+    if len(unwanted) > 0:
+        return False
+    return True
+
+def generate_audio(speaker, message):
+    try:
+        audio = generate(
+            text=message,
+            voice=speaker,
+            model="eleven_monolingual_v1"
+        )
+    except Exception as e:
+        print("Error:" + str(e))
+        return -1
+    file_name = speaker + str(uuid4()) + ".wav"
+    save(audio, file_name)
+    return file_name
+    
+
+def get_user_name(chat, user_name):
+    if not chat.safety_check(f"My name is {user_name}"):
+        print("Inappropriate name.")
+        return -1
+    if not _str_check(user_name):
+        print("Invalid name.")
+        return -2
+    return user_name
+
+def generate_video(triples,output_path):
+    video_clips = []
+    audio_clips = []
+    for _, audio_path, image_path in triples:
+        image = ImageClip(image_path)
+        audio = AudioFileClip(audio_path)
+        duration = audio.duration
+        image = image.set_duration(duration)
+        #txt_clip = TextClip(text, fontsize=24, color='white', stroke_width=3).set_pos(('left', 'top'))
+        video = CompositeVideoClip([image])#, txt_clip])
+        video = video.set_audio(audio)
+        video_clips.append(video)
+        audio_clips.append(audio)
+
+    final_video = concatenate_videoclips(video_clips, method="compose")
+    final_audio = concatenate_audioclips(audio_clips)
+    final_video = final_video.set_audio(final_audio)
+    final_video.write_videofile(output_path, fps=24, verbose=False, logger=None)
+    for _, audio_path, _ in triples:
+        os.remove(audio_path)