First Commit.

app.py

@@ -0,0 +1,150 @@
+import os
+import io
+from PIL import Image
+import gradio as gr
+
+import aws_utils
+
+AWS_BUCKET = os.getenv("AWS_BUCKET")
+os.environ["AWS_ACCESS_KEY_ID"] = os.getenv("AWS_ACCESS_KEY_ID")
+os.environ["AWS_SECRET_ACCESS_KEY"] = os.getenv("AWS_SECRET_ACCESS_KEY")
+os.environ["S3_BUCKET_NAME"] = os.getenv("AWS_BUCKET")
+
+
+def load_text_data(characters: list, current_index: int):
+    curr_char = characters[current_index]
+    return (
+        curr_char["name"],
+        curr_char["age"],
+        curr_char["gender"],
+        curr_char["description"],
+    )
+
+
+def load_data(character_data: list, current_index: int):
+    if current_index < len(character_data) - 1:
+        current_index += 1
+    else:
+        return [], *load_text_data(character_data, current_index)
+
+    images = []
+    name = character_data[current_index]["name"]
+    for idx in range(1, 5):
+        url = f"s3://{AWS_BUCKET}/{comic_id}/character_compositions/{name}/"
+        data = aws_utils.fetch_from_s3(url)
+        images.append(Image.open(io.BytesIO(data)))
+
+    return images, *load_text_data(character_data, current_index)
+
+
+def load_data_once(comic_id: str, current_index: int):
+    # Logic to load and return character images based on comic_id
+    # You can replace this with actual image paths or generation logic
+    print(f"Getting characters for comic id: {comic_id}")
+    characters = []
+    data = eval(
+        aws_utils.fetch_from_s3(
+            source=f"s3://{AWS_BUCKET}/{comic_id}/characters.json"
+        ).decode("utf-8")
+    )
+    for _, profile in data.items():
+        characters.append(profile)
+
+    images = []
+    # Loading the 0th frame of 0th scene in 0th episode.
+    name = characters[current_index]["name"]
+    for idx in range(1, 5):
+        url = f"s3://{AWS_BUCKET}/{comic_id}/character_compositions/{name}/"
+        data = aws_utils.fetch_from_s3(url)
+        images.append(Image.open(io.BytesIO(data)))
+
+    return images, *load_text_data(characters, current_index)
+
+
+def save_image(
+    selected_image,
+    comic_id: str,
+    character_data: list,
+    current_index: int,
+):
+    # Implement your AWS S3 save logic here
+    print(f"Saving image: {selected_image}")
+    name = character_data[current_index]["name"]
+    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}/characters",
+            img_bytes,
+            f"{name}.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_index = gr.State(0)
+    character_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 Character
+    with gr.Row():
+        name = gr.Textbox(label="Name", interactive=False)
+        age = gr.Textbox(label="Age", interactive=False)
+        gender = gr.Textbox(label="Gender", interactive=False)
+    description = gr.Textbox(label="description", 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_index],
+        outputs=[images, character_data, current_index, name, age, gender, description],
+    )
+
+    # 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,
+            character_data,
+            current_index,
+        ],
+        outputs=[],
+    )
+
+    next_button.click(
+        load_data,
+        inputs=[character_data, current_index],
+        outputs=[images, character_data, current_index, name, age, gender, description],
+    )
+
+demo.launch()

aws_utils.py

@@ -0,0 +1,136 @@
+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
+
+AWS_REGION = os.getenv("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

requirements.txt

@@ -0,0 +1 @@
+boto3==1.35.41