Below is a complete, clean, single-file Streamlit app that:
-
Loads a model output Excel (or multiple files) which contains model outputs (Actual, Predicted, Trend, Seasonal, Residuals, RMSE/MAE, MK results).
-
Runs the agricultural decision engine for the four examples (Irrigation, Crop planning, Fertilizer scheduling, Pest management).
-
Shows an interactive dashboard UI with tables, key metrics and plots.
-
Creates an auto-generated PDF report that contains summaries and embedded plots.
-
Exports each plot as PNG and offers them in a ZIP download.
-
Optionally sends a concise SMS alert (mobile-friendly summary) using Twilio if the user provides credentials.
-
Integrates cleanly so you can drop this into your existing rain-project flow or run standalone.
Save file as e.g. agri_decision_app.py and run:
pip install streamlit pandas numpy matplotlib scikit-learn reportlab openpyxl twilio
streamlit run agri_decision_app.py
(If you don't want SMS, you can skip installing twilio.)
What the app expects in the input Excel
Each Excel file should contain at least these columns (case-insensitive accepted):
-
Actual(observed rainfall) -
Predicted(model forecast) — can be multiple model columns likeARIMA_Pred,LSTM_Pred,BiLSTM_Pred -
Trend(trend component or linear trend values) -
Seasonal(seasonal component) -
Residuals(residuals) -
MK_Trend(Mann–Kendall label or similar) -
RMSE/MAEor model-specific RMSE/MAE columns (optional)
If column names vary, the app tries to be flexible.
The file — single complete app
# agri_decision_app.py
"""
Single-file Streamlit app:
- Load model output Excel(s)
- Generate agricultural decisions for:
1) Irrigation planning
2) Crop planning
3) Fertilizer scheduling
4) Pest management
- Dashboard: metrics, tables, plots
- PDF report generation with embedded plots
- Export plot PNGs and ZIP
- Optional SMS alerts via Twilio
"""
import os
import io
import math
import zipfile
import tempfile
from datetime import datetime
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import streamlit as st
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, mean_absolute_error
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
# Optional Twilio import occurs later only if SMS is requested
# from twilio.rest import Client
st.set_page_config(page_title="Agri Decision Engine", layout="wide")
# ---------------------------
# Helper utilities
# ---------------------------
def read_excel_flexible(uploaded_file):
"""Read uploaded file (streamlit uploader) or path string."""
try:
if hasattr(uploaded_file, "read"):
df = pd.read_excel(uploaded_file)
else:
df = pd.read_excel(uploaded_file)
return df
except Exception as e:
st.error(f"Failed to read Excel: {e}")
return None
def clean_cols(df):
"""Lowercase columns mapping to canonical names; also trim spaces."""
df = df.copy()
# strip whitespace from column names
df.columns = [c.strip() for c in df.columns]
col_map = {c: c.lower() for c in df.columns}
df = df.rename(columns=col_map)
return df
def safe_series(df, col_names):
"""Return first available series among col_names (list)."""
for c in col_names:
if c in df.columns:
return df[c].astype(float).reset_index(drop=True)
return None
def compute_rmse_mae(y_true, y_pred):
try:
mask = (~np.isnan(y_true)) & (~np.isnan(y_pred))
if mask.sum() == 0:
return np.nan, np.nan
y_true = np.asarray(y_true)[mask].astype(float)
y_pred = np.asarray(y_pred)[mask].astype(float)
rmse = math.sqrt(mean_squared_error(y_true, y_pred))
mae = mean_absolute_error(y_true, y_pred)
return rmse, mae
except Exception:
return np.nan, np.nan
def linear_trend(series):
"""Return slope and intercept of linear fit."""
y = np.asarray(series, dtype=float)
idx = np.arange(len(y)).reshape(-1,1)
mask = ~np.isnan(y)
if mask.sum() < 2:
return 0.0, float(np.nan)
reg = LinearRegression()
reg.fit(idx[mask], y[mask])
return float(reg.coef_[0]), float(reg.intercept_)
def mk_classify(mk_field):
"""Interpret MK field values into human labels."""
if mk_field is None or (isinstance(mk_field, float) and np.isnan(mk_field)):
return "no-data"
s = str(mk_field).lower()
if "incre" in s:
return "increasing"
if "decre" in s:
return "decreasing"
return s
def fill_default_columns(df):
"""Ensure presence of canonical series names by mapping."""
# Lower-case column names are expected
# possible names for each
mapping_candidates = {
"actual": ["actual", "rain", "observed", "rainfall"],
"predicted": ["predicted", "forecast", "pred"],
"seasonal": ["seasonal", "season"],
"residuals": ["residuals", "resid"],
"trend": ["trend", "linear_trend", "trend_component"],
"mk_trend": ["mk_trend", "mann_kendall", "mk"]
}
out = {}
for canon, candidates in mapping_candidates.items():
for c in candidates:
if c in df.columns:
out[canon] = df[c].astype(float).reset_index(drop=True)
break
else:
out[canon] = None
# Also collect model-specific preds and metrics
# Look for any column with 'pred' in name
preds = {c: df[c].astype(float).reset_index(drop=True) for c in df.columns if "pred" in c}
metrics = {c: df[c].astype(float).reset_index(drop=True) for c in df.columns if any(x in c for x in ["rmse","mae"])}
return out, preds, metrics
# ---------------------------
# Decision functions (simple rules)
# ---------------------------
def irrigation_decision_logic(actual, predicted, trend_series):
"""Return short irrigation decision string."""
# Use trend slope and next prediction vs average
slope, _ = linear_trend(trend_series if trend_series is not None else actual)
avg_actual = np.nanmean(actual) if actual is not None else np.nan
next_pred = float(predicted.iloc[-1]) if (predicted is not None and len(predicted) > 0) else np.nan
decisions = []
if not np.isnan(slope):
if slope > 0.01:
decisions.append("Long-term water demand is rising. Plan for increased storage and supply.")
elif slope < -0.01:
decisions.append("Long-term water demand is decreasing. Review storage expansion plans.")
else:
decisions.append("Long-term water demand is stable.")
else:
decisions.append("Trend data insufficient for long-term irrigation advice.")
if not np.isnan(next_pred) and not np.isnan(avg_actual):
if next_pred > avg_actual * 1.15:
decisions.append("Short-term forecast higher than average. Prepare additional releases from reservoirs.")
elif next_pred < avg_actual * 0.85:
decisions.append("Short-term forecast lower than average. Conserve water and delay non-essential releases.")
else:
decisions.append("No major short-term water action required.")
else:
decisions.append("Prediction data insufficient for short-term irrigation advice.")
return "\n".join(decisions)
def crop_planning_decision_logic(actual, predicted, mk_label, trend_series):
"""Return crop planning suggestions."""
slope, _ = linear_trend(trend_series if trend_series is not None else actual)
decisions = []
if not np.isnan(slope):
if slope < -0.01:
decisions.append("Rainfall trend decreasing. Recommend drought-tolerant and short-duration crops.")
elif slope > 0.01:
decisions.append("Rainfall trend increasing. Consider water-demanding crops or expanded cropping area.")
else:
decisions.append("Rainfall trend stable. Normal crop mix advisable.")
else:
decisions.append("Trend data insufficient for crop planning.")
mk = mk_classify(mk_label)
if mk == "increasing":
decisions.append("Mann–Kendall indicates increasing trend — adjust long-term cropping strategy accordingly.")
elif mk == "decreasing":
decisions.append("Mann–Kendall indicates decreasing trend — prioritize drought-adaptive varieties.")
else:
decisions.append("Mann–Kendall indicates no strong trend.")
return "\n".join(decisions)
def fertilizer_decision_logic(seasonal_series):
"""Use seasonal component to time fertilizer application."""
if seasonal_series is None or seasonal_series.isnull().all():
return "No seasonal information. Follow standard fertilizer schedule."
# find peak index(s)
peak_idx = int(np.nanargmax(seasonal_series.values))
decisions = []
decisions.append(f"Peak seasonal nutrient demand around index {peak_idx} (use this as reference).")
# measure seasonality strength
season_amp = (np.nanmax(seasonal_series) - np.nanmin(seasonal_series))
if season_amp > 0.2 * np.nanmax(np.abs(seasonal_series)):
decisions.append("Seasonality strong. Implement timely pre-peak fertilizer application.")
else:
decisions.append("Seasonality mild. Standard fertilizer program is sufficient.")
return "\n".join(decisions)
def pest_management_decision_logic(residuals, trend_series):
"""Use residual spikes and trend for pest warnings."""
decisions = []
slope, _ = linear_trend(trend_series if trend_series is not None else [])
if not np.isnan(slope) and slope > 0.005:
decisions.append("Rising trend suggests potential for increased pest/disease risk. Plan preventive monitoring.")
else:
decisions.append("No strong long-term increase in pest risk from trend analysis.")
if residuals is None or residuals.isnull().all():
decisions.append("No residual data to detect anomalies.")
return "\n".join(decisions)
resid = residuals.values
# check big spikes > 2*std
std = np.nanstd(resid)
max_abs = np.nanmax(np.abs(resid))
if std > 0 and max_abs > 2 * std:
decisions.append("Significant residual spikes detected. Investigate for outbreaks or extreme events.")
else:
decisions.append("Residuals show no large anomalies.")
return "\n".join(decisions)
# ---------------------------
# PDF & plotting utilities
# ---------------------------
def create_timeseries_plot(actual, preds_dict, title):
"""Create matplotlib figure with actual and multiple predictions (dict name->series)."""
fig, ax = plt.subplots(figsize=(10,4))
if actual is not None:
ax.plot(actual.values, label="Actual", linewidth=1.5)
for name, series in preds_dict.items():
if series is not None:
ax.plot(series.values, label=name)
ax.set_title(title)
ax.set_xlabel("Index")
ax.set_ylabel("Value")
ax.legend(fontsize=8)
plt.tight_layout()
return fig
def save_png_from_fig(fig, out_path):
fig.savefig(out_path, bbox_inches='tight')
plt.close(fig)
def save_pdf_report(pdf_path, summary_blocks, image_bytes_list):
c = canvas.Canvas(pdf_path, pagesize=A4)
width, height = A4
margin = 40
y = height - margin
c.setFont("Helvetica-Bold", 16)
c.drawString(margin, y, "Agricultural Decision Report")
y -= 24
c.setFont("Helvetica", 9)
c.drawString(margin, y, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
y -= 18
for block in summary_blocks:
lines = str(block).splitlines()
for line in lines:
if y < 120:
c.showPage()
y = height - margin
c.setFont("Helvetica", 9)
c.drawString(margin, y, line)
y -= 12
y -= 8
for label, img_bytes in image_bytes_list:
if y < 200:
c.showPage()
y = height - margin
c.setFont("Helvetica-Bold", 10)
c.drawString(margin, y, label)
y -= 14
tmpf = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
tmpf.write(img_bytes)
tmpf.flush()
tmpf.close()
try:
img_w = width - 2*margin
img_h = img_w * 0.45
c.drawImage(tmpf.name, margin, y - img_h, width=img_w, height=img_h)
y -= (img_h + 12)
except Exception as e:
c.setFont("Helvetica", 9)
c.drawString(margin, y, f"Could not embed image: {e}")
y -= 12
finally:
try:
os.unlink(tmpf.name)
except Exception:
pass
c.save()
# ---------------------------
# STREAMLIT UI
# ---------------------------
st.header("Upload model output Excel file(s)")
uploaded_files = st.file_uploader("Upload one or more Excel files (model outputs)", type=["xlsx","xls"], accept_multiple_files=True)
st.sidebar.header("Report & SMS options")
generate_pdf = st.sidebar.checkbox("Generate PDF report", value=True)
export_png_zip = st.sidebar.checkbox("Export plots as PNG & ZIP", value=True)
sms_toggle = st.sidebar.checkbox("Send SMS alerts (Twilio)", value=False)
if sms_toggle:
st.sidebar.text("Provide Twilio credentials")
twilio_sid = st.sidebar.text_input("Twilio Account SID", value="")
twilio_token = st.sidebar.text_input("Twilio Auth Token", value="", type="password")
twilio_from = st.sidebar.text_input("From number (Twilio)", value="")
sms_to_number = st.sidebar.text_input("Recipient number", value="")
process_button = st.button("Generate Decisions & Reports")
# ---------------------------
# Main processing
# ---------------------------
if process_button:
if not uploaded_files or len(uploaded_files) == 0:
st.error("Upload at least one model output Excel file.")
else:
all_summary_blocks = []
all_image_bytes = []
png_paths = []
# Accumulate dashboard table rows
dashboard_rows = []
for uploaded in uploaded_files:
try:
df_raw = read_excel_flexible(uploaded)
if df_raw is None:
continue
df = clean_cols(df_raw)
# map canonical series
canon, preds, metrics = fill_default_columns(df)
actual = canon.get("actual")
predicted_base = canon.get("predicted")
seasonal = canon.get("seasonal")
residuals = canon.get("residuals")
trend_series = canon.get("trend")
mk_field = canon.get("mk_trend")
# If predicted not present but model-specific preds exist, pick first
if predicted_base is None:
if preds:
# take first pred column
k = list(preds.keys())[0]
predicted_base = preds[k]
# Compute model metrics if not present
# Example: compute RMSE/MAE between actual and 'predicted' (or per-pred columns)
model_metrics = {}
# Based on all pred columns
if preds:
for name, series in preds.items():
rmse, mae = compute_rm = compute_rmse_mae = compute_rmse_mae if False else compute_rmse_mae # placeholder no-op to keep code readable (we'll compute below)
# Instead compute below properly
per_pred_metrics = {}
for name, series in preds.items():
try:
rmse, mae = compute_rmse_mae(actual.values if actual is not None else np.array([]), series.values)
except Exception:
rmse, mae = np.nan, np.nan
per_pred_metrics[name] = {"rmse": rmse, "mae": mae}
# Build decisions
irrigation_decision = irrigation_decision_logic(actual, predicted_base if predicted_base is not None else next(iter(preds.values())) if preds else None, trend_series)
crop_decision = crop_planning_decision_logic(actual, predicted_base if predicted_base is not None else None, mk_field, trend_series)
fertilizer_decision = fertilizer_decision_logic(seasonal)
pest_decision = pest_management_decision_logic(residuals, trend_series)
# Dashboard table row (summary)
row = {
"file_name": getattr(uploaded, "name", str(uploaded)),
"n_rows": len(df),
"irrigation_summary": irrigation_decision.split("\n")[0] if irrigation_decision else "",
"crop_summary": crop_decision.split("\n")[0] if crop_decision else "",
"fertilizer_summary": fertilizer_decision.split("\n")[0] if fertilizer_decision else "",
"pest_summary": pest_decision.split("\n")[0] if pest_decision else ""
}
dashboard_rows.append(row)
# Create combined timeseries plot (actual + all preds)
preds_for_plot = {}
if predicted_base is not None:
preds_for_plot["Predicted"] = predicted_base
# add named preds
for k,v in preds.items():
preds_for_plot[k] = v
fig = create_timeseries_plot(actual if actual is not None else pd.Series(dtype=float), preds_for_plot, title=f"Actual vs Predictions — {row['file_name']}")
buf = io.BytesIO()
fig.savefig(buf, format='png', bbox_inches='tight')
buf.seek(0)
img_bytes = buf.getvalue()
all_image_bytes.append((f"{row['file_name']} - Predictions", img_bytes))
if export_png_zip:
png_path = os.path.join(tempfile.gettempdir(), f"{row['file_name']}_pred.png")
with open(png_path, "wb") as f:
f.write(img_bytes)
png_paths.append(png_path)
st.subheader(f"File: {row['file_name']}")
st.image(img_bytes, use_column_width=True)
plt.close(fig)
# Also create residual plot if residuals present
if residuals is not None:
figr, axr = plt.subplots(figsize=(8,3))
axr.plot(residuals.values, label="Residuals")
axr.axhline(0, color='k', linewidth=0.6)
axr.set_title("Residuals")
axr.legend()
bufr = io.BytesIO()
figr.savefig(bufr, format='png', bbox_inches='tight')
bufr.seek(0)
all_image_bytes.append((f"{row['file_name']} - Residuals", bufr.getvalue()))
if export_png_zip:
p = os.path.join(tempfile.gettempdir(), f"{row['file_name']}_resid.png")
with open(p, "wb") as f:
f.write(bufr.getvalue())
png_paths.append(p)
st.image(bufr.getvalue(), width=700)
plt.close(figr)
# Add small textual summary on the page
st.markdown("**Irrigation advice:**")
st.info(irrigation_decision)
st.markdown("**Crop planning advice:**")
st.info(crop_decision)
st.markdown("**Fertilizer scheduling advice:**")
st.info(fertilizer_decision)
st.markdown("**Pest management advice:**")
st.info(pest_decision)
# Add per-pred metrics table
if per_pred_metrics:
metrics_df = pd.DataFrame.from_dict(per_pred_metrics, orient='index')
metrics_df = metrics_df.rename(columns={"rmse":"RMSE","mae":"MAE"})
st.write("Model metrics (per prediction column):")
st.dataframe(metrics_df)
else:
st.write("No per-model prediction columns found to compute RMSE/MAE.")
# Build summary block for PDF
summary_block = f"File: {row['file_name']}\nRows: {row['n_rows']}\nIrrigation: {irrigation_decision}\nCrop: {crop_decision}\nFertilizer: {fertilizer_decision}\nPest: {pest_decision}\n"
all_summary_blocks.append(summary_block)
except Exception as e:
st.error(f"Error processing file {getattr(uploaded, 'name', str(uploaded))}: {e}")
continue
# END for uploaded files
# Dashboard: aggregated table
if dashboard_rows:
st.header("Summary Dashboard")
dash_df = pd.DataFrame(dashboard_rows)
st.dataframe(dash_df)
# Generate PDF if requested
if generate_pdf and all_summary_blocks and all_image_bytes:
pdf_name = f"Agri_Report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
pdf_path = os.path.join(tempfile.gettempdir(), pdf_name)
try:
save_pdf_report(pdf_path, all_summary_blocks, all_image_bytes)
with open(pdf_path, "rb") as pf:
st.download_button("Download PDF Report", pf, file_name=pdf_name)
st.success("PDF report generated.")
except Exception as e:
st.error(f"PDF generation failed: {e}")
# Export PNGs as ZIP
if export_png_zip and png_paths:
zip_name = f"plots_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
zip_path = os.path.join(tempfile.gettempdir(), zip_name)
with zipfile.ZipFile(zip_path, 'w', compression=zipfile.ZIP_DEFLATED) as zf:
for p in png_paths:
zf.write(p, arcname=os.path.basename(p))
with open(zip_path, "rb") as zf:
st.download_button("Download PNG plots (ZIP)", zf, file_name=zip_name)
st.success("PNG zip ready.")
# SMS alerts via Twilio (optional)
if sms_toggle:
# Build short mobile-friendly summary
mobile_summary = []
for sblock in all_summary_blocks[:5]: # include up to first 5 files
# compress summary to short lines for SMS
lines = sblock.splitlines()
header = lines[0] if lines else ""
# take first 3 advice lines
advice_lines = []
for L in lines[1:6]:
if L.strip():
advice_lines.append(L.strip())
if len(advice_lines) >= 3:
break
mobile_summary.append(header + " | " + " ; ".join(advice_lines))
sms_text = "\n".join(mobile_summary)
st.text_area("SMS preview", sms_text, height=200)
if twilio_sid and twilio_token and twilio_from and sms_to_number:
try:
from twilio.rest import Client
client = Client(twilio_sid, twilio_token)
# Twilio SMS may require a single-line body and country code number formatting
msg = client.messages.create(body=sms_text[:1500], from_=twilio_from, to=sms_to_number)
st.success(f"SMS sent. SID: {msg.sid}")
except Exception as e:
st.error(f"Failed to send SMS via Twilio: {e}")
else:
st.warning("Provide Twilio credentials and numbers to send SMS.")
st.success("Processing finished.")
Quick usage notes
-
Upload one or more Excel files via the UI.
-
Adjust options in the sidebar:
-
Toggle PDF generation and PNG ZIP export.
-
Turn on SMS alerts if you want mobile notifications (enter Twilio credentials).
-
-
Click Generate Decisions & Reports.
-
View per-file plots and decision cards on the page.
-
Download the PDF report and PNG ZIP from the UI.
-
If SMS is enabled and configured, a concise mobile summary will be sent.
SMS / Twilio note
-
The app does not include Twilio credentials. You must supply them.
-
To get Twilio creds, sign up at twilio.com, get
Account SID,Auth Token, and a TwilioFromnumber. -
SMS cost applies per Twilio pricing.
If you'd like next:
-
I can add templates for SMS content (short/long).
-
I can hook the app to send emails instead of SMS.
-
I can add scheduled runs (cron) — requires deployment environment.
Do you want scheduled automatic runs next?
No comments:
Post a Comment