Initializing
Back to Projects
Year2024
DomainFullstack
AccessOpen Source
Complexity0 / 10
PythonCustomTkinterFastAPIImage ProcessingWebSocket
FullstackArchived

Aatmanova Advanced Image Processor

A dual-interface (Desktop GUI + Cloud API) image processing system with batch conversion, filters, watermarking, and real-time WebSocket monitoring.

# Aatmanova Advanced Image Processor

A comprehensive image processing system with two interfaces — a desktop GUI application and a cloud REST API. Both share the same core image processing engine with support for batch operations, format conversion, filters, watermarking, and QR code generation.

Architecture

Parsing system architecture diagram...

Tech Stack

ComponentTechnology
Desktop GUICustomTkinter (Modern Tkinter)
Web APIFastAPI + Uvicorn
Image ProcessingPillow (PIL)
Real-timeWebSocket
ConcurrencyProcessPoolExecutor (multiprocessing)
ContainerDocker + Docker Compose
File Watchingwatchdog (FolderWatcher)

Desktop GUI Features

The CustomTkinter-based desktop app provides four main tabs:

1. Process Tab (Batch Processing)

python
class App(ctk.CTk):
    def __init__(self):
        self.title("Aatmanova Pro")
        self.geometry("1100x750")
        
        # Sidebar navigation
        # - ⚔ Process: Batch processing
        # - šŸŽØ Design: Advanced filters
        # - šŸ› ļø Tools: Utilities
        # - šŸ“Š Dashboard: Statistics

Features:

  • Source/Output folder selection
  • Output format selection (JPEG, PNG, WebP, BMP, TIFF)
  • Profile presets (High Quality, Balanced, Compressed)
  • Batch processing with progress tracking
  • Queue management

2. Design Tab (Advanced Filters)

  • Brightness/Contrast adjustment
  • Saturation control
  • Blur/Sharpen filters
  • Rotation and flipping
  • Auto-enhance options

3. Tools Tab

  • QR Code generator
  • Watermark overlay
  • EXIF metadata stripper
  • Image compare tool

4. Dashboard Tab

  • Lifetime images processed
  • Storage saved (MB)
  • Recent activity log

Cloud API Features

The FastAPI server provides browser-based access:

REST Endpoints

python
@app.get("/api/settings")
async def get_settings():
    """Get current configuration"""
    return load_config()

@app.post("/api/process")
async def process_images(background_tasks: BackgroundTasks):
    """Start batch processing"""
    # Trigger processing in background

@app.post("/api/upload")
async def upload_image(file: UploadFile):
    """Upload single image for processing"""

@app.get("/api/download/{filename}")
async def download_result(filename: str):
    """Download processed image"""

WebSocket Real-time Logs

python
@app.websocket("/ws/logs")
async def websocket_logs(websocket: WebSocket):
    await websocket.accept()
    while True:
        await broadcast_log("Processing image 5 of 20...")

Usage:

javascript
const ws = new WebSocket("ws://localhost:8000/ws/logs");
ws.onmessage = (event) => {
    console.log(event.data); // "Processing image 5 of 20..."
};

Image Processing Core

python
class ImageProcessor:
    def __init__(self):
        self.supported_formats = ['.jpg', '.jpeg', '.png', '.webp', '.bmp', '.tiff']
    
    def process_image(self, input_path: str, output_path: str, settings: dict):
        with Image.open(input_path) as img:
            # 1. Resize if needed
            if settings.get('resize'):
                img = self._resize_image(img, settings)
            
            # 2. Apply filters
            if settings.get('filters'):
                img = self._apply_filters(img, settings)
            
            # 3. Add watermark
            if settings.get('watermark'):
                img = self._add_watermark(img, settings)
            
            # 4. Add QR code
            if settings.get('qr_code'):
                img = self._embed_qr(img, settings)
            
            # 5. Convert and save
            img.save(output_path, **self._get_save_params(settings))

Processing Options

FeatureParameters
Format Conversiontarget format, quality (1-100)
Resizemax_width, max_height, maintain aspect ratio
Filtersbrightness, contrast, saturation, blur, sharpen
Watermarktext, position, opacity, font size
QR Codedata, size, position
EXIF StripRemove all metadata

Parallel Processing

python
def process_batch(self, image_list: List[str], settings: dict):
    with ProcessPoolExecutor(max_workers=cpu_count()) as executor:
        futures = [
            executor.submit(self.process_image, img, output, settings)
            for img, output in image_list
        ]
        
        for future in as_completed(futures):
            result = future.result()
            yield result

Folder Watcher

Auto-process new images added to a folder:

python
class FolderWatcher:
    def __init__(self, watch_dir: str, callback):
        self.watch_dir = watch_dir
        self.observer = Observer()
        
    def start(self):
        self.observer.schedule(self._handler, self.watch_dir, recursive=False)
        self.observer.start()
        
    def _on_created(self, event):
        if event.is_directory:
            return
        # Trigger processing for new file

Docker Deployment

yaml
# docker-compose.yml
services:
  app:
    build: .
    volumes:
      - ./source:/app/source
      - ./output:/app/output
    environment:
      - SOURCE_DIR=/app/source
      - OUTPUT_DIR=/app/output
    ports:
      - "8000:8000"
dockerfile
# Dockerfile
FROM python:3.11-slim
RUN apt-get update && apt-get install -y libjpeg-dev zlib1g-dev
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "server.py"]

Configuration

json
{
  "source_dir": "/path/to/source",
  "output_dir": "/path/to/output",
  "output_format": "webp",
  "profile": "Balanced (Default)",
  "watermark": true,
  "watermark_text": "Ā© Aatmanova",
  "qr_code": false,
  "qr_data": "",
  "resize": true,
  "max_width": 1920,
  "max_height": 1080,
  "lifetime_saved_mb": 2048.5,
  "lifetime_images": 1523
}

Project Structure

code
Aatmanova Advanced Image Processor/
ā”œā”€ā”€ app.py                 # Desktop GUI (CustomTkinter)
ā”œā”€ā”€ server.py              # FastAPI web server
ā”œā”€ā”€ image_processor.py     # Core image processing
ā”œā”€ā”€ process_images.py      # Batch processing script
ā”œā”€ā”€ watcher.py             # Folder watching
ā”œā”€ā”€ config.json            # Configuration
ā”œā”€ā”€ requirements.txt       # Dependencies
ā”œā”€ā”€ Dockerfile             # Container definition
ā”œā”€ā”€ docker-compose.yml     # Docker Compose
ā”œā”€ā”€ static/                # Web UI assets
ā”œā”€ā”€ test_source/           # Test input images
└── test_output/           # Test output images

Key Design Decisions

Why Dual Interface?

  • Desktop app: For offline use, large batches, local file access
  • Cloud API: For remote access, integration with other services, mobile use

Why ProcessPoolExecutor?

Image processing is CPU-bound. Using multiprocessing instead of threading avoids Python's GIL limitation and utilizes all CPU cores for parallel batch processing.

Why CustomTkinter?

Modern look and feel with minimal code compared to PyQt. Dark mode support out of the box.

Use Cases

  1. Photographer workflows: Batch convert RAW exports to web-ready formats
  2. E-commerce: Resize product images to consistent dimensions
  3. Marketing: Add watermarks to batch images for brand consistency
  4. Archival: Convert legacy image formats to modern WebP

Architecture Feedback

Spotted a potential optimization or antipattern? Let me know.

Submit a Technical Suggestion