Recognize Industrial Components with GLM-4.5V and Hugging Face Transformers

                      
                     
"""
Production implementation for recognizing industrial components using GLM-4.5V and Hugging Face Transformers.
Provides secure, scalable operations for processing and recognizing components from images.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import requests
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, constr

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

class Config:
    model_url: str = os.getenv('MODEL_URL', 'http://localhost:8000/predict')
    retry_attempts: int = 3
    retry_delay: int = 2  # seconds

class RecognitionRequest(BaseModel):
    image_url: constr(strip_whitespace=True)

class RecognitionResponse(BaseModel):
    components: List[str]
    confidence_scores: List[float]

async def validate_input(data: RecognitionRequest) -> None:
    """Validate input data for recognition request.
    
    Args:
        data: RecognitionRequest object containing input data.
    Raises:
        ValueError: If validation fails.
    """
    if not data.image_url:
        raise ValueError('Image URL cannot be empty')

async def fetch_data(image_url: str) -> Dict[str, Any]:
    """Fetch data from the image URL for processing.
    
    Args:
        image_url: URL of the image to fetch.
    Returns:
        Dictionary containing the image data.
    Raises:
        HTTPException: If fetching fails.
    """
    try:
        response = requests.get(image_url)
        response.raise_for_status()  # Raise an error for bad responses
        logger.info('Fetched image data successfully.')
        return response.json()
    except requests.HTTPError as http_err:
        logger.error(f'HTTP error occurred: {http_err}')
        raise HTTPException(status_code=400, detail='Failed to fetch image data')

async def call_api(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Call external model API for component recognition.
    
    Args:
        payload: Data to send to the model API.
    Returns:
        Response from the model API.
    Raises:
        HTTPException: If the API call fails.
    """
    for attempt in range(Config.retry_attempts):
        try:
            logger.info(f'Calling model API, attempt {attempt + 1}.')
            response = requests.post(Config.model_url, json=payload)
            response.raise_for_status()  # Raise error for bad responses
            logger.info('Model API call successful.')
            return response.json()
        except requests.HTTPError as http_err:
            logger.warning(f'API call failed: {http_err}')
            if attempt < Config.retry_attempts - 1:
                time.sleep(Config.retry_delay)
            else:
                raise HTTPException(status_code=500, detail='Model API call failed')

async def transform_records(data: Dict[str, Any]) -> Tuple[List[str], List[float]]:
    """Transform API response data into a usable format.
    
    Args:
        data: Raw response data from model API.
    Returns:
        Tuple of component names and their confidence scores.
    """
    components = data.get('components', [])
    confidence_scores = data.get('scores', [])
    return components, confidence_scores

async def format_output(components: List[str], scores: List[float]) -> RecognitionResponse:
    """Format output into RecognitionResponse model.
    
    Args:
        components: List of recognized components.
        scores: List of confidence scores.
    Returns:
        RecognitionResponse object.
    """
    return RecognitionResponse(components=components, confidence_scores=scores)

app = FastAPI()

@app.post('/recognize', response_model=RecognitionResponse)
async def recognize_components(request: RecognitionRequest) -> RecognitionResponse:
    """Endpoint to recognize components from an image URL.
    
    Args:
        request: RecognitionRequest object.
    Returns:
        RecognitionResponse object containing recognized components.
    Raises:
        HTTPException: If errors occur in the process.
    """
    await validate_input(request)  # Validate input data
    image_data = await fetch_data(request.image_url)  # Fetch image data
    api_response = await call_api(image_data)  # Call model API
    components, scores = await transform_records(api_response)  # Transform response
    return await format_output(components, scores)  # Format output

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)