From efa04926491be37b5bba8acba9fe46cfffa807e9 Mon Sep 17 00:00:00 2001 From: Jordan Larson Date: Thu, 20 Nov 2025 16:50:02 -0500 Subject: [PATCH 1/4] initial monitoring dashboard added - requires addition of 'prediction-feedback' table in dynamodb --- backend/api.py | 222 +++++++++++++++++++++++++++ frontend/monitoring-dashboard/app.py | 98 ++++++++++++ 2 files changed, 320 insertions(+) create mode 100644 backend/api.py create mode 100644 frontend/monitoring-dashboard/app.py diff --git a/backend/api.py b/backend/api.py new file mode 100644 index 0000000..ef08502 --- /dev/null +++ b/backend/api.py @@ -0,0 +1,222 @@ +import fastapi +import boto3 +import os +import datetime +import random +import pickle +import json +import numpy as np +from typing import List +from pydantic import BaseModel +from dotenv import load_dotenv + +class MyFavorites(BaseModel): + items: List[str] + +class PredictionResponse(BaseModel): + user_id: int + req: MyFavorites + +# Load environment variables from .env file +load_dotenv() + +''' +To-Do: +- [ ] Connect to S3 w/ model +- [ ] Create a function to load the model from S3 +- [ ] Create a function to predict using the model +''' + +## Helper Functions +def load_supporting_tables_from_s3(table_name: str): + """ + Download and load supporting tables from S3 into memory without persisting it to disk. + """ + s3 = boto3.client("s3") + s3_bucket = 'readcrumbs' + + # Download model object as bytes into memory + response = s3.get_object(Bucket=s3_bucket, Key=table_name) + table_bytes = response['Body'].read().decode('utf-8') + table = json.loads(table_bytes) + + return table + +def load_model_from_s3(model_name: str): + """ + Download and load an ML model file from S3 into memory without persisting it to disk. + + Uses AWS credentials from environment variables if available: + - AWS_ACCESS_KEY_ID + - AWS_SECRET_ACCESS_KEY + - AWS_SESSION_TOKEN (optional, for temporary credentials) + + Falls back to IAM roles (if running on EC2, Lambda, ECS, etc.) or ~/.aws/credentials + + Required environment variables: + - S3_MODEL_BUCKET: S3 bucket name + - AWS_REGION: AWS region (optional, defaults to us-east-1) + + Args: + model_name (str): The key/path of the model file in the S3 bucket. + + Returns: + The loaded model object. + """ + s3 = boto3.client("s3") + s3_bucket = 'readcrumbs' + + # Download model object as bytes into memory + response = s3.get_object(Bucket=s3_bucket, Key=model_name) + model_bytes = response['Body'].read() + model = pickle.loads(model_bytes) + return model + +def predict_using_model(model, data: MyFavorites, n_recs: int = 10): + my_favs_ids = [title_to_index[f] for f in data.items] + fav_vectors = [model.item_factors[i] for i in my_favs_ids] + #Average the vectors + avg_vec = np.average(np.stack(fav_vectors), axis=0) + recommendations = np.argsort(np.dot(avg_vec, model.item_factors.T))[:n_recs] + return [index_to_title[i] for i in recommendations] + +def serialize_for_dynamodb(data): + """ + Recursively serialize data for DynamoDB. + Converts datetime objects to ISO format strings. + """ + if isinstance(data, datetime.datetime): + return data.isoformat() + elif isinstance(data, dict): + return {k: serialize_for_dynamodb(v) for k, v in data.items()} + elif isinstance(data, list): + return [serialize_for_dynamodb(item) for item in data] + else: + return data + +def get_dynamodb_table(): + """ + Get a DynamoDB table resource with proper credentials. + + Returns: + boto3 DynamoDB Table resource + """ + table_name = os.environ.get("DDB_TABLE") + if not table_name: + raise ValueError("DDB_TABLE environment variable not set.") + + region = os.environ.get("AWS_REGION", "us-east-1") + + # Get AWS credentials from environment variables + aws_access_key_id = os.environ.get("AWS_ACCESS_KEY_ID") + aws_secret_access_key = os.environ.get("AWS_SECRET_ACCESS_KEY") + aws_session_token = os.environ.get("AWS_SESSION_TOKEN") + + # Create boto3 session with explicit credentials if available + if aws_access_key_id and aws_secret_access_key: + session = boto3.Session( + aws_access_key_id=aws_access_key_id, + aws_secret_access_key=aws_secret_access_key, + aws_session_token=aws_session_token, + region_name=region + ) + dynamodb = session.resource("dynamodb") + else: + # Fall back to default credential chain (IAM roles, ~/.aws/credentials, etc.) + dynamodb = boto3.resource("dynamodb", region_name=region) + + return dynamodb.Table(table_name) + +def get_random_item_from_ddb(): + """ + Retrieve a random item from DynamoDB table. + + Returns: + dict: A random item from the table, or None if table is empty + """ + table = get_dynamodb_table() + + # Scan the table to get all items + # Note: For very large tables, this could be expensive. + # Consider optimizing with pagination or sampling if needed. + response = table.scan() + items = response.get('Items', []) + + # Handle pagination if there are more items + while 'LastEvaluatedKey' in response: + response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey']) + items.extend(response.get('Items', [])) + + if not items: + return None + + # Return a random item + return random.choice(items) + +def save_to_ddb(data): + """ + Save or update a dictionary of data to DynamoDB. + Uses user_id (integer) as the primary key. If user_id already exists, the item will be updated. + + Uses AWS credentials from environment variables if available: + - AWS_ACCESS_KEY_ID + - AWS_SECRET_ACCESS_KEY + - AWS_SESSION_TOKEN (optional, for temporary credentials) + + Falls back to IAM roles (if running on EC2, Lambda, ECS, etc.) or ~/.aws/credentials + + Required environment variables: + - DDB_TABLE: DynamoDB table name + - AWS_REGION: AWS region (optional, defaults to us-east-1) + + Args: + data: Dictionary containing user_id (int) and other fields. user_id is used as primary key. + """ + table = get_dynamodb_table() + + # Serialize data for DynamoDB (convert datetime objects, etc.) + serialized_data = serialize_for_dynamodb(data) + + # Ensure user_id exists (required as primary key) + if 'user_id' not in serialized_data: + raise ValueError("user_id is required in the request body") + + # Map user_id to pred-id (the table's primary key field name) + # Keep user_id in the data as well for reference + serialized_data['pred-id'] = serialized_data['user_id'] + + response = table.put_item(Item=serialized_data) + return response + +## API + +app = fastapi.FastAPI() + +model = load_model_from_s3("models/als_model-small-v1.pkl") +index_to_title = load_supporting_tables_from_s3("data/v1/index_to_title.json") +title_to_index = load_supporting_tables_from_s3("data/v1/title_to_index.json") + +@app.get("/health") +def health_check(): + return {"status": "ok"} + +@app.get("/random") +def get_random(): + """ + Get a random item from the DynamoDB table. + + Returns: + dict: A random item from the table + """ + random_item = get_random_item_from_ddb() + if random_item is None: + raise fastapi.HTTPException(status_code=404, detail="No items found in table") + return random_item + +@app.post("/predict") +def predict(request: PredictionResponse): + data = request.model_dump() + data['timestamp'] = datetime.datetime.now(datetime.timezone.utc) + data['prediction'] = predict_using_model(model, data) + save_to_ddb(data) + return {"status": "ok"} \ No newline at end of file diff --git a/frontend/monitoring-dashboard/app.py b/frontend/monitoring-dashboard/app.py new file mode 100644 index 0000000..69c130c --- /dev/null +++ b/frontend/monitoring-dashboard/app.py @@ -0,0 +1,98 @@ +import streamlit as st +import boto3 +import pandas as pd +import matplotlib.pyplot as plt +import seaborn as sns + +s3 = boto3.client('s3') +dynamodb = boto3.client('dynamodb') + +# Helper functions +def get_data_from_dynamodb(table_name): + response = dynamodb.scan(TableName=table_name) + return response['Items'] + +def convert_to_df(items): + return pd.DataFrame(items) + + +df = convert_to_df(get_data_from_dynamodb('prediction-logs')) + +# ---------------------------- Streamlit app ---------------------------- +st.title("Monitoring Dashboard") + +df = convert_to_df(get_data_from_dynamodb('prediction-logs')) + +# Convert columns to proper types +df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce') +df['user_id'] = pd.to_numeric(df['user_id'], errors='coerce') +df['prediction'] = df['prediction'].astype(str) +if 'req' in df: + df['req'] = df['req'].astype(str) + +st.header('Prediction Latency Over Time') + +if 'latency' in df.columns: + # Plot latency over time if available + fig1, ax1 = plt.subplots() + sns.lineplot(x='datetime', y='latency', data=df, ax=ax1) + ax1.set_title('Prediction Latency Over Time') + st.pyplot(fig1) +else: + st.info("No latency field present in data. Please ensure the backend logs the latency of predictions.") + +st.header('Prediction Distribution (Target Drift)') + +fig2, ax2 = plt.subplots() +sns.countplot(x='prediction', data=df, ax=ax2) +ax2.set_title('Distribution of Predicted Classes') +st.pyplot(fig2) + +st.header('Collect User Feedback') + +st.write("Click below to rate the most recent model prediction and help track accuracy.") + +user_id_input = st.text_input("User ID", "") +recent = None +if user_id_input: + try: + uid = int(user_id_input) + cur_user_rows = df[df['user_id'] == uid] + if not cur_user_rows.empty: + recent = cur_user_rows.sort_values('datetime', ascending=False).iloc[0] # get latest + st.write(f"Last prediction for User {user_id_input}:") + st.code(dict(recent), language='json') + except Exception: + st.warning("Please enter a valid numeric user ID.") + +if recent is not None: + feedback = st.radio("Was this prediction correct?", ['Yes', 'No']) + feedback_submitted = st.button("Submit Feedback") + if feedback_submitted: + + feedback_table = 'prediction-feedback' + record = { + 'user_id': {'N': str(recent['user_id'])}, + 'datetime': {'S': str(recent['datetime'])}, + 'prediction': {'S': str(recent['prediction'])}, + 'feedback': {'S': feedback} + } + try: + dynamodb.put_item(TableName=feedback_table, Item=record) + st.success("Thank you for your feedback!") + except Exception as e: + st.error(f"Failed to submit feedback: {e}") + +# Calculate live accuracy from feedback + feedback_items = convert_to_df(get_data_from_dynamodb('prediction-feedback')) + feedback_items['feedback'] = feedback_items['feedback'].astype(str) + if not feedback_items.empty: + acc = (feedback_items['feedback'] == 'Yes').mean() + st.metric("Live Model Accuracy (from feedback)", f"{acc:.2%}") + else: + st.info("No feedback yet; accuracy cannot be computed.") + + + + + From f4da484635c3ca9751eace49f794c76506972de0 Mon Sep 17 00:00:00 2001 From: Jordan Larson Date: Fri, 21 Nov 2025 17:30:33 -0500 Subject: [PATCH 2/4] changed file structure to match main; added Dockerfile --- monitoring/Dockerfile | 21 +++++++++++++++++++ .../app.py | 0 monitoring/requirements.txt | 5 +++++ 3 files changed, 26 insertions(+) create mode 100644 monitoring/Dockerfile rename {frontend/monitoring-dashboard => monitoring}/app.py (100%) create mode 100644 monitoring/requirements.txt diff --git a/monitoring/Dockerfile b/monitoring/Dockerfile new file mode 100644 index 0000000..badb45c --- /dev/null +++ b/monitoring/Dockerfile @@ -0,0 +1,21 @@ +# Use Python 3.11 slim image as base +FROM python:3.11-slim + +# Set working directory +WORKDIR /app + +# Copy requirements file +COPY requirements.txt . + +# Install Python dependencies +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application code +COPY app.py . + +# Expose Streamlit port +EXPOSE 8501 + +# Run Streamlit app +CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"] + diff --git a/frontend/monitoring-dashboard/app.py b/monitoring/app.py similarity index 100% rename from frontend/monitoring-dashboard/app.py rename to monitoring/app.py diff --git a/monitoring/requirements.txt b/monitoring/requirements.txt new file mode 100644 index 0000000..261174b --- /dev/null +++ b/monitoring/requirements.txt @@ -0,0 +1,5 @@ +streamlit +boto3 +pandas +matplotlib.pyplot +seaborn \ No newline at end of file From 68033d03b2c5bd91407ef2772e0763ae69a6d0fa Mon Sep 17 00:00:00 2001 From: Jordan Larson Date: Sat, 22 Nov 2025 14:18:11 -0500 Subject: [PATCH 3/4] changed feedback question "Are these recommendations relevant to you?" --- monitoring/app.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/monitoring/app.py b/monitoring/app.py index 69c130c..a2b6d30 100644 --- a/monitoring/app.py +++ b/monitoring/app.py @@ -66,7 +66,7 @@ def convert_to_df(items): st.warning("Please enter a valid numeric user ID.") if recent is not None: - feedback = st.radio("Was this prediction correct?", ['Yes', 'No']) + feedback = st.radio("Are these recommendations relevant to you?", ['Yes', 'No']) feedback_submitted = st.button("Submit Feedback") if feedback_submitted: From 99aac700eb1f8ec7be657b4af970d48734e2db36 Mon Sep 17 00:00:00 2001 From: Jordan Larson Date: Sat, 22 Nov 2025 14:19:15 -0500 Subject: [PATCH 4/4] removed backend folder to prevent conflict --- backend/api.py | 222 ------------------------------------------------- 1 file changed, 222 deletions(-) delete mode 100644 backend/api.py diff --git a/backend/api.py b/backend/api.py deleted file mode 100644 index ef08502..0000000 --- a/backend/api.py +++ /dev/null @@ -1,222 +0,0 @@ -import fastapi -import boto3 -import os -import datetime -import random -import pickle -import json -import numpy as np -from typing import List -from pydantic import BaseModel -from dotenv import load_dotenv - -class MyFavorites(BaseModel): - items: List[str] - -class PredictionResponse(BaseModel): - user_id: int - req: MyFavorites - -# Load environment variables from .env file -load_dotenv() - -''' -To-Do: -- [ ] Connect to S3 w/ model -- [ ] Create a function to load the model from S3 -- [ ] Create a function to predict using the model -''' - -## Helper Functions -def load_supporting_tables_from_s3(table_name: str): - """ - Download and load supporting tables from S3 into memory without persisting it to disk. - """ - s3 = boto3.client("s3") - s3_bucket = 'readcrumbs' - - # Download model object as bytes into memory - response = s3.get_object(Bucket=s3_bucket, Key=table_name) - table_bytes = response['Body'].read().decode('utf-8') - table = json.loads(table_bytes) - - return table - -def load_model_from_s3(model_name: str): - """ - Download and load an ML model file from S3 into memory without persisting it to disk. - - Uses AWS credentials from environment variables if available: - - AWS_ACCESS_KEY_ID - - AWS_SECRET_ACCESS_KEY - - AWS_SESSION_TOKEN (optional, for temporary credentials) - - Falls back to IAM roles (if running on EC2, Lambda, ECS, etc.) or ~/.aws/credentials - - Required environment variables: - - S3_MODEL_BUCKET: S3 bucket name - - AWS_REGION: AWS region (optional, defaults to us-east-1) - - Args: - model_name (str): The key/path of the model file in the S3 bucket. - - Returns: - The loaded model object. - """ - s3 = boto3.client("s3") - s3_bucket = 'readcrumbs' - - # Download model object as bytes into memory - response = s3.get_object(Bucket=s3_bucket, Key=model_name) - model_bytes = response['Body'].read() - model = pickle.loads(model_bytes) - return model - -def predict_using_model(model, data: MyFavorites, n_recs: int = 10): - my_favs_ids = [title_to_index[f] for f in data.items] - fav_vectors = [model.item_factors[i] for i in my_favs_ids] - #Average the vectors - avg_vec = np.average(np.stack(fav_vectors), axis=0) - recommendations = np.argsort(np.dot(avg_vec, model.item_factors.T))[:n_recs] - return [index_to_title[i] for i in recommendations] - -def serialize_for_dynamodb(data): - """ - Recursively serialize data for DynamoDB. - Converts datetime objects to ISO format strings. - """ - if isinstance(data, datetime.datetime): - return data.isoformat() - elif isinstance(data, dict): - return {k: serialize_for_dynamodb(v) for k, v in data.items()} - elif isinstance(data, list): - return [serialize_for_dynamodb(item) for item in data] - else: - return data - -def get_dynamodb_table(): - """ - Get a DynamoDB table resource with proper credentials. - - Returns: - boto3 DynamoDB Table resource - """ - table_name = os.environ.get("DDB_TABLE") - if not table_name: - raise ValueError("DDB_TABLE environment variable not set.") - - region = os.environ.get("AWS_REGION", "us-east-1") - - # Get AWS credentials from environment variables - aws_access_key_id = os.environ.get("AWS_ACCESS_KEY_ID") - aws_secret_access_key = os.environ.get("AWS_SECRET_ACCESS_KEY") - aws_session_token = os.environ.get("AWS_SESSION_TOKEN") - - # Create boto3 session with explicit credentials if available - if aws_access_key_id and aws_secret_access_key: - session = boto3.Session( - aws_access_key_id=aws_access_key_id, - aws_secret_access_key=aws_secret_access_key, - aws_session_token=aws_session_token, - region_name=region - ) - dynamodb = session.resource("dynamodb") - else: - # Fall back to default credential chain (IAM roles, ~/.aws/credentials, etc.) - dynamodb = boto3.resource("dynamodb", region_name=region) - - return dynamodb.Table(table_name) - -def get_random_item_from_ddb(): - """ - Retrieve a random item from DynamoDB table. - - Returns: - dict: A random item from the table, or None if table is empty - """ - table = get_dynamodb_table() - - # Scan the table to get all items - # Note: For very large tables, this could be expensive. - # Consider optimizing with pagination or sampling if needed. - response = table.scan() - items = response.get('Items', []) - - # Handle pagination if there are more items - while 'LastEvaluatedKey' in response: - response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey']) - items.extend(response.get('Items', [])) - - if not items: - return None - - # Return a random item - return random.choice(items) - -def save_to_ddb(data): - """ - Save or update a dictionary of data to DynamoDB. - Uses user_id (integer) as the primary key. If user_id already exists, the item will be updated. - - Uses AWS credentials from environment variables if available: - - AWS_ACCESS_KEY_ID - - AWS_SECRET_ACCESS_KEY - - AWS_SESSION_TOKEN (optional, for temporary credentials) - - Falls back to IAM roles (if running on EC2, Lambda, ECS, etc.) or ~/.aws/credentials - - Required environment variables: - - DDB_TABLE: DynamoDB table name - - AWS_REGION: AWS region (optional, defaults to us-east-1) - - Args: - data: Dictionary containing user_id (int) and other fields. user_id is used as primary key. - """ - table = get_dynamodb_table() - - # Serialize data for DynamoDB (convert datetime objects, etc.) - serialized_data = serialize_for_dynamodb(data) - - # Ensure user_id exists (required as primary key) - if 'user_id' not in serialized_data: - raise ValueError("user_id is required in the request body") - - # Map user_id to pred-id (the table's primary key field name) - # Keep user_id in the data as well for reference - serialized_data['pred-id'] = serialized_data['user_id'] - - response = table.put_item(Item=serialized_data) - return response - -## API - -app = fastapi.FastAPI() - -model = load_model_from_s3("models/als_model-small-v1.pkl") -index_to_title = load_supporting_tables_from_s3("data/v1/index_to_title.json") -title_to_index = load_supporting_tables_from_s3("data/v1/title_to_index.json") - -@app.get("/health") -def health_check(): - return {"status": "ok"} - -@app.get("/random") -def get_random(): - """ - Get a random item from the DynamoDB table. - - Returns: - dict: A random item from the table - """ - random_item = get_random_item_from_ddb() - if random_item is None: - raise fastapi.HTTPException(status_code=404, detail="No items found in table") - return random_item - -@app.post("/predict") -def predict(request: PredictionResponse): - data = request.model_dump() - data['timestamp'] = datetime.datetime.now(datetime.timezone.utc) - data['prediction'] = predict_using_model(model, data) - save_to_ddb(data) - return {"status": "ok"} \ No newline at end of file