68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""Script to upload images to MinIO server."""
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
import pandas as pd
|
|
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.")
|