Skip to main content

Fase 1: Preprocesamiento de Imágenes

Propósito​

La Fase 1 del pipeline IRIS se encarga del preprocesamiento y corrección geométrica de imágenes de documentos. Esta fase es crítica para mejorar la calidad de las imágenes antes de procesarlas con OCR.

Arquitectura del Servicio​

Componentes Principales​

Tecnologías Utilizadas​

  • FastAPI: Framework web para la API REST
  • OpenCV: Biblioteca principal para procesamiento de imágenes
  • NumPy: Operaciones matemáticas y arrays
  • Pillow: Manipulación adicional de imágenes
  • scikit-image: Algoritmos complementarios de visión computacional

Funcionalidades​

1. Detección Automática de Documentos​

El servicio utiliza algoritmos avanzados de detección de contornos para identificar automáticamente los bordes del documento en la imagen.

Algoritmo de Detección:

def detect_document_corners(image):
    """
    Detecta automáticamente las esquinas del documento en la imagen
    
    Args:
        image: Imagen de entrada (numpy array)
        
    Returns:
        corners: Array de 4 puntos representando las esquinas del documento
    """
    # Convertir a escala de grises
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    
    # Aplicar filtro Gaussiano para reducir ruido
    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
    
    # Detectar bordes usando Canny
    edges = cv2.Canny(blurred, 50, 150)
    
    # Encontrar contornos
    contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    
    # Filtrar y encontrar el contorno más grande (documento)
    largest_contour = max(contours, key=cv2.contourArea)
    
    # Aproximar el contorno a un polígono
    epsilon = 0.02 * cv2.arcLength(largest_contour, True)
    corners = cv2.approxPolyDP(largest_contour, epsilon, True)
    
    return corners.reshape(4, 2)

Características:

  • Detección robusta: Funciona con documentos en ángulos diversos
  • Tolerancia a iluminación: Maneja condiciones de luz variables
  • Múltiples formatos: Soporta documentos rectangulares y cuadrados
  • Validación automática: Verifica que se detecten exactamente 4 esquinas

2. Corrección Geométrica (Unwarping)​

Una vez detectadas las esquinas, el sistema aplica transformaciones geométricas para corregir la perspectiva y obtener una imagen "enderezada" del documento.

Transformación de Perspectiva:

def apply_perspective_correction(image, corners):
    """
    Aplica corrección de perspectiva para enderezar el documento
    
    Args:
        image: Imagen original
        corners: Esquinas detectadas del documento
        
    Returns:
        corrected_image: Imagen con perspectiva corregida
    """
    # Ordenar esquinas en sentido horario (top-left, top-right, bottom-right, bottom-left)
    ordered_corners = order_corners(corners)
    
    # Calcular dimensiones del documento corregido
    width, height = calculate_output_dimensions(ordered_corners)
    
    # Definir puntos de destino para un rectángulo perfecto
    dst_points = np.float32([
        [0, 0],
        [width, 0],
        [width, height],
        [0, height]
    ])
    
    # Calcular matriz de transformación
    matrix = cv2.getPerspectiveTransform(ordered_corners.astype(np.float32), dst_points)
    
    # Aplicar transformación
    corrected = cv2.warpPerspective(image, matrix, (width, height))
    
    return corrected

Beneficios:

  • Documentos rectangulares: Convierte documentos inclinados en rectangulares perfectos
  • Dimensiones consistentes: Normaliza el tamaño de salida
  • Mejora OCR: Incrementa significativamente la precisión del OCR posterior
  • Eliminación de distorsión: Corrige distorsiones de cámara y ángulo

3. Mejoramiento de Calidad​

Además de la corrección geométrica, la Fase 1 incluye mejoras adicionales de calidad de imagen.

Filtros Aplicados:

def enhance_image_quality(image):
    """
    Aplica filtros para mejorar la calidad de la imagen
    
    Args:
        image: Imagen corregida geométricamente
        
    Returns:
        enhanced_image: Imagen con calidad mejorada
    """
    # Conversión a escala de grises optimizada
    if len(image.shape) == 3:
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    else:
        gray = image
    
    # Ecualización de histograma adaptativa
    clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
    enhanced = clahe.apply(gray)
    
    # Filtro de reducción de ruido
    denoised = cv2.fastNlMeansDenoising(enhanced)
    
    # Sharpening opcional para texto
    kernel = np.array([[-1,-1,-1], [-1,9,-1], [-1,-1,-1]])
    sharpened = cv2.filter2D(denoised, -1, kernel)
    
    return sharpened

Mejoras Incluidas:

  • Ecualización de histograma: Mejora contraste y brillo
  • Reducción de ruido: Elimina artefactos de cámara
  • Sharpening: Realza bordes del texto para mejor OCR
  • Normalización: Estandariza niveles de intensidad

API Endpoints​

POST /process​

Procesa una imagen de documento aplicando detección automática y corrección geométrica.

Request:

curl -X POST "http://localhost:8001/process" \
     -H "Content-Type: multipart/form-data" \
     -F "file=@document.jpg" \
     -F "enhance_quality=true"

Parameters:

  • file (required): Archivo de imagen del documento
  • enhance_quality (optional): Aplicar mejoras adicionales de calidad (default: true)
  • output_format (optional): Formato de salida (jpeg, png) (default: jpeg)
  • output_quality (optional): Calidad de compresión JPEG (1-100) (default: 95)

Response:

{
    "success": true,
    "processing_time": 2.34,
    "original_dimensions": {
        "width": 1920,
        "height": 1080
    },
    "corrected_dimensions": {
        "width": 2480,
        "height": 3508
    },
    "corners_detected": [
        [245, 180],
        [1675, 195],
        [1685, 925],
        [235, 910]
    ],
    "confidence_score": 0.92,
    "output_image": "base64_encoded_image_data"
}

GET /health​

Verifica el estado del servicio.

Response:

{
    "status": "healthy",
    "service": "image-processor",
    "version": "1.0.0",
    "dependencies": {
        "opencv": "4.8.0",
        "numpy": "1.24.0"
    },
    "uptime": "2h 45m 30s"
}

GET /metrics​

Obtiene métricas de performance del servicio.

Response:

{
    "total_processed": 1247,
    "average_processing_time": 2.15,
    "success_rate": 0.978,
    "detection_accuracy": 0.945,
    "uptime_percentage": 99.8
}

Configuración​

Variables de Entorno​

# Configuración del servicio
IMAGE_PROCESSOR_PORT=8001
IMAGE_PROCESSOR_HOST=0.0.0.0
IMAGE_PROCESSOR_WORKERS=4

# Configuración de OpenCV
OPENCV_NUM_THREADS=4
OPENCV_OPENCL_ENABLED=true

# Configuración de detección
DETECTION_MIN_CONTOUR_AREA=10000
DETECTION_EPSILON_FACTOR=0.02
DETECTION_GAUSSIAN_BLUR_SIZE=5

# Configuración de corrección
CORRECTION_OUTPUT_DPI=300
CORRECTION_MAX_OUTPUT_SIZE=4096
CORRECTION_INTERPOLATION=INTER_CUBIC

# Configuración de calidad
QUALITY_ENHANCEMENT_ENABLED=true
QUALITY_CLAHE_CLIP_LIMIT=3.0
QUALITY_DENOISING_STRENGTH=10

Docker Configuration​

FROM python:3.11-slim

# Instalar dependencias del sistema para OpenCV
RUN apt-get update && apt-get install -y \
    libglib2.0-0 \
    libsm6 \
    libxext6 \
    libxrender-dev \
    libgomp1 \
    libglib2.0-0 \
    libgtk-3-0 \
    libavcodec-dev \
    libavformat-dev \
    libswscale-dev \
    && rm -rf /var/lib/apt/lists/*

# Configurar directorio de trabajo
WORKDIR /app

# Instalar dependencias Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copiar código fuente
COPY . .

# Exponer puerto
EXPOSE 8001

# Comando de inicio
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8001"]

Monitoreo y Debugging​

Logs Estructurados​

import logging
import json
from datetime import datetime

logger = logging.getLogger(__name__)

def log_processing_metrics(image_shape, processing_time, success, corners=None):
    """Log structured metrics for monitoring"""
    log_data = {
        "timestamp": datetime.utcnow().isoformat(),
        "service": "image-processor",
        "phase": "1",
        "input_dimensions": {
            "width": image_shape[1],
            "height": image_shape[0]
        },
        "processing_time_seconds": processing_time,
        "success": success,
        "corners_detected": len(corners) == 4 if corners is not None else False
    }
    
    if success:
        logger.info(f"[PHASE_1] Processing completed: {json.dumps(log_data)}")
    else:
        logger.error(f"[PHASE_1] Processing failed: {json.dumps(log_data)}")

Métricas de Performance​

  • Tiempo de procesamiento promedio: < 3 segundos por imagen
  • Tasa de detección exitosa: > 95% en condiciones normales
  • Precisión de esquinas: ± 5 píxeles en imágenes de alta resolución
  • Throughput: 20-30 imágenes por minuto por worker

Troubleshooting Common Issues​

1. Detección de Esquinas Fallida​

Síntomas:

  • No se detectan 4 esquinas
  • Contorno detectado no es rectangular

Soluciones:

# Ajustar parámetros de detección
DETECTION_EPSILON_FACTOR=0.01  # Más estricto
DETECTION_MIN_CONTOUR_AREA=5000  # Menor área mínima

# Pre-procesamiento adicional
def enhance_edge_detection(image):
    # Aplicar filtro bilateral para mantener bordes
    bilateral = cv2.bilateralFilter(image, 9, 75, 75)
    
    # Usar adaptiveThreshold para mejor detección
    thresh = cv2.adaptiveThreshold(bilateral, 255, 
                                  cv2.ADAPTIVE_THRESH_GAUSSIAN_C, 
                                  cv2.THRESH_BINARY, 11, 2)
    return thresh

2. Calidad de Imagen Degradada​

Síntomas:

  • Imagen borrosa después del procesamiento
  • Pérdida de detalles del texto

Soluciones:

# Usar interpolación de mayor calidad
CORRECTION_INTERPOLATION=cv2.INTER_LANCZOS4

# Aumentar resolución de salida
CORRECTION_OUTPUT_DPI=600

# Aplicar sharpening adaptativo
def adaptive_sharpening(image):
    # Detectar bordes para aplicar sharpening selectivo
    edges = cv2.Canny(image, 50, 150)
    kernel = np.array([[0,-1,0], [-1,5,-1], [0,-1,0]])
    sharpened = cv2.filter2D(image, -1, kernel)
    
    # Combinar usando máscara de bordes
    result = np.where(edges[..., None] > 0, sharpened, image)
    return result

Performance Optimization​

GPU Acceleration (Opcional)​

# Configuración para uso de GPU con OpenCV
import cv2

def enable_gpu_acceleration():
    """Habilitar aceleración GPU si está disponible"""
    if cv2.cuda.getCudaEnabledDeviceCount() > 0:
        # Configurar OpenCV para usar GPU
        cv2.setUseOptimized(True)
        cv2.setNumThreads(0)  # Usar todos los cores disponibles
        
        logger.info("[GPU] CUDA acceleration enabled")
        return True
    else:
        logger.info("[CPU] Using CPU-only processing")
        return False

Batch Processing​

async def process_multiple_images(images: List[bytes]) -> List[dict]:
    """
    Procesar múltiples imágenes en batch para mejor throughput
    """
    tasks = []
    for image_data in images:
        task = asyncio.create_task(process_single_image(image_data))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

Caching de Resultados​

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_process_image(image_hash: str, image_data: bytes) -> dict:
    """
    Cache de resultados para imágenes procesadas recientemente
    """
    # Procesar imagen si no está en cache
    result = process_image_internal(image_data)
    return result

def process_with_cache(image_data: bytes) -> dict:
    # Generar hash de la imagen
    image_hash = hashlib.md5(image_data).hexdigest()
    return cached_process_image(image_hash, image_data)

La Fase 1 es fundamental para el éxito del pipeline completo de IRIS, ya que proporciona la base de calidad necesaria para que las fases posteriores puedan extraer información precisa de los documentos.