# 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
Tech Stack
| Component | Technology |
|---|---|
| Desktop GUI | CustomTkinter (Modern Tkinter) |
| Web API | FastAPI + Uvicorn |
| Image Processing | Pillow (PIL) |
| Real-time | WebSocket |
| Concurrency | ProcessPoolExecutor (multiprocessing) |
| Container | Docker + Docker Compose |
| File Watching | watchdog (FolderWatcher) |
Desktop GUI Features
The CustomTkinter-based desktop app provides four main tabs:
1. Process Tab (Batch Processing)
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: StatisticsFeatures:
- 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
@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
@app.websocket("/ws/logs")
async def websocket_logs(websocket: WebSocket):
await websocket.accept()
while True:
await broadcast_log("Processing image 5 of 20...")Usage:
const ws = new WebSocket("ws://localhost:8000/ws/logs");
ws.onmessage = (event) => {
console.log(event.data); // "Processing image 5 of 20..."
};Image Processing Core
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
| Feature | Parameters |
|---|---|
| Format Conversion | target format, quality (1-100) |
| Resize | max_width, max_height, maintain aspect ratio |
| Filters | brightness, contrast, saturation, blur, sharpen |
| Watermark | text, position, opacity, font size |
| QR Code | data, size, position |
| EXIF Strip | Remove all metadata |
Parallel Processing
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 resultFolder Watcher
Auto-process new images added to a folder:
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 fileDocker Deployment
# 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
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
{
"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
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 imagesKey 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
- Photographer workflows: Batch convert RAW exports to web-ready formats
- E-commerce: Resize product images to consistent dimensions
- Marketing: Add watermarks to batch images for brand consistency
- Archival: Convert legacy image formats to modern WebP
Architecture Feedback
Spotted a potential optimization or antipattern? Let me know.