Control Industrial Robots via Natural Language with ROS-LLM and FastAPI

                      
                     
"""
Production implementation for controlling industrial robots via natural language using ROS-LLM and FastAPI.
Provides secure, scalable operations for robot control commands.
"""

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field
import os
import logging
import aiohttp
import asyncio
from typing import Dict, Any

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


class Config:
    """
    Configuration class to manage environment variables.
    """
    robot_api_url: str = os.getenv('ROBOT_API_URL', 'http://localhost:5000/api')


config = Config()


class CommandRequest(BaseModel):
    """
    Pydantic model for command requests to robots.
    """
    command: str = Field(..., example="move forward")
    parameters: Dict[str, Any] = Field(..., example={"speed": 1})


async def validate_input(data: CommandRequest) -> bool:
    """Validate command input data.
    
    Args:
        data: CommandRequest object to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if not data.command:
        logger.error('Missing command in request')
        raise ValueError('Missing command')
    return True


def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent injection attacks.
    
    Args:
        data: Dictionary containing command parameters
    Returns:
        Sanitized dictionary
    """
    sanitized_data = {k: str(v).strip() for k, v in data.items()}
    logger.info('Sanitized input data')
    return sanitized_data


async def fetch_data(url: str) -> Dict[str, Any]:
    """Fetch data from an external API asynchronously.
    
    Args:
        url: URL to fetch data from
    Returns:
        JSON response as a dictionary
    Raises:
        Exception: If request fails
    """
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                response.raise_for_status()  # Raise an error for bad responses
                data = await response.json()
                logger.info('Fetched data from %s', url)
                return data
    except Exception as e:
        logger.error('Failed to fetch data: %s', e)
        raise


def save_to_db(data: Dict[str, Any]) -> None:
    """Simulate saving command data to a database.
    
    Args:
        data: Command data to save
    Raises:
        Exception: If save fails
    """
    # For demo purposes, we'll just log instead of saving to a real database.
    logger.info('Saving command to database: %s', data)


async def call_api(command: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """Call the robot control API with the command and parameters.
    
    Args:
        command: Command to send to the robot
        params: Parameters associated with the command
    Returns:
        Response from the robot API
    Raises:
        HTTPException: If the API call fails
    """
    url = f'{config.robot_api_url}/{command}'
    logger.info('Calling API at %s with params %s', url, params)
    try:
        response = await fetch_data(url)
        logger.info('Received response: %s', response)
        return response
    except Exception as e:
        logger.error('API call failed: %s', e)
        raise HTTPException(status_code=500, detail='API call failed')


async def process_command(request: CommandRequest) -> Dict[str, Any]:
    """Process the incoming command request.
    
    Args:
        request: CommandRequest object containing command details
    Returns:
        Response from the robot control API
    """
    await validate_input(request)  # Validate the incoming request
    params = sanitize_fields(request.parameters)  # Sanitize parameters
    response = await call_api(request.command, params)  # Call the robot API
    save_to_db(params)  # Save parameters to the database
    return response


app = FastAPI()

@app.post('/control', response_model=Dict[str, Any])
async def control_robot(request: CommandRequest):
    """Endpoint to control the robot via natural language commands.
    
    Args:
        request: CommandRequest object
    Returns:
        Response from the robot
    Raises:
        HTTPException: If an error occurs
    """
    try:
        response = await process_command(request)  # Process the command
        return response
    except ValueError as ve:
        raise HTTPException(status_code=400, detail=str(ve))  # Handle validation errors
    except Exception as e:
        logger.error('Unexpected error: %s', e)
        raise HTTPException(status_code=500, detail='Internal server error')


if __name__ == '__main__':
    # Example usage of the FastAPI application
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)