Detect Factory Defects via Text Prompts with SAM 3 and Roboflow Inference

                      
                     
"""
Production implementation for detecting factory defects via text prompts with SAM 3 and Roboflow inference.
Provides secure, scalable operations to identify anomalies in product quality.
"""
from typing import Dict, Any, List
import os
import logging
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, constr
from time import sleep

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

# Configuration for environment variables
class Config:
    roboflow_api_key: str = os.getenv('ROBOFLOW_API_KEY')
    roboflow_project: str = os.getenv('ROBOFLOW_PROJECT')
    roboflow_version: str = os.getenv('ROBOFLOW_VERSION')

# Initialize FastAPI app
app = FastAPI()

# Input data model
class DefectRequest(BaseModel):
    prompt: constr(strip_whitespace=True, min_length=1)  # Ensure prompt is non-empty

def validate_input(data: Dict[str, Any]) -> None:
    """
    Validate request data.

    Args:
        data: Input to validate
    Raises:
        ValueError: If validation fails
    """
    if 'prompt' not in data:
        raise ValueError('Missing prompt field')

def fetch_data(prompt: str) -> Dict[str, Any]:
    """
    Fetch defect predictions using Roboflow API.

    Args:
        prompt: The text prompt for defect detection
    Returns:
        JSON response from Roboflow API
    Raises:
        HTTPException: If API call fails
    """
    url = f'https://api.roboflow.com/{Config.roboflow_project}/{Config.roboflow_version}/detect'
    headers = {'Authorization': f'Bearer {Config.roboflow_api_key}'}
    payload = {'prompt': prompt}
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()  # Raises an error for bad responses
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f'Error fetching data from Roboflow: {e}')
        raise HTTPException(status_code=500, detail='Error fetching data from Roboflow')

def transform_records(data: Dict[str, Any]) -> List[str]:
    """
    Transform raw defect data into usable format.

    Args:
        data: Raw data returned from Roboflow API
    Returns:
        List of defect descriptions
    """
    return [item['description'] for item in data.get('predictions', [])]

def process_batch(prompts: List[str]) -> List[str]:
    """
    Process a batch of prompts and return defect descriptions.

    Args:
        prompts: List of prompts to send to Roboflow
    Returns:
        List of defect descriptions
    """
    results = []
    for prompt in prompts:
        try:
            raw_data = fetch_data(prompt)
            transformed_data = transform_records(raw_data)
            results.extend(transformed_data)
        except Exception as e:
            logger.warning(f'Failed to process prompt {prompt}: {e}')  # Continue processing even if one fails
            results.append(f'Error processing prompt: {prompt}')  # Log error
    return results

@app.post('/detect_defects/', response_model=List[str])
async def detect_defects(request: DefectRequest):
    """
    Endpoint to detect defects from text prompts.

    Args:
        request: Request containing the defect prompt
    Returns:
        List of detected defects
    Raises:
        HTTPException: If validation or processing fails
    """
    try:
        # Validate input
        validate_input(request.dict())
        # Process the prompt
        results = process_batch([request.prompt])
        return results
    except ValueError as ve:
        logger.error(f'Validation error: {ve}')
        raise HTTPException(status_code=400, detail=str(ve))
    except Exception as e:
        logger.error(f'Unexpected error: {e}')
        raise HTTPException(status_code=500, detail='Internal Server Error')

if __name__ == '__main__':
    import uvicorn
    # Start the server using Uvicorn for FastAPI
    uvicorn.run(app, host='0.0.0.0', port=8000)