Extract Structured Data from Engineering Diagrams with dots.mocr and spaCy

                      
                     
"""
Production implementation for extracting structured data from engineering diagrams using dots.mocr and spaCy.
This implementation securely extracts, processes, and saves data from diagram images.
"""

from typing import Dict, Any, List
import os
import logging
import spacy
import requests
from dots_mocr import dots_mocr

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """Configuration class for environment variables."""
    mocr_api_key: str = os.getenv('MOCR_API_KEY')
    db_url: str = os.getenv('DATABASE_URL')

# Validate input data
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'image_url' not in data:
        raise ValueError('Missing image_url')  # Must provide image URL
    return True

# Sanitize fields
async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input data fields.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data
    """
    return {key: str(value).strip() for key, value in data.items()}

# Normalize data for processing
async def normalize_data(raw_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Normalize raw data for structured processing.
    
    Args:
        raw_data: List of raw data entries
    Returns:
        Normalized data entries
    """
    return [dict(item, normalized=True) for item in raw_data]  # Normalize flag

# Fetch data from dots.mocr
async def fetch_data(image_url: str) -> Dict[str, Any]:
    """Fetch structured data using dots.mocr API.
    
    Args:
        image_url: URL of the diagram image
    Returns:
        Extracted data from the image
    Raises:
        Exception: If API call fails
    """
    headers = {'Authorization': f'Bearer {Config.mocr_api_key}'}
    response = requests.post('https://api.dots.mocr/v1/extract', json={'url': image_url}, headers=headers)
    if response.status_code != 200:
        raise Exception('Failed to fetch data from dots.mocr')
    return response.json()

# Transform records for storage
async def transform_records(data: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Transform extracted data to required format.
    
    Args:
        data: Data extracted from the diagram
    Returns:
        Transformed data ready for storage
    """
    return [{'key': item['key'], 'value': item['value']} for item in data.get('results', [])]

# Save to database
async def save_to_db(records: List[Dict[str, Any]]) -> None:
    """Save processed records to the database.
    
    Args:
        records: List of records to save
    Raises:
        Exception: If database operation fails
    """
    # Simulating a database call
    logger.info(f'Saving {len(records)} records to the database.')
    # Actual database saving logic would go here

# Handle errors gracefully
async def handle_errors(func):
    """Decorator to handle errors in async functions.
    
    Args:
        func: The function to decorate
    Returns:
        Wrapped function with error handling
    """
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {str(e)}')
            raise
    return wrapper

# Main orchestrator class
class DiagramExtractor:
    """Orchestrator for extracting data from engineering diagrams."""

    @handle_errors
    async def process_diagram(self, image_url: str) -> None:
        """Main processing function for a single diagram image.
        
        Args:
            image_url: URL of the diagram image
        """
        await validate_input({'image_url': image_url})  # Validate input
        sanitized_data = await sanitize_fields({'image_url': image_url})  # Sanitize
        raw_data = await fetch_data(sanitized_data['image_url'])  # Fetch data from API
        normalized_data = await normalize_data(raw_data)  # Normalize data
        transformed_data = await transform_records(normalized_data)  # Transform
        await save_to_db(transformed_data)  # Save to DB

# Main block
if __name__ == '__main__':
    # Example usage
    extractor = DiagramExtractor()
    import asyncio
    image_url = 'https://example.com/diagram.png'  # Example image URL
    asyncio.run(extractor.process_diagram(image_url))