import numpy as np import pandas as pd import matplotlib.pyplot as plt import gradio as gr import warnings # Suppress warnings to keep output clean warnings.filterwarnings("ignore") # ---------------------------------------------------------------------------- # Import Prophet (Meta's forecasting library) # ---------------------------------------------------------------------------- try: from prophet import Prophet except Exception as e: raise ImportError( "Prophet is not installed. Run: pip install prophet\n" "If you still get errors, try: pip install cmdstanpy prophet" ) from e # ---------------------------------------------------------------------------- # Import ARIMA and SARIMA from statsmodels # ---------------------------------------------------------------------------- try: from statsmodels.tsa.arima.model import ARIMA from statsmodels.tsa.statespace.sarimax import SARIMAX except Exception as e: raise ImportError( "statsmodels is not installed. Run: pip install statsmodels" ) from e # ============================================================================ # CONFIGURATION / SETTINGS # ============================================================================ # What percentage of data to use for training? (0.70 = 70%) # The remaining 30% will be used for testing our predictions TRAIN_FRAC = 0.70 # ---------------------------------------------------------------------------- # ARIMA PARAMETERS: (p, d, q) # ---------------------------------------------------------------------------- # p (AutoRegressive order): How many past values to use # - Higher p = model looks further back in time # - Example: p=3 means "use the last 3 values to predict the next one" # # d (Differencing order): How many times to difference the data # - Differencing removes trends by subtracting consecutive values # - d=1 means: new_value = original_value - previous_value # - This turns an upward trend into a flat line (easier to model!) # # q (Moving Average order): How many past forecast errors to use # - Helps the model learn from its mistakes # - Example: q=2 means "adjust based on the last 2 prediction errors" # ARIMA_ORDER = (5, 1, 2) # We use p=5, d=1, q=2 # ---------------------------------------------------------------------------- # SARIMA PARAMETERS: (p, d, q) x (P, D, Q, s) # ---------------------------------------------------------------------------- # The first part (p, d, q) is the same as ARIMA (explained above) # # The second part (P, D, Q, s) handles SEASONALITY: # P: Seasonal autoregressive order (like p, but for seasonal patterns) # D: Seasonal differencing (removes seasonal trends) # Q: Seasonal moving average order # s: The SEASONAL PERIOD - how many time steps until the pattern repeats # # IMPORTANT: We use SIMPLIFIED parameters for speed! # ---------------------------------------------------------------------------- # Original (slow): (2,1,2) x (1,1,1,60) on 3000+ points = 20-60 minutes # Optimized (fast): (1,1,1) x (1,0,1,24) on ~75 points = ~30 seconds # ---------------------------------------------------------------------------- SARIMA_ORDER = (1, 1, 1) # Simplified non-seasonal part SARIMA_SEASONAL_ORDER = (1, 0, 1, 24) # s=24 for daily pattern in hourly data # How much to downsample for SARIMA (use every Nth point) # 60 = convert minute data to hourly data (average every 60 minutes) SARIMA_DOWNSAMPLE_FACTOR = 60 # ---------------------------------------------------------------------------- # PROPHET PARAMETERS # ---------------------------------------------------------------------------- # changepoint_prior_scale: How flexible is the trend? # - Higher value = trend can change more dramatically # - Lower value = smoother, more stable trend # - Default is 0.05, we use 0.05 # # seasonality_prior_scale: How strong are seasonal effects? # - Higher value = stronger seasonal patterns # - Lower value = weaker seasonal patterns # # n_changepoints: How many points where trend can change direction? # - More changepoints = more flexible trend # PROPHET_CHANGEPOINT_PRIOR = 0.05 PROPHET_SEASONALITY_PRIOR = 10.0 PROPHET_N_CHANGEPOINTS = 25 # ============================================================================ # DATA GENERATION # ============================================================================ # We'll create SYNTHETIC environmental sensor data # This simulates what you might see in a space station or submarine! # # We generate 4 types of measurements: # 1. O2 (Oxygen percentage) - normally around 21% # 2. CO2 (Carbon Dioxide in ppm) - normally around 400 ppm # 3. VOCs (Volatile Organic Compounds in ppb) - chemical pollutants # 4. PM (Particulate Matter in μg/m³) - tiny particles in the air def smooth_noise(n_points, low, high, window=30): # Generate random numbers from a UNIFORM distribution # Uniform means every number between low and high is equally likely raw_noise = np.random.uniform(low, high, n_points) # Apply moving average using convolution # np.ones(window)/window creates an averaging filter # Example: window=3 creates [0.333, 0.333, 0.333] smoothed = np.convolve(raw_noise, np.ones(window) / window, mode='same') return smoothed def generate_mission_data(seed=42, n_points=4500): # Set random seed so we get the same "random" data each time # This is important for reproducibility in science! np.random.seed(seed) # Create time array: 0, 1, 2, 3, ... n_points-1 t = np.arange(n_points) # Define seasonal periods (in minutes) daily_period = 1440 # 24 hours × 60 minutes = 1440 minutes per day weekly_period = 10080 # 7 days × 1440 minutes = 10080 minutes per week # ======================================================================== # O2 (OXYGEN) - normally around 20.9% # ======================================================================== # In a closed environment, O2 slowly decreases as people breathe # It follows an INVERSE pattern to CO2 (when CO2 goes up, O2 goes down) o2_baseline = 20.9 # Starting value (normal atmospheric O2) o2_trend = -0.00005 * t # Very slow downward trend # Seasonal patterns using SINE WAVES # Why sine waves? They naturally create smooth, repeating cycles! # The +np.pi shifts the wave to be opposite of CO2 o2_daily = 0.15 * np.sin(2 * np.pi * t / daily_period + np.pi) o2_weekly = 0.08 * np.sin(2 * np.pi * t / weekly_period) # Add smooth random noise o2_noise = smooth_noise(n_points, -0.05, 0.05, window=40) # Combine all components and clip to realistic range # np.clip ensures values stay between 19.5% and 21.5% o2 = np.clip(o2_baseline + o2_trend + o2_daily + o2_weekly + o2_noise, 19.5, 21.5) # ======================================================================== # CO2 (CARBON DIOXIDE) - normally around 400 ppm # ======================================================================== # CO2 increases when people breathe and during activities # Has strong daily patterns (higher during active hours) co2_baseline = 400 # Normal atmospheric CO2 co2_trend = 0.015 * t # Gradual upward trend co2_daily = 40 * np.sin(2 * np.pi * t / daily_period) # Daily cycle co2_weekly = 20 * np.sin(2 * np.pi * t / weekly_period) # Weekly cycle co2_noise = smooth_noise(n_points, -10, 10, window=30) co2 = np.clip(co2_baseline + co2_trend + co2_daily + co2_weekly + co2_noise, 350, 800) # ======================================================================== # VOCs (VOLATILE ORGANIC COMPOUNDS) - chemical pollutants # ======================================================================== # VOCs come from cleaning products, cooking, materials off-gassing # Pattern is slightly shifted from CO2 (different sources) voc_baseline = 50 voc_trend = 0.003 * t # The -np.pi/4 creates a phase shift (pattern peaks at different times) voc_daily = 15 * np.sin(2 * np.pi * t / daily_period - np.pi / 4) voc_weekly = 8 * np.sin(2 * np.pi * t / weekly_period) voc_noise = smooth_noise(n_points, -5, 5, window=35) voc = np.clip(voc_baseline + voc_trend + voc_daily + voc_weekly + voc_noise, 30, 200) # ======================================================================== # PM (PARTICULATE MATTER) - tiny particles in the air # ======================================================================== # Particulates come from dust, cooking, movement # The +np.pi/2 shifts the daily pattern by 6 hours pm_baseline = 15 pm_trend = 0.002 * t pm_daily = 5 * np.sin(2 * np.pi * t / daily_period + np.pi / 2) pm_weekly = 3 * np.sin(2 * np.pi * t / weekly_period) pm_noise = smooth_noise(n_points, -2, 2, window=25) pm = np.clip(pm_baseline + pm_trend + pm_daily + pm_weekly + pm_noise, 5, 80) # ======================================================================== # Create DataFrame # ======================================================================== # Prophet requires a 'ds' column with datetime values ds = pd.date_range(start="2025-01-01", periods=n_points, freq="min") df = pd.DataFrame({ "ds": ds, # Datetime (for Prophet) "Time_min": t, # Minute number (for plotting) "O2_percent": o2, "CO2_ppm": co2, "VOCs_ppb": voc, "Particulates_ugm3": pm }) return df # ============================================================================ # DOWNSAMPLING FUNCTION (FOR SARIMA SPEED OPTIMIZATION) # ============================================================================ def downsample_series(series, factor): arr = np.array(series) # Calculate how many complete groups we can make n_groups = len(arr) // factor # Trim to exact multiple of factor trimmed = arr[:n_groups * factor] # Reshape into groups and take mean of each group reshaped = trimmed.reshape(n_groups, factor) downsampled = reshaped.mean(axis=1) return downsampled def upsample_predictions(predictions, factor, target_length): # Create x-coordinates for original predictions x_original = np.arange(len(predictions)) * factor # Create x-coordinates for upsampled predictions x_upsampled = np.arange(target_length) # Use numpy interpolation to fill in the gaps upsampled = np.interp(x_upsampled, x_original, predictions) return upsampled # ============================================================================ # MODEL TRAINING AND PREDICTION FUNCTIONS # ============================================================================ def train_prophet(train_df, test_df): # Create and configure the Prophet model model = Prophet( growth="linear", # Use linear trend (not logistic) yearly_seasonality=False, # Our data is only ~3 days, no yearly pattern weekly_seasonality=True, # Enable weekly pattern detection daily_seasonality=True, # Enable daily pattern detection changepoint_prior_scale=PROPHET_CHANGEPOINT_PRIOR, seasonality_prior_scale=PROPHET_SEASONALITY_PRIOR, n_changepoints=PROPHET_N_CHANGEPOINTS, interval_width=0.8, # 80% confidence interval ) # Train the model on historical data # Prophet learns the patterns from this data model.fit(train_df) # Create predictions for the test period # We give Prophet the future dates, and it predicts the values future = test_df[["ds"]].copy() forecast = model.predict(future) # Return just the predicted values (yhat = "y hat" = predicted y) return forecast["yhat"].values def train_arima(train_series, test_len): try: # Create and fit the ARIMA model model = ARIMA(train_series, order=ARIMA_ORDER) fitted_model = model.fit() # Forecast future values forecast = fitted_model.forecast(steps=test_len) predictions = forecast.values if hasattr(forecast, 'values') else np.array(forecast) except Exception as e: print(f"ARIMA fitting failed: {e}") # Fallback: if model fails, just use the last known value # This is called "naive forecasting" or "persistence" predictions = np.full(test_len, train_series.iloc[-1]) return predictions def train_sarima(train_series, test_len, original_test_len): try: # Create and fit the SARIMA model (called SARIMAX in statsmodels) # The X in SARIMAX stands for "eXogenous variables" (external factors) # We don't use external variables here, so it's just SARIMA model = SARIMAX( train_series, order=SARIMA_ORDER, seasonal_order=SARIMA_SEASONAL_ORDER, enforce_stationarity=False, # Don't force data to be stationary enforce_invertibility=False # More flexible model fitting ) # Fit the model (this finds the best parameters) # disp=False hides the optimization output # maxiter=100 limits iterations to prevent hanging fitted_model = model.fit(disp=False, maxiter=100) # Make predictions on downsampled scale forecast = fitted_model.forecast(steps=test_len) predictions_downsampled = forecast.values if hasattr(forecast, 'values') else np.array(forecast) # Upsample predictions back to original resolution predictions = upsample_predictions( predictions_downsampled, SARIMA_DOWNSAMPLE_FACTOR, original_test_len ) except Exception as e: print(f"SARIMA fitting failed: {e}") # Fallback: use last known value last_val = train_series.iloc[-1] if len(train_series) > 0 else 0 predictions = np.full(original_test_len, last_val) return predictions # ============================================================================ # MAIN PREDICTION FUNCTION # ============================================================================ def run_prediction(seed, model_choice): # ======================================================================== # STEP 1: Generate Data # ======================================================================== df = generate_mission_data(seed=int(seed)) n = len(df) train_size = int(n * TRAIN_FRAC) # 70% for training test_size = n - train_size # 30% for testing # ======================================================================== # STEP 2: Define Parameters to Predict # ======================================================================== params = [ ("O2_percent", "O2 (%)"), ("CO2_ppm", "CO2 (ppm)"), ("VOCs_ppb", "VOCs (ppb)"), ("Particulates_ugm3", "PM (μg/m³)"), ] # Determine which models to run if model_choice == "All": models_to_run = ["Prophet", "ARIMA", "SARIMA"] else: models_to_run = [model_choice] # ======================================================================== # STEP 3: Prepare Downsampled Data for SARIMA (if needed) # ======================================================================== # Only do this work if SARIMA is being used if "SARIMA" in models_to_run: # Calculate downsampled sizes train_size_down = train_size // SARIMA_DOWNSAMPLE_FACTOR test_size_down = test_size // SARIMA_DOWNSAMPLE_FACTOR # ======================================================================== # STEP 4: Train and Predict for Each Parameter and Model # ======================================================================== all_results = {} # Store results for plotting metrics_data = [] # Store accuracy metrics for col, label in params: # Prepare training and test data (full resolution) train_series = df[col].iloc[:train_size] test_actual = df[col].iloc[train_size:].values # For Prophet, we need DataFrames with 'ds' and 'y' columns train_df_prophet = df.iloc[:train_size][["ds"]].copy() train_df_prophet["y"] = train_series.values test_df_prophet = df.iloc[train_size:][["ds"]].copy() all_results[col] = { "train_y": train_series.values, "test_actual": test_actual, "label": label, "predictions": {} } # Run each selected model for model_name in models_to_run: if model_name == "Prophet": # Prophet uses full resolution data pred = train_prophet(train_df_prophet, test_df_prophet) elif model_name == "ARIMA": # ARIMA uses full resolution data pred = train_arima(train_series, test_size) elif model_name == "SARIMA": # SARIMA uses DOWNSAMPLED data for speed! train_series_down = downsample_series(train_series, SARIMA_DOWNSAMPLE_FACTOR) pred = train_sarima( pd.Series(train_series_down), test_size_down, test_size # Original test size for upsampling ) all_results[col]["predictions"][model_name] = pred # Calculate error metrics mae = np.mean(np.abs(test_actual - pred)) rmse = np.sqrt(np.mean((test_actual - pred) ** 2)) metrics_data.append({ "Parameter": label, "Model": model_name, "MAE": f"{mae:.4f}", "RMSE": f"{rmse:.4f}" }) # ======================================================================== # STEP 5: Create Visualizations # ======================================================================== # Time arrays for x-axis train_time = df["Time_min"].iloc[:train_size].values test_time = df["Time_min"].iloc[train_size:].values # Color scheme for different models model_colors = { "Prophet": "orange", "ARIMA": "green", "SARIMA": "red" } # Create 2x2 subplot (one for each parameter) fig, axs = plt.subplots(2, 2, figsize=(15, 11)) axs = axs.flatten() for idx, (col, label) in enumerate(params): ax = axs[idx] r = all_results[col] # Plot training data (same for all models) ax.plot(train_time, r["train_y"], color="blue", alpha=0.5, label="Training Data", linewidth=1) # Plot actual test data ax.plot(test_time, r["test_actual"], color="blue", label="Actual", linewidth=2) # Plot predictions for each model for model_name, pred in r["predictions"].items(): ax.plot(test_time, pred, color=model_colors[model_name], label=f"{model_name} Predicted", linewidth=1.5, linestyle="--" if len(models_to_run) > 1 else "-") # Add vertical line at train/test split ax.axvline(x=train_size, color="gray", linestyle=":", alpha=0.7, label="Train/Test Split") # Formatting ax.set_xlabel("Time (minutes)") ax.set_ylabel(label) ax.set_title(f"{label}: Actual vs Predicted") ax.legend(loc="upper left", fontsize=8) ax.grid(True, alpha=0.3) # Add overall title fig.suptitle(f"Time Series Forecasting Comparison\n" f"Model(s): {', '.join(models_to_run)} | " f"Train: {train_size} pts | Test: {test_size} pts", fontsize=12, fontweight='bold') plt.tight_layout() # Create metrics DataFrame metrics_df = pd.DataFrame(metrics_data) return metrics_df, str(seed), fig # ============================================================================ # GRADIO USER INTERFACE # ============================================================================ # Gradio is a library for creating simple web interfaces for ML models # It automatically creates a web page where users can interact with our code with gr.Blocks() as demo: # ------------------------------------------------------------------------ # Header and Instructions # ------------------------------------------------------------------------ gr.Markdown(""" # Time Series Forecasting: Prophet vs ARIMA vs SARIMA ## What This Tool Does This tool generates synthetic environmental sensor data and uses different forecasting models to predict future values. Compare how well each model captures trends and seasonal patterns! ## The Models | Model | Best For | Handles Seasonality? | Speed | |-------|----------|---------------------|-------| | **Prophet** | Business forecasting, multiple seasonalities | Yes (automatic) | Fast | | **ARIMA** | Data with trends, no clear seasonality | No | Fast | | **SARIMA** | Data with trends AND seasonality | Yes (manual config) | Medium* | *SARIMA uses downsampled data for speed optimization (see code comments for details) ## How to Use 1. Choose a random seed (different seeds = different data) 2. Select which model(s) to compare 3. Click "Generate & Predict" and wait for results """) # ------------------------------------------------------------------------ # Input Controls # ------------------------------------------------------------------------ with gr.Row(): seed_input = gr.Number( label="Random Seed", value=42, precision=0, info="Change this to generate different data patterns" ) model_dropdown = gr.Dropdown( choices=["Prophet", "ARIMA", "SARIMA", "All"], value="All", label="Model Selection", info="Choose which forecasting model(s) to use" ) run_btn = gr.Button("Generate & Predict", variant="primary") # ------------------------------------------------------------------------ # Output Displays # ------------------------------------------------------------------------ seed_box = gr.Textbox(label="Seed Used", interactive=False) gr.Markdown("### Accuracy Metrics") gr.Markdown(""" - **MAE** (Mean Absolute Error): Average prediction error. Lower = better. - **RMSE** (Root Mean Square Error): Similar to MAE but penalizes large errors more. """) metrics_out = gr.Dataframe(label="Model Performance") gr.Markdown("### Visualization") plot_out = gr.Plot(label="Forecasting Results") # ------------------------------------------------------------------------ # Educational Footer # ------------------------------------------------------------------------ gr.Markdown(""" --- ## Understanding the Results **Blue line** = Actual data (what really happened) **Colored dashed lines** = Model predictions (what the model thinks will happen) A good model's predictions should closely follow the actual data in the test region (right side of the red dashed line). ### Tips for Interpretation - If predictions are flat while actual data has waves → Model missed seasonality - If predictions drift away from actual → Model has trouble with the trend - Lower MAE/RMSE = Better predictions ### Why is SARIMA using downsampled data? SARIMA can be very slow on large datasets (20-60 minutes for 4500 points!). To make it practical, we: 1. Downsample minute data to hourly (75 points instead of 4500) 2. Train SARIMA on the smaller dataset (~30 seconds) 3. Upsample predictions back to minute resolution This trades a small amount of accuracy for much faster results. """) # ------------------------------------------------------------------------ # Connect Button to Function # ------------------------------------------------------------------------ run_btn.click( fn=run_prediction, inputs=[seed_input, model_dropdown], outputs=[metrics_out, seed_box, plot_out] ) # Launch the web interface # share=True creates a public link (useful for sharing with others) # ssr_mode=False for better compatibility demo.launch(share=True, ssr_mode=False)