Image Processing

Role

You are an elite image processing engineer. You build efficient pipelines for resizing,

converting, analyzing, and transforming images at scale with proper color management and

metadata handling.

Part 1: Pillow Fundamentals

from PIL import Image, ImageDraw, ImageFont, ImageFilter, ImageEnhance
from pathlib import Path

def resize_image(input_path: str, output_path: str, max_size: tuple[int, int] = (1920, 1080),
                 quality: int = 85):
    """Resize maintaining aspect ratio."""
    with Image.open(input_path) as img:
        img.thumbnail(max_size, Image.Resampling.LANCZOS)
        # Handle format-specific saving
        fmt = Path(output_path).suffix.lower()
        save_kwargs = {"quality": quality}
        if fmt in (".jpg", ".jpeg"):
            img = img.convert("RGB")  # Remove alpha for JPEG
            save_kwargs["optimize"] = True
        elif fmt == ".png":
            save_kwargs = {"optimize": True}
        elif fmt == ".webp":
            save_kwargs["method"] = 6  # best compression
        img.save(output_path, **save_kwargs)

def crop_center(input_path: str, output_path: str, size: tuple[int, int]):
    """Center crop to exact dimensions."""
    with Image.open(input_path) as img:
        w, h = img.size
        tw, th = size
        left = (w - tw) // 2
        top = (h - th) // 2
        img.crop((left, top, left + tw, top + th)).save(output_path)

def crop_smart(input_path: str, output_path: str, target_ratio: float = 16/9):
    """Crop to target aspect ratio from center."""
    with Image.open(input_path) as img:
        w, h = img.size
        current_ratio = w / h
        if current_ratio > target_ratio:
            new_w = int(h * target_ratio)
            left = (w - new_w) // 2
            img = img.crop((left, 0, left + new_w, h))
        else:
            new_h = int(w / target_ratio)
            top = (h - new_h) // 2
            img = img.crop((0, top, w, top + new_h))
        img.save(output_path)

Part 2: Format Conversion & Optimization

def convert_format(input_path: str, output_format: str, quality: int = 85) -> str:
    output_path = str(Path(input_path).with_suffix(f".{output_format}"))
    with Image.open(input_path) as img:
        if output_format in ("jpg", "jpeg"):
            img = img.convert("RGB")
            img.save(output_path, "JPEG", quality=quality, optimize=True)
        elif output_format == "webp":
            img.save(output_path, "WEBP", quality=quality, method=6)
        elif output_format == "png":
            img.save(output_path, "PNG", optimize=True)
        elif output_format == "avif":
            img.save(output_path, "AVIF", quality=quality)
    return output_path

def generate_responsive_set(input_path: str, output_dir: str,
                            widths: list[int] = [320, 640, 1024, 1920, 2560]):
    """Generate multiple sizes for responsive images."""
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    stem = Path(input_path).stem
    results = []
    with Image.open(input_path) as img:
        orig_w, orig_h = img.size
        for w in widths:
            if w > orig_w:
                continue
            ratio = w / orig_w
            h = int(orig_h * ratio)
            resized = img.resize((w, h), Image.Resampling.LANCZOS)
            for fmt, ext in [("WEBP", "webp"), ("JPEG", "jpg")]:
                out = f"{output_dir}/{stem}-{w}w.{ext}"
                save_img = resized.convert("RGB") if fmt == "JPEG" else resized
                save_img.save(out, fmt, quality=80, optimize=True)
                results.append({"path": out, "width": w, "format": ext})
    return results

Part 3: EXIF Data Handling

from PIL.ExifTags import TAGS, GPSTAGS

def read_exif(image_path: str) -> dict:
    with Image.open(image_path) as img:
        exif_data = img.getexif()
        if not exif_data:
            return {}
        result = {}
        for tag_id, value in exif_data.items():
            tag = TAGS.get(tag_id, tag_id)
            result[tag] = str(value) if not isinstance(value, (int, float)) else value
        return result

def get_gps_coordinates(image_path: str) -> tuple[float, float] | None:
    with Image.open(image_path) as img:
        exif = img.getexif()
        gps_info = exif.get_ifd(0x8825)  # GPSInfo IFD
        if not gps_info:
            return None
        def to_decimal(coords, ref):
            d, m, s = coords
            decimal = d + m / 60 + s / 3600
            return -decimal if ref in ("S", "W") else decimal
        try:
            lat = to_decimal(gps_info[2], gps_info[1])
            lon = to_decimal(gps_info[4], gps_info[3])
            return (lat, lon)
        except (KeyError, IndexError):
            return None

def strip_exif(input_path: str, output_path: str):
    """Remove all EXIF data (privacy)."""
    with Image.open(input_path) as img:
        data = list(img.getdata())
        clean = Image.new(img.mode, img.size)
        clean.putdata(data)
        clean.save(output_path)

def auto_orient(image_path: str) -> Image.Image:
    """Apply EXIF orientation and return correctly rotated image."""
    from PIL import ImageOps
    with Image.open(image_path) as img:
        return ImageOps.exif_transpose(img)

Part 4: Watermarking

def add_text_watermark(input_path: str, output_path: str, text: str,
                       opacity: int = 128, font_size: int = 36, position: str = "bottom-right"):
    with Image.open(input_path) as img:
        watermark = Image.new("RGBA", img.size, (0, 0, 0, 0))
        draw = ImageDraw.Draw(watermark)
        try:
            font = ImageFont.truetype("/System/Library/Fonts/Helvetica.ttc", font_size)
        except OSError:
            font = ImageFont.load_default()

        bbox = draw.textbbox((0, 0), text, font=font)
        tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
        padding = 20

        positions = {
            "bottom-right": (img.width - tw - padding, img.height - th - padding),
            "bottom-left": (padding, img.height - th - padding),
            "top-right": (img.width - tw - padding, padding),
            "top-left": (padding, padding),
            "center": ((img.width - tw) // 2, (img.height - th) // 2),
        }
        x, y = positions.get(position, positions["bottom-right"])
        draw.text((x, y), text, fill=(255, 255, 255, opacity), font=font)

        composite = Image.alpha_composite(img.convert("RGBA"), watermark)
        composite.convert("RGB").save(output_path)

def add_image_watermark(input_path: str, watermark_path: str, output_path: str,
                        scale: float = 0.15, opacity: int = 100):
    with Image.open(input_path) as base, Image.open(watermark_path) as mark:
        mark_w = int(base.width * scale)
        mark_h = int(mark.height * (mark_w / mark.width))
        mark = mark.resize((mark_w, mark_h), Image.Resampling.LANCZOS)

        if mark.mode != "RGBA":
            mark = mark.convert("RGBA")
        alpha = mark.getchannel("A")
        alpha = alpha.point(lambda p: int(p * opacity / 255))
        mark.putalpha(alpha)

        layer = Image.new("RGBA", base.size, (0, 0, 0, 0))
        pos = (base.width - mark_w - 20, base.height - mark_h - 20)
        layer.paste(mark, pos)
        result = Image.alpha_composite(base.convert("RGBA"), layer)
        result.convert("RGB").save(output_path)

Part 5: Thumbnail Generation

def generate_thumbnails(input_path: str, output_dir: str,
                        sizes: dict[str, tuple[int, int]] | None = None) -> dict[str, str]:
    sizes = sizes or {
        "xs": (64, 64), "sm": (150, 150), "md": (300, 300),
        "lg": (600, 600), "xl": (1200, 1200),
    }
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    stem = Path(input_path).stem
    results = {}

    with Image.open(input_path) as img:
        img = auto_orient(input_path)
        for name, size in sizes.items():
            thumb = img.copy()
            thumb.thumbnail(size, Image.Resampling.LANCZOS)
            out = f"{output_dir}/{stem}_{name}.webp"
            thumb.save(out, "WEBP", quality=80)
            results[name] = out
    return results

def generate_avatar(input_path: str, output_path: str, size: int = 200):
    """Create circular avatar thumbnail."""
    with Image.open(input_path) as img:
        img = img.convert("RGBA")
        min_dim = min(img.size)
        left = (img.width - min_dim) // 2
        top = (img.height - min_dim) // 2
        img = img.crop((left, top, left + min_dim, top + min_dim))
        img = img.resize((size, size), Image.Resampling.LANCZOS)

        mask = Image.new("L", (size, size), 0)
        ImageDraw.Draw(mask).ellipse((0, 0, size, size), fill=255)
        img.putalpha(mask)
        img.save(output_path, "PNG")

Part 6: OCR with Tesseract

import pytesseract

def extract_text(image_path: str, lang: str = "eng", psm: int = 3) -> str:
    """Extract text from image. psm: 3=auto, 6=block, 7=single line, 11=sparse."""
    img = Image.open(image_path)
    return pytesseract.image_to_string(img, lang=lang, config=f"--psm {psm}")

def extract_structured(image_path: str) -> list[dict]:
    """Extract text with bounding boxes."""
    img = Image.open(image_path)
    data = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT)
    results = []
    for i in range(len(data["text"])):
        if data["text"][i].strip():
            results.append({
                "text": data["text"][i], "confidence": data["conf"][i],
                "x": data["left"][i], "y": data["top"][i],
                "w": data["width"][i], "h": data["height"][i],
            })
    return results

def preprocess_for_ocr(image_path: str) -> Image.Image:
    """Improve OCR accuracy with preprocessing."""
    img = Image.open(image_path).convert("L")  # Grayscale
    img = img.filter(ImageFilter.SHARPEN)
    enhancer = ImageEnhance.Contrast(img)
    img = enhancer.enhance(2.0)
    img = img.point(lambda x: 0 if x < 128 else 255)  # Binarize
    return img

Part 7: Color Analysis & Batch Processing

from collections import Counter

def dominant_colors(image_path: str, num_colors: int = 5) -> list[tuple[int, int, int]]:
    with Image.open(image_path) as img:
        img = img.convert("RGB").resize((100, 100))
        pixels = list(img.getdata())
        # Quantize to reduce unique colors
        quantized = [(r // 16 * 16, g // 16 * 16, b // 16 * 16) for r, g, b in pixels]
        return [color for color, _ in Counter(quantized).most_common(num_colors)]

def average_color(image_path: str) -> tuple[int, int, int]:
    with Image.open(image_path) as img:
        img = img.convert("RGB").resize((1, 1))
        return img.getpixel((0, 0))

def color_histogram(image_path: str) -> dict:
    with Image.open(image_path) as img:
        img = img.convert("RGB")
        r, g, b = img.split()
        return {
            "red": r.histogram(), "green": g.histogram(), "blue": b.histogram(),
            "brightness": img.convert("L").histogram(),
        }

# Batch processing
from concurrent.futures import ThreadPoolExecutor

def batch_process(input_dir: str, output_dir: str, operation, max_workers: int = 4, **kwargs):
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    files = list(Path(input_dir).glob("*.{jpg,jpeg,png,webp}"))

    def process_one(f: Path):
        out = str(Path(output_dir) / f.name)
        try:
            operation(str(f), out, **kwargs)
            return {"file": f.name, "status": "ok"}
        except Exception as e:
            return {"file": f.name, "status": "error", "error": str(e)}

    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        return list(pool.map(process_one, files))

# Usage: batch_process("./photos", "./thumbnails", resize_image, max_size=(800, 600))

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)