Added gradio file.

.gitignore

@@ -0,0 +1 @@
+.env

app.py

@@ -0,0 +1,228 @@
+import os
+import io
+from PIL import Image
+import gradio as gr
+
+import aws_utils
+from dotenv import load_dotenv
+load_dotenv()
+
+AWS_BUCKET = os.environ.get("AWS_BUCKET")
+
+
+def list_current_dir(bucket_name: str, folder_path: str = "") -> list:
+    response = aws_utils.S3_CLIENT.list_objects_v2(
+        Bucket=bucket_name, Prefix=folder_path, Delimiter="/"
+    )
+
+    # Check if the bucket contains objects
+    folders = []
+    if "CommonPrefixes" in response:
+        for prefix in response["CommonPrefixes"]:
+            folders.append(prefix["Prefix"])
+    return folders
+
+
+def load_text_data(
+    episodes: list, current_episode: int, current_scene: int, current_frame: int
+):
+    curr_frame = episodes[current_episode]["scenes"][current_scene]["frames"][
+        current_frame
+    ]
+    return (
+        episodes,
+        current_episode,
+        current_scene,
+        current_frame,
+        str(current_episode + 1),
+        str(current_scene + 1),
+        str(current_frame + 1),
+        curr_frame["description"],
+        curr_frame["narration"],
+        curr_frame["audio_cue_character"],
+        curr_frame["audio_cue_text"],
+    )
+
+
+def load_data(
+    episodes: list, current_episode: int, current_scene: int, current_frame: int
+):
+    if current_frame + 1 < len(
+        episodes[current_episode]["scenes"][current_scene]["frames"]
+    ):
+        current_frame += 1
+    else:
+        if current_scene + 1 < len(episodes[current_episode]["scenes"]):
+            current_scene += 1
+            current_frame = 0
+        else:
+            if current_episode + 1 < len(episodes):
+                current_episode += 1
+                current_scene = 0
+                current_frame = 0
+            else:
+                return [], current_episode, current_scene, current_frame
+
+    images = []
+    # Loading the 0th frame of 0th scene in 0th episode.
+    for comps in episodes[current_episode]["scenes"][current_scene]["frames"][
+        current_frame
+    ]["compositions"]:
+        data = aws_utils.fetch_from_s3(comps["image"])
+        images.append(Image.open(io.BytesIO(data)))
+
+    return images, *load_text_data(
+        episodes, current_episode, current_scene, current_frame
+    )
+
+
+def load_data_once(
+    comic_id: str, current_episode: int, current_scene: int, current_frame: int
+):
+    # Logic to load and return images based on comic_id and episode
+    # You can replace this with actual image paths or generation logic
+    print(f"Getting episodes for comic id: {comic_id}")
+    episodes = {}
+    idx = 0
+    for folder in list_current_dir(AWS_BUCKET, f"{comic_id}/"):
+        if "episode" in folder:
+            json_path = f"s3://{AWS_BUCKET}/{folder}episode.json"
+            episodes[idx] = eval(
+                aws_utils.fetch_from_s3(source=json_path).decode("utf-8")
+            )
+            idx += 1
+
+    images = []
+    # Loading the 0th frame of 0th scene in 0th episode.
+    for comps in episodes[current_episode]["scenes"][current_scene]["frames"][
+        current_frame
+    ]["compositions"]:
+        data = aws_utils.fetch_from_s3(comps["image"])
+        images.append(Image.open(io.BytesIO(data)))
+
+    return images, *load_text_data(
+        episodes, current_episode, current_scene, current_frame
+    )
+
+
+def save_image(
+    selected_image,
+    comic_id: str,
+    current_episode: int,
+    current_scene: int,
+    current_frame: int,
+):
+    # Implement your AWS S3 save logic here
+    print(f"Saving image: {selected_image}")
+    with Image.open(selected_image[0]) as img:
+        # Convert and save as JPG
+        img_bytes = io.BytesIO()
+        img.convert("RGB").save(img_bytes, "JPEG")
+        img_bytes.seek(0)
+
+        aws_utils.save_to_s3(
+            AWS_BUCKET,
+            f"{comic_id}/episode-{current_episode}/images/scene-{current_scene}",
+            img_bytes,
+            f"{current_frame}.jpg",
+        )
+    print("Image saved successfully!")
+
+
+# Function to handle image selection and enable the save button
+def select_image(selected_image_index, images):
+    # Get the selected image from its index
+    selected_image = images[selected_image_index]
+    return gr.update(interactive=True), selected_image
+
+
+with gr.Blocks() as demo:
+    selected_image = gr.State(None)
+    current_episode = gr.State(0)
+    current_scene = gr.State(0)
+    current_frame = gr.State(0)
+    episodes_data = gr.State({})
+
+    with gr.Row():
+        comic_id = gr.Textbox(label="Enter Comic ID:", placeholder="Enter Comic ID")
+        load_button = gr.Button("Load Data")
+
+    images = gr.Gallery(
+        label="Select an Image", elem_id="image_select", columns=4, height=300
+    )
+
+    # Display information about current Image
+    with gr.Row():
+        episode = gr.Textbox(label="Current Episode", interactive=False)
+        scene = gr.Textbox(label="Current Scene", interactive=False)
+        frame = gr.Textbox(label="Current Frame", interactive=False)
+
+    image_description = gr.Textbox(label="Description", interactive=False)
+    narration = gr.Textbox(label="narration", interactive=False)
+    with gr.Row():
+        character = gr.Textbox(label="Character", interactive=False)
+        dialouge = gr.Textbox(label="dialouge", interactive=False)
+
+    # buttons to interact with the data
+    with gr.Row():
+        save_button = gr.Button("Save Image")
+        next_button = gr.Button("Next Image")
+
+    load_button.click(
+        load_data_once,
+        inputs=[comic_id, current_episode, current_scene, current_frame],
+        outputs=[
+            images,
+            episodes_data,
+            current_episode,
+            current_scene,
+            current_frame,
+            episode,
+            scene,
+            frame,
+            image_description,
+            narration,
+            character,
+            dialouge
+        ],
+    )
+
+    # When an image is clicked
+    images.select(
+        select_image,
+        inputs=[gr.Number(), images],
+        outputs=[save_button, selected_image],
+    )
+
+    save_button.click(
+        save_image,
+        inputs=[
+            selected_image,
+            comic_id,
+            current_episode,
+            current_scene,
+            current_frame,
+        ],
+        outputs=[],
+    )
+
+    next_button.click(
+        load_data,
+        inputs=[episodes_data, current_episode, current_scene, current_frame],
+        outputs=[
+            images,
+            episodes_data,
+            current_episode,
+            current_scene,
+            current_frame,
+            episode,
+            scene,
+            frame,
+            image_description,
+            narration,
+            character,
+            dialouge
+        ],
+    )
+
+demo.launch()

aws_utils.py

@@ -0,0 +1,138 @@
+import os
+import json
+from io import BytesIO
+from typing import Union
+from urllib.parse import urlparse
+
+import boto3
+from botocore.client import Config
+from botocore.exceptions import NoCredentialsError
+from dotenv import load_dotenv
+load_dotenv()
+
+AWS_REGION = os.environ.get("AWS_REGION")
+
+# Initialize the S3 client
+S3_CLIENT = boto3.client(
+    "s3", region_name=AWS_REGION, config=Config(signature_version="s3v4")
+)
+
+
+def save_to_s3(
+    bucket_name: str,
+    folder_name: str,
+    content: Union[str, dict, BytesIO],
+    file_name: str,
+) -> str:
+    """
+    Save a file to an S3 bucket, determining the content type based on the input type.
+
+    Args:
+        bucket_name (str): The name of the S3 bucket.
+        folder_name (str): The folder path in the S3 bucket.
+        content (Union[str, dict, BytesIO]): The content to save, can be a string, dictionary, or BytesIO.
+        file_name (str): The file name under which the content should be saved.
+
+    Returns:
+        str: The S3 URL of the uploaded file, or an error message if credentials are not available.
+    """
+    # Ensure the folder name ends with a '/'
+    # if not folder_name.endswith('/'):
+    #     folder_name += '/'
+    # Determine file name and content type based on the input
+    if isinstance(content, str):
+        file_content = content
+        content_type = "text/plain"
+    elif isinstance(content, dict):
+        file_content = json.dumps(content)
+        content_type = "application/json"
+    elif isinstance(content, BytesIO):
+        file_content = content
+        content_type = "image/jpeg"
+    else:
+        print(
+            "Invalid content type. Content must be a string, dictionary, or BytesIO."
+        )
+        raise ValueError("Content must be either a string, dictionary, or BytesIO.")
+
+    # Ensure the folder name ends with a '/'
+    s3_file_path = f"{folder_name.rstrip('/')}/{file_name}"
+
+    try:
+        # Upload the file to S3
+        S3_CLIENT.put_object(
+            Bucket=bucket_name,
+            Key=s3_file_path,
+            Body=file_content,
+            ContentType=content_type,
+        )
+        s3_url = f"s3://{bucket_name}/{s3_file_path}"
+        print(f"File successfully uploaded to {s3_url}")
+        return s3_url
+
+    except NoCredentialsError:
+        print("AWS credentials not available.")
+        return "Error: AWS credentials not available."
+
+
+def fetch_from_s3(source: Union[str, dict], region_name: str = "ap-south-1") -> bytes:
+    """
+    Fetch a file's content from S3 given a source URL or dictionary with bucket and key.
+
+    Args:
+        source (Union[str, dict]): The source S3 URL or a dictionary with 'bucket_name' and 'file_key'.
+        region_name (str): The AWS region name for the S3 client (default is 'ap-south-1').
+
+    Returns:
+        bytes: The content of the file fetched from S3.
+    """
+    print(f"Fetching file from S3. Source: {source}")
+    s3_client = boto3.client("s3", region_name=region_name)
+
+    # Parse the source depending on its type
+    if isinstance(source, str):
+        parsed_url = urlparse(source)
+        bucket_name = parsed_url.netloc.split(".")[0]
+        file_path = parsed_url.path.lstrip("/")
+    elif isinstance(source, dict):
+        bucket_name = source.get("bucket_name")
+        file_path = source.get("file_key")
+        if not bucket_name or not file_path:
+            print("Dictionary input must contain 'bucket_name' and 'file_key'.")
+            raise ValueError(
+                "Dictionary input must contain 'bucket_name' and 'file_key'."
+            )
+    else:
+        print("Source must be a string URL or a dictionary.")
+        raise ValueError("Source must be a string URL or a dictionary.")
+
+    print(f"Attempting to download from bucket: {bucket_name}, path: {file_path}")
+    try:
+        response = s3_client.get_object(Bucket=bucket_name, Key=file_path)
+        file_content = response["Body"].read()
+        print(f"File fetched successfully from {bucket_name}/{file_path}")
+        return file_content
+    except Exception as e:
+        print(f"Failed to fetch file from S3: {e}")
+        raise
+
+
+def list_s3_objects(bucket_name: str, folder_path: str = "") -> list:
+    """
+    Lists a content of the given a directory URL.
+
+    Args:
+        bucket_name (str): The name of the S3 bucket.
+        folder_name (str): The folder path in the S3 bucket.
+
+    Returns:
+        list: The list of files found inside the given directory URL.
+    """
+    response = S3_CLIENT.list_objects_v2(Bucket=bucket_name, Prefix=folder_path)
+    # Check if the bucket contains objects
+    objects = []
+    if "Contents" in response:
+        for obj in response["Contents"]:
+            objects.append(obj["Key"])
+
+    return objects