Add startup and runtime diagnostics

app.py

@@ -3,6 +3,10 @@ import requests
 import pandas as pd
 import numpy as np
 import json
+import platform
+import sys
+import traceback
+import faulthandler
 from datetime import datetime, timedelta
 
 from huggingface_hub import HfApi, hf_hub_download
@@ -23,6 +27,9 @@ AARHUS_LON = 10.2108
 HF_TOKEN = os.environ.get("HF_TOKEN")
 
 COPENHAGEN_TZ = ZoneInfo("Europe/Copenhagen")
+APP_NAME = "dmi-collector"
+
+faulthandler.enable()
 
 # Extended feature set from PLAN.md
 FORECAST_FEATURES = [
@@ -61,6 +68,65 @@ OBSERVATION_FEATURES = [
 ]
 
 
+def log_event(message, **fields):
+    """Emit a structured log line that shows up clearly in HF runtime logs."""
+    timestamp = datetime.utcnow().isoformat(timespec="seconds") + "Z"
+    details = " ".join(f"{key}={fields[key]!r}" for key in sorted(fields))
+    if details:
+        print(f"[{APP_NAME}] {timestamp} {message} {details}", flush=True)
+    else:
+        print(f"[{APP_NAME}] {timestamp} {message}", flush=True)
+
+
+def log_exception(context, exc):
+    """Log an exception with a full traceback."""
+    log_event(f"{context} failed", error=str(exc), error_type=type(exc).__name__)
+    print(traceback.format_exc(), flush=True)
+
+
+def install_global_logging():
+    """Install process-level exception logging."""
+    def handle_exception(exc_type, exc_value, exc_traceback):
+        if issubclass(exc_type, KeyboardInterrupt):
+            sys.__excepthook__(exc_type, exc_value, exc_traceback)
+            return
+        log_event("uncaught_exception", error=str(exc_value), error_type=exc_type.__name__)
+        print("".join(traceback.format_exception(exc_type, exc_value, exc_traceback)), flush=True)
+
+    sys.excepthook = handle_exception
+
+
+def log_startup():
+    """Log runtime context during process startup."""
+    log_event(
+        "startup",
+        python=sys.version.split()[0],
+        platform=platform.platform(),
+        cwd=os.getcwd(),
+        hf_token_present=bool(HF_TOKEN),
+        dataset=DATASET_NAME,
+        predictions_dataset=PREDICTIONS_DATASET,
+    )
+
+
+def run_logged(name, fn, *args, **kwargs):
+    """Run a function with start/end/error logging."""
+    log_event(f"{name} started")
+    try:
+        result = fn(*args, **kwargs)
+        if isinstance(result, pd.DataFrame):
+            log_event(f"{name} completed", rows=len(result), columns=list(result.columns))
+        else:
+            log_event(f"{name} completed", result_type=type(result).__name__)
+        return result
+    except Exception as exc:
+        log_exception(name, exc)
+        raise
+
+
+install_global_logging()
+
+
 def now_cph():
     """Current time in Copenhagen timezone."""
     return datetime.now(COPENHAGEN_TZ)
@@ -86,6 +152,7 @@ def fetch_forecasts_for_period(start_date, end_date):
     Fetch historical forecast runs for Aarhus from Open-Meteo.
     Returns DataFrame with all features.
     """
+    log_event("fetch_forecasts_for_period", start_date=str(start_date), end_date=str(end_date))
     all_forecasts = []
     run_hours = [0, 3, 6, 9, 12, 15, 18, 21]
 
@@ -155,16 +222,19 @@ def fetch_forecasts_for_period(start_date, end_date):
         time.sleep(0.1)
 
     if not all_forecasts:
+        log_event("fetch_forecasts_for_period no_data", start_date=str(start_date), end_date=str(end_date))
         return None
 
     df = pd.DataFrame(all_forecasts)
     df = df.drop_duplicates(subset=['target_timestamp'], keep='first')
     df = df.sort_values('target_timestamp').reset_index(drop=True)
+    log_event("fetch_forecasts_for_period done", rows=len(df))
     return df
 
 
 def fetch_future_forecasts():
     """Fetch future forecasts - 48 hours ahead."""
+    log_event("fetch_future_forecasts started")
     now = now_cph()
     today = now.date()
     
@@ -229,15 +299,17 @@ def fetch_future_forecasts():
                     forecasts.append(row)
         
         if not forecasts:
+            log_event("fetch_future_forecasts no_data")
             return None
             
         df = pd.DataFrame(forecasts)
         df = df.drop_duplicates(subset=['target_timestamp'], keep='first')
         df = df.sort_values('target_timestamp').reset_index(drop=True)
+        log_event("fetch_future_forecasts done", rows=len(df))
         return df
         
     except Exception as e:
-        print(f"❌ Error fetching future forecasts: {e}")
+        log_exception("fetch_future_forecasts", e)
         return None
 
 
@@ -248,6 +320,7 @@ def fetch_observations_for_period(start_date, end_date):
     """
     Fetch actual weather observations for Aarhus from Open-Meteo archive.
     """
+    log_event("fetch_observations_for_period", start_date=str(start_date), end_date=str(end_date))
     url = "https://archive-api.open-meteo.com/v1/archive"
 
     cph_today = now_cph().date()
@@ -297,9 +370,10 @@ def fetch_observations_for_period(start_date, end_date):
         current_hour = now_cph().replace(minute=0, second=0, microsecond=0)
         df = df[df['target_timestamp'] <= current_hour]
         
+        log_event("fetch_observations_for_period done", rows=len(df))
         return df
     except Exception as e:
-        print(f"❌ Error fetching observations: {e}")
+        log_exception("fetch_observations_for_period", e)
         return None
 
 
@@ -374,12 +448,14 @@ def build_training_matrix(forecasts_df, observations_df):
     Adds all derived features.
     """
     if forecasts_df is None or observations_df is None:
+        log_event("build_training_matrix missing_inputs")
         return None
     
     # Merge on target_timestamp
     merged = pd.merge(forecasts_df, observations_df, on='target_timestamp', how='inner')
     
     if len(merged) == 0:
+        log_event("build_training_matrix no_matches")
         return None
     
     # Add temporal features
@@ -400,6 +476,7 @@ def build_training_matrix(forecasts_df, observations_df):
     current_hour = now_cph().replace(minute=0, second=0, microsecond=0)
     merged = merged[merged['target_timestamp'] <= current_hour]
     
+    log_event("build_training_matrix done", rows=len(merged), columns=len(merged.columns))
     return merged
 
 
@@ -408,6 +485,7 @@ def build_training_matrix(forecasts_df, observations_df):
 # =============================================================================
 def upload_to_dataset(local_path, filename, dataset_name, repo_type="dataset"):
     """Upload a file to Hugging Face dataset."""
+    log_event("upload_to_dataset", local_path=local_path, filename=filename, dataset_name=dataset_name)
     try:
         api = HfApi()
         api.upload_file(
@@ -417,14 +495,16 @@ def upload_to_dataset(local_path, filename, dataset_name, repo_type="dataset"):
             repo_type=repo_type,
             token=HF_TOKEN
         )
+        log_event("upload_to_dataset done", filename=filename, dataset_name=dataset_name)
         return True
     except Exception as e:
-        print(f"❌ Upload error: {e}")
+        log_exception("upload_to_dataset", e)
         return False
 
 
 def load_from_dataset(filename, dataset_name, repo_type="dataset"):
     """Load a file from Hugging Face dataset."""
+    log_event("load_from_dataset", filename=filename, dataset_name=dataset_name)
     try:
         path = hf_hub_download(
             repo_id=dataset_name,
@@ -432,9 +512,10 @@ def load_from_dataset(filename, dataset_name, repo_type="dataset"):
             repo_type=repo_type,
             token=HF_TOKEN
         )
+        log_event("load_from_dataset done", filename=filename, dataset_name=dataset_name, path=path)
         return path
     except Exception as e:
-        print(f"❌ Download error: {e}")
+        log_exception("load_from_dataset", e)
         return None
 
 
@@ -500,6 +581,7 @@ def load_existing_training_matrix():
 # =============================================================================
 def backfill_historical_data():
     """Backfill historical data from 2025-11-01 to now."""
+    log_event("backfill_historical_data entered")
     init_dataset_if_needed()
 
     start_date = datetime(2025, 11, 1).date()
@@ -562,6 +644,7 @@ def backfill_historical_data():
 # =============================================================================
 def update_daily():
     """Daily update - fetch last 7 days and append to dataset."""
+    log_event("update_daily entered")
     init_dataset_if_needed()
 
     end_date = now_cph().date()
@@ -618,6 +701,7 @@ def update_daily():
 # =============================================================================
 def load_model_bundle(target_name):
     """Load model bundle from dataset."""
+    log_event("load_model_bundle", target_name=target_name)
     try:
         path = hf_hub_download(
             repo_id=DATASET_NAME,
@@ -627,7 +711,7 @@ def load_model_bundle(target_name):
         )
         return joblib.load(path)
     except Exception as e:
-        print(f"❌ Could not load model bundle for {target_name}: {e}")
+        log_exception(f"load_model_bundle[{target_name}]", e)
         return None
 
 
@@ -658,8 +742,9 @@ def predict_with_bundle(bundle, df):
 
 def generate_predictions():
     """Generate predictions for all targets."""
+    log_event("generate_predictions entered")
     current_time = now_cph()
-    print(f"🔮 Generating predictions: {current_time}")
+    log_event("generate_predictions clock", current_time=str(current_time))
 
     future_forecasts = fetch_future_forecasts()
     if future_forecasts is None or len(future_forecasts) == 0:
@@ -676,7 +761,7 @@ def generate_predictions():
         with open(registry_path, 'r') as f:
             registry = json.load(f)
     except Exception as e:
-        print(f"❌ Could not load model registry: {e}")
+        log_exception("generate_predictions model_registry", e)
         registry = None
 
     results = future_forecasts.copy()
@@ -741,6 +826,7 @@ def generate_predictions():
 # =============================================================================
 def verify_predictions():
     """Verify past predictions with actual observations."""
+    log_event("verify_predictions entered")
     try:
         # Load predictions
         pred_path, pred_filename = load_first_available_dataset_file(
@@ -803,29 +889,48 @@ def verify_predictions():
 # =============================================================================
 def run_scheduler():
     """Background scheduler for automated tasks."""
+    log_event("scheduler starting")
     # Every 3 hours: fetch new forecast run
-    schedule.every(3).hours.do(lambda: fetch_forecasts_for_period(now_cph().date() - timedelta(days=1), now_cph().date()))
+    schedule.every(3).hours.do(
+        lambda: run_logged(
+            "scheduled_fetch_forecasts",
+            fetch_forecasts_for_period,
+            now_cph().date() - timedelta(days=1),
+            now_cph().date(),
+        )
+    )
     
     # Every hour: fetch new observations
-    schedule.every().hour.do(lambda: fetch_observations_for_period(now_cph().date() - timedelta(days=1), now_cph().date()))
+    schedule.every().hour.do(
+        lambda: run_logged(
+            "scheduled_fetch_observations",
+            fetch_observations_for_period,
+            now_cph().date() - timedelta(days=1),
+            now_cph().date(),
+        )
+    )
     
     # Every 3 hours: generate new predictions
-    schedule.every(3).hours.do(generate_predictions)
+    schedule.every(3).hours.do(lambda: run_logged("scheduled_generate_predictions", generate_predictions))
     
     # Every hour: verify past predictions
-    schedule.every().hour.do(verify_predictions)
+    schedule.every().hour.do(lambda: run_logged("scheduled_verify_predictions", verify_predictions))
     
     # Daily: rebuild training matrix
-    schedule.every().day.at("06:00").do(update_daily)
+    schedule.every().day.at("06:00").do(lambda: run_logged("scheduled_update_daily", update_daily))
     
     while True:
-        schedule.run_pending()
+        try:
+            schedule.run_pending()
+        except Exception as exc:
+            log_exception("scheduler.run_pending", exc)
         time.sleep(60)
 
 
 # =============================================================================
 # GRADIO UI
 # =============================================================================
+log_event("building_gradio_ui")
 with gr.Blocks(title="DMI Aarhus Collector") as demo:
     gr.Markdown("""
     # 🌤️ DMI Aarhus Weather Collector
@@ -841,16 +946,49 @@ with gr.Blocks(title="DMI Aarhus Collector") as demo:
         btn_predict = gr.Button("🔮 Generate Predictions", variant="primary")
         btn_verify = gr.Button("✅ Verify Predictions", variant="secondary")
 
-    btn_backfill.click(backfill_historical_data, outputs=status)
-    btn_daily.click(update_daily, outputs=status)
-    btn_predict.click(generate_predictions, outputs=status)
-    btn_verify.click(verify_predictions, outputs=status)
+    def backfill_handler():
+        try:
+            return run_logged("gradio_backfill_historical_data", backfill_historical_data)
+        except Exception as exc:
+            return f"❌ backfill_historical_data failed: {exc}"
+
+    def daily_handler():
+        try:
+            return run_logged("gradio_update_daily", update_daily)
+        except Exception as exc:
+            return f"❌ update_daily failed: {exc}"
+
+    def predict_handler():
+        try:
+            return run_logged("gradio_generate_predictions", generate_predictions)
+        except Exception as exc:
+            return f"❌ generate_predictions failed: {exc}"
+
+    def verify_handler():
+        try:
+            return run_logged("gradio_verify_predictions", verify_predictions)
+        except Exception as exc:
+            return f"❌ verify_predictions failed: {exc}"
+
+    btn_backfill.click(backfill_handler, outputs=status)
+    btn_daily.click(daily_handler, outputs=status)
+    btn_predict.click(predict_handler, outputs=status)
+    btn_verify.click(verify_handler, outputs=status)
+
+log_event("gradio_ui_ready")
 
 if __name__ == "__main__":
-    scheduler_thread = threading.Thread(target=run_scheduler, daemon=True)
-    scheduler_thread.start()
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        ssr_mode=False
-    )
+    log_startup()
+    try:
+        scheduler_thread = threading.Thread(target=run_scheduler, daemon=True, name="collector-scheduler")
+        scheduler_thread.start()
+        log_event("scheduler_thread_started", thread_name=scheduler_thread.name, is_alive=scheduler_thread.is_alive())
+        log_event("launching_gradio", server_name="0.0.0.0", server_port=7860, ssr_mode=False)
+        demo.launch(
+            server_name="0.0.0.0",
+            server_port=7860,
+            ssr_mode=False
+        )
+    except Exception as exc:
+        log_exception("__main__", exc)
+        raise