feat: entirely remove PIL from this project

Author: anna-singleton-resolver

examples/utils/image_generation.py

@@ -1,31 +1,26 @@
 """Ultra-fast random image creation utilities for maximum throughput."""
 
 import asyncio
-import io
 import random
 import time
 from collections.abc import AsyncIterator
 
-from PIL import Image, ImageDraw
+import cv2 as cv
+import numpy as np
 
 from resolver_athena_client.client.models import ImageData
 
 # Global cache for reusable objects and constants
-_image_cache: dict[
-    tuple[int, int], tuple[Image.Image, ImageDraw.ImageDraw]
-] = {}
+_image_cache: dict[tuple[int, int], np.ndarray] = {}
 _rng = random.Random()  # noqa: S311 - Not used for cryptographic purposes
 
 
-def _get_cached_image(
-    width: int, height: int
-) -> tuple[Image.Image, ImageDraw.ImageDraw]:
-    """Get cached image and draw objects, creating if needed."""
+def _get_cached_image(width: int, height: int) -> np.ndarray:
+    """Get cached image array, creating if needed."""
     key = (width, height)
     if key not in _image_cache:
-        img = Image.new("RGB", (width, height), (0, 0, 0))
-        draw = ImageDraw.Draw(img)
-        _image_cache[key] = (img, draw)
+        img = np.zeros((height, width, 3), dtype=np.uint8)
+        _image_cache[key] = img
     return _image_cache[key]
 
 
@@ -45,8 +40,9 @@ def create_random_image(
         PNG image bytes
 
     """
-    # Get cached image and draw objects
-    image, draw = _get_cached_image(width, height)
+    # Get cached image array
+    image = _get_cached_image(width, height)
+    img = image.copy()
 
     # Random background color
     bg_r, bg_g, bg_b = (
@@ -56,21 +52,24 @@ def create_random_image(
     )
 
     # Fill with background color
-    draw.rectangle([0, 0, width, height], fill=(bg_r, bg_g, bg_b))
+    img[:, :] = (bg_b, bg_g, bg_r)  # OpenCV uses BGR
 
     # Add single accent rectangle for visual variation
-    accent_color = (255 - bg_r, 255 - bg_g, 255 - bg_b)
+    accent_color = (255 - bg_b, 255 - bg_g, 255 - bg_r)  # BGR
     x1, y1 = width // 4, height // 4
     x2, y2 = (width * 3) // 4, (height * 3) // 4
-    draw.rectangle([x1, y1, x2, y2], fill=accent_color)
+    img = cv.rectangle(img, (x1, y1), (x2, y2), accent_color, thickness=-1)
 
     if img_format.upper() == "RAW_UINT8":
-        return image.tobytes()
+        return img.tobytes()
 
-    # Convert to PNG bytes
-    buffer = io.BytesIO()
-    image.save(buffer, format=img_format)
-    return buffer.getvalue()
+    # Convert to PNG/JPEG bytes
+    ext = f".{img_format.lower()}"
+    success, buf = cv.imencode(ext, img)
+    if not success:
+        err = f"Failed to encode image as {img_format}"
+        raise RuntimeError(err)
+    return buf.tobytes()
 
 
 def create_batch_images(
@@ -90,29 +89,32 @@ def create_batch_images(
 
     """
     images: list[bytes] = []
-    image, draw = _get_cached_image(width, height)
+    image = _get_cached_image(width, height)
 
     # Pre-calculate accent rectangle coordinates
     x1, y1 = width // 4, height // 4
     x2, y2 = (width * 3) // 4, (height * 3) // 4
 
     for _ in range(count):
+        img = image.copy()
         # Random background
         bg_r, bg_g, bg_b = (
             _rng.randint(0, 255),
             _rng.randint(0, 255),
             _rng.randint(0, 255),
         )
-        draw.rectangle([0, 0, width, height], fill=(bg_r, bg_g, bg_b))
+        img[:, :] = (bg_b, bg_g, bg_r)  # OpenCV uses BGR
 
         # Complement accent color
-        accent_color = (255 - bg_r, 255 - bg_g, 255 - bg_b)
-        draw.rectangle([x1, y1, x2, y2], fill=accent_color)
+        accent_color = (255 - bg_b, 255 - bg_g, 255 - bg_r)  # BGR
+        img = cv.rectangle(img, (x1, y1), (x2, y2), accent_color, thickness=-1)
 
         # Convert to PNG bytes
-        buffer = io.BytesIO()
-        image.save(buffer, format="PNG")
-        images.append(buffer.getvalue())
+        success, buf = cv.imencode(".png", img)
+        if not success:
+            msg = "Failed to encode image as PNG"
+            raise RuntimeError(msg)
+        images.append(buf.tobytes())
 
     return images
 

pyproject.toml

@@ -27,8 +27,7 @@ dependencies = [
     "grpcio-tools>=1.74.0",
     "httpx>=0.25.0",
     "numpy>=2.2.6",
-    "opencv-python-headless>=4.13.0.92",
-    "pillow>=11.3.0",
+    "opencv-python-headless>=4.13.0.92"
 ]
 
 [project.optional-dependencies]

tests/client/transformers/test_core.py

@@ -1,9 +1,8 @@
 """Test core transformation functions."""
 
-from io import BytesIO
-
+import cv2 as cv
+import numpy as np
 import pytest
-from PIL import Image
 
 from resolver_athena_client.client.consts import (
     EXPECTED_HEIGHT,
@@ -19,11 +18,24 @@
 def create_test_image(
     width: int = 100, height: int = 100, mode: str = "RGB"
 ) -> bytes:
-    """Create a test image with specified dimensions."""
-    img = Image.new(mode, (width, height), color="red")
-    img_bytes = BytesIO()
-    img.save(img_bytes, format="PNG")
-    return img_bytes.getvalue()
+    """Create a test image with specified dimensions using OpenCV."""
+    # Map mode to OpenCV color shape
+    if mode == "RGB":
+        color = (255, 0, 0)  # Red in RGB
+        img = np.full((height, width, 3), color, dtype=np.uint8)
+    elif mode == "L":
+        color = 76  # Red in grayscale
+        img = np.full((height, width), color, dtype=np.uint8)
+    else:
+        err = f"Unsupported mode: {mode}"
+        raise ValueError(err)
+
+    success, buf = cv.imencode(".png", img)
+    if not success:
+        err = "Failed to encode image to PNG"
+        raise RuntimeError(err)
+
+    return buf.tobytes()
 
 
 @pytest.mark.asyncio

tests/client/transformers/test_hash_pipeline.py

@@ -1,10 +1,10 @@
 """Tests for hash list behavior throughout the transformation pipeline."""
 
 import hashlib
-from io import BytesIO
 
+import cv2
+import numpy as np
 import pytest
-from PIL import Image
 
 from resolver_athena_client.client.consts import EXPECTED_HEIGHT, EXPECTED_WIDTH
 from resolver_athena_client.client.models import ImageData
@@ -21,10 +21,17 @@
 
 def create_test_png_image(width: int = 200, height: int = 200) -> bytes:
     """Create a test PNG image with specified dimensions."""
-    img = Image.new("RGB", (width, height), color=(255, 0, 0))
-    buffer = BytesIO()
-    img.save(buffer, format="PNG")
-    return buffer.getvalue()
+
+    # Create a red RGB image using numpy
+    img = np.zeros((height, width, 3), dtype=np.uint8)
+    img[:] = (255, 0, 0)  # Red color
+
+    # Encode image as PNG to memory
+    success, buffer = cv2.imencode(".png", img)
+    if not success:
+        err = "Failed to encode image as PNG"
+        raise RuntimeError(err)
+    return buffer.tobytes()
 
 
 @pytest.mark.asyncio

tests/functional/conftest.py

@@ -36,9 +36,7 @@ def _create_base_test_image_opencv(width: int, height: int) -> np.ndarray:
     # Add an accent rectangle for visual variation
     x1, y1 = width // 4, height // 4
     x2, y2 = (width * 3) // 4, (height * 3) // 4
-    cv2.rectangle(img_bgr, (x1, y1), (x2, y2), (200, 100, 50), -1)
-
-    return img_bgr
+    return cv2.rectangle(img_bgr, (x1, y1), (x2, y2), (200, 100, 50), -1)
 
 
 SUPPORTED_TEST_FORMATS = [
@@ -56,7 +54,6 @@ def _create_base_test_image_opencv(width: int, height: int) -> np.ndarray:
     "tiff",
     "pic",
     "raw_uint8",
-
     # pxm - OpenCV2 has issues with this format, the docs state its
     # supported, but pxm is also used to mean PBM/PGM/PPM which are supported,
     # so its unclear if this format is truly supported.
@@ -138,26 +135,21 @@ def valid_formatted_image(
             return f.read()
 
     # Convert format using OpenCV2 and cache to disk
-    try:
-        # Encode image in the target format
-        if image_format in ["pgm", "pbm"]:
-            # PGM and PBM are grayscale, so convert the image to grayscale
-            gray_image = cv2.cvtColor(base_image, cv2.COLOR_BGR2GRAY)
-            success, encoded = cv2.imencode(f".{image_format}", gray_image)
-        else:
-            success, encoded = cv2.imencode(f".{image_format}", base_image)
-
-        if not success:
-            pytest.fail(
-                f"OpenCV failed to encode image in {image_format} format"
-            )
-
-        image_bytes = encoded.tobytes()
-
-        # Cache the image to disk
-        with image_path.open("wb") as f:
-            _ = f.write(image_bytes)
-
-        return image_bytes
-    except Exception as e:
-        pytest.fail(f"Failed to create {image_format} image with OpenCV: {e}")
+    # Encode image in the target format
+    if image_format in ["pgm", "pbm"]:
+        # PGM and PBM are grayscale, so convert the image to grayscale
+        gray_image = cv2.cvtColor(base_image, cv2.COLOR_BGR2GRAY)
+        success, encoded = cv2.imencode(f".{image_format}", gray_image)
+    else:
+        success, encoded = cv2.imencode(f".{image_format}", base_image)
+
+    if not success:
+        pytest.fail(f"OpenCV failed to encode image in {image_format} format")
+
+    image_bytes = encoded.tobytes()
+
+    # Cache the image to disk
+    with image_path.open("wb") as f:
+        _ = f.write(image_bytes)
+
+    return image_bytes

tests/test_classify_single.py

@@ -1,12 +1,12 @@
 """Tests for the classify_single method in AthenaClient."""
 
-import io
 import uuid
 from unittest.mock import AsyncMock, Mock
 
+import cv2 as cv
 import grpc.aio
+import numpy as np
 import pytest
-from PIL import Image
 
 from resolver_athena_client.client.athena_client import AthenaClient
 from resolver_athena_client.client.athena_options import AthenaOptions
@@ -196,10 +196,11 @@ async def test_classify_single_error_handling(
     # Create a simple valid image for testing
 
     # Create a simple 1x1 pixel image
-    img = Image.new("RGB", (1, 1), color="red")
-    img_bytes = io.BytesIO()
-    img.save(img_bytes, format="PNG")
-    valid_image_data = ImageData(img_bytes.getvalue())
+    img_arr = np.ndarray((1, 1, 3), dtype=np.uint8)
+    img_arr.fill(255)
+    success, img = cv.imencode(".png", img_arr, [])
+    assert success, "Failed to encode test image"
+    valid_image_data = ImageData(img.tobytes())
 
     # Enable resizing
     athena_client.options.resize_images = True