Feat: multiple parameters to optimzie the model

GoogleDrive_API.py

@@ -10,7 +10,7 @@ class GoogleDrive_API:
         self.PARENT_FOLDER_ID = "1r-MlnEpWHx3b1fxHDnHcZ2-Wh_Y89676"
 
         self.service = self.authenticate(SERVICE_ACCOUNT_FILE_path)
-        self.clear_files()
+        self.delete_all_files()
 
     def authenticate(self, SERVICE_ACCOUNT_FILE):
         if len(SERVICE_ACCOUNT_FILE) > 0:
@@ -54,13 +54,16 @@ class GoogleDrive_API:
         )
         return results.get("files", [])
 
-    def clear_files(self):
+    def delete_file(self, file_id):
+        self.service.files().delete(fileId=file_id).execute()
+
+    def delete_all_files(self):
         items = self.get_files()
 
         # Delete each file
         for item in items:
             file_id = item["id"]
-            self.service.files().delete(fileId=file_id).execute()
+            self.delete_file(file_id)
 
     def upload_file(self, file_name: str, file_path: str):
         file_metadata = {
@@ -68,11 +71,7 @@ class GoogleDrive_API:
             "parents": [self.PARENT_FOLDER_ID],
         }
         media = MediaFileUpload(file_path)
-        file = (
-            self.service.files()
-            .create(body=file_metadata, media_body=media, fields="id")
-            .execute()
-        )
+        self.service.files().create(body=file_metadata, media_body=media, fields="id").execute()
         print(rf"{file_path} uploaded.")
 
     def download_file(self, file_name: str, file_path: str):
@@ -99,4 +98,4 @@ class GoogleDrive_API:
                     status, done = downloader.next_chunk()
                     print(f"Download {int(status.progress() * 100)}%.")
         else:
-            print(rf"Fine {file_name} is Not Found")
+            print(rf"{file_name} is Not Found")

Helper_functions.py

@@ -0,0 +1,86 @@
+import shutil
+import subprocess
+import os
+import json
+import pytz
+import random
+from datetime import datetime
+from PIL import Image
+
+
+def execute_terminal_command(command: str):
+    try:
+        process = subprocess.Popen(
+            command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+        )
+        output, error = process.communicate()
+        output = str(output.decode("utf-8"))
+        print(rf"Command executed successfully: {command}")
+        return output
+    except Exception as e:
+        return None, str(e)
+
+
+def correct_path(path: str):
+    return path[1:] if path.startswith("\\") else path
+
+
+def write_file(data_list: list, file_path: str, file_name: str = ""):
+    if len(file_name) > 0:
+        file_path = rf"{file_path}\{file_name}"
+
+    # Writing JSON data
+    with open(file_path, "w") as file:
+        json.dump(data_list, file)
+
+
+def read_file(file_path: str):
+    with open(file_path, 'r') as file:
+        data = json.load(file)
+    return data
+
+
+def copy_file(source_path: str, destination_path: str):
+    try:
+        shutil.copyfile(source_path, destination_path)
+    except Exception as e:
+        print(rf"Failed to copy {source_path} to {destination_path}")
+        raise e
+
+
+def read_image(image_path: str):
+    try:
+        image = Image.open(image_path)
+        return image
+    except IOError:
+        print("Unable to load image")
+        return None
+
+
+def get_current_time():
+    # Get the current datetime in UTC timezone
+    current_datetime_utc = datetime.now(pytz.utc)
+    # Convert UTC datetime to Egypt timezone
+    egypt_timezone = pytz.timezone("Africa/Cairo")
+    current_datetime_local = current_datetime_utc.astimezone(egypt_timezone)
+
+    return str(current_datetime_local.strftime("%Y-%m-%d %H:%M:%S %Z"))
+
+
+def get_random_str(sz: int):
+    result: str = ""
+    while len(result) < sz:
+        result += str(random.randint(0, 9))
+
+    return result
+
+
+def create_folder(path: str, Replace_if_exist = True):
+    try:
+        if Replace_if_exist and os.path.exists(path):
+            shutil.rmtree(path)
+
+        os.makedirs(path, exist_ok=False)
+        print(f"Folder '{path}' created successfully.")
+    except Exception as e:
+        print(f"Failed to create folder '{path}'. Error: {e}")

Kaggle_API.py

@@ -1,129 +1,77 @@
-import os
-import shutil
-import subprocess
-import json
 import time
-import pytz
-from datetime import datetime
-from PIL import Image
+
+from Helper_functions import *
 
 
 class API_Connection:
-    def __init__(self, GD_connection, kaggle_username: str = "", kaggle_key: str = ""):
+    def __init__(self, gd_connection, kaggle_username: str = "", kaggle_key: str = ""):
         os.environ["KAGGLE_USERNAME"] = kaggle_username
         os.environ["KAGGLE_KEY"] = kaggle_key
 
-        self.GoogleDrive_connection = GD_connection
+        self.GoogleDrive_connection = gd_connection
 
-        self.PROJECT_PATH = r""
-        self.NOTEBOOK_ID = "amirmoris/pix2pix-model"
+        self.NOTEBOOK_ID = "amirmorris/pix2pix-model"
         self.DATASET_NAME = "dataset"
 
-    def execute_terminal_command(self, command: str):
-        try:
-            process = subprocess.Popen(
-                command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
-            )
-            output, error = process.communicate()
-            output = str(output.decode("utf-8"))
-            error = str(error.decode("utf-8"))
-            print(rf"Command executed successfully: {command}")
-            return output
-        except Exception as e:
-            return None, str(e)
-
-    def correct_path(self, path: str):
-        return path[1:] if path.startswith("\\") else path
-
-    def create_folder(self, path: str):
-        path = self.correct_path(path)
-        if os.path.exists(path):
-            shutil.rmtree(path)
-        self.execute_terminal_command(rf"mkdir {path}")
-
     def pull_kaggle_notebook(self, notebook_path: str):
         command = rf"kaggle kernels pull {self.NOTEBOOK_ID} -p {notebook_path} -m"
-        return self.execute_terminal_command(command)
+        return execute_terminal_command(command)
 
     def push_kaggle_notebook(self, notebook_path: str):
         command = rf"kaggle kernels push -p {notebook_path}"
-        return self.execute_terminal_command(command)
+        return execute_terminal_command(command)
 
     def get_notebook_status(self):
         command = rf"kaggle kernels status {self.NOTEBOOK_ID}"
-        return self.execute_terminal_command(command)
+        return execute_terminal_command(command)
 
     def run(self, notebook_path: str):
-        notebook_path = self.correct_path(notebook_path)
         self.pull_kaggle_notebook(notebook_path)
         return self.push_kaggle_notebook(notebook_path)
 
-    def write_file(self, data: list, file_path: str, file_name: str = ""):
-        if len(file_name) > 0:
-            file_path = rf"{file_path}\{file_name}"
-
-        file_path = self.correct_path(file_path)
-        # Writing JSON data
-        with open(file_path, "w") as file:
-            for idx in range(len(data)):
-                json_string = json.dumps(data[idx]) + (
-                    "\n" if idx < len(data) - 1 else ""
-                )
-                file.write(json_string)
-
-    def read_image(self, image_path: str):
-        try:
-            image = Image.open(image_path)
-            return image
-        except IOError:
-            print("Unable to load image")
-            return None
-
-    def get_notebook_output(self, output_path: str):
-        output_path = self.correct_path(output_path)
-        command = rf"kaggle kernels output {self.NOTEBOOK_ID} -p {output_path}"
-        return self.execute_terminal_command(command)
-
-    def generate_image(
-        self,
-        input_image_name: str,
-        edit_instruction: str,
-        output_image_name: str,
-    ):
-        if len(input_image_name) == 0 or len(edit_instruction) == 0:
-            return False, rf"Missing Input"
+    def generate_image(self, input_image_name: str, edit_instruction: str, output_image_name: str,
+                    steps: int, seed: int, cfgtext: float, cfgimage: float, resolution: int
+                       ):
+
+        if len(input_image_name) == 0:
+            return False, rf"Missing Input: input_image"
+
+        if len(edit_instruction) == 0:
+            return False, rf"Missing Input: edit_instruction"
 
         if len(output_image_name) == 0:
-            return False, rf"Missing Output"
+            return False, rf"Output Error: Missing output_image path"
 
-        current_time = self.get_current_time()
+        current_time = get_current_time()
         print(rf"Start Time : {current_time}")
 
-        dataset_path = self.correct_path(rf"{self.PROJECT_PATH}\{self.DATASET_NAME}")
-        notebook_path = self.correct_path(rf"{self.PROJECT_PATH}\notebook")
+        dataset_path = correct_path(self.DATASET_NAME)
+        notebook_path = correct_path("notebook")
 
-        self.create_folder(dataset_path)
+        create_folder(dataset_path)
 
         # copy image to the dataset
-        shutil.copyfile(
-            rf"local_dataset\{input_image_name}",
-            rf"{dataset_path}\{input_image_name}",
-        )
+        copy_file(rf"local_dataset\{input_image_name}", rf"{dataset_path}\{input_image_name}")
 
         data = [
             {
                 "time": current_time,
-                "status": "IDLE",
-                "edit": edit_instruction,
+                "edit_instruction": edit_instruction,
                 "input_image_path": input_image_name,
                 "output_image_path": output_image_name,
+                "steps": steps,
+                "seed": seed,
+                "cfg-text": cfgtext,
+                "cfg-image": cfgimage,
+                "resolution": resolution
             }
         ]
 
-        self.write_file(data, dataset_path, "data.jsonl")
+        write_file(data, dataset_path, "data.json")
+
         # update dataset
         self.GoogleDrive_connection.upload_file(
-            "data.jsonl", rf"{self.DATASET_NAME}\data.jsonl"
+            "data.json", rf"{self.DATASET_NAME}\data.json"
         )
         self.GoogleDrive_connection.upload_file(
             input_image_name, rf"{self.DATASET_NAME}\{input_image_name}"
@@ -150,23 +98,13 @@ class API_Connection:
         self.GoogleDrive_connection.download_file(
             output_image_name, rf"{dataset_path}\{output_image_name}"
         )
-        output_image = self.read_image(rf"{dataset_path}\{output_image_name}")
-        # clear input and output
+        output_image = read_image(rf"{dataset_path}\{output_image_name}")
 
         if output_image is None:
             return False, "An error occured while running, no output image found"
 
         return True, output_image
 
-    def get_current_time(self):
-        # Get the current datetime in UTC timezone
-        current_datetime_utc = datetime.now(pytz.utc)
-        # Convert UTC datetime to Egypt timezone
-        egypt_timezone = pytz.timezone("Africa/Cairo")
-        current_datetime_local = current_datetime_utc.astimezone(egypt_timezone)
-
-        return str(current_datetime_local.strftime("%Y-%m-%d %H:%M:%S %Z"))
-
 
 def main():
     pass

app.py

@@ -1,30 +1,63 @@
 import os
-import random
-from PIL import Image
+import math
 import gradio as gr
 
+from Helper_functions import *
 from Kaggle_API import API_Connection
 from GoogleDrive_API import GoogleDrive_API
 
+DEFAULT_VALUES = {
+    "input_image": None,
+    "edit_instruction": "",
+    "steps": 100,
+    "randomize_seed": "Fix Seed",
+    "seed": 1371,
+    "randomize_cfg": "Fix CFG",
+    "text_cfg_scale": 7.5,
+    "image_cfg_scale": 1.5,
+    "resolution": 512,
+    "edited_image": None
+}
+
+def generate_button_clicked(*args):
+    # set kaggle-api variables
+    kaggle_username = os.environ["kaggle_username"]
+    kaggle_key = os.environ["kaggle_key"]
+
+    input_keys = list(DEFAULT_VALUES.keys())
+    values = dict(zip(input_keys, list(args)))
+
+    for key in values:
+        if values[key] is None:
+            values[key] = DEFAULT_VALUES[key]
+
+    if values["randomize_seed"]:
+        values["randomize_seed"] = random.randint(1, 100000)
+
+    if values["randomize_cfg"]:
+        values["text_cfg_scale"] = round(random.uniform(6.0, 9.0), ndigits=2)
+        values["image_cfg_scale"] = round(random.uniform(1.2, 1.8), ndigits=2)
+
+    # parameters for the model
+    input_image = values["input_image"]
+    edit_instruction = values["edit_instruction"]
+    steps = values["steps"]
+    seed = values["seed"]
+    cfgtext = values["text_cfg_scale"]
+    cfgimage = values["image_cfg_scale"]
+    resolution = 2 ** int(math.log2(values["resolution"]))
 
-def generate_button_clicked(input_image: Image = None, edit_instruction: str = ""):
     if input_image is None:
         raise gr.Error("Missing Input: input_image")
-    if edit_instruction == 0:
+    if len(edit_instruction) == 0:
         raise gr.Error("Missing Input: edit_instruction")
 
-    # set kaggle-api variables
-    kaggle_username = os.environ["kaggle_username"]
-    kaggle_key = os.environ["kaggle_key"]
-
-    GoogleDrive_connection = GoogleDrive_API()
+    GoogleDrive_connection = GoogleDrive_API("service_account.json")
     api_connection = API_Connection(GoogleDrive_connection, kaggle_username, kaggle_key)
 
-    api_connection.create_folder(rf"{api_connection.PROJECT_PATH}\local_dataset")
+    create_folder("local_dataset")
 
-    image_ID = ""
-    while len(image_ID) < 4:
-        image_ID += str(random.randint(0, 9))
+    image_ID = get_random_str(4)
 
     input_image_name = rf"input_image_{image_ID}.png"
     output_image_name = rf"output_image_{image_ID}.png"
@@ -32,22 +65,17 @@ def generate_button_clicked(input_image: Image = None, edit_instruction: str = "
     input_image.save(rf"local_dataset\{input_image_name}")
 
     status, img = api_connection.generate_image(
-        input_image_name, edit_instruction, output_image_name
+        input_image_name, edit_instruction, output_image_name,
+        steps, seed, cfgtext, cfgimage, resolution
     )
-    print(rf"Finish Time : {api_connection.get_current_time()}")
-    if status == False:
-        raise gr.Error(img)  # img represents the error
+    print(rf"End Time : {get_current_time()}")
+    if not status:
+        raise gr.Error(img)
 
     return img
 
-
 def reset_button_clicked():
-    return None, "", None
-
-
-SCOPES = ["https://www.googleapis.com/auth/drive"]
-SERVICE_ACCOUNT_FILE = "service_account.json"
-PARENT_FOLDER_ID = "1r-MlnEpWHx3b1fxHDnHcZ2-Wh_Y89676"
+    return list(DEFAULT_VALUES.values())
 
 
 def main():
@@ -69,17 +97,46 @@ def main():
                 with gr.Row():
                     reset_button = gr.Button("Reset")
 
+        with gr.Row():
+            steps = gr.Number(value=DEFAULT_VALUES["steps"], precision=0, label="Steps", interactive=True)
+            randomize_seed = gr.Radio(
+                ["Fix Seed", "Randomize Seed"],
+                value=DEFAULT_VALUES["randomize_seed"],
+                type="index",
+                show_label=False,
+                interactive=True,
+            )
+            seed = gr.Number(value=DEFAULT_VALUES["seed"], precision=0, label="Seed", interactive=True)
+            randomize_cfg = gr.Radio(
+                ["Fix CFG", "Randomize CFG"],
+                value=DEFAULT_VALUES["randomize_cfg"],
+                type="index",
+                show_label=False,
+                interactive=True,
+            )
+            text_cfg_scale = gr.Number(value=DEFAULT_VALUES["text_cfg_scale"], label=f"Text CFG", interactive=True)
+            image_cfg_scale = gr.Number(value=DEFAULT_VALUES["image_cfg_scale"], label=f"Image CFG", interactive=True)
+            resolution = gr.Number(value=DEFAULT_VALUES["resolution"], label=f"Resolution", interactive=True)
+
+
         generate_button.click(
             fn=generate_button_clicked,
             inputs=[
                 input_image,
                 instruction,
+                steps,
+                randomize_seed,
+                seed,
+                randomize_cfg,
+                text_cfg_scale,
+                image_cfg_scale,
+                resolution
             ],
             outputs=edited_image,
         )
         reset_button.click(
             fn=reset_button_clicked,
-            outputs=[input_image, instruction, edited_image],
+            outputs=[input_image, instruction, edited_image, steps, randomize_seed, seed, randomize_cfg, text_cfg_scale, image_cfg_scale, resolution],
         )
 
     # Launch Gradio interface

pix2pix-model.ipynb

@@ -1 +1 @@
-{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":8109869,"sourceType":"datasetVersion","datasetId":4790518},{"sourceId":8131022,"sourceType":"datasetVersion","datasetId":4805885}],"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!git clone \"https://github.com/timothybrooks/instruct-pix2pix/\"","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cd /kaggle/working/instruct-pix2pix","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"\\nInstalling dependencies ...\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!conda env create -f /kaggle/working/instruct-pix2pix/environment.yaml > /dev/null 2>&1","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"mkdir /root/.kaggle/","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cp -r /kaggle/input/api-connections/kaggle.json /root/.kaggle/","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cp -r /kaggle/input/api-connections/service_account.json /kaggle/working/instruct-pix2pix","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cp -r /kaggle/input/pix2pix-checkpoints/checkpoints /kaggle/working/instruct-pix2pix","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import os\nimport subprocess\nimport shutil\nimport json\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport matplotlib.image as mpimg\nfrom google.oauth2 import service_account\nfrom googleapiclient.discovery import build\nfrom googleapiclient.http import MediaFileUpload, MediaIoBaseDownload\n\n\nDATASET_PATH = \"/kaggle/working/dataset\"\n\nclass GoogleDrive_API:\n    def __init__(self):\n        self.SCOPES = [\"https://www.googleapis.com/auth/drive\"]\n        self.SERVICE_ACCOUNT_FILE = \"service_account.json\"\n        self.PARENT_FOLDER_ID = \"1r-MlnEpWHx3b1fxHDnHcZ2-Wh_Y89676\"\n        self.service = self.authenticate()\n\n    def authenticate(self):\n        credentials = service_account.Credentials.from_service_account_file(\n            self.SERVICE_ACCOUNT_FILE, scopes=self.SCOPES\n        )\n        service = build(\"drive\", \"v3\", credentials=credentials)\n\n        return service\n\n    def get_files(self):\n        # List all files in the folder\n        results = (\n            self.service.files()\n            .list(\n                q=f\"'{self.PARENT_FOLDER_ID}' in parents and trashed=false\",\n                fields=\"files(id, name)\",\n            )\n            .execute()\n        )\n        return results.get(\"files\", [])\n\n    def clear_files(self):\n        items = self.get_files()\n\n        # Delete each file\n        for item in items:\n            file_id = item[\"id\"]\n            self.service.files().delete(fileId=file_id).execute()\n\n    def upload_file(self, file_path):\n        file_metadata = {\n            \"name\": os.path.basename(file_path),\n            \"parents\": [self.PARENT_FOLDER_ID],\n        }\n        media = MediaFileUpload(file_path)\n        file = (\n            self.service.files()\n            .create(body=file_metadata, media_body=media, fields=\"id\")\n            .execute()\n        )\n        print(\"File uploaded. File ID:\", file.get(\"id\"))\n\n    def download_file(self, file_name, file_path):\n        results = (\n            self.service.files()\n            .list(\n                q=f\"'{self.PARENT_FOLDER_ID}' in parents and name='{file_name}' and trashed=false\",\n                fields=\"files(id)\",\n            )\n            .execute()\n        )\n        items = results.get(\"files\", [])\n\n        if items:\n            # Get the file ID\n            file_id = items[0][\"id\"]\n\n            request = self.service.files().get_media(fileId=file_id)\n\n            with open(file_path, \"wb\") as file:\n                downloader = MediaIoBaseDownload(file, request)\n                done = False\n                while not done:\n                    status, done = downloader.next_chunk()\n                    print(f\"Download {int(status.progress() * 100)}%.\")\n        else:\n            print(rf\"Fine {file_name} is Not Found\")\n\ndef execute_terminal_command(command: str):\n    try:\n        process = subprocess.Popen(\n            command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE\n        )\n        output, error = process.communicate()\n        output = str(output.decode(\"utf-8\"))\n        error = str(error.decode(\"utf-8\"))\n        print(rf\"Command executed successfully: {command}\")\n        return output\n    except Exception as e:\n        return None, str(e)\n        \ndef read_file(file_path: str):\n    with open(file_path, \"r\") as file:\n        data = json.load(file)\n    return data\n\ndef write_file(data: list, file_path: str, file_name: str = \"\"):\n    # Writing JSON data\n    with open(rf\"{file_path}/{file_name}\", \"w\") as file:\n        for idx in range(len(data)):\n            json_string = json.dumps(data[idx]) + (\"\\n\" if idx < len(data) - 1 else \"\")\n            file.write(json_string)\n            \ndef read_image(image_path: str):\n    try:\n        image = Image.open(image_path)\n        return image\n    except IOError:\n        print(\"Unable to load image\")\n        return None\n    \ndef generate_image(input_image: str, edit_instruction: str, output_path: str):\n    command = fr'/opt/conda/envs/ip2p/bin/python3 /kaggle/working/instruct-pix2pix/edit_cli.py --input {input_image} --output {output_path} --edit \"{edit_instruction}\"'\n    return execute_terminal_command(command)\n\n\ndef main():\n    if not os.path.exists(DATASET_PATH):\n        execute_terminal_command(rf\"mkdir {DATASET_PATH}\")\n    \n    # download dataset\n    GoogleDrive_connection = GoogleDrive_API()\n    GoogleDrive_connection.download_file(\"data.jsonl\", rf\"{DATASET_PATH}/data.jsonl\")\n    \n    # input\n    data = read_file(rf\"{DATASET_PATH}/data.jsonl\")\n    # get data\n    edit_instruction = data[\"edit\"]\n    input_image_path = data[\"input_image_path\"]\n    output_image_path = data[\"output_image_path\"]\n    \n    GoogleDrive_connection.download_file(input_image_path, rf\"{DATASET_PATH}/{input_image_path}\")\n\n    input_image = read_image(rf\"{DATASET_PATH}/{input_image_path}\")\n    # run model\n    ######################################################\n    generate_image(rf\"{DATASET_PATH}/{input_image_path}\", edit_instruction, rf\"{DATASET_PATH}/{output_image_path}\")\n    ######################################################\n\n    # update dataset\n    GoogleDrive_connection.upload_file(rf\"{DATASET_PATH}/{output_image_path}\")\n    \n    # Load the images\n    img1 = mpimg.imread(rf\"{DATASET_PATH}/{input_image_path}\")\n    img2 = mpimg.imread(rf\"{DATASET_PATH}/{output_image_path}\")\n\n    # Display the images side by side\n    plt.figure(figsize=(10, 5))\n    plt.subplot(1, 2, 1)\n    plt.imshow(img1)\n    plt.axis('off')\n    plt.title('Input Image')\n\n    plt.subplot(1, 2, 2)\n    plt.imshow(img2)\n    plt.axis('off')\n    plt.title(rf\"Edited Image: {edit_instruction}\")\n        \n    plt.show()\n        \nif __name__ == \"__main__\":\n    main()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}
+{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":8109869,"sourceType":"datasetVersion","datasetId":4790518},{"sourceId":8149113,"sourceType":"datasetVersion","datasetId":4805885}],"dockerImageVersionId":30683,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!git clone \"https://github.com/timothybrooks/instruct-pix2pix/\"","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cd /kaggle/working/instruct-pix2pix","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"\\nInstalling Dependencies ...\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!conda env create -f /kaggle/working/instruct-pix2pix/environment.yaml > /dev/null 2>&1","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"\\nDependencies Installed\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"mkdir /root/.kaggle/","metadata":{"execution":{"iopub.status.busy":"2024-04-17T21:42:47.289791Z","iopub.execute_input":"2024-04-17T21:42:47.290713Z","iopub.status.idle":"2024-04-17T21:42:48.315343Z","shell.execute_reply.started":"2024-04-17T21:42:47.290662Z","shell.execute_reply":"2024-04-17T21:42:48.313748Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cp -r /kaggle/input/api-connections/kaggle2.json /root/.kaggle/kaggle.json","metadata":{"execution":{"iopub.status.busy":"2024-04-17T21:42:49.638661Z","iopub.execute_input":"2024-04-17T21:42:49.639066Z","iopub.status.idle":"2024-04-17T21:42:50.73111Z","shell.execute_reply.started":"2024-04-17T21:42:49.639033Z","shell.execute_reply":"2024-04-17T21:42:50.729935Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cp -r /kaggle/input/api-connections/service_account.json /kaggle/working/instruct-pix2pix","metadata":{"execution":{"iopub.status.busy":"2024-04-17T21:42:51.962757Z","iopub.execute_input":"2024-04-17T21:42:51.963135Z","iopub.status.idle":"2024-04-17T21:42:52.986254Z","shell.execute_reply.started":"2024-04-17T21:42:51.963102Z","shell.execute_reply":"2024-04-17T21:42:52.984997Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cp -r /kaggle/input/pix2pix-checkpoints/checkpoints /kaggle/working/instruct-pix2pix","metadata":{"execution":{"iopub.status.busy":"2024-04-17T21:42:54.554271Z","iopub.execute_input":"2024-04-17T21:42:54.554668Z","iopub.status.idle":"2024-04-17T21:44:02.002039Z","shell.execute_reply.started":"2024-04-17T21:42:54.554637Z","shell.execute_reply":"2024-04-17T21:44:02.000754Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import os\nimport subprocess\nimport shutil\nimport json\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport matplotlib.image as mpimg\nfrom google.oauth2 import service_account\nfrom googleapiclient.discovery import build\nfrom googleapiclient.http import MediaFileUpload, MediaIoBaseDownload\n\n\nDATASET_PATH = \"/kaggle/working/dataset\"\n\nclass GoogleDrive_API:\n    def __init__(self):\n        self.SCOPES = [\"https://www.googleapis.com/auth/drive\"]\n        self.SERVICE_ACCOUNT_FILE = \"service_account.json\"\n        self.PARENT_FOLDER_ID = \"1r-MlnEpWHx3b1fxHDnHcZ2-Wh_Y89676\"\n        self.service = self.authenticate()\n\n    def authenticate(self):\n        credentials = service_account.Credentials.from_service_account_file(\n            self.SERVICE_ACCOUNT_FILE, scopes=self.SCOPES\n        )\n        service = build(\"drive\", \"v3\", credentials=credentials)\n\n        return service\n\n    def get_files(self):\n        # List all files in the folder\n        results = (\n            self.service.files()\n            .list(\n                q=f\"'{self.PARENT_FOLDER_ID}' in parents and trashed=false\",\n                fields=\"files(id, name)\",\n            )\n            .execute()\n        )\n        return results.get(\"files\", [])\n\n    def upload_file(self, file_path):\n        file_metadata = {\n            \"name\": os.path.basename(file_path),\n            \"parents\": [self.PARENT_FOLDER_ID],\n        }\n        media = MediaFileUpload(file_path)\n        file = (\n            self.service.files()\n            .create(body=file_metadata, media_body=media, fields=\"id\")\n            .execute()\n        )\n        print(\"File uploaded. File ID:\", file.get(\"id\"))\n\n    def download_file(self, file_name, file_path):\n        results = (\n            self.service.files()\n            .list(\n                q=f\"'{self.PARENT_FOLDER_ID}' in parents and name='{file_name}' and trashed=false\",\n                fields=\"files(id)\",\n            )\n            .execute()\n        )\n        items = results.get(\"files\", [])\n\n        if items:\n            # Get the file ID\n            file_id = items[0][\"id\"]\n\n            request = self.service.files().get_media(fileId=file_id)\n\n            with open(file_path, \"wb\") as file:\n                downloader = MediaIoBaseDownload(file, request)\n                done = False\n                while not done:\n                    status, done = downloader.next_chunk()\n                    print(f\"Download {int(status.progress() * 100)}%.\")\n        else:\n            print(rf\"{file_name} is Not Found\")\n\ndef execute_terminal_command(command: str):\n    try:\n        process = subprocess.Popen(\n            command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE\n        )\n        output, error = process.communicate()\n        output = str(output.decode(\"utf-8\"))\n        error = str(error.decode(\"utf-8\"))\n        print(rf\"Command executed successfully: {command}\")\n        return output\n    except Exception as e:\n        print(\"Command Failed !\")\n        return None, str(e)\n        \ndef read_file(file_path: str):\n    with open(file_path, 'r') as file:\n        data = json.load(file)\n    return data[0]\n\ndef write_file(data: list, file_path: str, file_name: str = \"\"):\n    # Writing JSON data\n    with open(rf\"{file_path}/{file_name}\", \"w\") as file:\n        for idx in range(len(data)):\n            json_string = json.dumps(data[idx]) + (\"\\n\" if idx < len(data) - 1 else \"\")\n            file.write(json_string)\n            \ndef read_image(image_path: str):\n    try:\n        image = Image.open(image_path)\n        return image\n    except IOError:\n        print(\"Unable to load image\")\n        return None\n    \ndef generate_image(input_image, edit_instruction, output_path, steps, seed, cfgtext, cfgimage, resolution):\n    inputs = fr'--input {input_image} --output {output_path} --edit \"{edit_instruction}\" --steps {steps} --seed {seed} --cfg-text {cfgtext} --cfg-image {cfgimage} --resolution {resolution}'\n    command = fr\"/opt/conda/envs/ip2p/bin/python3 edit_cli.py  {inputs}\"\n    return execute_terminal_command(command)\n\n\ndef main():\n    if not os.path.exists(DATASET_PATH):\n        execute_terminal_command(rf\"mkdir {DATASET_PATH}\")\n    \n    # download dataset\n    GoogleDrive_connection = GoogleDrive_API()\n    GoogleDrive_connection.download_file(\"data.json\", rf\"{DATASET_PATH}/data.json\")\n    \n    # input\n    data = read_file(rf\"{DATASET_PATH}/data.json\")\n    # get data\n    time = data[\"time\"]\n    print(fr\"Start Time: {time}\")\n    edit_instruction = data[\"edit_instruction\"]\n    input_image_path = data[\"input_image_path\"]\n    output_image_path = data[\"output_image_path\"]\n    steps = data[\"steps\"]\n    seed = data[\"seed\"]\n    cfgtext = data[\"cfg-text\"]\n    cfgimage = data[\"cfg-image\"]\n    resolution = data[\"resolution\"]\n\n    GoogleDrive_connection.download_file(input_image_path, rf\"{DATASET_PATH}/{input_image_path}\")\n\n    input_image = read_image(rf\"{DATASET_PATH}/{input_image_path}\")\n    # run model\n    ######################################################\n    generate_image(rf\"{DATASET_PATH}/{input_image_path}\", edit_instruction, rf\"{DATASET_PATH}/{output_image_path}\", steps, seed, cfgtext, cfgimage, resolution)\n    ######################################################\n    \n    # update dataset\n    GoogleDrive_connection.upload_file(rf\"{DATASET_PATH}/{output_image_path}\")\n    \n    # Load the images\n    img1 = mpimg.imread(rf\"{DATASET_PATH}/{input_image_path}\")\n    img2 = mpimg.imread(rf\"{DATASET_PATH}/{output_image_path}\")\n\n    # Display the images side by side\n    plt.figure(figsize=(10, 5))\n    plt.subplot(1, 2, 1)\n    plt.imshow(img1)\n    plt.axis('off')\n    plt.title('Input Image')\n\n    plt.subplot(1, 2, 2)\n    plt.imshow(img2)\n    plt.axis('off')\n    plt.title(rf\"Edited Image: {edit_instruction}\")\n        \n    plt.show()\n        \nif __name__ == \"__main__\":\n    main()","metadata":{"execution":{"iopub.status.busy":"2024-04-17T22:04:07.306536Z","iopub.execute_input":"2024-04-17T22:04:07.306923Z","iopub.status.idle":"2024-04-17T22:06:11.475935Z","shell.execute_reply.started":"2024-04-17T22:04:07.306896Z","shell.execute_reply":"2024-04-17T22:06:11.474759Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}

service_account.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "graduation-project-420420",
+  "private_key_id": "514ceeaa4abc9a06525ab7f5b224cb9421da5c43",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDYskZd2uJSE/wN\n6xnh90qoi93WidUNN/xOloOzwlFeWk6+5/UP1OQ4WcInnIZae2wO2q4rAZL2vnsb\n7h3joQMicoqKBKQ58qPEO0gb7yXirAQW6XmTrPnJl0EakdGTQFurQynFEsjVa9fQ\nmFjTQtMXa3X48tlb5ip94TwgnNcw3+DtHmkU8Sg5RAV1+Tl8dMUEQ7F2ZwtcggOu\nDKzRcekr89f8fSBeq6S8a4UaAnVGiy03MGk6GOVQAF7LlOPGjBs97wkH7Qt9OLMp\nYcMQ3mFZOdgpJnYrP0pM1C/LASJ4ysv9GAVWKU+cpOC2zmj17HO6Qe6i6IEF82s8\nSj6bOawDAgMBAAECggEAYfMHcpInffFRnBr6ipUEtrXuMvT/jCeT0RQg+QarUApj\nEND5iUvyUvRjmhMa+dhTKk23r3HZM9Y7AI98DDHyIEaFD8j0UMs4MlGXkSTiy5h0\nisTDMzjbeb94aqGiCTzskrfbsByDElFabowGjovkre/hlgNzs6IQ7/B5BMbQUGL0\n1bXvFW9x1n5XK8Rbd3ls74F7Qob3tSMOv7zZ2MRHv/VtzOFVgobFipwG615+LJ1G\nuYtbQ2qSYdc82s/801OLVhuUDJzaaadrPORQH6ThwX+KqJeN91egVYfG1d/nwUuF\n0EqUbuUQrGJD90iobNzz/04Xoej/7Y9NGxbvE1/uKQKBgQD5x6bxsOMIwoXD9wdd\ntq2e7yiaQZ+YCwD2iwPbkkZvVBzHBe0fd2qTdl8e7F43VjJwyARcKHZ8AncjfQeG\nezvJFL2TLPi/8XEI41ZDqrH4Cb+Dsp6m5pRqIfxdoiVH10vcdyKQOKP4KOuy4+GV\n+PEdkh1qEXzZj+i2HNLGfpCJmwKBgQDeF7cbNKi2pwcTDuzjLzIZrYEkv24oO7l5\nexAqnn3xw1281kRZlilZePPJfZidMkRAidtySaxrluHXxk2Fo9ZHu7hDI6mtZ9Kq\nxJCHeakhaW0zotJ4SQN07zNnR+6+bTgh7ZisSOKnsQUkQLJC4/pShUzjdHGHE1Mf\nz6ERv0/huQKBgQDganMh/61b93m7obzqg/DmNTYjsgYz0t8go7HpmzjRWZ8pabes\nZ9xHdTjtTa0wxq83CV5xdqFNteFY70LNnvcOKSZSjztHKSRHugmNblXL5uY860UD\n63M5ECPv9HjOXS95e0iAj4MyuYFbDsdluW7ZBKiM3f1wc42/4ZG2aYxvNQKBgQDA\nY8LW6nMMZvyV4eQLmaLI46hv8PS4HaS6O4TxpoJroFXfc62Z/Z6sRz7IWOiPpC/D\nyKg1BOU7/mR9yub00cyaL1+bqZUfBkVUcRrSgO2mZ/GVPYMdeAJqlZvt6D5E79Ak\nc/JBrJHLumk90l2Z3GvIeHj2wLRn0rG7aN2axiOo4QKBgQD0f6HGlvVsc4OoqZMk\nj+ynqAmOcKtanD36GAFtRN6Ztb/7VsyX3LJe2KS71MXhnjN/hb6ug4jROjqMPI+q\nqT6zXJgRNu2n35NYDtzg8LD9J7dYM4vTeQsKF+leFT6XRQpkOBxvbgvGjaNrTCwN\nJCmUCo24od5hKOmqiQQCEMJS5g==\n-----END PRIVATE KEY-----\n",
+  "client_email": "graduation-project@graduation-project-420420.iam.gserviceaccount.com",
+  "client_id": "106719380856521616350",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/graduation-project%40graduation-project-420420.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}