app0.py

import gradio as gr
import pandas as pd
import numpy as np
import joblib
import plotly.graph_objects as go
import plotly.express as px
from huggingface_hub import hf_hub_download
import os
from pathlib import Path
import warnings
warnings.filterwarnings('ignore')
import re
from groq import Groq

# Initialize Groq client
groq_api_key = os.getenv("GROQ_API_KEY")
if groq_api_key:
    client = Groq(api_key=groq_api_key)
else:
    print("⚠️  GROQ_API_KEY not found. Chat functionality will be limited.")
    client = None

# Try to import xgboost, but fallback to scikit-learn
try:
    import xgboost as xgb
    XGB_AVAILABLE = True
    print("✅ XGBoost is available")
except ImportError:
    XGB_AVAILABLE = False
    print("⚠️  XGBoost not available, using scikit-learn models")
    from sklearn.ensemble import RandomForestRegressor

def create_dummy_model(model_type):
    """Create a realistic dummy model that has all required methods"""
    class RealisticDummyModel:
        def __init__(self, model_type):
            self.model_type = model_type
            self.n_features_in_ = 9
            self.feature_names_in_ = [
                'floor_area_sqm', 'storey_level', 'flat_age', 'remaining_lease',
                'transaction_year', 'flat_type_encoded', 'town_encoded',
                'flat_model_encoded', 'dummy_feature'
            ]
            # Add methods that might be called by joblib or other code
            self.get_params = lambda deep=True: {}
            self.set_params = lambda **params: self

        def predict(self, X):
            # Realistic prediction logic
            if isinstance(X, np.ndarray) and len(X.shape) == 2:
                X = X[0]  # Take first row if it's a 2D array

            floor_area = X[0]
            storey_level = X[1]
            flat_age = X[2]
            town_encoded = X[6]
            flat_type_encoded = X[5]

            base_price = floor_area * (4800 + town_encoded * 200)
            storey_bonus = storey_level * 2500
            age_discount = flat_age * 1800

            price = base_price + storey_bonus - age_discount + 35000
            if storey_level > 20: price += 15000
            if flat_age < 10: price += 20000

            return np.array([max(300000, price)])

    return RealisticDummyModel(model_type)()

def safe_joblib_load(filepath):
    """Safely load joblib file with error handling"""
    try:
        model = joblib.load(filepath)
        print(f"✅ Successfully loaded model from {filepath}")

        # Check if model has required methods
        if not hasattr(model, 'predict'):
            print("❌ Loaded object doesn't have predict method")
            return None

        # Add missing methods if needed
        if not hasattr(model, 'get_params'):
            model.get_params = lambda deep=True: {}
        if not hasattr(model, 'set_params'):
            model.set_params = lambda **params: model

        return model

    except Exception as e:
        print(f"❌ Error loading model from {filepath}: {e}")
        return None

def load_models():
    """Load models with robust error handling"""
    models = {}

    # Try to load XGBoost model
    try:
        xgboost_path = hf_hub_download(
            repo_id="Lesterchia174/HDB_Price_Predictor",
            filename="best_model_xgboost.joblib",
            repo_type="space"
        )
        models['xgboost'] = safe_joblib_load(xgboost_path)
        if models['xgboost'] is None:
            print("⚠️  Creating dummy model for XGBoost")
            models['xgboost'] = create_dummy_model("xgboost")
        else:
            print("✅ XGBoost model loaded and validated")

    except Exception as e:
        print(f"❌ Error downloading XGBoost model: {e}")
        print("⚠️  Creating dummy model for XGBoost")
        models['xgboost'] = create_dummy_model("xgboost")

    return models

def load_data():
    """Load data using Hugging Face Hub"""
    try:
        data_path = hf_hub_download(
            repo_id="Lesterchia174/HDB_Price_Predictor",
            filename="base_hdb_resale_prices_2015Jan-2025Jun_processed.csv",
            repo_type="space"
        )
        df = pd.read_csv(data_path)
        print("✅ Data loaded successfully via Hugging Face Hub")
        return df
    except Exception as e:
        print(f"❌ Error loading data: {e}")
        return create_sample_data()

def create_sample_data():
    """Create sample data if real data isn't available"""
    np.random.seed(42)
    towns = ['ANG MO KIO', 'BEDOK', 'TAMPINES', 'WOODLANDS', 'JURONG WEST']
    flat_types = ['4 ROOM', '5 ROOM', 'EXECUTIVE']
    flat_models = ['Improved', 'Model A', 'New Generation']

    data = []
    for _ in range(100):
        town = np.random.choice(towns)
        flat_type = np.random.choice(flat_types)
        flat_model = np.random.choice(flat_models)
        floor_area = np.random.randint(85, 150)
        storey = np.random.randint(1, 25)
        age = np.random.randint(0, 40)

        base_price = floor_area * 5000
        town_bonus = towns.index(town) * 20000
        storey_bonus = storey * 2000
        age_discount = age * 1500
        flat_type_bonus = flat_types.index(flat_type) * 30000

        resale_price = base_price + town_bonus + storey_bonus - age_discount + flat_type_bonus
        resale_price = max(300000, resale_price + np.random.randint(-20000, 20000))

        data.append({
            'town': town, 'flat_type': flat_type, 'flat_model': flat_model,
            'floor_area_sqm': floor_area, 'storey_level': storey,
            'flat_age': age, 'resale_price': resale_price
        })

    return pd.DataFrame(data)

def preprocess_input(user_input, model_type='xgboost'):
    """Preprocess user input for prediction with correct feature mapping"""
    # Flat type mapping
    flat_type_mapping = {'1 ROOM': 1, '2 ROOM': 2, '3 ROOM': 3, '4 ROOM': 4,
                         '5 ROOM': 5, 'EXECUTIVE': 6, 'MULTI-GENERATION': 7}

    # Town mapping
    town_mapping = {
        'SENGKANG': 0, 'WOODLANDS': 1, 'TAMPINES': 2, 'PUNGGOL': 3,
        'JURONG WEST': 4, 'YISHUN': 5, 'BEDOK': 6, 'HOUGANG': 7,
        'CHOA CHU KANG': 8, 'ANG MO KIO': 9
    }

    # Flat model mapping
    flat_model_mapping = {
        'Model A': 0, 'Improved': 1, 'New Generation': 2,
        'Standard': 3, 'Premium': 4
    }

    # Create input array with features
    input_features = [
        user_input['floor_area_sqm'],           # Feature 1
        user_input['storey_level'],             # Feature 2
        user_input['flat_age'],                 # Feature 3
        99 - user_input['flat_age'],            # Feature 4: remaining_lease
        2025,                                   # Feature 5: transaction_year
        flat_type_mapping.get(user_input['flat_type'], 4),  # Feature 6: flat_type_ordinal
        town_mapping.get(user_input['town'], 0),           # Feature 7: town_encoded
        flat_model_mapping.get(user_input['flat_model'], 0), # Feature 8: flat_model_encoded
        1                                       # Feature 9: (placeholder)
    ]

    return np.array([input_features])

def create_market_insights_chart(data, user_input, predicted_price):
    """Create market insights visualization"""
    if data is None or len(data) == 0:
        return None

    similar_properties = data[
        (data['flat_type'] == user_input['flat_type']) &
        (data['town'] == user_input['town'])
    ]

    if len(similar_properties) < 5:
        similar_properties = data[data['flat_type'] == user_input['flat_type']]

    if len(similar_properties) > 0:
        fig = px.scatter(similar_properties, x='floor_area_sqm', y='resale_price',
                         color='flat_model',
                         title=f"Market Position: {user_input['flat_type']} in {user_input['town']}",
                         labels={'floor_area_sqm': 'Floor Area (sqm)', 'resale_price': 'Resale Price (SGD)'})

        # Add model prediction
        fig.add_trace(go.Scatter(x=[user_input['floor_area_sqm']], y=[predicted_price],
                                 mode='markers',
                                 marker=dict(symbol='star', size=20, color='red',
                                             line=dict(width=2, color='darkred')),
                                 name='XGBoost Prediction'))

        fig.update_layout(template="plotly_white", height=400, showlegend=True)
        return fig
    return None

def predict_hdb_price(town, flat_type, flat_model, floor_area_sqm, storey_level, flat_age):
    """Main prediction function for Gradio with robust error handling"""
    user_input = {
        'town': town,
        'flat_type': flat_type,
        'flat_model': flat_model,
        'floor_area_sqm': floor_area_sqm,
        'storey_level': storey_level,
        'flat_age': flat_age
    }

    try:
        processed_input = preprocess_input(user_input)

        # Get prediction with error handling
        try:
            predicted_price = max(0, float(models['xgboost'].predict(processed_input)[0]))
        except Exception as e:
            print(f"❌ XGBoost prediction error: {e}")
            predicted_price = 400000  # Fallback value

        # Create insights
        remaining_lease = 99 - flat_age
        price_per_sqm = predicted_price / floor_area_sqm

        insights = f"""
        **Property Summary:**
        - Location: {town}
        - Type: {flat_type}
        - Model: {flat_model}
        - Area: {floor_area_sqm} sqm
        - Floor: Level {storey_level}
        - Age: {flat_age} years
        - Remaining Lease: {remaining_lease} years
        - Price per sqm: ${price_per_sqm:,.0f}

        **Predicted Price: ${predicted_price:,.0f}**

        **Financing Eligibility:**
        """

        if remaining_lease >= 60:
            insights += "✅ Bank loan eligible"
        elif remaining_lease >= 20:
            insights += "⚠️ HDB loan eligible only"
        else:
            insights += "❌ Limited financing options"

        # Create chart
        chart = create_market_insights_chart(data, user_input, predicted_price)

        return f"${predicted_price:,.0f}", chart, insights

    except Exception as e:
        error_msg = f"Prediction failed. Error: {str(e)}"
        print(error_msg)
        return "Error: Prediction failed", None, error_msg

def extract_parameters_from_query(query):
    """Extract HDB parameters from natural language query using LLM"""
    if not groq_api_key or client is None:
        return {"error": "Please set GROQ_API_KEY environment variable to use chat functionality."}

    try:
        # System prompt to guide the LLM
        system_prompt = """You are an expert at extracting parameters for HDB price prediction from natural language queries.
        Extract the following parameters if mentioned in the query:
        - town (e.g., Ang Mo Kio, Bedok, Tampines)
        - flat_type (e.g., 3 ROOM, 4 ROOM, 5 ROOM, EXECUTIVE)
        - flat_model (e.g., Improved, Model A, New Generation, Standard, Premium)
        - floor_area_sqm (floor area in square meters)
        - storey_level (floor level)
        - flat_age (age of flat in years)

        Return only a JSON object with the extracted parameters. If a parameter is not mentioned, set it to null.
        Example: {"town": "ANG MO KIO", "flat_type": "4 ROOM", "flat_model": "Improved", "floor_area_sqm": 95, "storey_level": 8, "flat_age": 15}"""

        # Query the LLM
        completion = client.chat.completions.create(
            model="llama-3.3-70b-versatile",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query}
            ],
            temperature=0.1,
            max_tokens=200
        )

        # Extract and parse the JSON response
        response = completion.choices[0].message.content
        # Clean the response to extract just the JSON
        json_match = re.search(r'\{.*\}', response, re.DOTALL)
        if json_match:
            import json
            params = json.loads(json_match.group())
            return params
        else:
            return {"error": "Could not extract parameters from query"}

    except Exception as e:
        return {"error": f"Error processing query: {str(e)}"}

def is_small_talk(query):
    """Check if the query is small talk/casual conversation"""
    small_talk_keywords = [
        'hello', 'hi', 'hey', 'good morning', 'good afternoon', 'good evening',
        'how are you', 'how are things', "what's up", 'how do you do',
        'thank you', 'thanks', 'bye', 'goodbye', 'see you', 'nice to meet you',
        'who are you', 'what can you do', 'help', 'tell me about yourself'
    ]
    
    query_lower = query.lower()
    return any(keyword in query_lower for keyword in small_talk_keywords)

def handle_small_talk(query):
    """Handle small talk queries with appropriate responses"""
    query_lower = query.lower()
    
    if any(greeting in query_lower for greeting in ['hello', 'hi', 'hey', 'good morning', 'good afternoon', 'good evening']):
        return "Hello! 👋 I'm your HDB price assistant. How can I help you today?"
    
    elif any(how_are_you in query_lower for how_are_you in ['how are you', 'how are things', "what's up", 'how do you do']):
        return "I'm doing great, thanks for asking! I'm here to help you with HDB price predictions and information. What can I assist you with today?"
    
    elif any(thanks in query_lower for thanks in ['thank you', 'thanks']):
        return "You're welcome! 😊 Is there anything else you'd like to know about HDB prices?"
    
    elif any(bye in query_lower for bye in ['bye', 'goodbye', 'see you']):
        return "Goodbye! 👋 Feel free to come back if you have more questions about HDB prices!"
    
    elif 'who are you' in query_lower:
        return "I'm an AI assistant specialized in helping with HDB resale price predictions and information. I can estimate property values based on various factors like location, flat type, size, and age."
    
    elif 'what can you do' in query_lower or 'help' in query_lower:
        return "I can help you with:\n- Predicting HDB resale prices\n- Answering questions about HDB properties\n- Providing market insights\n\nJust tell me about the property you're interested in (location, type, size, etc.) and I'll give you an estimate!"
    
    elif 'tell me about yourself' in query_lower:
        return "I'm an AI assistant powered by machine learning models trained on HDB resale data. I can provide price estimates and insights about public housing in Singapore. My goal is to help you make informed decisions about HDB properties!"
    
    else:
        return "I'm here to help with HDB price predictions and information. How can I assist you today?"


def chat_with_llm(query, chat_history):
    """Handle chat queries about HDB pricing and small talk"""
    if not groq_api_key or client is None:
        return "Please set GROQ_API_KEY...", chat_history

    # 1. First, check for small talk
    if is_small_talk(query):
        response = handle_small_talk(query)
        chat_history.append((query, response))
        return response, chat_history

    # 2. Check if the query is a clear request for a general explanation/trend (not a specific price)
    is_general_query = any(keyword in query.lower() for keyword in [
        'trend', 'overview', 'how are', 'what are', 'like in', 'average', 
        'over the years', 'market', 'compare'
    ])
    
    # 3. If it's a general query, use the LLM to answer it directly
    if is_general_query:
        try:
            completion = client.chat.completions.create(
                model="llama-3.3-70b-versatile",
                messages=[
                    {
                        "role": "system", 
                        "content": "You are a helpful assistant specialized in HDB (Housing & Development Board) properties in Singapore. Provide accurate, helpful information about HDB prices, policies, and market trends. Use the provided context if available."
                    },
                    {
                        "role": "user", 
                        "content": f"Based on general HDB market knowledge, answer this question: {query}"
                    }
                ],
                temperature=0.3,
                max_tokens=500
            )
            response = completion.choices[0].message.content
            chat_history.append((query, response))
            return response, chat_history
        except Exception as e:
            error_msg = f"I encountered an error. Please try again later."
            chat_history.append((query, error_msg))
            return error_msg, chat_history

    # 4. If it's not clearly general, try to extract parameters for a specific prediction
    params = extract_parameters_from_query(query)

    if "error" in params:
        # If extraction failed, fall back to general Q&A
        return answer_general_hdb_question(query, chat_history)

    # 5. Check what we got back from parameter extraction
    extracted_params = {k: v for k, v in params.items() if v is not None}
    required_for_prediction = ['town', 'flat_type', 'floor_area_sqm', 'storey_level', 'flat_age']

    # 6. If the user only provided a town or one other parameter, it's likely a general question.
    if len(extracted_params) < 3: # e.g., if only 'town' and 'flat_type' are provided
        # Ask a clarifying question or provide a general overview
        if 'town' in extracted_params:
            town = extracted_params['town']
            # You could add a pre-generated fact here, e.g., average price for that town from the dataset
            response = f"You asked about {town}. HDB prices can vary widely based on flat type, size, age, and specific location within the town. "
            response += f"For example, are you interested in 4-Room or 5-Room flats? What's your budget or preferred size? "
            response += "Alternatively, I can give you a prediction if you provide more details like flat type, size, and age."
        else:
            response = "I specialize in HDB price predictions and information. Could you provide more details about the property you're interested in (e.g., town, flat type, size) so I can give you a accurate estimate or information?"
        chat_history.append((query, response))
        return response, chat_history

    # 7. If we have most parameters, ask for the missing ones specifically
    missing_params = [param for param in required_for_prediction if params.get(param) is None]
    if missing_params:
        missing_list = ", ".join(missing_params)
        response = f"I'd be happy to predict a price for you. I just need a few more details: {missing_list}."
        chat_history.append((query, response))
        return response, chat_history

    # 8. If we have all parameters, make the prediction!
    # ... (rest of the prediction code remains the same)


    
    # If we have all parameters, make a prediction
    try:
        # Convert string numbers to appropriate types
        if isinstance(params['floor_area_sqm'], str):
            params['floor_area_sqm'] = float(params['floor_area_sqm'])
        if isinstance(params['storey_level'], str):
            params['storey_level'] = int(params['storey_level'])
        if isinstance(params['flat_age'], str):
            params['flat_age'] = int(params['flat_age'])

        # Make prediction
        price, chart, insights = predict_hdb_price(
            params['town'], params['flat_type'], params['flat_model'],
            params['floor_area_sqm'], params['storey_level'], params['flat_age']
        )

        # Format response
        response = f"Based on your query:\n\n"
        response += f"📍 Town: {params['town']}\n"
        response += f"🏠 Flat Type: {params['flat_type']}\n"
        response += f"📐 Floor Area: {params['floor_area_sqm']} sqm\n"
        response += f"🏢 Storey Level: {params['storey_level']}\n"
        response += f"📅 Flat Age: {params['flat_age']} years\n\n"
        response += f"💰 Predicted Price: {price}\n\n"
        response += insights

        chat_history.append((query, response))
        return response, chat_history

    except Exception as e:
        error_msg = f"Error making prediction: {str(e)}"
        chat_history.append((query, error_msg))
        return error_msg, chat_history

# Preload models and data
print("Loading models and data...")
models = load_models()
data = load_data()

# Define Gradio interface
towns_list = [
    'SENGKANG', 'WOODLANDS', 'TAMPINES', 'PUNGGOL', 'JURONG WEST',
    'YISHUN', 'BEDOK', 'HOUGANG', 'CHOA CHU KANG', 'ANG MO KIO'
]

flat_types = ['3 ROOM', '4 ROOM', '5 ROOM', 'EXECUTIVE', '2 ROOM', '1 ROOM']
flat_models = ['Model A', 'Improved', 'New Generation', 'Standard', 'Premium']

# Create Gradio interface with chatbot
with gr.Blocks(title="🏠 HDB Price Predictor + Chat", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🏠 HDB Price Predictor + AI Chat")
    gr.Markdown("Predict HDB resale prices using XGBoost model or chat with our AI assistant")

    with gr.Tab("Traditional Interface"):
        with gr.Row():
            with gr.Column():
                town = gr.Dropdown(label="Town", choices=sorted(towns_list), value="ANG MO KIO")
                flat_type = gr.Dropdown(label="Flat Type", choices=sorted(flat_types), value="4 ROOM")
                flat_model = gr.Dropdown(label="Flat Model", choices=sorted(flat_models), value="Improved")
                floor_area_sqm = gr.Slider(label="Floor Area (sqm)", minimum=30, maximum=200, value=95, step=5)
                storey_level = gr.Slider(label="Storey Level", minimum=1, maximum=50, value=8, step=1)
                flat_age = gr.Slider(label="Flat Age (years)", minimum=0, maximum=99, value=15, step=1)

                predict_btn = gr.Button("🔮 Predict Price", variant="primary")

            with gr.Column():
                predicted_price = gr.Label(label="💰 Predicted Price")
                insights = gr.Markdown(label="📋 Property Summary")

        with gr.Row():
            chart_output = gr.Plot(label="📈 Market Insights")

        # Connect button to function
        predict_btn.click(
            fn=predict_hdb_price,
            inputs=[town, flat_type, flat_model, floor_area_sqm, storey_level, flat_age],
            outputs=[predicted_price, chart_output, insights]
        )

    with gr.Tab("AI Chat Assistant"):
        gr.Markdown("💬 Chat with our AI assistant to get HDB price predictions using natural language!")
        gr.Markdown("Example: 'What would be the price of a 4-room model A flat in Ang Mo Kio with 95 sqm, on the 8th floor, that's 15 years old?'")
        gr.Markdown("You can also say hello, ask how I am, or ask general questions about HDB!")

        chatbot = gr.Chatbot(label="HDB Price Chatbot", height=500)
        msg = gr.Textbox(label="Your question", placeholder="Type your message here...")
        clear = gr.Button("Clear Chat")

        def respond(message, chat_history):
            response, updated_history = chat_with_llm(message, chat_history)
            return updated_history

        msg.submit(respond, [msg, chatbot], [chatbot])
        clear.click(lambda: None, None, [chatbot], queue=False)

# To run in Colab
if __name__ == "__main__":
    # Let Gradio automatically find an available port
    demo.launch(server_name="0.0.0.0", share=True)