init

gui.py

@@ -0,0 +1,75 @@
+import gradio as gr
+import voiceClone as vc
+import videoGenerate as vg
+
+model_list = []
+output_audio = None
+image_error = None
+
+# Function to create model from audio file
+def create_voice_model(title, audio_file):
+    global model_list
+    new_model = vc.create_model(audio_file, title)
+    if new_model:
+        model_list.append((title, new_model))  # Store as (title, id)
+    return f"Voice model {title} created"
+
+def update_dropdown_choices():
+  return gr.Dropdown(choices=[title for title, _ in model_list])
+
+def on_model_select(selected_item):
+    return next((model_id for title, model_id in model_list if title == selected_item), None)
+
+def generate_tts(model_id, text):
+    return vc.tts(model_id, text)
+
+def create_talking_face(audio, image, aspect_ratio, resolution, text_prompt, seed):
+    output_path = vg.generate_video(audio, image, aspect_ratio, resolution, text_prompt, seed)
+    return output_path
+
+
+# Gradio UI components
+with gr.Blocks() as app:
+    gr.Markdown(" # Deepfake Generator")
+    gr.Markdown(""" ## Voice Cloning """)
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Markdown("Generate Voice Model")
+            audio_input = gr.Audio(sources=["upload", "microphone"], label="Record Voice Sample", type="filepath", interactive = True)
+            title_input = gr.Textbox(label="Model Title", placeholder="Enter model title")
+            output_textbox = gr.Label(label="Output", value = "")
+            generate_model_button = gr.Button("Generate Voice Model")
+            generate_model_button.click(create_voice_model, inputs=[title_input, audio_input],  outputs=output_textbox)
+
+        with gr.Column(scale=1):
+            gr.Markdown("Generate TTS")
+            update_models = gr.Button("Update Models")
+            reference_id = gr.Textbox(label="Model ID", interactive=False, visible=False)
+            model_dropdown = gr.Dropdown(label="Select Model", choices=[], interactive=True)
+            model_dropdown.change(fn=on_model_select, inputs=model_dropdown, outputs= reference_id)
+            update_models.click(update_dropdown_choices, outputs=model_dropdown)
+            text_input = gr.Textbox(label="Text for TTS", placeholder="Enter text to synthesize", lines=3)
+            tts_output = gr.Audio(label="TTS Output", type="filepath", interactive=False)
+            generate_tts_button = gr.Button("Generate TTS")
+            generate_tts_button.click(generate_tts, inputs=[reference_id, text_input], outputs=tts_output)
+
+
+    gr.Markdown(""" ## Avatar Creation""")
+    with gr.Row():
+        with gr.Column(scale=1):
+            image_input = gr.Image(label="Upload Image", type="filepath", sources=["upload", "webcam"], interactive=True)
+            aspect_ratio = gr.Radio(choices= ["1:1", "16:9", "9:16"], value= "1:1", label="Aspect Ratio")
+            resolution = gr.Radio(choices= ["540p", "720p"], value= "720p", label="Resolution")
+            text_prompt = gr.Textbox(label="Text Prompt", placeholder="Enter text prompt to describe your avatar", lines=3)
+            seed = gr.Slider(minimum=1, maximum=10000, value=None, label="Optional seed for generation (integer)")
+        with gr.Column(scale=1):
+            output_video = gr.Video(label="Talking Head")
+            generate_video_button = gr.Button("Generate Talking Face Avatar")
+            generate_video_button.click(create_talking_face, inputs=[tts_output, image_input, aspect_ratio, resolution, text_prompt, seed], outputs=output_video)
+
+if __name__ == "__main__":
+    app.launch()
+
+
+
+

requirements.txt

@@ -0,0 +1,4 @@
+requests
+ffmpeg
+fish-audio-sdk
+

videoGenerate.py

@@ -0,0 +1,120 @@
+import os
+import time
+import logging
+import requests
+
+logger = logging.getLogger()
+logging.basicConfig(level=logging.INFO)
+
+api_key = os.getenv("HEDRA_API_KEY")
+
+class Session(requests.Session):
+    def __init__(self, api_key: str):
+        super().__init__()
+
+        self.base_url: str = "https://api.hedra.com/web-app/public"
+        self.headers["x-api-key"] = api_key
+
+    #@override
+    def prepare_request(self, request: requests.Request) -> requests.PreparedRequest:
+        request.url = f"{self.base_url}{request.url}"
+        return super().prepare_request(request)
+
+
+def generate_video(audio, image, aspect_ratio, resolution, text_prompt, seed):
+    global api_key
+    # Load environment variables from .env file 
+    if not api_key:
+        print("HEDRA_API_KEY not found in environment variables or .env file.")
+        return
+
+
+    # Initialize Hedra client
+    session = Session(api_key=api_key)
+
+    logger.info("testing against %s", session.base_url)
+    model_id = session.get("/models").json()[0]["id"]
+    logger.info("got model id %s", model_id)
+
+    image_response = session.post(
+        "/assets",
+        json={"name": os.path.basename(image), "type": "image"},
+    )
+    if not image_response.ok:
+        logger.error(
+            "error creating image: %d %s",
+            image_response.status_code,
+            image_response.json(),
+        )
+    image_id = image_response.json()["id"]
+    with open(image, "rb") as f:
+        session.post(f"/assets/{image_id}/upload", files={"file": f}).raise_for_status()
+    logger.info("uploaded image %s", image_id)
+
+    audio_id = session.post(
+        "/assets", json={"name": os.path.basename(audio), "type": "audio"}
+    ).json()["id"]
+    with open(audio, "rb") as f:
+        session.post(f"/assets/{audio_id}/upload", files={"file": f}).raise_for_status()
+    logger.info("uploaded audio %s", audio_id)
+
+    generation_request_data = {
+        "type": "video",
+        "ai_model_id": model_id,
+        "start_keyframe_id": image_id,
+        "audio_id": audio_id,
+        "generated_video_inputs": {
+            "text_prompt": text_prompt,
+            "resolution": resolution,
+            "aspect_ratio": aspect_ratio,
+        },
+    }
+
+    # Add optional parameters if provided
+    if seed is not None:
+        generation_request_data["generated_video_inputs"]["seed"] = seed
+
+    generation_response = session.post(
+        "/generations", json=generation_request_data
+    ).json()
+    logger.info(generation_response)
+    generation_id = generation_response["id"]
+    while True:
+        status_response = session.get(f"/generations/{generation_id}/status").json()
+        logger.info("status response %s", status_response)
+        status = status_response["status"]
+
+        # --- Check for completion or error to break the loop ---
+        if status in ["complete", "error"]:
+            break
+
+        time.sleep(5)
+
+    # --- Process final status (download or log error) ---
+    if status == "complete" and status_response.get("url"):
+        download_url = status_response["url"]
+        # Use asset_id for filename if available, otherwise use generation_id
+        output_filename_base = status_response.get("asset_id", generation_id)
+        output_filename = f"{output_filename_base}.mp4"
+        logger.info(f"Generation complete. Downloading video from {download_url} to {output_filename}")
+        try:
+            # Use a fresh requests get, not the session, as the URL is likely presigned S3
+            with requests.get(download_url, stream=True) as r:
+                r.raise_for_status() # Check if the request was successful
+                with open(output_filename, 'wb') as f:
+                    for chunk in r.iter_content(chunk_size=8192):
+                        f.write(chunk)
+            logger.info(f"Successfully downloaded video to {output_filename}")
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Failed to download video: {e}")
+        except IOError as e:
+            logger.error(f"Failed to save video file: {e}")
+    elif status == "error":
+        logger.error(f"Video generation failed: {status_response.get('error_message', 'Unknown error')}")
+    else:
+        # This case might happen if loop breaks unexpectedly or API changes
+        logger.warning(f"Video generation finished with status '{status}' but no download URL was found.")
+
+    return output_filename if 'output_filename' in locals() else None
+
+

voiceClone.py

@@ -0,0 +1,38 @@
+from fish_audio_sdk  import Session, TTSRequest
+import os
+import tempfile
+from dotenv import load_dotenv
+
+load_dotenv()
+
+fish_api_key = os.getenv("FISH_API_KEY")
+session = Session(fish_api_key)
+
+def create_model(audio_file, title:str="test"):
+    with open(audio_file, "rb") as voice:
+        print(voice)
+        model = session.create_model(
+            title=title,
+            description=" ",
+            voices=[voice.read()]
+        )
+    return model.id
+
+def tts(model_id, input_text):
+    audio_chunks = []
+    tts_request = TTSRequest(reference_id=model_id, text=input_text)
+    #print(tts_request)
+    # Collect audio chunks for TTS
+    for chunk in session.tts(tts_request):
+        audio_chunks.append(chunk)
+
+    # Write audio chunks to a temporary file to serve as output
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tts_audio:
+        for chunk in audio_chunks:
+            tts_audio.write(chunk)
+        audio_path = tts_audio.name
+
+    return audio_path
+
+
+