import gradio as gr import torch import numpy as np import pandas as pd import plotly.graph_objects as go import requests import os from datetime import datetime, timezone, timedelta from transformers import TimesFm2_5ModelForPrediction, AutoModelForCausalLM, AutoTokenizer from typing import List, Optional from ui_components import CSS, get_header_html, make_metric_cards, make_day_cards, get_status_hero, AQI_BANDS, AQI_TIPS from viz_components import make_gauge, make_plot, make_map TIMESFM_MODEL_ID = "mahwizzzz/skyloom" ADVISOR_MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct" print("Loading Models …") device = "cpu" forecaster_model = TimesFm2_5ModelForPrediction.from_pretrained( TIMESFM_MODEL_ID, dtype=torch.float32, ) forecaster_model.eval() print("Loading Advisor (Qwen2.5-0.5B)...") advisor_tokenizer = AutoTokenizer.from_pretrained(ADVISOR_MODEL_ID) advisor_model = AutoModelForCausalLM.from_pretrained( ADVISOR_MODEL_ID, dtype=torch.float32, ) advisor_model.eval() print("Models ready on CPU") def pm25_to_aqi(pm25: float) -> float: pm25 = max(pm25, 0) breakpoints = [ (0.0, 12.0, 0, 50), (12.1, 35.4, 51, 100), (35.5, 55.4, 101, 150), (55.5, 150.4, 151, 200), (150.5, 250.4, 201, 300), (250.5, 350.4, 301, 400), (350.5, 500.4, 401, 500) ] for lo, hi, aqi_lo, aqi_hi in breakpoints: if pm25 <= hi: return ((aqi_hi - aqi_lo) / (hi - lo)) * (pm25 - lo) + aqi_lo return 500.0 API_KEY = os.getenv("OPENWEATHER_API_KEY", "") KARACHI_LAT, KARACHI_LON = 24.8607, 67.0011 def fetch_data(hours=24): if not API_KEY: return [30 + np.random.normal(0, 5) for _ in range(hours)] url = ( f"http://api.openweathermap.org/data/2.5/air_pollution/history" f"?lat={KARACHI_LAT}&lon={KARACHI_LON}" f"&start={int((datetime.now()-timedelta(hours=hours)).timestamp())}" f"&end={int(datetime.now().timestamp())}&appid={API_KEY}" ) try: r = requests.get(url).json() return [e['components']['pm2_5'] for e in r['list']][-hours:] except: return [30 + np.random.normal(0, 5) for _ in range(hours)] def get_forecast(history, horizon=24): inputs = [torch.tensor(history, dtype=torch.float32)] with torch.no_grad(): out = forecaster_model(past_values=inputs, prediction_length=horizon, return_dict=True) return out.mean_predictions[0].numpy().flatten() def ask_advisor(msg, history, fc): current_aqi = fc[0] if len(fc) > 0 else 0 category = "Good" for lo, hi, cat, color, bg in AQI_BANDS: if current_aqi <= hi: category = cat break tip = AQI_TIPS.get(category, "") system_prompt = ( f"You are Skyloom, an air-quality assistant for Karachi, Pakistan. " f"Current AQI in Karachi: {current_aqi:.0f}, category: {category}. " f"Recommendation for this AQI level: {tip} " f"Answer the user's question directly based ONLY on this AQI information. " f"Keep your answer to 1-2 sentences. Do not discuss unrelated topics." ) chat = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": msg}, ] prompt = advisor_tokenizer.apply_chat_template(chat, tokenize=False, add_generation_prompt=True) inputs = advisor_tokenizer(prompt, return_tensors="pt") with torch.no_grad(): outputs = advisor_model.generate( **inputs, max_new_tokens=80, do_sample=False, repetition_penalty=1.15, no_repeat_ngram_size=3, pad_token_id=advisor_tokenizer.eos_token_id, ) new_tokens = outputs[0][inputs["input_ids"].shape[1]:] answer = advisor_tokenizer.decode(new_tokens, skip_special_tokens=True).strip() if not answer: answer = tip history = history + [{"role": "user", "content": msg}, {"role": "assistant", "content": answer}] return history, "" def update_ui(horizon): try: pm_hist = fetch_data(24) aqi_hist = [pm25_to_aqi(p) for p in pm_hist] fc = get_forecast(aqi_hist, horizon) current_aqi = fc[0] status_hero = get_status_hero(current_aqi) gauge_fig = make_gauge(current_aqi) metric_html = make_metric_cards(pm_hist[-1], fc) day_html = make_day_cards(fc) trend_plot = make_plot(aqi_hist, fc, window="24h") map_fig = make_map(current_aqi) share_html = make_share_card(current_aqi) return status_hero, gauge_fig, metric_html, day_html, trend_plot, map_fig, share_html, fc.tolist() except Exception as e: print(f"Error: {e}") empty = go.Figure() return "
Error loading data
", empty, "", "", empty, empty, "", [] def update_trend(window, fc_state): """Re-render the trend chart when user changes the window toggle.""" try: pm_hist = fetch_data(24) aqi_hist = [pm25_to_aqi(p) for p in pm_hist] fc = np.array(fc_state) if fc_state else np.array(aqi_hist) return make_plot(aqi_hist, fc, window=window) except Exception as e: print(f"Trend error: {e}") return go.Figure() def make_share_card(aqi): category = "Good" color = "#10b981" tip = "" for lo, hi, cat, col, bg in AQI_BANDS: if aqi <= hi: category, color = cat, col tip = AQI_TIPS.get(cat, "") break now = datetime.now().strftime("%d %b %Y, %H:%M") return f"""
Karachi AQI Snapshot
{now}
🌬️
{aqi:.0f} AQI
{category}
{tip}
""" with gr.Blocks(title="Karachi Air Quality") as demo: fc_state = gr.State([]) with gr.Column(elem_id="main-container"): gr.HTML(get_header_html()) with gr.Tabs(): with gr.Tab("🏠 Dashboard"): status_hero = gr.HTML( "
" "

Current Observation

--

Loading…

" ) with gr.Row(): # Left column with gr.Column(scale=1): with gr.Column(elem_classes="mobile-card"): gr.Markdown("### Air Quality Index") gauge = gr.Plot(show_label=False) horizon = gr.Slider( label="Forecast Horizon (hours)", minimum=1, maximum=72, value=24, step=1 ) btn = gr.Button("🔄 Update Forecast", elem_id="predict-btn") with gr.Column(scale=1): with gr.Column(elem_classes="mobile-card"): gr.Markdown("### Current Observation") metrics = gr.HTML() with gr.Column(elem_classes="mobile-card"): gr.Markdown("### Weekly Outlook") day_cards = gr.HTML() # Trend chart with window toggle with gr.Column(elem_classes="mobile-card"): with gr.Row(): gr.Markdown("### AQI Trend") window_radio = gr.Radio( choices=["24h", "7d", "30d"], value="24h", label="", show_label=False, elem_id="window-radio", ) plot = gr.Plot(show_label=False) with gr.Row(): with gr.Column(scale=2, elem_classes="mobile-card"): gr.Markdown("### 📍 Monitoring Station Karachi") map_plot = gr.Plot(show_label=False) with gr.Column(scale=1, elem_classes="mobile-card"): gr.Markdown("### 📤 Share Snapshot") share_card = gr.HTML() with gr.Tab("🤖 AI Advisor"): with gr.Column(elem_classes="mobile-card"): gr.Markdown("### Skyloom Health Advisor") chatbot = gr.Chatbot(height=420) with gr.Row(): msg_input = gr.Textbox( placeholder="Ask about health precautions…", scale=4, show_label=False ) send = gr.Button("Ask", variant="primary", scale=1) gr.Examples( examples=[ "Is it safe to exercise outdoors?", "Should I wear a mask?", "Can I open my windows?", "Is it safe for children to play outside?", ], inputs=msg_input, ) outputs_full = [status_hero, gauge, metrics, day_cards, plot, map_plot, share_card, fc_state] btn.click(update_ui, inputs=horizon, outputs=outputs_full) demo.load(update_ui, inputs=horizon, outputs=outputs_full) window_radio.change(update_trend, inputs=[window_radio, fc_state], outputs=plot) send.click(ask_advisor, inputs=[msg_input, chatbot, fc_state], outputs=[chatbot, msg_input]) msg_input.submit(ask_advisor, inputs=[msg_input, chatbot, fc_state], outputs=[chatbot, msg_input]) if __name__ == "__main__": demo.queue(max_size=5) demo.launch(css=CSS)