frist commit

app.py

@@ -1,79 +1,45 @@
-from typing import Union
-
 from fastapi import FastAPI
-from fastapi import Request
-
-app = FastAPI()
+from fastapi.middleware.cors import CORSMiddleware
+from scripts.s3 import download_model_from_s3
+from router.disaster import router as disaster_router
+from router.sentiment import router as sentiment_router
+from router.image_clf import router as image_router
+from utils.logger import logger
+
+
+app = FastAPI(
+    title="ML API",
+    description="ML API for sentiment analysis and image classification",
+    version="0.0.1",
+    openapi_url="/openapi.json"
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+MODEL_PATH = "ml-models/"  
+sentiment_model_path = "tinybert-sentiment-analysis/"
+disaster_model_path = "tinybert-disaster-tweet/"
+image_model_path = "vit-human-pose-classification/"
+
+logger.info("Ensuring models are downloaded...")
+download_model_from_s3(MODEL_PATH + sentiment_model_path, sentiment_model_path)
+download_model_from_s3(MODEL_PATH + disaster_model_path, disaster_model_path)
+download_model_from_s3(MODEL_PATH + image_model_path, image_model_path)
+logger.info("All models are ready.")
 
 
 @app.get("/")
 def read_root():
-    return {"Hello": "World!!!"}
-    # return "This is string"
-
-
-@app.get("/items/{item_id}")
-def read_item(item_id: int, q: Union[str, None] = None):
-    return {"item_id": item_id, "q": q}
-
-
-@app.get("/get_sentiment/{text}")
-def get_sentiment(text: str, user_id: Union[str, None] = None):
-    return {"text": text, 
-            "sentiment": "positive", 
-            "user_id": user_id}
-
-@app.get("/get_sentiment_v2/{text}/{ip}")
-def get_sentiment_v2(text: str, ip: str, user_id: Union[str, None] = None):
-    return {"ip": ip,
-            "text": text, 
-            "sentiment": "positive", 
-            "user_id": user_id}
-
-
-@app.post("/get_twitter_sentiment")
-def get_twitter_sentiment(text: str, ip: str, user_id: Union[str, None] = None):
-    """
-    Analyze the sentiment of a given text from Twitter.
-
-    Parameters:
-    - text (str): The text to analyze.
-    - ip (str): The IP address of the user.
-    - user_id (Union[str, None]): Optional user ID for tracking.
-
-    Returns:
-    - A JSON object containing the IP, text, sentiment analysis result, and user ID.
-    """
-    return {
-        "ip": ip,
-        "text": text,
-        "sentiment": "normal",  # Placeholder for sentiment analysis
-        "user_id": user_id
-    }
-
-@app.post("/get_twitter_sentiment_v2")
-async def get_twitter_sentiment_v2(request: Request):
-    """
-    Analyze the sentiment of a given text from Twitter asynchronously.
-
-    This endpoint accepts a JSON body with the following fields:
-    - text (str): The text to analyze.
-    - ip (str): The IP address of the user.
-    - user_id (Union[str, None]): Optional user ID for tracking.
-
-    Returns:
-    - A JSON object containing the IP, text, sentiment analysis result, and user ID.
-    """
-    data = await request.json()
-
-    text: str = data.get('text')
-    ip = data.get('ip')
-    user_id = data.get('user_id')
-
-    return {
-        "ip": ip,
-        "text": text,
-        "sentiment": "normal",  # Placeholder for sentiment analysis
-        "user_id": user_id
-    }
+    return {"Status": "Running"}
+  
 
+app.include_router(disaster_router, prefix="/api/v1", tags=["Disaster"])
+app.include_router(sentiment_router, prefix="/api/v1", tags=["Sentiment"])
+app.include_router(image_router, prefix="/api/v1", tags=["Image"])

requirements.txt

@@ -1,2 +1,5 @@
+boto3
+torch
+transformers
 fastapi
 uvicorn[standard]

router/disaster.py

@@ -0,0 +1,40 @@
+import os
+import time
+from typing import Union
+from fastapi import APIRouter
+from scripts.data_model import ClassificationInput, ClassificationOutput
+from utils.pipeline import load_model
+
+router = APIRouter()
+
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  
+MODEL_PATH = os.path.join(BASE_DIR, "ml_models", "tinybert-disaster-tweet/")
+
+@router.post(
+    "/disaster_classification",
+    response_model=ClassificationOutput,
+    summary="Disaster Classification",
+    description="Classify the disaster of a given text using a pre-trained model."
+)
+def disaster_classification(input: ClassificationInput)-> ClassificationOutput:
+    try:
+        pipe = load_model(MODEL_PATH)
+        start = time.time()
+        output = pipe(input.text)
+        end = time.time()
+        prediction_time = int((end-start)*1000)
+
+        labels = [x['label'] for x in output]
+        scores = [x['score'] for x in output]
+
+        return ClassificationOutput(
+            user_id=input.user_id,
+            text=input.text,
+            model_name="tinybert-disaster-tweet",
+            sentiment=labels,
+            score=scores,
+            prediction_time=prediction_time
+        )
+        
+    except Exception as e:
+        return {"error": f"Failed to process text classification: {str(e)}"}, 500

router/image_clf.py

@@ -0,0 +1,39 @@
+import os
+import time
+from typing import Union
+from fastapi import APIRouter
+from scripts.data_model import ImageInput, ImageOutput
+from utils.pipeline import load_model
+
+router = APIRouter()
+
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  
+MODEL_PATH = os.path.join(BASE_DIR, "ml_models", "vit-human-pose-classification/")
+
+@router.post(
+    "/image_classification",
+    response_model=ImageOutput,
+    summary="Image Classification",
+    description="Classify the image using a pre-trained model."
+)
+def image_classification(input: ImageInput)-> ImageOutput:
+    try:
+        pipe = load_model(MODEL_PATH, is_image_model=True)
+        start = time.time()
+        output = pipe(input.url)
+        end = time.time()
+        prediction_time = int((end-start)*1000)
+
+        labels_and_scores = [{"label": x['label'], "score": x['score']} for x in output]
+
+        return ImageOutput(
+            user_id=input.user_id,
+            url=input.url,
+            model_name="vit-human-pose-classification",
+            labels=[x['label'] for x in labels_and_scores],
+            scores=[x['score'] for x in labels_and_scores],
+            prediction_time=prediction_time
+        )
+   
+    except Exception as e:
+        return {"error": f"Failed to process text classification: {str(e)}"}, 500

router/sentiment.py

@@ -0,0 +1,40 @@
+import os
+import time
+from typing import Union
+from fastapi import APIRouter
+from scripts.data_model import ClassificationInput, ClassificationOutput
+from utils.pipeline import load_model
+
+router = APIRouter()
+
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  
+MODEL_PATH = os.path.join(BASE_DIR, "ml_models", "tinybert-sentiment-analysis/")
+
+@router.post(
+    "/sentiment_classification",
+    response_model=ClassificationOutput,
+    summary="Sentiment Classification",
+    description="Classify the sentiment of a given text using a pre-trained model."
+)
+def sentiment_classification(input: ClassificationInput)-> ClassificationOutput:
+    try:
+        pipe = load_model(MODEL_PATH)
+        start = time.time()
+        output = pipe(input.text)
+        end = time.time()
+        prediction_time = int((end-start)*1000)
+
+        labels = [x['label'] for x in output]
+        scores = [x['score'] for x in output]
+
+        return ClassificationOutput(
+            user_id=input.user_id,
+            text=input.text,
+            model_name="tinybert-sentiment-analysis",
+            sentiment=labels,
+            score=scores,
+            prediction_time=prediction_time
+        )
+    
+    except Exception as e:
+        return {"error": f"Failed to process text classification: {str(e)}"}, 500

router/upload_image.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+from fastapi.responses import FileResponse
+
+# router = APIRouter()
+
+# @router.get("/image/{image_name}")
+# async def get_image(image_name: str):
+#     return FileResponse(f"image/{image_name}")

scripts/data_model.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+from typing import List
+
+
+class ClassificationInput(BaseModel):
+    user_id: str
+    text: List[str]
+
+class ImageInput(BaseModel):
+    user_id: str
+    url: List[str]
+
+
+class ClassificationOutput(BaseModel):
+    user_id: str
+    text: List[str]
+    model_name: str
+    sentiment: List[str]
+    score: List[float]
+    prediction_time: int
+
+class ImageOutput(BaseModel):
+    user_id: str
+    url: List[str]
+    model_name: str
+    label: List[str]
+    score: List[float]
+    prediction_time: int

scripts/s3.py

@@ -0,0 +1,61 @@
+import os
+import boto3
+from pathlib import Path
+from utils import logger
+
+aws_access_key = os.getenv("AWS_ACCESS_KEY_ID") #"AKIA2ZIOMY3VPQVDJNDG"
+aws_key_pw =  os.getenv("AWS_SECRET_ACCESS_KEY") #"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"
+BUCKET_NAME =  os.getenv("BUCKET_NAME") #"mlops-tiny-bert-19032025"
+s3 = boto3.client(
+    "s3",
+    aws_access_key_id=aws_access_key,
+    aws_secret_access_key=aws_key_pw,    
+)
+
+
+def download_model_from_s3(local_path: Path, s3_prefix: str):
+    try:
+        if os.path.exists(local_path) and os.listdir(local_path):
+            logger.info(f"Model {local_path} already exists. Skipping download.")
+            return
+
+        logger.info(f"Downloading model from S3: {s3_prefix} to {local_path}")
+        os.makedirs(local_path, exist_ok=True)
+        paginator = s3.get_paginator("list_objects_v2")
+
+        for result in paginator.paginate(Bucket=BUCKET_NAME, Prefix=s3_prefix):
+            if "Contents" in result:
+                for key in result["Contents"]:
+                    s3_key = key["Key"]
+                    local_file = os.path.join(local_path, os.path.relpath(s3_key, s3_prefix))
+                   
+                    os.makedirs(os.path.dirname(local_file), exist_ok=True)
+                    logger.info(f"Downloading {s3_key} to {local_file}")
+                    s3.download_file(BUCKET_NAME, s3_key, local_file)
+                    logger.info(f"Downloaded {s3_key} to {local_file}")
+    except Exception as e:
+        logger.error(f"Failed to download model from S3: {e}")
+        raise RuntimeError(f"Error downloading model from S3: {e}")
+
+def upload_image_to_s3(
+        file_name, 
+        s3_prefix="ml-images", 
+        object_name=None
+    ):
+    if object_name is None:
+        object_name = os.path.basename(file_name)
+
+    object_name = f"{s3_prefix}/{object_name}"
+    s3.upload_file(file_name, BUCKET_NAME, object_name)
+    logger.info(f"Uploaded {file_name} to s3://{BUCKET_NAME}/{object_name}")
+
+    response = s3.generate_presigned_url(
+        'get_object',
+        Params={
+            "Bucket": BUCKET_NAME,
+            "Key": object_name
+        },
+        ExpiresIn=3600
+    )
+    logger.info(f"Generated presigned URL for {object_name}: {response}")
+    return response

utils/logger.py

@@ -0,0 +1,13 @@
+import logging
+
+def get_logger(name: str):
+    logger = logging.getLogger(name)
+    if not logger.handlers: 
+        handler = logging.StreamHandler()
+        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+        logger.setLevel(logging.INFO)
+    return logger
+
+logger = get_logger(__name__)

utils/pipeline.py

@@ -0,0 +1,20 @@
+import torch
+from transformers import pipeline, AutoImageProcessor
+
+# Set the device to GPU if available, otherwise use CPU
+device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
+
+def load_model(local_path, is_image_model=False):
+    """
+    Load a model from the specified local path.
+    
+    Args:
+        local_path (str): The local path to the model.
+        is_image_model (bool): Flag indicating if the model is an image model.
+    Returns:
+        pipeline: The loaded model pipeline.
+    """
+    if is_image_model:
+        image_processor = AutoImageProcessor.from_pretrained(local_path, use_fast=True)
+        return pipeline("image-classification", model=local_path, device=device, image_processor=image_processor)
+    return pipeline("text-classification", model=local_path, device=device)