prepare.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import csv\n",
    "import os\n",
    "\n",
    "jsonl_file = \"./jsonl/multi2_out_2.jsonl\"\n",
    "csv_file = \"G:\\\\Scrapes\\\\experiment_light\\\\database_name\"\n",
    "new_path = \"G:\\\\Scrapes\\\\experiment_light\"  # New parent directory path\n",
    "\n",
    "\n",
    "# Initialize counters\n",
    "skipped_entries = 0\n",
    "created_entries = 0\n",
    "# Initialize a line counter\n",
    "jsonl_line_count = 0\n",
    "# Dictionary to store file extensions based on IDs\n",
    "extensions = {}\n",
    "\n",
    "\n",
    "# Check for extensions of files in the new_path directory\n",
    "for filename in os.listdir(new_path):\n",
    "    name, extension = os.path.splitext(filename)\n",
    "    extensions[name] = extension\n",
    "\n",
    "\n",
    "# Open JSONL file for reading and CSV file for writing\n",
    "with open(jsonl_file, \"r\") as jsonl, open(csv_file, \"w\", newline=\"\") as csvf:\n",
    "    # Create a CSV writer\n",
    "    csv_writer = csv.writer(csvf)\n",
    "\n",
    "    # Define the CSV header\n",
    "    header = [\"name\", \"path\", \"flag\", \"flag_pred\", \"label\", \"label_pred\", \"score\", \"score_pred\", \"show\"]\n",
    "    csv_writer.writerow(header)\n",
    "\n",
    "\n",
    "    for line in jsonl:\n",
    "        data = json.loads(line)\n",
    "        jsonl_line_count+=1\n",
    "        \n",
    "        # Check if \"accept\" field is empty\n",
    "        if not data[\"accept\"]:\n",
    "            skipped_entries += 1  # Increment the skipped entries count\n",
    "            continue  # Skip this entry if \"accept\" is empty\n",
    "        \n",
    "        # Extract values from the JSON data\n",
    "        path = data[\"path\"]\n",
    "        #name = os.path.splitext(os.path.basename(path))[0]  # Extract the file name without extension\n",
    "        accept = data[\"accept\"]\n",
    "        label = 1 if 1 in accept else 2  # Convert accept values to label\n",
    "       \n",
    "        # Extract the file extension from the \"path\" field\n",
    "        name = os.path.splitext(os.path.basename(path))[0]\n",
    "        \n",
    "        # Replace the entire path with the new path format\n",
    "        path = os.path.join(new_path, f\"{name}{os.path.splitext(path)[1]}\")\n",
    "       \n",
    "       \n",
    "        # Check if the ID is already in the extensions dictionary\n",
    "        if name in extensions:\n",
    "            \n",
    "            \n",
    "            \n",
    "            # Use the stored extension\n",
    "            file_extension = extensions[name]\n",
    "\n",
    "            #print(\"Name -> \",name,\"is in extensions so the file extension will be ->\",file_extension)\n",
    "        else:\n",
    "            # Extract the file extension from the path\n",
    "            file_extension = os.path.splitext(path)[1]\n",
    "            # Store the extension in the dictionary for future use\n",
    "            extensions[name] = file_extension\n",
    "        \n",
    "        # Create a new path with the specified parent directory and the same file name and extension\n",
    "        #new_file_path = os.path.join(new_path, f\"{name}{file_extension}\")\n",
    "        \n",
    "        # Update the file extension in the path\n",
    "        path = os.path.join(new_path, f\"{name}{file_extension}\")\n",
    "        print(\"check extension changes -> \",os.path.splitext(path)[1])\n",
    "\n",
    "        # Create a CSV row with the updated path\n",
    "        csv_row = [name, path, 0, 0, label, 0, 0, 0, True]\n",
    "        \n",
    "        # Write the CSV row\n",
    "        csv_writer.writerow(csv_row)\n",
    "        created_entries += 1  # Increment the created entries count\n",
    "\n",
    "print(f\"CSV conversion complete.\")\n",
    "print(f\"Skipped entries: {skipped_entries}\")\n",
    "print(f\"Created entries: {created_entries}\")\n",
    "print(f\"Total lines in the JSONL file: {jsonl_line_count}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import open_clip\n",
    "config = open_clip.get_model_config('ViT-B-32')\n",
    "dim = config['embed_dim']\n",
    "print(dim)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import open_clip\n",
    "open_clip.list_pretrained()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install open-clip-torch>=2.23.0\n",
    "!pip install timm>=0.9.8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from huggingface_hub import login\n",
    "\n",
    "login(token=\"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\",write_permission=True,add_to_git_credential=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import open_clip\n",
    "from open_clip import create_model_from_pretrained, get_tokenizer,get_model_config # works on open-clip-torch>=2.23.0, timm>=0.9.8\n",
    "\n",
    "\n",
    "model, preprocess = create_model_from_pretrained('hf-hub:timm/ViT-B-16-SigLIP-512')\n",
    "tokenizer = get_tokenizer('hf-hub:timm/ViT-B-16-SigLIP-512')\n",
    "config=get_model_config(\"ViT-B-16-SigLIP-512\")[\"embed_dim\"]\n",
    "print(config)\n",
    "#preprocess"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "CustomTextCLIP.encode_image() got an unexpected keyword argument 'batch'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[7], line 28\u001b[0m\n\u001b[0;32m     26\u001b[0m     images_tensor \u001b[38;5;241m=\u001b[39m torch\u001b[38;5;241m.\u001b[39mcat(batch_images, dim\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m0\u001b[39m)\u001b[38;5;241m.\u001b[39mto(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mcuda\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m     27\u001b[0m     \u001b[38;5;66;03m# Now you can pass 'images_tensor' to your model\u001b[39;00m\n\u001b[1;32m---> 28\u001b[0m     image_features \u001b[38;5;241m=\u001b[39m \u001b[43mmodel\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mencode_image\u001b[49m\u001b[43m(\u001b[49m\u001b[43mimages_tensor\u001b[49m\u001b[43m,\u001b[49m\u001b[43mbatch\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;241;43m2\u001b[39;49m\u001b[43m)\u001b[49m\n\u001b[0;32m     32\u001b[0m \u001b[38;5;66;03m# # with open('/home/peepa/workfiles/captionrepos/capstone/siglip-test/camera_txt', 'r') as file:\u001b[39;00m\n\u001b[0;32m     33\u001b[0m \u001b[38;5;66;03m# #     labels_list = file.read().split('\\n')\u001b[39;00m\n\u001b[0;32m     34\u001b[0m \u001b[38;5;66;03m# labels_list =['jpeg artifacts', 'image with no jpeg artifacts']\u001b[39;00m\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m     48\u001b[0m \u001b[38;5;66;03m# max_prob_label = max(zipped_list, key=lambda x: x[1])\u001b[39;00m\n\u001b[0;32m     49\u001b[0m \u001b[38;5;66;03m# print(\"Label with max probability: \", max_prob_label)\u001b[39;00m\n",
      "\u001b[1;31mTypeError\u001b[0m: CustomTextCLIP.encode_image() got an unexpected keyword argument 'batch'"
     ]
    }
   ],
   "source": [
    "\n",
    "import torch\n",
    "import torch.nn.functional as F\n",
    "from urllib.request import urlopen\n",
    "from PIL import Image\n",
    "from open_clip import create_model_from_pretrained, get_tokenizer # works on open-clip-torch>=2.23.0, timm>=0.9.8\n",
    "\n",
    "model, preprocess = create_model_from_pretrained('hf-hub:timm/ViT-B-16-SigLIP-512')\n",
    "model.to(\"cuda\")\n",
    "tokenizer = get_tokenizer('hf-hub:timm/ViT-B-16-SigLIP-512')\n",
    "\n",
    "paths = [r\"P:\\datasets\\digital_Art_hd\\base\\img_000.webp\",r\"P:\\datasets\\digital_Art_hd\\base\\img_001.webp\"]\n",
    "def load_and_preprocess_image(image_path, preprocessor):\n",
    "    try:\n",
    "        image = Image.open(image_path)\n",
    "        image = preprocessor(image).unsqueeze(0)  # Add batch dimension\n",
    "        return image\n",
    "    except Exception as e:\n",
    "        print(f\"Failed to process image {image_path}: {str(e)}\")\n",
    "        return None\n",
    "    \n",
    "images=[]\n",
    "batch_images = [load_and_preprocess_image(path, preprocess) for path in paths]\n",
    "batch_images = [img for img in batch_images if img is not None]  # Filter out None values if any\n",
    "\n",
    "if batch_images:\n",
    "    images_tensor = torch.cat(batch_images, dim=0).to(\"cuda\")\n",
    "    # Now you can pass 'images_tensor' to your model\n",
    "    image_features = model.encode_image(images_tensor,batch_size=2)\n",
    "\n",
    "\n",
    "\n",
    "# # with open('/home/peepa/workfiles/captionrepos/capstone/siglip-test/camera_txt', 'r') as file:\n",
    "# #     labels_list = file.read().split('\\n')\n",
    "# labels_list =['jpeg artifacts', 'image with no jpeg artifacts']\n",
    "# # print(labels_list)\n",
    "# text = tokenizer(labels_list, context_length=model.context_length)\n",
    "\n",
    "# with torch.no_grad(), torch.cuda.amp.autocast():\n",
    "#     image_features = model.encode_image(image)\n",
    "#     text_features = model.encode_text(text)\n",
    "#     image_features = F.normalize(image_features, dim=-1)\n",
    "#     text_features = F.normalize(text_features, dim=-1)\n",
    "\n",
    "#     text_probs = torch.sigmoid(image_features @ text_features.T * model.logit_scale.exp() + model.logit_bias)\n",
    "\n",
    "# zipped_list = list(zip(labels_list, [round(p.item(), 3) for p in text_probs[0]]))\n",
    "# print(\"Label Probabilities: \", zipped_list)\n",
    "# max_prob_label = max(zipped_list, key=lambda x: x[1])\n",
    "# print(\"Label with max probability: \", max_prob_label)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import tensorflow as tf\n",
    "import tensorflow_hub as hub\n",
    "from PIL import Image\n",
    "import io\n",
    "\n",
    "# Load the image\n",
    "image_path = \"C:\\\\Users\\\\Daniel\\\\Pictures\\\\655dda7b7c2c248222eb46e84441f531.jpg\"\n",
    "with open(image_path, \"rb\") as f:\n",
    "    image_bytes = f.read()\n",
    "\n",
    "# Load the model\n",
    "model = hub.load('https://tfhub.dev/google/vila/image/1')\n",
    "predict_fn = model.signatures['serving_default']\n",
    "\n",
    "# Make a prediction\n",
    "predictions = predict_fn(image_bytes=tf.constant(image_bytes))\n",
    "aesthetic_score = predictions['predictions']\n",
    "\n",
    "print(aesthetic_score[0])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "##########\n",
    "import tensorflow as tf\n",
    "import tensorflow_hub as hub\n",
    "\n",
    "import requests\n",
    "from PIL import Image\n",
    "from io import BytesIO\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_handle = 'https://tfhub.dev/google/vila/image/1'\n",
    "model = hub.load(model_handle)\n",
    "predict_fn = model.signatures['serving_default']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_image_from_url(img_url):\n",
    "  \"\"\"Returns an image with shape [1, height, width, num_channels].\"\"\"\n",
    "  user_agent = {'User-agent': 'Colab Sample (https://tensorflow.org)'}\n",
    "  response = requests.get(img_url, headers=user_agent)\n",
    "  image_bytes = BytesIO(response.content)\n",
    "  image = Image.open(image_bytes)\n",
    "  return image, response.content\n",
    "\n",
    "def show_image(image, title=''):\n",
    "  image_size = image.size\n",
    "  plt.imshow(image)\n",
    "  plt.axis('on')\n",
    "  plt.title(title)\n",
    "  plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "image_url = 'https://cdn.discordapp.com/attachments/756702984955625536/1159185356609028106/image.png?ex=651ef6fd&is=651da57d&hm=d1f42e95335a8e58f36cb8cc17ef84681a3b81d84db40aeebec2a0df7e18e632&' #@param {type: 'string'}\n",
    "\n",
    "image, image_bytes = load_image_from_url(image_url)\n",
    "\n",
    "show_image(image)\n",
    "prediction = predict_fn(tf.constant(image_bytes))\n",
    "print(\"predicted MOS in [0, 1]: \", prediction['predictions'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Directories for input and output\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "import os\n",
    "import shutil\n",
    "import tensorflow as tf\n",
    "import tensorflow_hub as hub\n",
    "\n",
    "def get_aesthetic_score(image_path):\n",
    "    with open(image_path, \"rb\") as f:\n",
    "        image_bytes = f.read()\n",
    "    \n",
    "    predictions = predict_fn(image_bytes=tf.constant(image_bytes))\n",
    "    score = predictions['predictions'].numpy()[0][0]\n",
    "    return score\n",
    "\n",
    "# Load the model\n",
    "model = hub.load('https://tfhub.dev/google/vila/image/1')\n",
    "predict_fn = model.signatures['serving_default']\n",
    "\n",
    "# Directories for input and output\n",
    "image_directory = \"G:\\\\Scrapes\\\\photo\"\n",
    "output_directory = \"G:\\\\Scrapes\\\\villa\"\n",
    "\n",
    "if not os.path.exists(output_directory):\n",
    "    os.makedirs(output_directory)\n",
    "\n",
    "# Define category folders and create them if they don't exist\n",
    "category_folders = {\n",
    "    1: os.path.join(output_directory, \"Category_0-0.1\"),\n",
    "    2: os.path.join(output_directory, \"Category_0.1-0.2\"),\n",
    "    3: os.path.join(output_directory, \"Category_0.2-0.3\"),\n",
    "    4: os.path.join(output_directory, \"Category_0.3-0.4\"),\n",
    "    5: os.path.join(output_directory, \"Category_0.4-0.5\"),\n",
    "    6: os.path.join(output_directory, \"Category_0.5-0.6\"),\n",
    "    7: os.path.join(output_directory, \"Category_0.6-0.7\"),\n",
    "    8: os.path.join(output_directory, \"Category_0.7-0.8\"),\n",
    "    9: os.path.join(output_directory, \"Category_0.8-0.9\"),\n",
    "    10: os.path.join(output_directory, \"Category_0.9-1\")\n",
    "}\n",
    "\n",
    "for folder in category_folders.values():\n",
    "    if not os.path.exists(folder):\n",
    "        os.makedirs(folder)\n",
    "\n",
    "# Process each image\n",
    "for image_file in os.listdir(image_directory):\n",
    "    if not image_file.lower().endswith(('png', 'jpg', 'jpeg', 'webp')):\n",
    "        continue\n",
    "\n",
    "    full_path = os.path.join(image_directory, image_file)\n",
    "    score = get_aesthetic_score(full_path)\n",
    "\n",
    "    if 0 <= score < 0.1:\n",
    "        category = 1\n",
    "    elif 0.1 <= score < 0.2:\n",
    "        category = 2\n",
    "    elif 0.2 <= score < 0.3:\n",
    "        category = 3\n",
    "    elif 0.3 <= score < 0.4:\n",
    "        category = 4\n",
    "    elif 0.4 <= score < 0.5:\n",
    "        category = 5\n",
    "    elif 0.5 <= score < 0.6:\n",
    "        category = 6\n",
    "    elif 0.6 <= score < 0.7:\n",
    "        category = 7\n",
    "    elif 0.7 <= score < 0.8:\n",
    "        category = 8\n",
    "    elif 0.8 <= score < 0.9:\n",
    "        category = 9\n",
    "    else:\n",
    "        category = 10\n",
    "\n",
    "    # Copy and rename the image\n",
    "    new_filename = f\"{os.path.splitext(image_file)[0]}_{score:.2f}{os.path.splitext(image_file)[1]}\"\n",
    "    dest_path = os.path.join(category_folders[category], new_filename)\n",
    "    shutil.copy2(full_path, dest_path)\n",
    "\n",
    "    print(f\"Copied {image_file} to {category_folders[category]} as {new_filename}\")\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.13"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}