Serve Lightweight Vision Models on Industrial Cameras with TFLite and Triton Inference Server

                      
                     
"""
Production implementation for serving lightweight vision models
using TFLite and Triton Inference Server.
Provides secure, scalable operations.
"""
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, conlist
from typing import Dict, Any
import os
import logging
import httpx
import asyncio

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

class Config:
    """Configuration settings loaded from environment variables."""
    triton_server_url: str = os.getenv('TRITON_SERVER_URL', 'http://localhost:8000')
    model_name: str = os.getenv('MODEL_NAME', 'my_model')

class ImageData(BaseModel):
    """Data model for image input."""
    image: conlist(float, min_items=1, max_items=300 * 300 * 3)  # Flattened image array

async def validate_input(data: ImageData) -> bool:
    """Validate input data for the image.
    
    Args:
        data: Input data to validate.
    Returns:
        True if valid, raises ValueError otherwise.
    Raises:
        ValueError: If validation fails.
    """
    if len(data.image) != 300 * 300 * 3:
        raise ValueError('Input image must be of size 300x300 with 3 channels.')
    return True

async def fetch_data(image_data: ImageData) -> Dict[str, Any]:
    """Fetch predictions from Triton Inference Server based on image data.
    
    Args:
        image_data: Image data to process.
    Returns:
        Server response containing predictions.
    Raises:
        HTTPException: If request fails.
    """
    try:
        logger.info('Fetching predictions from Triton server.')
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f'{Config.triton_server_url}/v2/models/{Config.model_name}/infer',
                json={"inputs": [{"name": "input", "shape": [1, 3, 300, 300], "data": image_data.image}]}
            )
        response.raise_for_status()  # Raise an error for bad responses
        return response.json()
    except httpx.HTTPStatusError as e:
        logger.error(f'Error fetching data: {e}')
        raise HTTPException(status_code=e.response.status_code, detail=str(e))

async def process_batch(data: ImageData) -> Dict[str, Any]:
    """Main processing function for input data.
    
    Args:
        data: Image data to process.
    Returns:
        Predictions from the model.
    Raises:
        ValueError: If validation fails.
    """
    await validate_input(data)  # Validate input data
    predictions = await fetch_data(data)  # Fetch predictions
    return predictions

app = FastAPI()

@app.post('/predict', response_model=Dict[str, Any])
async def predict(request: Request, image_data: ImageData):
    """Predict endpoint to receive image data and return model predictions.
    
    Args:
        request: Incoming HTTP request.
        image_data: Image data for prediction.
    Returns:
        Model predictions.
    Raises:
        HTTPException: If any errors occur.
    """
    try:
        logger.info('Received prediction request.')
        predictions = await process_batch(image_data)
        return predictions
    except Exception as e:
        logger.error(f'Error in predict: {e}')
        raise HTTPException(status_code=500, detail='Internal Server Error')

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