"""
Generate Interactive CI Graphs Dashboard
=========================================
This script creates an interactive HTML dashboard with:
1. Historic CI trends by field/season (from RDS file)
2. Current and last week statistics (box plots, heatmaps, scatter)
3. Field and season selection dropdowns
Data Sources:
- Historic: laravel_app/storage/app/esa/Data/extracted_ci/cumulative_vals/All_pivots_Cumulative_CI_quadrant_year_v2.rds
- Current: laravel_app/storage/app/esa/weekly_mosaic/week_*.tif
Usage:
python generate_ci_graphs_dashboard.py [project] [--current-week W] [--previous-week W] [--output-dir DIR]
Example:
python generate_ci_graphs_dashboard.py esa --current-week 43 --previous-week 42 --output-dir output
"""
import argparse
import json
import warnings
from pathlib import Path
from datetime import datetime
import numpy as np
import pandas as pd
import rasterio
from rasterio.mask import mask
import geopandas as gpd
from shapely.geometry import box, shape
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
warnings.filterwarnings('ignore')
class CIGraphsDashboard:
"""Create interactive CI graphs dashboard with historic and current data."""
def __init__(self, project='esa', current_week=None, previous_week=None, output_dir='output'):
"""Initialize dashboard generator."""
self.project = project
self.current_week = current_week
self.previous_week = previous_week
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
# Data paths
self.base_dir = Path(__file__).parent.parent / 'laravel_app' / 'storage' / 'app' / project
self.weekly_mosaic_dir = self.base_dir / 'weekly_mosaic'
self.rds_file = self.base_dir / 'Data' / 'extracted_ci' / 'cumulative_vals' / 'All_pivots_Cumulative_CI_quadrant_year_v2.rds'
self.pivot_geojson = self.base_dir / 'Data' / 'pivot.geojson'
# Load data
self.field_gdf = None
self.historic_data = None
self.current_week_data = None
self.previous_week_data = None
print(f"Initialized CIGraphsDashboard for project: {project}")
print(f"Data directory: {self.base_dir}")
print(f"RDS file: {self.rds_file}")
def load_rds_file(self):
"""Load RDS file with historic CI values using pandas and pyreadr."""
try:
import pyreadr
print(f"Loading RDS file: {self.rds_file}")
result = pyreadr.read_r(str(self.rds_file))
# RDS files are stored as dict of dataframes
# Get the first (and usually only) dataframe
df_name = list(result.keys())[0]
self.historic_data = result[df_name]
print(f"Loaded historic data shape: {self.historic_data.shape}")
print(f"Columns: {self.historic_data.columns.tolist()}")
print(f"First few rows:\n{self.historic_data.head()}")
return self.historic_data
except ImportError:
print("pyreadr not installed. Attempting alternative approach...")
# Fallback: try using rpy2 to read RDS
try:
import rpy2.robjects as robjects
from rpy2.robjects import pandas2ri
pandas2ri.activate()
robjects.r(f'data <- readRDS("{str(self.rds_file)}")')
self.historic_data = pandas2ri.rpy2py(robjects.r('data'))
print(f"Loaded historic data shape: {self.historic_data.shape}")
print(f"Columns: {self.historic_data.columns.tolist()}")
return self.historic_data
except ImportError:
print("rpy2 not installed either. Trying CSV fallback...")
# Try to find a CSV version of the data
csv_file = self.rds_file.with_suffix('.csv')
if csv_file.exists():
self.historic_data = pd.read_csv(csv_file)
print(f"Loaded historic data from CSV: {csv_file}")
return self.historic_data
else:
raise ValueError(f"Could not load RDS file: {self.rds_file}\n"
"Install pyreadr or rpy2 to read RDS files.")
def load_field_boundaries(self):
"""Load field boundaries GeoJSON."""
print(f"Loading field boundaries: {self.pivot_geojson}")
self.field_gdf = gpd.read_file(self.pivot_geojson)
print(f"Loaded {len(self.field_gdf)} fields")
print(f"Columns: {self.field_gdf.columns.tolist()}")
return self.field_gdf
def load_weekly_ci_data(self, week_num):
"""Load CI data from weekly mosaic GeoTIFF."""
# Try multiple file naming patterns
possible_files = [
self.weekly_mosaic_dir / f"week_{week_num}.tif",
self.weekly_mosaic_dir / f"week_{week_num}_2025.tif",
self.weekly_mosaic_dir / f"week_{week_num:02d}.tif",
self.weekly_mosaic_dir / f"week_{week_num:02d}_2025.tif",
]
week_file = None
for f in possible_files:
if f.exists():
week_file = f
break
if week_file is None:
print(f"Warning: Week file not found for week {week_num}. Tried: {possible_files}")
return None
print(f"Loading week {week_num} data: {week_file}")
try:
with rasterio.open(week_file) as src:
# CI is typically in band 5
ci_band = src.read(5)
profile = src.profile
# Extract CI values for each field
field_ci_stats = {}
for idx, row in self.field_gdf.iterrows():
field_name = row.get('pivot_name', row.get('PIVOT', f'field_{idx}'))
try:
# Get geometry and extract CI values
geom = [row.geometry]
# Use rasterio mask to extract values within field boundary
masked_array, _ = mask(src, geom, crop=True, indexes=5)
# Remove masked/invalid values
valid_values = masked_array[masked_array > 0]
if len(valid_values) > 0:
field_ci_stats[field_name] = {
'mean': float(np.mean(valid_values)),
'median': float(np.median(valid_values)),
'std': float(np.std(valid_values)),
'min': float(np.min(valid_values)),
'max': float(np.max(valid_values)),
'q25': float(np.percentile(valid_values, 25)),
'q75': float(np.percentile(valid_values, 75)),
'count': len(valid_values),
'values': valid_values.tolist() # Store all values for heatmap
}
except Exception as e:
print(f"Could not extract CI for field {field_name}: {e}")
continue
return field_ci_stats
except Exception as e:
print(f"Error loading week {week_num}: {e}")
return None
def extract_current_week_data(self):
"""Extract CI statistics from current and previous week GeoTIFFs."""
print(f"\nExtracting current week data (week {self.current_week})...")
self.current_week_data = self.load_weekly_ci_data(self.current_week)
if self.previous_week:
print(f"Extracting previous week data (week {self.previous_week})...")
self.previous_week_data = self.load_weekly_ci_data(self.previous_week)
return self.current_week_data, self.previous_week_data
def create_historic_trend_chart(self):
"""Create line chart for historic CI trends by field and season."""
if self.historic_data is None:
print("No historic data loaded")
return None
# Prepare data - assumes columns include: pivot_name, year, season, mean_ci (or similar)
print("Creating historic trend chart...")
print(f"Historic data columns: {self.historic_data.columns.tolist()}")
# Create Plotly figure
fig = go.Figure()
# Get unique fields
field_col = next((col for col in ['pivot_name', 'PIVOT', 'field']
if col in self.historic_data.columns), None)
if field_col is None:
print("Warning: Could not find field column in historic data")
return None
unique_fields = self.historic_data[field_col].unique()
# Add traces for each field
for field in unique_fields[:10]: # Limit to first 10 for clarity
field_data = self.historic_data[self.historic_data[field_col] == field]
# Try to find CI value column
ci_col = next((col for col in field_data.columns
if 'ci' in col.lower() or 'mean' in col.lower()),
field_data.columns[-1])
if field_data.shape[0] > 0:
x_label = 'year' if 'year' in field_data.columns else field_data.columns[0]
fig.add_trace(go.Scatter(
x=field_data[x_label].astype(str),
y=field_data[ci_col],
mode='lines+markers',
name=str(field),
hovertemplate=f"{field}
Value: %{{y:.3f}}
Previous: %{x:.3f}
Current: %{y:.3f}
Week: %{x}
Mean CI: %{z:.3f}
Historic Trends & Current Week Statistics