Update app.py

app.py

@@ -5,7 +5,6 @@ from typing import List, Dict, Any, Tuple
 
 import gradio as gr
 import torch
-import pandas as pd               # <-- NEW: for real CSV analytics
 import regex as re2  # robust control-char sanitizer
 
 from settings import SNAPSHOT_PATH, PERSIST_CONTENT
@@ -48,7 +47,12 @@ from huggingface_hub import login
 
 from safety import safety_filter, refusal_reply
 from retriever import init_retriever, retrieve_context
-from decision_math import compute_operational_numbers
+from decision_math import compute_operitional_numbers as compute_operational_numbers  # in case of rename
+try:
+    # prefer the original name if present
+    from decision_math import compute_operational_numbers as compute_operational_numbers
+except Exception:
+    pass
 from prompt_templates import build_system_preamble
 from upload_ingest import extract_text_from_files
 from session_rag import SessionRAG
@@ -218,9 +222,6 @@ def _load_snapshot(path=SNAPSHOT_PATH):
 init_retriever()
 _session_rag = SessionRAG()
 
-# In-memory stash of uploaded DataFrames (name -> pd.DataFrame)
-_SESSION_FRAMES: Dict[str, pd.DataFrame] = {}   # <-- NEW
-
 # ---------- Executive pre-compute (MDSi block) ----------
 def _mdsi_block():
     base_capacity = capacity_projection(18, 48, 6)
@@ -234,51 +235,236 @@ def _mdsi_block():
         "outcomes_summary": outcomes
     }, indent=2)
 
-# ---------- DataFrame -> JSON summary (generic, schema-free) ----------
-def _summarize_frames_for_prompt(frames: Dict[str, pd.DataFrame], max_cols: int = 12, max_groups: int = 10) -> str:
-    """
-    Build a JSON block with concrete, generic stats from uploaded DataFrames.
-    Works for arbitrary CSVs (no static schema).
-    """
-    def safe_num_cols(df: pd.DataFrame):
-        return [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
-    def likely_group_cols(df: pd.DataFrame):
-        cand = [c for c in df.columns if any(k in str(c).lower()
-                for k in ["settlement", "community", "facility", "site", "region", "zone", "program", "service", "specialty", "hospital"])]
-        return cand[:2]
-    out = {"files": []}
-    for name, df in (frames or {}).items():
+# ---------- Scenario-agnostic dynamic analytics (multi-file) ----------
+import pandas as _pd
+from collections import Counter
+import itertools as _it
+import numpy as _np
+
+_NUM_SAMPLE_ROWS = 50000  # cap per file for speed
+
+def _read_table(path: str) -> _pd.DataFrame:
+    try:
+        if path.lower().endswith((".xlsx", ".xls")):
+            return _pd.read_excel(path)
+        return _pd.read_csv(path, low_memory=False, nrows=_NUM_SAMPLE_ROWS)
+    except Exception:
+        return _pd.DataFrame()
+
+def _profile_schema(df: _pd.DataFrame) -> Dict[str, Any]:
+    if df.empty:
+        return {"numeric": [], "categorical": [], "datetime": [], "textlike": []}
+    numeric, categorical, datetime, textlike = [], [], [], []
+    for c in df.columns:
+        s = df[c]
+        if _pd.api.types.is_numeric_dtype(s):
+            numeric.append(c)
+        elif _pd.api.types.is_datetime64_any_dtype(s):
+            datetime.append(c)
+        else:
+            uniq = s.astype(str).nunique(dropna=True)
+            if uniq <= max(50, int(0.03 * max(1, len(s)))):
+                categorical.append(c)
+            else:
+                textlike.append(c)
+    return {"numeric": numeric, "categorical": categorical, "datetime": datetime, "textlike": textlike}
+
+def _safe_num(s: _pd.Series) -> _pd.Series:
+    if not _pd.api.types.is_numeric_dtype(s):
+        return _pd.to_numeric(s, errors="coerce")
+    return s
+
+def _top_numeric_by_variance(df: _pd.DataFrame, numeric_cols: List[str], k=5) -> List[str]:
+    scores = []
+    for c in numeric_cols:
+        x = _safe_num(df[c])
         try:
-            rec = {"name": name, "shape": [int(df.shape[0]), int(df.shape[1])], "columns": list(map(str, df.columns[:max_cols]))}
-            num_cols = safe_num_cols(df)
-            if num_cols:
-                # count, mean, std, min, 25%, 50%, 75%, max for each numeric column
-                desc = df[num_cols].describe().to_dict()
-                # convert numpy types to natives for JSON
-                for k, v in desc.items():
-                    for m, val in v.items():
-                        try:
-                            v[m] = float(val)
-                        except Exception:
-                            try:
-                                v[m] = int(val)
-                            except Exception:
-                                pass
-                rec["numeric_summary"] = desc
-            groups = []
-            for gcol in likely_group_cols(df):
+            scores.append((c, _np.nanvar(x.values)))
+        except Exception:
+            scores.append((c, _np.nan))
+    scores.sort(key=lambda t: (t[1] if _np.isfinite(t[1]) else -1), reverse=True)
+    return [c for c, _ in scores[:k]]
+
+def _top_categories(df: _pd.DataFrame, cat_cols: List[str], k=3) -> Dict[str, List[Tuple[str,int]]]:
+    out = {}
+    for c in cat_cols[:6]:
+        vc = Counter(df[c].astype(str).fillna("<NA>")).most_common(k)
+        out[c] = vc
+    return out
+
+def _infer_candidate_keys(dfs_named: List[Tuple[str, _pd.DataFrame]]) -> List[str]:
+    all_cols = []
+    for name, df in dfs_named:
+        all_cols.extend(list(map(str, df.columns)))
+    counts = Counter([c.strip() for c in all_cols])
+    bad = set(["value","values","count","total","sum","mean","median","date","timestamp","index"])
+    return [c for c, n in counts.items() if n >= 2 and c.lower() not in bad]
+
+def _try_joins(dfs_named: List[Tuple[str, _pd.DataFrame]], keys: List[str], max_pairs=3) -> List[str]:
+    previews = []
+    pairs = list(_it.combinations(range(len(dfs_named)), 2))
+    shown = 0
+    for i, j in pairs:
+        if shown >= max_pairs:
+            break
+        name_i, dfi = dfs_named[i]
+        name_j, dfj = dfs_named[j]
+        for k in keys:
+            if k in dfi.columns and k in dfj.columns:
                 try:
-                    gb = df.groupby(gcol).size().sort_values(ascending=False).head(max_groups)
-                    # ensure JSON-serializable
-                    groups.append({"by": str(gcol), "size_top": {str(k): int(v) for k, v in gb.to_dict().items()}})
+                    merged = dfi[[k]].dropna().merge(dfj[[k]].dropna(), on=k, how="inner")
+                    previews.append(f"- Join {name_i} ↔ {name_j} on `{k}` → matches: {len(merged):,}")
+                    shown += 1
+                    if shown >= max_pairs:
+                        break
                 except Exception:
-                    pass
-            if groups:
-                rec["groups"] = groups
-            out["files"].append(rec)
-        except Exception:
-            continue
-    return json.dumps(out, indent=2)
+                    continue
+    return previews
+
+def _scenario_tokens(text: str) -> List[str]:
+    t = (text or "").lower()
+    t = re.sub(r"[^a-z0-9_ -]+", " ", t)
+    toks = [w for w in t.split() if len(w) >= 3]
+    out, seen = [], set()
+    for w in toks:
+        if w not in seen:
+            seen.add(w); out.append(w)
+    return out
+
+def _extract_intents(text: str) -> Dict[str, Any]:
+    toks = _scenario_tokens(text)
+    intents = {
+        "rank": any(x in toks for x in ["rank","top","longest","highest","lowest","shortest","worst","best"]),
+        "agg_words": [w for w in toks if w in set(["mean","average","avg","median","p50","p90","sum","total"])],
+        "n_top": 5
+    }
+    return intents
+
+def _pick_dims_from_tokens(df: _pd.DataFrame, cat_cols: List[str], toks: List[str]) -> List[str]:
+    scored = []
+    for c in cat_cols:
+        score = sum(1 for t in toks if t in c.lower())
+        scored.append((score, c))
+    scored.sort(key=lambda t: (t[0], -len(t[1])), reverse=True)
+    picked = [c for s, c in scored if s > 0][:3]
+    if not picked:
+        picked = cat_cols[:3]
+    return picked
+
+def _pick_metrics_from_tokens(df: _pd.DataFrame, num_cols: List[str], toks: List[str]) -> List[str]:
+    scored = []
+    for c in num_cols:
+        score = sum(1 for t in toks if t in c.lower())
+        scored.append((score, c))
+    scored.sort(key=lambda t: (t[0], -len(t[1])), reverse=True)
+    picked = [c for s, c in scored if s > 0][:3]
+    if not picked:
+        picked = _top_numeric_by_variance(df, num_cols, k=3)
+    return picked
+
+def _mk_table(md_title: str, df: _pd.DataFrame, limit=10) -> str:
+    if df.empty: return ""
+    return f"{md_title}\n" + df.head(limit).to_markdown(index=False)
+
+def compute_dynamic_analytics_block(arts: List[Dict[str, Any]], scenario_text: str) -> str:
+    dfs_named: List[Tuple[str, _pd.DataFrame]] = []
+    for a in arts or []:
+        p = a.get("path"); n = a.get("name") or "table"
+        if not p: continue
+        if not str(p).lower().endswith((".csv",".xlsx",".xls")): continue
+        d = _read_table(p)
+        if d.empty: continue
+        d = d.copy()
+        d.columns = [str(c).strip().replace("\n"," ").replace("\r"," ") for c in d.columns]
+        dfs_named.append((n, d))
+
+    if not dfs_named:
+        return ""
+
+    overview_rows = []
+    for n, d in dfs_named:
+        overview_rows.append({"File": n, "Rows": len(d), "Columns": d.shape[1]})
+    overview_md = _pd.DataFrame(overview_rows).to_markdown(index=False)
+
+    per_table_blocks = []
+    toks = _scenario_tokens(scenario_text)
+    intents = _extract_intents(scenario_text)
+
+    for n, d in dfs_named:
+        prof = _profile_schema(d)
+        num_cols = prof["numeric"]
+        cat_cols = prof["categorical"]
+
+        top_num = _top_numeric_by_variance(d, num_cols, k=5) if num_cols else []
+        num_sum = _pd.DataFrame()
+        if top_num:
+            stat_rows = []
+            for c in top_num:
+                x = _safe_num(d[c])
+                try_mean = float(_np.nanmean(x)) if x.size else _np.nan
+                try_median = float(_np.nanmedian(x)) if x.size else _np.nan
+                try_p90 = float(_np.nanpercentile(x.dropna(), 90)) if x.dropna().size else _np.nan
+                stat_rows.append({
+                    "Metric": c,
+                    "count": int(x.count()),
+                    "mean": try_mean,
+                    "median": try_median,
+                    "p90": try_p90
+                })
+            num_sum = _pd.DataFrame(stat_rows)
+
+        cat_info = _top_categories(d, cat_cols, k=5) if cat_cols else {}
+        cat_md = []
+        for c, vc in cat_info.items():
+            parts = ", ".join([f"{val} ({cnt})" for val, cnt in vc])
+            cat_md.append(f"- {c}: {parts}")
+
+        rank_tables = []
+        if intents.get("rank") and num_cols and cat_cols:
+            dims = _pick_dims_from_tokens(d, cat_cols, toks)
+            mets = _pick_metrics_from_tokens(d, num_cols, toks)
+            for gcol in dims[:2]:
+                for mcol in mets[:2]:
+                    try:
+                        g = (
+                            d.groupby(gcol, as_index=False)[mcol]
+                             .mean(numeric_only=True)
+                             .rename(columns={mcol: f"avg({mcol})"})
+                             .sort_values(f"avg({mcol})", ascending=False)
+                             .head(intents["n_top"])
+                        )
+                        rank_tables.append(_mk_table(f"Top {intents['n_top']} by avg({mcol}) — grouped by {gcol}:", g))
+                    except Exception:
+                        continue
+
+        block_parts = [f"### {n}"]
+        if not num_sum.empty:
+            block_parts.append(_mk_table("Numeric summary (top-variance metrics):", num_sum))
+        if cat_md:
+            block_parts.append("Top categories:\n" + "\n".join(cat_md))
+        for rt in rank_tables:
+            if rt: block_parts.append(rt)
+
+        per_table_blocks.append("\n\n".join([p for p in block_parts if p]))
+
+    keys = _infer_candidate_keys(dfs_named)
+    join_md = ""
+    if keys:
+        joins = _try_joins(dfs_named, keys, max_pairs=3)
+        if joins:
+            join_md = "Join previews:\n" + "\n".join(joins)
+
+    parts = [
+        "Computed Analytics Block (auto-generated, scenario-agnostic):",
+        "",
+        "Dataset overview:",
+        overview_md,
+        "",
+        "\n\n".join(per_table_blocks)
+    ]
+    if join_md:
+        parts.extend(["", join_md])
+    return "\n".join(parts)
 
 # ---------- Dynamic Phase 1 question generator ----------
 def _extract_present_domains(artifacts: List[Dict[str, Any]]) -> Dict[str, bool]:
@@ -391,7 +577,6 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
             ans = "I am ClarityOps, your strategic decision making AI partner."
             return history + [(user_msg, ans)], awaiting_answers
 
-        # ---- Ingest uploads FIRST (files alone can trigger scenario mode)
         artifacts = []
         if uploaded_files_paths:
             ing = extract_text_from_files(uploaded_files_paths)
@@ -401,21 +586,8 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
                 _session_rag.add_docs(chunks)
             if artifacts:
                 _session_rag.register_artifacts(artifacts)
-            # NEW: Read CSVs into DataFrames and stash in-memory for analytics
-            for a in (artifacts or []):
-                try:
-                    if a.get("kind") == "csv" and a.get("path") and a.get("name"):
-                        # read the whole CSV with automatic dtype inference; fallback to strings
-                        try:
-                            df = pd.read_csv(a["path"])
-                        except Exception:
-                            df = pd.read_csv(a["path"], dtype=str, low_memory=False)
-                        _SESSION_FRAMES[str(a["name"])] = df
-                except Exception:
-                    pass
-            log_event("uploads_added", None, {"chunks": len(chunks), "artifacts": len(artifacts), "dfs": len(_SESSION_FRAMES)})
+            log_event("uploads_added", None, {"chunks": len(chunks), "artifacts": len(artifacts)})
 
-        # CSV columns helper (works in both modes)
         if re.search(r"\b(columns?|headers?)\b", (safe_in or "").lower()):
             cols = _session_rag.get_latest_csv_columns()
             if cols:
@@ -424,7 +596,6 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
         scenario_mode = is_scenario_triggered(safe_in, uploaded_files_paths)
 
         if not scenario_mode:
-            # ---------- Normal conversational chat ----------
             out = cohere_chat(safe_in, history) if USE_HOSTED_COHERE else None
             if not out:
                 model, tokenizer = load_local_model()
@@ -447,9 +618,7 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
             })
             return history + [(user_msg, safe_out)], awaiting_answers
 
-        # ---------- Scenario Mode ----------
         if not awaiting_answers:
-            # PHASE 1: dynamic questions (no assumptions)
             phase1 = build_dynamic_clarifications(scenario_text=safe_in, artifacts=artifacts or _session_rag.artifacts)
             phase1 = _sanitize_text(phase1)
             log_event("assistant_reply", None, {
@@ -460,7 +629,7 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
             })
             return history + [(user_msg, phase1)], True
 
-        # PHASE 2: build rich system preamble + feed to LLM
+        # ---------- Phase 2 ----------
         session_snips = "\n---\n".join(_session_rag.retrieve(
             "diabetes screening Indigenous Métis mobile program cost throughput outcomes logistics",
             k=6
@@ -475,16 +644,32 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
         user_lower = (safe_in or "").lower()
         mdsi_extra = _mdsi_block() if ("diabetes" in user_lower or "mdsi" in user_lower or "mobile screening" in user_lower) else ""
 
-        # NEW: Real computed stats from CSVs for the model to use
-        computed_from_csvs = _summarize_frames_for_prompt(_SESSION_FRAMES)
-        artifact_block = "Computed Blocks From Uploaded Data (JSON):\n" + computed_from_csvs
+        arts = _session_rag.artifacts or []
+        if arts:
+            arts_summ = []
+            for a in arts:
+                nm = a.get("name") or "<unnamed>"
+                cols = ", ".join(a.get("columns") or [])[:600]
+                rows = a.get("n_rows_sampled") or 0
+                arts_summ.append(f"- {nm}: columns[{cols}] sample_rows={rows}")
+            artifact_block = "Uploaded Data Files (summarized):\n" + "\n".join(arts_summ)
+        else:
+            artifact_block = "Uploaded Data Files (summarized):\n- <none>"
+
+        # NEW: scenario-agnostic, multi-file analytics block
+        analytics_block = compute_dynamic_analytics_block(arts, safe_in)
 
         scenario_block = safe_in if len((safe_in or "")) > 0 else ""
         system_preamble = build_system_preamble(
             snapshot=snapshot,
             policy_context=policy_context,
             computed_numbers=computed,
-            scenario_text=scenario_block + f"\n\n{artifact_block}" + (f"\n\nExecutive Pre-Computed Blocks:\n{mdsi_extra}" if mdsi_extra else ""),
+            scenario_text=(
+                scenario_block
+                + f"\n\n{artifact_block}"
+                + (f"\n\n{analytics_block}" if analytics_block else "")
+                + (f"\n\nExecutive Pre-Computed Blocks:\n{mdsi_extra}" if mdsi_extra else "")
+            ),
             session_snips=session_snips
         )
 
@@ -492,8 +677,7 @@ def clarityops_reply(user_msg, history, tz, uploaded_files_paths, awaiting_answe
             "\n\n[INSTRUCTION TO MODEL]\n"
             "Produce **Phase 2** only now: start with 'Structured Analysis' and follow the exact section order "
             "(Prioritization, Capacity, Cost, Clinical Benefits, ClarityOps Top 3 Recommendations). "
-            "Use the JSON computed blocks from the uploaded files + the user's latest answers as authoritative. "
-            "Show calculations, units, and a brief Provenance. If required data is still missing, output INSUFFICIENT_DATA.\n"
+            "Use uploaded files + the user's latest answers as authoritative. Show calculations, units, and a brief Provenance.\n"
         )
 
         augmented_user = SYSTEM_MASTER + "\n\n" + system_preamble + "\n\nUser scenario & answers:\n" + safe_in + directive
@@ -658,8 +842,6 @@ with gr.Blocks(theme=theme, css=custom_css, analytics_enabled=False) as demo:
                concurrency_limit=2, queue=True)
 
     def _on_clear():
-        # also clear in-memory DataFrames
-        _SESSION_FRAMES.clear()
         return (
             [], "", [], False,
             gr.update(visible=True),
@@ -674,3 +856,4 @@ if __name__ == "__main__":
     demo.launch(server_name="0.0.0.0", server_port=port, show_api=False, max_threads=8)
 
 
+