""" Enhanced Streamlit application for engine predictive maintenance. Intended to be run locally or deployed as a Hugging Face Space. Features an interactive, modern UI with real-time predictions, visualizations, and detailed insights. """ from __future__ import annotations import os import pandas as pd import plotly.graph_objects as go import plotly.express as px import streamlit as st import config from inference import predict_engine_condition def _is_running_in_hf_space() -> bool: """Check if app is running in Hugging Face Space.""" # HF Spaces set SPACE_ID or SYSTEM environment variable return os.getenv("SPACE_ID") is not None or os.getenv("SYSTEM") == "spaces" def _get_default_source() -> str: """Decide whether to load the model from HF or local based on env vars.""" # In HF Space, always use HF model if _is_running_in_hf_space(): return "hf" if config.HF_TOKEN and config.HF_MODEL_REPO: return "hf" return "local" def create_gauge_chart(value: float, title: str, color: str) -> go.Figure: """Create a gauge chart for sensor readings.""" fig = go.Figure(go.Indicator( mode="gauge+number", value=value, domain={'x': [0, 1], 'y': [0, 1]}, title={'text': title, 'font': {'size': 16}}, gauge={ 'axis': {'range': [None, 100]}, 'bar': {'color': color}, 'steps': [ {'range': [0, 50], 'color': "lightgray"}, {'range': [50, 80], 'color': "gray"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 90 } } )) fig.update_layout(height=200, margin=dict(l=20, r=20, t=40, b=20)) return fig def create_sensor_comparison_chart(sensor_data: dict) -> go.Figure: """Create a radar chart comparing sensor values.""" categories = list(sensor_data.keys()) values = list(sensor_data.values()) # Normalize values for better visualization (0-100 scale) max_values = { "Engine_RPM": 4000, "Lub_Oil_Pressure": 10, "Fuel_Pressure": 30, "Coolant_Pressure": 10, "Lub_Oil_Temperature": 150, "Coolant_Temperature": 150, } normalized_values = [ (v / max_values.get(k, 100)) * 100 for k, v in zip(categories, values) ] fig = go.Figure() fig.add_trace(go.Scatterpolar( r=normalized_values + [normalized_values[0]], # Close the loop theta=[k.replace("_", " ") for k in categories] + [categories[0].replace("_", " ")], fill='toself', name='Current Readings', line_color='#1f77b4' )) fig.update_layout( polar=dict( radialaxis=dict( visible=True, range=[0, 100] )), showlegend=True, height=400, title="Sensor Readings Overview" ) return fig def main() -> None: # MUST be first Streamlit command st.set_page_config( page_title="Engine Predictive Maintenance", page_icon="đŸ› ī¸", layout="wide", initial_sidebar_state="expanded" ) # Custom CSS for better styling st.markdown(""" """, unsafe_allow_html=True) # Compact Header st.markdown('

đŸ› ī¸ Engine Predictive Maintenance

', unsafe_allow_html=True) st.markdown('

AI-Powered Engine Health Monitoring & Failure Prediction

', unsafe_allow_html=True) # Sidebar with st.sidebar: # In HF Space, always use HF model and hide selection is_in_space = _is_running_in_hf_space() default_source = _get_default_source() if is_in_space: # In Space: completely hide model source selection, always use HF source = "hf" # Don't show any configuration UI in Space else: # Local development: show configuration section st.header("âš™ī¸ Configuration") source = st.radio( "đŸ“Ļ Model Source:", options=["local", "hf"], index=0 if default_source == "hf" else 1, format_func=lambda x: "🤖 Hugging Face Hub" if x == "hf" else "💾 Local File", help="Select where to load the trained model from" ) st.markdown("---") st.header("📊 Quick Stats") if is_in_space: # In Space: always show HF model status if config.HF_TOKEN and config.HF_MODEL_REPO: st.success("✅ Model Ready") st.caption(f"Loading from: {config.HF_MODEL_REPO}") else: st.error("❌ Configuration Missing") st.caption("Set HF_TOKEN as Space secret in Settings") st.markdown(""" **To fix:** 1. Go to Space Settings 2. Add secret: `HF_TOKEN` 3. Restart Space """) else: # Local development: check local model if os.path.exists(config.BEST_MODEL_LOCAL_PATH): st.success("✅ Model Available") st.caption("Trained model found locally") else: st.warning("âš ī¸ Model Not Found") st.caption("Run training script first") st.markdown("---") st.header("â„šī¸ About") st.markdown(""" This application uses machine learning to predict engine failures based on: - Engine RPM - Oil & Fuel Pressures - Coolant Pressure - Temperature Readings **Status**: 0 = Normal | 1 = Requires Maintenance """) st.markdown("---") st.caption("Built with â¤ī¸ using Streamlit & Scikit-learn") # Balanced layout - inputs on left, larger visualization on right col_input, col_viz = st.columns([1, 1.2]) with col_input: # Input form with st.form(key="engine_form", clear_on_submit=False): st.markdown("### 🔧 Sensor Inputs") # 2 columns for inputs col_a, col_b = st.columns(2) with col_a: engine_rpm = st.number_input( "âš™ī¸ Engine RPM", min_value=0.0, max_value=4000.0, value=800.0, step=10.0, help="Revolutions per minute" ) lub_oil_pressure = st.number_input( "đŸ›ĸī¸ Lub Oil Pressure", min_value=0.0, max_value=10.0, value=3.0, step=0.1, help="bar/kPa" ) fuel_pressure = st.number_input( "â›Ŋ Fuel Pressure", min_value=0.0, max_value=30.0, value=10.0, step=0.1, help="bar/kPa" ) with col_b: coolant_pressure = st.number_input( "💧 Coolant Pressure", min_value=0.0, max_value=10.0, value=2.0, step=0.1, help="bar/kPa" ) lub_oil_temp = st.number_input( "đŸŒĄī¸ Lub Oil Temp", min_value=0.0, max_value=150.0, value=80.0, step=0.5, help="°C" ) coolant_temp = st.number_input( "đŸŒĄī¸ Coolant Temp", min_value=0.0, max_value=150.0, value=80.0, step=0.5, help="°C" ) submitted = st.form_submit_button("🚀 Predict Engine Condition", use_container_width=True) with col_viz: st.markdown("### 📊 Sensor Visualization") # Real-time sensor visualization sensor_data = { "Engine_RPM": engine_rpm, "Lub_Oil_Pressure": lub_oil_pressure, "Fuel_Pressure": fuel_pressure, "Coolant_Pressure": coolant_pressure, "Lub_Oil_Temperature": lub_oil_temp, "Coolant_Temperature": coolant_temp, } # Larger, more readable radar chart radar_fig = create_sensor_comparison_chart(sensor_data) radar_fig.update_layout(height=450, margin=dict(l=40, r=40, t=50, b=40)) # Larger and more readable st.plotly_chart(radar_fig, use_container_width=True, config={'displayModeBar': False}) # Prediction results if submitted: inputs = { "Engine_RPM": engine_rpm, "Lub_Oil_Pressure": lub_oil_pressure, "Fuel_Pressure": fuel_pressure, "Coolant_Pressure": coolant_pressure, "Lub_Oil_Temperature": lub_oil_temp, "Coolant_Temperature": coolant_temp, } # Check if HF_TOKEN is set when using HF model if source == "hf" and not config.HF_TOKEN: st.error("❌ **HF_TOKEN not configured**") st.markdown(""" **To fix this:** 1. Go to Space Settings → Repository secrets 2. Add secret: `HF_TOKEN` with your Hugging Face token 3. Restart the Space Get token from: https://huggingface.co/settings/tokens """) st.stop() with st.spinner("🤖 Loading model and analyzing sensor data..."): try: result = predict_engine_condition(inputs=inputs, source=source) except Exception as e: st.error( f"❌ **Prediction Failed**\n\n" f"Error: {str(e)}\n\n" f"**Troubleshooting:**\n" f"- Ensure the model is trained: `python src/train.py`\n" f"- Check model file exists: `models/best_model.joblib`\n" f"- Verify HF credentials if using Hugging Face Hub" ) return pred_label = result["prediction"] prob_faulty = result["probability_faulty"] prob_normal = 1 - prob_faulty # Compact results section st.markdown("---") result_col1, result_col2 = st.columns([1.5, 1]) with result_col1: if pred_label == 1: st.markdown( f'
' f'

🚨 MAINTENANCE REQUIRED

' f'

Engine is LIKELY FAULTY - Fault Probability: {prob_faulty:.1%}

' f'
', unsafe_allow_html=True ) with st.expander("🔧 Recommended Actions", expanded=False): st.markdown(""" - Schedule immediate engine inspection - Verify sensor readings are accurate - Review maintenance history - Consult maintenance specialist """) else: st.markdown( f'
' f'

✅ ENGINE HEALTHY

' f'

Engine is OPERATING NORMALLY - Fault Probability: {prob_faulty:.1%}

' f'
', unsafe_allow_html=True ) st.success("✅ All sensors within normal ranges. Continue regular monitoring.") with result_col2: # Compact probability gauge fig = go.Figure(go.Indicator( mode="gauge+number", value=prob_faulty * 100, domain={'x': [0, 1], 'y': [0, 1]}, title={'text': "Fault Risk %", 'font': {'size': 16}}, gauge={ 'axis': {'range': [None, 100]}, 'bar': {'color': "darkred" if pred_label == 1 else "darkgreen"}, 'steps': [ {'range': [0, 30], 'color': "lightgreen"}, {'range': [30, 70], 'color': "yellow"}, {'range': [70, 100], 'color': "lightcoral"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 70 } } )) fig.update_layout(height=200, margin=dict(l=10, r=10, t=30, b=10)) st.plotly_chart(fig, use_container_width=True, config={'displayModeBar': False}) # Compact metrics col_m1, col_m2 = st.columns(2) with col_m1: st.metric("Normal", f"{prob_normal:.0%}") with col_m2: st.metric("Fault", f"{prob_faulty:.0%}") # Compact Footer st.markdown("---") st.markdown("""

đŸ› ī¸ Predictive Maintenance System | Built with Streamlit, Scikit-learn & Plotly | Developed by Anant Tripathi

âš ī¸ Use as decision-support tool, not replacement for expert diagnostics

""", unsafe_allow_html=True) if __name__ == "__main__": main()