Spaces:
Sleeping
Sleeping
File size: 1,783 Bytes
a321b61 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
#!/usr/bin/env python3
"""Validate forecast results"""
import sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import polars as pl
from pathlib import Path
# Find the most recent forecast file in Windows temp directory
temp_dir = Path(r"C:\Users\evgue\AppData\Local\Temp\gradio")
forecast_files = list(temp_dir.glob("**/forecast_*.parquet"))
if not forecast_files:
print("[ERROR] No forecast files found", flush=True)
sys.exit(1)
# Get the most recent file
latest_forecast = max(forecast_files, key=lambda p: p.stat().st_mtime)
print(f"Examining: {latest_forecast.name}", flush=True)
print(f"Full path: {latest_forecast}", flush=True)
# Load and examine the forecast
df = pl.read_parquet(latest_forecast)
print(f"\n[OK] Forecast loaded successfully", flush=True)
print(f"Shape: {df.shape} (rows x columns)", flush=True)
print(f"\nColumns: {df.columns}", flush=True)
print(f"\nData types:\n{df.dtypes}", flush=True)
# Check for expected structure
print(f"\n--- Validation ---", flush=True)
assert 'timestamp' in df.columns, "Missing timestamp column"
print("[OK] timestamp column present", flush=True)
# Check for forecast columns (median, q10, q90)
forecast_cols = [c for c in df.columns if c != 'timestamp']
print(f"[OK] Found {len(forecast_cols)} forecast columns", flush=True)
# Check number of rows (should be 168 for 7 days)
expected_rows = 168 # 7 days * 24 hours
print(f"[OK] Rows: {len(df)} (expected: {expected_rows})", flush=True)
# Display first few rows
print(f"\n--- First 5 rows ---", flush=True)
print(df.head(5))
# Display summary statistics
print(f"\n--- Summary Statistics ---", flush=True)
print(df.select([c for c in df.columns if c != 'timestamp']).describe())
print(f"\n[SUCCESS] Smoke test validation complete!", flush=True)
|