Update app.py

app.py

@@ -1,45 +1,27 @@
-import os
 import gradio as gr
-from fastapi import FastAPI, UploadFile, File
-import io
-from PIL import Image
-import json
-import pandas as pd
 import numpy as np
+import pandas as pd
 import matplotlib.pyplot as plt
 import torch
 from transformers import AutoImageProcessor, AutoModelForImageClassification
+from PIL import Image
 from difflib import get_close_matches
-from datasets import load_dataset
-import nltk  # For stemming/lemmatization
-
-# Download nltk resources (if not already present)
-try:
-    from nltk.stem import PorterStemmer  # Or LancasterStemmer
-    stemmer = PorterStemmer()
-    nltk.data.find("corpora/wordnet")
-    from nltk.stem import WordNetLemmatizer
-    lemmatizer = WordNetLemmatizer()
-except LookupError:
-    import nltk
-    nltk.download('wordnet')
-    nltk.download('omw-1.4')
-    from nltk.stem import WordNetLemmatizer
-
-    lemmatizer = WordNetLemmatizer()
-
-    from nltk.stem import PorterStemmer  # Or LancasterStemmer
-    stemmer = PorterStemmer()
-
-
+from typing import Optional, Dict, Any
+import json
+import io
+from datasets import load_dataset  # Import the datasets library
 
 # -------------------------------------------------
 # Configuration
 # -------------------------------------------------
+
+# Define insulin types and their durations and peak times
 INSULIN_TYPES = {
-    "Rapid-Acting": {"onset": 0.25, "duration": 4, "peak_time": 1.0},
+    "Rapid-Acting": {"onset": 0.25, "duration": 4, "peak_time": 1.0},  # Onset in hours, duration in hours, peak time in hours
     "Long-Acting": {"onset": 2, "duration": 24, "peak_time": 8},
 }
+
+#Define basal rates
 DEFAULT_BASAL_RATES = {
     "00:00-06:00": 0.8,
     "06:00-12:00": 1.0,
@@ -50,6 +32,7 @@ DEFAULT_BASAL_RATES = {
 # -------------------------------------------------
 # Load Food Data from Hugging Face Dataset
 # -------------------------------------------------
+
 def load_food_data(dataset_name="Anupam007/Diabetic"):
     try:
         dataset = load_dataset(dataset_name)
@@ -61,6 +44,13 @@ def load_food_data(dataset_name="Anupam007/Diabetic"):
         # Remove unnamed columns
         food_data = food_data.loc[:, ~food_data.columns.str.contains('^unnamed')]
 
+        # Check if "fiber_grams" and "sugar_grams" are not in the dataset
+        if "fiber_grams" not in food_data.columns:
+            food_data['fiber_grams'] = None
+        if "sugar_grams" not in food_data.columns:
+            food_data['sugar_grams'] = None
+
+
         # Normalize food_name column to lowercase: Crucial for matching
         if 'food_name' in food_data.columns:
             food_data['food_name'] = food_data['food_name'].str.lower()
@@ -76,6 +66,8 @@ def load_food_data(dataset_name="Anupam007/Diabetic"):
                 'serving_amount': [29],
                 'serving_unit': ['g'],
                 'carbohydrate_grams': [15],
+                'fiber_grams': [0], #added column
+                'sugar_grams': [0], #added column
                 'notes': ['default']
             })
 
@@ -91,11 +83,13 @@ def load_food_data(dataset_name="Anupam007/Diabetic"):
         food_data = pd.DataFrame({
             'food_category': ['starch'],
             'food_subcategory': ['bread'],
-            'food_name': ['white bread'], # lowercase default
+            'food_name': ['white bread'],
             'serving_description': ['servingsize'],
             'serving_amount': [29],
             'serving_unit': ['g'],
             'carbohydrate_grams': [15],
+            'fiber_grams': [0], #added column
+            'sugar_grams': [0], #added column
             'notes': ['default']
         })
         return food_data
@@ -103,13 +97,14 @@ def load_food_data(dataset_name="Anupam007/Diabetic"):
 # -------------------------------------------------
 # Load Food Classification Model
 # -------------------------------------------------
-
 try:
-    # Load model directly
-    from transformers import AutoImageProcessor, AutoModelForImageClassification
-
-    processor = AutoImageProcessor.from_pretrained("rajistics/finetuned-indian-food")
-    model = AutoModelForImageClassification.from_pretrained("rajistics/finetuned-indian-food")
+    processor = AutoImageProcessor.from_pretrained("google/vit-base-patch16-224")
+    model = AutoModelForImageClassification.from_pretrained(
+        "google/vit-base-patch16-224",
+        torch_dtype=torch.float16,  # Use float16 for efficiency
+        device_map="cpu",  # Run on CPU
+        low_cpu_mem_usage=True  # Optimize memory usage
+    )
     model_loaded = True #Flag for error handling in other defs
 except Exception as e:
     print(f"Model Load Error: {e}") # include e in print statement
@@ -118,16 +113,8 @@ except Exception as e:
     model = None
 
 def classify_food(image):
-
-    inputs = processor(images=image, return_tensors="pt")
-    print(f"Processed image keys: {inputs.keys()}")  # Print the keys of the inputs dictionary
-    if 'pixel_values' in inputs:
-        print(f"Pixel values shape: {inputs['pixel_values'].shape}")
-        print(f"Pixel values type: {inputs['pixel_values'].dtype}")
-        print(f"First few pixel values: {inputs['pixel_values'][0, :5]}") # Print a small slice
-    else:
-        print("Pixel values not found in inputs!")
-
+    """Classify food image using the pre-trained model"""
+    print("classify_food function called") # Check if this function is even called
     try:
         if not model_loaded:
             print("Model not loaded, returning 'Unknown'")
@@ -158,214 +145,73 @@ def classify_food(image):
         return "Unknown"  # If an exception arises make sure to create a default case
 
 # -------------------------------------------------
-# Nutrition Data Retrieval with Enhanced Matching
+# USDA API Integration - REMOVED for local HF Spaces deployment
 # -------------------------------------------------
 
-def lemmatize_food_name(food_name):
-    """Lemmatize the food name to its base form."""
-    return lemmatizer.lemmatize(food_name)
-
-def stem_food_name(food_name):
-    """Stem the food name to its root form."""
-    return stemmer.stem(food_name)
-
-def get_food_nutrition(food_name: str, food_data, portion_size: float = 1.0) -> dict:
-    """
-    Retrieves nutrition information for a given food name from a local database,
-    with enhanced matching including stemming, lemmatization, and synonym lookup.
-    """
-    food_name_lower = food_name.lower()
-    food_names = food_data['food_name'].str.lower().tolist()
-
-    print(f"Searching for: {food_name_lower}")
-
-    # 1. Exact Match
-    if food_name_lower in food_names:
-        print("Exact match found.")
-        return _extract_nutrition_data(food_data, food_name_lower, portion_size)
-
-    # 2. Close Matches
-    close_matches = get_close_matches(food_name_lower, food_names, n=1, cutoff=0.6)
-    if close_matches:
-        print(f"Close match found: {close_matches[0]}")
-        return _extract_nutrition_data(food_data, close_matches[0], portion_size)
-
-    # 3. Stemming and Lemmatization
-    stemmed_name = stem_food_name(food_name_lower)
-    lemmatized_name = lemmatize_food_name(food_name_lower)
-
-    stemmed_matches = get_close_matches(stemmed_name, [stem_food_name(name) for name in food_names], n=1, cutoff=0.6)
-    lemmatized_matches = get_close_matches(lemmatized_name, [lemmatize_food_name(name) for name in food_names], n=1, cutoff=0.6)
-
-    if stemmed_matches:
-        original_name = food_data['food_name'].iloc[[food_names.index(stemmed_matches[0])]].values[0]
-        print(f"Stemmed match found: {stemmed_matches[0]} -> {original_name}")
-        return _extract_nutrition_data(food_data, original_name, portion_size)
-
-    if lemmatized_matches:
-         original_name = food_data['food_name'].iloc[[food_names.index(lemmatized_matches[0])]].values[0]
-         print(f"Lemmatized match found: {lemmatized_matches[0]} -> {original_name}")
-         return _extract_nutrition_data(food_data, original_name, portion_size)
-
-    # 4. Synonym Lookup (if 'synonyms' column exists)
-    if 'synonyms' in food_data.columns:
-        for index, row in food_data.iterrows():
-            synonyms = row.get('synonyms', '')
-            if synonyms:
-                synonym_list = [s.strip().lower() for s in synonyms.split(',')]
-                if food_name_lower in synonym_list:
-                    print(f"Synonym match found: {food_name_lower} in synonyms for {row['food_name']}")
-                    return _extract_nutrition_data(food_data, row['food_name'].lower(), portion_size)
-
-    print(f"No nutrition information found for {food_name} using any matching method.")
-    return None
-
-def _extract_nutrition_data(food_data: pd.DataFrame, matched_food_name: str, portion_size: float) -> dict:
-    """Extracts nutrition data from a matched food item."""
-    matched_row = food_data[food_data['food_name'] == matched_food_name].iloc[0]
-
-    carb_col = 'carbohydrate_grams'
-    amount_col = 'serving_amount'
-    unit_col = 'serving_unit'
-    fiber_col = 'fiber_grams'
-    sugar_col = 'sugar_grams'  # Add Sugar
-
-    base_carbs = matched_row.get(carb_col, 0.0)
-    base_fiber = matched_row.get(fiber_col, 0.0)  # Get Fiber
-    base_sugar = matched_row.get(sugar_col, 0.0)  # Get Sugar
-
+def get_food_nutrition(food_name: str, food_data, portion_size: float = 1.0) -> Optional[Dict[str, Any]]:
+    """Get carbohydrate content for the given food, and also sugar and fiber"""
+    print("get_food_nutrition function called")  # Ensure the function is called
     try:
-        base_carbs = float(base_carbs)
-        base_fiber = float(base_fiber)  # Convert Fiber to float
-        base_sugar = float(base_sugar)  # Convert Sugar to float
-    except (ValueError, TypeError):
-        base_carbs = 0.0
-        base_fiber = 0.0  # Default Fiber
-        base_sugar = 0.0  # Default Sugar
-
-    serving_size = f"{matched_row.get(amount_col, 'Unknown')} {matched_row.get(unit_col, '')}"
-    adjusted_carbs = base_carbs * portion_size
-    adjusted_fiber = base_fiber * portion_size  # Adjust Fiber
-    adjusted_sugar = base_sugar * portion_size  # Adjust Sugar
-
-    print(f"Carbs: {base_carbs}, Fiber: {base_fiber}, Sugar: {base_sugar}") # Debug print
-
-    return {
-        'matched_food': matched_row['food_name'],
-        'category': matched_row.get('food_category', 'Unknown'),
-        'subcategory': matched_row.get('food_subcategory', 'Unknown'),
-        'base_carbs': base_carbs,
-        'adjusted_carbs': adjusted_carbs,
-        'base_fiber': base_fiber,        # Include Base Fiber
-        'adjusted_fiber': adjusted_fiber,  # Include Adjusted Fiber
-        'base_sugar': base_sugar,      # Include Base Sugar
-        'adjusted_sugar': adjusted_sugar, # Include Adjusted Sugar
-        'serving_size': serving_size,
-        'portion_multiplier': portion_size,
-        'notes': matched_row.get('notes', '')
-    }
-
+        # First try the local CSV database
+        food_name_lower = food_name.lower()  # Ensure input is also lowercase
+        food_names = food_data['food_name'].str.lower().tolist()
+
+        print(f"Searching for: {food_name_lower}")
+        matches = get_close_matches(food_name_lower, food_names, n=1, cutoff=0.5)
+
+        print(f"Matches found: {matches}")
+
+        if matches:
+            matched_row = food_data[food_data['food_name'].str.lower() == matches[0]]
+
+            if not matched_row.empty:
+                row = matched_row.iloc[0]
+
+                print(f"Matched row from CSV: {row}")
+
+                carb_col = 'carbohydrate_grams'
+                amount_col = 'serving_amount'
+                unit_col = 'serving_unit'
+                fiber_col = 'fiber_grams'  # Added Fiber column
+                sugar_col = 'sugar_grams'  # Added Sugar column
+
+                base_carbs = row.get(carb_col, 0.0) if pd.notna(row.get(carb_col, None)) else 0.0
+                base_fiber = row.get(fiber_col, 0.0) if pd.notna(row.get(fiber_col, None)) else 0.0
+                base_sugar = row.get(sugar_col, 0.0) if pd.notna(row.get(sugar_col, None)) else 0.0
+
+
+                if amount_col not in row or unit_col not in row or pd.isna(row[amount_col]) or pd.isna(row[unit_col]):
+                    serving_size = "Unknown"
+                    print(f"Warning: '{amount_col}' or '{unit_col}' is missing in CSV")
+                else:
+                    serving_size = f"{row[amount_col]} {row[unit_col]}"
+
+                adjusted_carbs = base_carbs * portion_size
+                adjusted_fiber = base_fiber * portion_size
+                adjusted_sugar = base_sugar * portion_size
+
+                return {
+                    'matched_food': row['food_name'],
+                    'category': row['food_category'] if 'food_category' in row and not pd.isna(row['food_category']) else 'Unknown',
+                    'subcategory': row['food_subcategory'] if 'food_subcategory' in row and not pd.isna(row['food_subcategory']) else 'Unknown',
+                    'base_carbs': base_carbs,
+                    'adjusted_carbs': adjusted_carbs,
+                    'base_fiber': base_fiber,
+                    'adjusted_fiber': adjusted_fiber,
+                    'base_sugar': base_sugar,
+                    'adjusted_sugar': adjusted_sugar,
+                    'serving_size': serving_size,
+                    'portion_multiplier': portion_size,
+                    'notes': row['notes'] if 'notes' in row and not pd.isna(row['notes']) else ''
+                }
+
+        print(f"No match found in CSV for {food_name}")
+        print(f"No nutrition information found for {food_name} in the local database.")
+        return None
+    except Exception as e:
+        print(f"Error in get_food_nutrition: {e}")
+        return None
 
-# -------------------------------------------------
-# Insulin and Glucose Calculations
-# -------------------------------------------------
-def get_basal_rate(current_time_hour, basal_rates):
-    """Gets the appropriate basal rate for a given time of day."""
-    for interval, rate in basal_rates.items():
-        try: # add a try and except to handle values in intervals that do not have the format "start-end"
-            parts = interval.split(":")[0].split("-")
-            if len(parts) == 2:  # Check if there are two parts (start and end)
-                start_hour, end_hour = map(int, parts)
-                if start_hour <= current_time_hour < end_hour or (start_hour <= current_time_hour and end_hour == 24):
-                    return rate
-        except Exception as e: # include exception in exception handling
-            print(f"Warning: Invalid interval format: {interval}. Skipping. Error: {e}")  #Inform user of formatting issues
-
-    return 0  # Default if no matching interval
-
-def insulin_activity(t, insulin_type, bolus_dose, bolus_duration=0):
-    """Models insulin activity over time."""
-    insulin_data = INSULIN_TYPES.get(insulin_type)
-    if not insulin_data:
-        return 0  # Or raise an error
-
-    # Simple exponential decay model (replace with a more sophisticated model)
-    peak_time = insulin_data['peak_time'] # Time in hours at which insulin activity is at max level
-    duration = insulin_data['duration'] # Total time for which insulin stays in effect
-    if t < peak_time:
-      activity = (bolus_dose * t / peak_time) * np.exp(1- t/peak_time) # rising activity
-    elif t < duration:
-      activity = bolus_dose * np.exp((peak_time - t) / (duration - peak_time)) # decaying activity
-    else:
-      activity = 0
-
-    if bolus_duration > 0: # Extended Bolus
-      if 0 <= t <= bolus_duration:
-          # Linear release of insulin over bolus_duration
-          effective_dose = bolus_dose / bolus_duration
-          duration = INSULIN_TYPES.get(insulin_type)['duration']
-          if t < duration:
-              activity = effective_dose
-          else:
-              activity = 0
-      else:
-          activity = 0
-
-    return activity
-
-def calculate_active_insulin(insulin_history, current_time):
-    """Calculates remaining active insulin from previous doses."""
-    active_insulin = 0
-    for dose_time, dose_amount, insulin_type, bolus_duration in insulin_history:
-        elapsed_time = current_time - dose_time
-        remaining_activity = insulin_activity(elapsed_time, insulin_type, dose_amount, bolus_duration)
-        active_insulin += remaining_activity
-    return active_insulin
-
-def calculate_insulin_needs(carbs, glucose_current, glucose_target, tdd, weight, insulin_type="Rapid-Acting", override_correction_dose = None):
-    """Calculate insulin needs for Type 1 diabetes"""
-    if tdd <= 0:
-        return {
-            'error': 'Total Daily Dose (TDD) must be greater than 0'
-        }
-    insulin_data = INSULIN_TYPES.get(insulin_type)
-    if not insulin_data:
-        return {
-            'error': "Invalid insulin type. Choose from" + ", ".join(INSULIN_TYPES.keys())
-        }
-
-    # Refined calculations
-    icr = (450 if weight <= 45 else 500) / tdd
-    isf = 1700 / tdd
-
-    # Calculate correction dose
-    glucose_difference = glucose_current - glucose_target
-    correction_dose = glucose_difference / isf
-
-    if override_correction_dose is not None:  # Check for None
-        correction_dose = override_correction_dose
-
-    # Calculate carb dose
-    carb_dose = carbs / icr
-
-    # Calculate total bolus
-    total_bolus = max(0, carb_dose + correction_dose)
-
-    # Calculate basal
-    basal_dose = weight * 0.5
-
-    return {
-        'icr': round(icr, 2),
-        'isf': round(isf, 2),
-        'correction_dose': round(correction_dose, 2),
-        'carb_dose': round(carb_dose, 2),
-        'total_bolus': round(total_bolus, 2),
-        'basal_dose': round(basal_dose, 2),
-        'insulin_type': insulin_type,
-        'insulin_onset': insulin_data['onset'],
-        'insulin_duration': insulin_data['duration'],
-        'peak_time': insulin_data['peak_time'],
-    }
 
 def create_detailed_report(nutrition_info, insulin_info, current_basal_rate):
     """Create a detailed report of carbs and insulin calculations"""
@@ -380,12 +226,12 @@ def create_detailed_report(nutrition_info, insulin_info, current_basal_rate):
     ------------------------
     Standard Serving Size: {nutrition_info['serving_size']}
     Carbs per Serving: {nutrition_info['base_carbs']}g
+    Adjusted Carbs: {nutrition_info['adjusted_carbs']}g
     Fiber per Serving: {nutrition_info['base_fiber']}g
+    Adjusted Fiber: {nutrition_info['adjusted_fiber']}g
     Sugar per Serving: {nutrition_info['base_sugar']}g
+    Adjusted Sugar: {nutrition_info['adjusted_sugar']}g
     Portion Multiplier: {nutrition_info['portion_multiplier']}x
-    Total Carbs: {nutrition_info['adjusted_carbs']}g
-    Total Fiber: {nutrition_info['adjusted_fiber']}g
-    Total Sugar: {nutrition_info['adjusted_sugar']}g
     Notes: {nutrition_info['notes']}
     """
 
@@ -425,17 +271,22 @@ def diabetes_dashboard(initial_glucose, food_image, stress_level, sleep_hours, t
         food_name = classify_food(food_image)  # This line is now inside the function
         print(f"Classified food name: {food_name}")  # Debugging: What is classified as? # Corrected indentation
         nutrition_info = get_food_nutrition(food_name, food_data, portion_size)  # Changed to pass in data
-
         if not nutrition_info:
-            return (
-                f"Could not find nutrition information for: {food_name} in the local database",
-                "No insulin calculations available",
-                None,
-                None,
-                None,
-                None,
-                None  # Fiber and Sugar Output
-            )
+            # Try with generic categories if specific food not found
+            generic_terms = food_name.split()
+            for term in generic_terms:
+                nutrition_info = get_food_nutrition(term, food_data, portion_size)  # Changed to pass in data
+                if nutrition_info:
+                    break
+
+            if not nutrition_info:
+                return (
+                    f"Could not find nutrition information for: {food_name} in the local database",  # Removed USDA ref
+                    "No insulin calculations available",
+                    None,
+                    None,
+                    None
+                )
 
         # 2. Insulin Calculations
         try:
@@ -455,7 +306,7 @@ def diabetes_dashboard(initial_glucose, food_image, stress_level, sleep_hours, t
         )
 
         if 'error' in insulin_info:
-            return insulin_info['error'], None, None, None, None, None, None
+            return insulin_info['error'], None, None, None, None
 
         # 3. Create detailed reports
         current_time_for_basal = 12 #Arbitrary number to pull from Basal Rates Dict
@@ -516,13 +367,11 @@ def diabetes_dashboard(initial_glucose, food_image, stress_level, sleep_hours, t
             insulin_details,
             insulin_info['basal_dose'],
             insulin_info['total_bolus'],
-            fig,
-            nutrition_info['base_fiber'],     # Base Fiber Output
-            nutrition_info['base_sugar']       # Base Sugar Output
+            fig
         )
 
     except Exception as e:
-        return f"Error: {str(e)}", None, None, None, None, None, None
+        return f"Error: {str(e)}", None, None, None, None
 
 # -------------------------------------------------
 # Gradio Interface Setup
@@ -570,8 +419,6 @@ with gr.Blocks() as app: # using Blocks API to manually design the layout
         basal_dose_output = gr.Number(label="Basal Insulin Dose (units/day)")
         bolus_dose_output = gr.Number(label="Bolus Insulin Dose (units)")
         glucose_plot_output = gr.Plot(label="Glucose Prediction")
-        fiber_output = gr.Number(label="Fiber (g)")
-        sugar_output = gr.Number(label="Sugar (g)")
 
     calculate_button.click(
         fn=diabetes_dashboard,
@@ -597,9 +444,7 @@ with gr.Blocks() as app: # using Blocks API to manually design the layout
             insulin_details_output,
             basal_dose_output,
             bolus_dose_output,
-            glucose_plot_output,
-            fiber_output,
-            sugar_output
+            glucose_plot_output
         ]
     )