added script to upload images to minio

scripts/upload_images_to_minio.py

@@ -0,0 +1,67 @@
+"""Script to upload images to MinIO server."""
+
+from pathlib import Path
+import pandas as pd
+from uuid import uuid4
+from PIL import Image
+from dotenv import load_dotenv
+
+from data_store.repositories import ImageRepository
+
+UPLOAD_DIR = Path("/Volumes/BW-PSSD/maersk_data/images")
+
+
+def upload_images_to_minio(
+    image_repo: ImageRepository,
+    upload_dir: Path,
+) -> None:
+    """Upload images from the specified directory to MinIO server."""
+    # Check input
+    if not isinstance(image_repo, ImageRepository):
+        raise ValueError("image_repo must be an instance of ImageRepository.")
+    if not image_repo.is_connected:
+        raise ConnectionError("image_repo must be connected to MinIO server.")
+    if not isinstance(upload_dir, Path):
+        raise ValueError("upload_dir must be a pathlib.Path instance.")
+    if not upload_dir.is_dir():
+        raise NotADirectoryError(f"upload_dir must be a valid directory: {upload_dir}")
+    if not upload_dir.exists():
+        raise FileNotFoundError(f"upload_dir does not exist: {upload_dir}")
+    # Scan directory for images
+    images = list(upload_dir.glob("*.png"))
+    if not images:
+        raise FileNotFoundError(f"No PNG images found in directory: {upload_dir}")
+    print(f"Found {len(images)} images to upload.")
+    # Initialize list to store image name and id mapping
+    image_name_id_map: dict[str, str] = {}
+    # Begin uploading images
+    for image_path in images:
+        try:
+            # Load image
+            img = Image.open(image_path)
+            # Upload image to MinIO
+            img_id = uuid4()
+            image_repo.put(img, img_id)
+            print(f"Uploaded {image_path.name} with ID {img_id}.")
+        except Exception as e:
+            print(f"Failed to upload {image_path.name}: {e}")
+            continue
+        else:
+            # Store mapping
+            image_name_id_map[image_path.name] = str(img_id)
+    # Save mapping to CSV
+    df = pd.DataFrame(
+        list(image_name_id_map.items()), columns=["image_name", "image_id"]
+    )
+    csv_path = Path(__file__).parent / "image_name_id_map.csv"
+    df.to_csv(csv_path, index=False)
+
+
+if __name__ == "__main__":
+    # Load environment variables from .env file
+    load_dotenv()
+    # Instantiate ImageRepository
+    with ImageRepository() as image_repo:
+        # Upload images from the specified directory
+        upload_images_to_minio(image_repo, UPLOAD_DIR)
+    print("Image upload process completed.")