Spaces:
Running
Running
| """The public page of a governed system: the signed model, its monitoring page, and the dossier. | |
| Built for the AI Governance Practicum (AI Governance Collective). One app.py serves all three organizations; space_config.json says which.""" | |
| import json, os, pickle, io, warnings | |
| import numpy as np, pandas as pd | |
| import gradio as gr | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| warnings.filterwarnings("ignore") | |
| HERE = os.path.dirname(os.path.abspath(__file__)) | |
| # Hugging Face's free tier runs Gradio Spaces on ZeroGPU, which refuses to start an app that declares no GPU function. | |
| # This app needs no GPU: the declaration below satisfies the check and is never called. On a plain CPU Space it is skipped. | |
| try: | |
| import spaces | |
| def _gpu_not_used(): | |
| return "This app runs on the CPU." | |
| except Exception: | |
| pass | |
| CFG = json.load(open(os.path.join(HERE, "space_config.json"))) | |
| REF = json.load(open(os.path.join(HERE, "reference.json"))) | |
| MODEL = pickle.load(open(os.path.join(HERE, CFG["model_file"]), "rb")) | |
| RECAL = pickle.load(open(os.path.join(HERE, CFG["recal_file"]), "rb")) if CFG.get("recal_file") else None | |
| REASONS = pd.read_csv(os.path.join(HERE, CFG["reasons_file"])) if CFG.get("reasons_file") else None | |
| EXAMPLES = pd.read_csv(os.path.join(HERE, "examples.csv"), keep_default_na=False) | |
| INPUTS, KEPT, SHARE = CFG["inputs"], CFG["kept"], CFG["share"] | |
| CAL_FLOOR = 25 # a calibration band is read only when it expects at least this many outcomes; a smaller band says nothing either way | |
| SPACE_ID = os.environ.get("SPACE_ID", "") | |
| WEEK6 = CFG["week6"] | |
| # ----------------------------------------------------------------------------- scoring | |
| def score_frame(X): | |
| """The production score. Where the Week 6 memo ratified fitted steps (the recalibration and the odds correction), they run here.""" | |
| X = X[INPUTS] | |
| if RECAL is not None: | |
| pf = RECAL.predict_proba(X)[:, 1]; f = CFG["odds_factor"] | |
| return pf * f / (pf * f + 1 - pf) | |
| return MODEL.predict_proba(X)[:, 1] | |
| def contributions(X): | |
| """Each input's fixed weight times the record's value on the scaled, remover-adjusted inputs. The record's own reasons.""" | |
| Z = X[INPUTS] | |
| for name, step in MODEL.steps[:-1]: | |
| Z = step.transform(Z) | |
| coef = MODEL.named_steps["model"].coef_[0] | |
| return pd.DataFrame(np.asarray(Z) * coef, columns=KEPT, index=X.index) | |
| def allowed_reasons(con): | |
| """Drop the columns a reason may never cite: a medication the record marks No (Alderwyn, Week 6 memo, ruling 5).""" | |
| drugs = CFG.get("drug_fields", []) | |
| if not drugs: return con | |
| keep = [c for c in con.index if not any(c == d + "_No" for d in drugs)] | |
| return con[keep] | |
| def allowed_reasons_frame(frame): | |
| drugs = CFG.get("drug_fields", []) | |
| return frame[[c for c in frame.columns if not any(c == d + "_No" for d in drugs)]] if drugs else frame | |
| def readable(col): | |
| if REASONS is not None: | |
| r = REASONS.loc[(REASONS.input == col) | ((REASONS.input + "_" + REASONS.value.astype(str)) == col), "reason"] | |
| if len(r): return r.iloc[0] | |
| for c in CFG["categorical"]: | |
| if col.startswith(c + "_"): return f"{c} is {col[len(c) + 1:]}" | |
| return col | |
| def coerce(df): | |
| """Read a file the way the working file is read: numeric inputs as numbers, categorical inputs as text.""" | |
| df = df.copy() | |
| for c in CFG["numeric"]: | |
| if c in df: df[c] = pd.to_numeric(df[c], errors="coerce") | |
| for c in CFG["categorical"]: | |
| if c in df: df[c] = df[c].astype(str) | |
| return df | |
| def groups_of(frame): | |
| g = {} | |
| for gname, d in CFG["groups"].items(): | |
| col = d["col"] | |
| if col not in frame: continue | |
| if "bands" in d: | |
| v = pd.to_numeric(frame[col], errors="coerce"); lab = pd.Series(["unrecorded"] * len(frame), index=frame.index, dtype=object) | |
| for lo, hi, name in d["bands"]: lab[(v >= lo) & (v <= hi)] = name | |
| g[gname] = lab | |
| else: | |
| s = frame[col].astype(str); g[gname] = s.map(d["map"]).fillna(s) if d.get("map") else s | |
| return pd.DataFrame(g, index=frame.index) | |
| def pct(x, d=1): return f"{x * 100:.{d}f} percent" | |
| # ----------------------------------------------------------------------------- the signed version (the collapsed panel at the top) | |
| def model_header(): | |
| a = CFG["artifacts"][CFG["model_file"]] | |
| L = [f"**Organization:** {CFG['org']}. **Signed by:** {CFG['caio']} ({CFG['memo_ref']}). **Kept by:** {CFG['lead']}.", | |
| f"**File:** `{CFG['model_file']}`, {a['bytes']:,} bytes, SHA-256 `{a['sha256']}`."] | |
| if RECAL is not None: | |
| r = CFG["artifacts"][CFG["recal_file"]] | |
| L.append(f"**Fitted steps the memo ratified:** the isotonic recalibration (`{CFG['recal_file']}`, SHA-256 `{r['sha256']}`), then the odds correction to the {int(CFG['calibration']['book_rate']*100)} percent book rate (factor {CFG['odds_factor']}). Every score on this page has both applied.") | |
| L.append(f"**What it is:** a logistic regression with {len(KEPT)} fitted columns from {len(INPUTS)} input fields. The weights are fixed. The same record gets the same score every time.") | |
| L.append(f"**Operating point:** the top {int(SHARE*100)} percent of the ranked file gets {CFG['action']}. On the Week 6 test set ({WEEK6['test_rows']:,} {CFG['units']}) that was {WEEK6['flagged']:,} {CFG['units']}, of which {WEEK6['caught']:,} were {CFG['positive_label']}. Test-set ROC-AUC {WEEK6['roc_auc']:.3f}.") | |
| L.append(f"**Libraries:** scikit-learn {CFG['versions']['sklearn']}, fairlearn {CFG['versions']['fairlearn']}. The model does not load on another version of scikit-learn, by design.") | |
| return "\n\n".join(L) | |
| def weights_table(n=25): | |
| return pd.DataFrame([dict(column=c, weight=round(w, 4), meaning=readable(c)) for c, w in CFG["weights"][:n]]) | |
| # ----------------------------------------------------------------------------- tab 1: one record | |
| CAT = [c for c in INPUTS if c in CFG["categorical"]] | |
| NUM = [c for c in INPUTS if c in CFG["numeric"]] | |
| FORM = CAT + NUM # the order of the form's controls | |
| def _typical(): | |
| d = {c: REF["numeric"][c]["median"] for c in NUM} | |
| for c in CAT: | |
| r = REF["categorical"][c]; d[c] = max(r, key=r.get) | |
| return d | |
| def _case_row(i): | |
| row = EXAMPLES.iloc[i] | |
| return {c: (float(row[c]) if c in NUM else str(row[c])) for c in FORM} | |
| PRESETS = {f"Typical {CFG['unit']} (training-file medians and most common values)": _typical()} | |
| for i in range(len(EXAMPLES)): | |
| PRESETS[f"{EXAMPLES['case'].iloc[i].capitalize()} (Week 6 case, rank {EXAMPLES['week6_rank'].iloc[i]} of {WEEK6['test_rows']:,})"] = _case_row(i) | |
| FIRST = list(PRESETS)[0] | |
| def apply_preset(name): | |
| d = PRESETS[name] | |
| return [d[c] for c in FORM] | |
| def field_label(c): | |
| """The input's name, with the meaning of its codes where the program's group definitions give one (SEX: 1 male, 2 female).""" | |
| for d in CFG["groups"].values(): | |
| if d["col"] == c and d.get("map"): | |
| return c + " (" + ", ".join(f"{k} {v}" for k, v in d["map"].items()) + ")" | |
| return c | |
| def choices_for(c): | |
| ch = sorted(REF["categorical"][c], key=str) | |
| for v in EXAMPLES[c].astype(str).unique(): | |
| if v not in ch: ch.append(v) | |
| return ch | |
| def score_one(*vals): | |
| row = {c: v for c, v in zip(FORM, vals)} | |
| X = coerce(pd.DataFrame([row])) | |
| p = float(score_frame(X)[0]); cut = REF["cut_score"]; inside = p >= cut | |
| verdict = (f"## Score: **{p:.1%}**\n\n" | |
| f"### {'馃敶 Above the cut' if inside else '馃煝 Below the cut'} (Week 6 cut = {cut:.1%}, the score of the last {CFG['unit']} inside the top {int(SHARE*100)} percent)\n\n" | |
| f"At the operating point the {CFG['memo_ref']} set, this {CFG['unit']} {'would' if inside else 'would not'} be {CFG['flagged_word']}: {CFG['action']}. " | |
| f"The score says where the record ranks. It is not a decision about the {CFG['person']}.") | |
| con = allowed_reasons(contributions(X).iloc[0]) | |
| up = con[con > 0].sort_values(ascending=False).head(6); down = con[con < 0].sort_values().head(4) | |
| items = list(up.items()) + list(down.items()) | |
| labs = [readable(c) for c, _ in items][::-1]; v = [float(x) for _, x in items][::-1] | |
| fig, ax = plt.subplots(figsize=(11, 0.45 * len(v) + 1.8)) | |
| ax.barh(range(len(v)), v, color=["#d62728" if x > 0 else "#2ca02c" for x in v]) | |
| ax.set_yticks(range(len(v))); ax.set_yticklabels(labs, fontsize=9); ax.axvline(0, color="black", lw=0.8) | |
| ax.set_xlabel(f"Toward {CFG['positive_label']} (red) or away from it (green): the fixed weight times this record's value", fontsize=9) | |
| ax.set_title(f"Why {CFG['system']} scored this {CFG['unit']} {p:.1%}: the record's own reasons", fontsize=11, loc="left") | |
| fig.tight_layout() | |
| return verdict, fig | |
| # ----------------------------------------------------------------------------- tab 2: batch upload | |
| def score_batch(file, top_n): | |
| if file is None: return None, f"Upload a CSV of {CFG['units']} to score (the model's input columns, one row per {CFG['unit']}).", None | |
| try: df = coerce(pd.read_csv(file.name if hasattr(file, "name") else file, low_memory=False, keep_default_na=False, na_values=[])) | |
| except Exception as e: return None, f"Could not read the file: {e}", None | |
| missing = [c for c in INPUTS if c not in df.columns] | |
| if missing: return None, "The file is missing input columns: " + ", ".join(missing) + ". Nothing was scored.", None | |
| s = score_frame(df); order = np.argsort(-s, kind="stable"); k = int(round(SHARE * len(df))) | |
| idc = CFG.get("id_col"); ids = df[idc].values if idc and idc in df.columns else np.arange(1, len(df) + 1) | |
| con = allowed_reasons_frame(contributions(df)); top = [readable(con.columns[i]) for i in np.argmax(con.values, axis=1)] | |
| out = pd.DataFrame({(idc if idc and idc in df.columns else "row"): ids, "score": np.round(s, 4), "ranked_in": np.where(np.arange(len(df)) >= 0, "", ""), "top_reason": top}) | |
| out = out.iloc[order].reset_index(drop=True); out.insert(0, "rank", np.arange(1, len(df) + 1)); out["ranked_in"] = np.where(out["rank"] <= k, "yes", "no") | |
| path = os.path.join("/tmp", "ranked_list.csv"); out.to_csv(path, index=False) | |
| msg = (f"Scored **{len(df):,}** {CFG['units']}. The top {int(SHARE*100)} percent, the operating point, is **{k:,}** {CFG['units']}. " | |
| f"Showing the top **{int(top_n)}** by score (the team's capacity). The download carries every record with its score, rank, whether it is ranked in, and its own top reason.") | |
| return out.head(int(top_n)), msg, path | |
| # ----------------------------------------------------------------------------- tab 2: monitoring | |
| def psi(ref_shares, new_shares): | |
| keys = set(ref_shares) | set(new_shares); eps = 1e-4; v = 0.0 | |
| for k in keys: | |
| a, b = max(ref_shares.get(k, 0.0), eps), max(new_shares.get(k, 0.0), eps); v += (b - a) * np.log(b / a) | |
| return float(v) | |
| def drift_table(df): | |
| rows = [] | |
| for c in INPUTS: | |
| if c not in df.columns: continue | |
| watch = c in CFG["watch_fields"] | |
| if c in REF["numeric"]: | |
| r = REF["numeric"][c]; v = pd.to_numeric(df[c], errors="coerce"); edges = r["edges"] | |
| if len(edges) < 3: continue | |
| def shares(x): | |
| b = np.clip(np.searchsorted(edges, x, side="right") - 1, 0, len(edges) - 2); return {str(i): float((b == i).mean()) for i in range(len(edges) - 1)} | |
| ref_sh = r["shares"]; new_sh = shares(v.dropna().values) | |
| idx = psi(ref_sh, new_sh) | |
| rows.append(dict(input=c, kind="number", training_file=f"mean {r['mean']:,.2f}, median {r['median']:,.2f}", this_file=f"mean {v.mean():,.2f}, median {v.median():,.2f}", missing=int(v.isna().sum()), stability_index=round(idx, 3), reading=reading(idx), watched="yes" if watch else "")) | |
| elif c in REF["categorical"]: | |
| r = REF["categorical"][c]; vc = df[c].astype(str).value_counts(normalize=True); new_sh = {str(k): float(x) for k, x in vc.items()} | |
| idx = psi(r, new_sh); top_r = max(r, key=r.get); top_n = max(new_sh, key=new_sh.get) if new_sh else "" | |
| new_levels = [k for k in new_sh if k not in r] | |
| rows.append(dict(input=c, kind="category", training_file=f"most common {top_r} ({pct(r[top_r])})", this_file=f"most common {top_n} ({pct(new_sh.get(top_n, 0))})" + (f"; new levels: {', '.join(new_levels)}" if new_levels else ""), | |
| missing=int((df[c].astype(str) == "nan").sum()), stability_index=round(idx, 3), reading=reading(idx), watched="yes" if watch else "")) | |
| t = pd.DataFrame(rows) | |
| return t.sort_values(["watched", "stability_index"], ascending=[False, False]).reset_index(drop=True) if len(t) else t | |
| def reading(idx): | |
| return "stable" if idx < 0.10 else ("watch" if idx < 0.25 else "SHIFTED") | |
| def group_table(df, s, flag, y): | |
| """Each group's rate at the cut beside the Week 6 test-set reading and the tripwire. With outcomes: the ruled check (false alarm or miss). Without: the share ranked in.""" | |
| G = groups_of(df); rows = []; check = CFG["group_check"]; line = CFG["group_tripwire"] | |
| if y is not None: | |
| pos, neg = y == 1, y == 0 | |
| port = (((flag == 1) & neg).sum() / max(neg.sum(), 1)) if check == "false_alarm" else (((flag == 0) & pos).sum() / max(pos.sum(), 1)) | |
| for gname in G.columns: | |
| d = CFG["groups"][gname]; order = [b[2] for b in d["bands"]] if "bands" in d else d.get("order", []) | |
| for lvl in sorted(G[gname].unique(), key=lambda s: (order.index(s) if s in order else len(order), str(s))): | |
| m = (G[gname] == lvl).values; n = int(m.sum()) | |
| wk6 = REF["group_rates"].get(gname, {}).get(str(lvl), {}) | |
| covered = n >= CFG["min_group"] # the memo's group floor, applied to the month's file | |
| if y is not None: | |
| rate = (((flag == 1) & neg & m).sum() / max((neg & m).sum(), 1)) if check == "false_alarm" else (((flag == 0) & pos & m).sum() / max((pos & m).sum(), 1)) | |
| base = wk6.get("false_alarm" if check == "false_alarm" else "miss", np.nan) | |
| over = rate > port * (1 + line) | |
| result = (f"not judged (under the group floor of {CFG['min_group']:,})" if not covered else ("TRIP" if over else "inside the line")) | |
| rows.append({"group": gname, "level": lvl, "records this file": n, ("false alarm rate" if check == "false_alarm" else "miss rate"): pct(rate), "whole file": pct(port), "line": pct(port * (1 + line)), "Week 6 test set": pct(base) if base == base else "", "watched": "yes" if str(lvl) in CFG["watch_groups"] else "", "result": result}) | |
| else: | |
| rate = float(flag[m].mean()); base = wk6.get("flagged_share", np.nan) | |
| rows.append({"group": gname, "level": lvl, "records this file": n, "share ranked in": pct(rate), "whole file": pct(SHARE), "Week 6 test set": pct(base) if base == base else "", "watched": "yes" if str(lvl) in CFG["watch_groups"] else "", "result": "no outcomes yet: share ranked in only" if covered else f"not judged (under the group floor of {CFG['min_group']:,})"}) | |
| return pd.DataFrame(rows) | |
| def calibration_table(s, y): | |
| cal = CFG["calibration"]; rows = [] | |
| for lo, hi, name in cal["bands"]: | |
| m = (s >= lo) & (s < hi) | |
| if m.sum() == 0: rows.append(dict(band=name, records=0, predicted="", observed="", gap_share_of_observed="", result="empty band")); continue | |
| mp, ob = float(s[m].mean()), float(y[m].mean()); gap = abs(mp - ob); expected = mp * m.sum() | |
| if expected < CAL_FLOOR: res = f"not read (fewer than {CAL_FLOOR} expected outcomes in the band)" | |
| else: res = "no outcomes in band" if ob == 0 else ("pass" if gap / ob <= cal["tripwire"] else "TRIP") | |
| rows.append(dict(band=name, records=int(m.sum()), predicted=pct(mp, 2), observed=pct(ob, 2), gap_share_of_observed=(pct(gap / ob, 1) if ob else ""), result=res)) | |
| return pd.DataFrame(rows) | |
| def monitor(file): | |
| empty = (None, None, None) | |
| if file is None: return "Upload a monthly file: one row per " + CFG["unit"] + ", the model's input columns, the measurement fields, and the outcome column once it is known.", *empty | |
| try: df = coerce(pd.read_csv(file.name if hasattr(file, "name") else file, low_memory=False, keep_default_na=False, na_values=[])) | |
| except Exception as e: return f"Could not read the file: {e}", *empty | |
| missing = [c for c in INPUTS if c not in df.columns] | |
| if missing: return "The file is missing input columns: " + ", ".join(missing) + ". The job would hold this file out of the cycle (the input contract).", *empty | |
| s = score_frame(df); order = np.argsort(-s, kind="stable"); k = int(round(SHARE * len(df))); flag = np.zeros(len(df), int); flag[order[:k]] = 1 | |
| y = (df[CFG["outcome"]].astype(str) == str(CFG["positive"])).astype(int).values if CFG["outcome"] in df.columns else None | |
| above_cut = float((s >= REF["cut_score"]).mean()) | |
| L = [f"**File:** {len(df):,} {CFG['units']}, {df.shape[1]} columns. Outcome column `{CFG['outcome']}`: " + ("present, so the ruled checks run." if y is not None else "absent, so only the input and score readings run. Upload again when the outcomes are known."), | |
| f"**Scores:** mean {s.mean():.4f} (training-file positive rate {pct(REF['positive_rate'])}). {pct(above_cut)} of this file scores above the Week 6 cut of {REF['cut_score']:.4f}, against {pct(SHARE, 0)} by construction on the test set. The top {int(SHARE*100)} percent of this file is {k:,} {CFG['units']}.", | |
| f"**Inputs:** the stability index compares each input's distribution with the training file's ({REF['rows']:,} rows). Under 0.10 is stable, 0.10 to 0.25 is watch, over 0.25 is shifted. These lines are the usual working lines, provisional until the Week 8 monitoring plan sets the program's own. Watched fields: {CFG['watch_fields_label']}."] | |
| if y is not None: | |
| pos = y == 1 | |
| L.append(f"**At the cut:** {int((flag & pos).sum()):,} of the {int(pos.sum()):,} {CFG['positive_label']} {CFG['units']} are ranked in ({pct((flag & pos).sum() / max(pos.sum(), 1))}); the whole-file {'false alarm' if CFG['group_check'] == 'false_alarm' else 'miss'} rate is the line every group is read against, plus {int(CFG['group_tripwire']*100)} percent ({CFG['memo_ref']}).") | |
| cal = calibration_table(s, y) if (CFG.get("calibration") and y is not None) else None | |
| if cal is not None: | |
| worst = [r for r in cal.result if r == "TRIP"]; read = [r for r in cal.result if r in ("TRIP", "pass")] | |
| L.append(f"**Calibration:** predicted against observed by score band, line {int(CFG['calibration']['tripwire']*100)} percent of observed. Bands that expect fewer than {CAL_FLOOR} outcomes are not read. " + (f"{len(read)} band(s) read, every one inside the line." if not worst else f"{len(worst)} of {len(read)} band(s) read are over the line: the calibration condition (ruling 1) is not met for this file.")) | |
| return "\n\n".join(L), drift_table(df), group_table(df, s, flag, y), cal | |
| # ----------------------------------------------------------------------------- tab 4: the dossier | |
| def dossier_markdown(): | |
| L = [f"## The dossier: {CFG['system']} at {CFG['org']}", "", "The six deliverables of the practicum, in course order. Each opens in the browser.", ""] | |
| for f, label in CFG["docs"]: | |
| p = os.path.join(HERE, "docs", f) | |
| if SPACE_ID: url = f"https://huggingface.co/spaces/{SPACE_ID}/resolve/main/docs/{f}" | |
| else: url = f"docs/{f}" | |
| L.append(f"- [{label}]({url})" + ("" if os.path.exists(p) else " (not uploaded yet)")) | |
| L += ["", f"**Monitoring targets from the {CFG['memo_ref']}:**"] + [f"- {m}" for m in CFG["monitoring_lines"]] | |
| L += ["", f"**Data:** [{CFG['data_file']}]({CFG['data_url']}) in the program's dataset repository (MD5 `{CFG['md5']}`). The data is not copied into this page."] | |
| return "\n".join(L) | |
| # ----------------------------------------------------------------------------- the page | |
| INTRO = f""" | |
| # {CFG['system']} {CFG['version']}: {CFG['org']} | |
| **Governance analyst: [YOUR NAME]** 路 a teaching deployment for the *AI Governance Practicum*, Track 1 | |
| --- | |
| {CFG['system']} estimates, for each {CFG['unit']}, the likelihood of {CFG['positive_label']}, ranks the file, and sends the top {int(SHARE*100)} percent to a person for {CFG['action']}. This page runs the signed version ({CFG['memo_ref']}). It is for **education**, not real decisions. A score says where a {CFG['person']}'s record ranks on one list. It is not a decision about the {CFG['person']}. | |
| """ | |
| with gr.Blocks(title=f"{CFG['org']} {CFG['system']}") as demo: | |
| gr.Markdown(INTRO) | |
| with gr.Accordion("About the signed version: the file, its hash, the fitted steps, the weights", open=False): | |
| gr.Markdown(model_header()) | |
| gr.Markdown("**The fixed weights.** The largest by size, with what each column means. A positive weight pushes the score up.") | |
| gr.Dataframe(weights_table(), interactive=False, wrap=True) | |
| with gr.Tab(f"1 路 Single {CFG['unit']}"): | |
| gr.Markdown(f"Score one {CFG['unit']}, then read the **reasons**. Load a Week 6 case to fill the form, or edit any field and score again.") | |
| with gr.Row(): | |
| preset = gr.Dropdown(list(PRESETS), label="Load an example", value=FIRST) | |
| d0 = PRESETS[FIRST] | |
| widgets = {} | |
| with gr.Accordion("Categorical inputs", open=True): | |
| for i in range(0, len(CAT), 4): | |
| with gr.Row(): | |
| for c in CAT[i:i + 4]: | |
| widgets[c] = gr.Dropdown(choices_for(c), value=d0[c], label=field_label(c), allow_custom_value=True) | |
| with gr.Accordion("Numeric inputs", open=(len(CAT) == 0)): | |
| for i in range(0, len(NUM), 4): | |
| with gr.Row(): | |
| for c in NUM[i:i + 4]: | |
| widgets[c] = gr.Number(value=d0[c], label=c) | |
| ordered = [widgets[c] for c in FORM] | |
| score_btn = gr.Button(f"Score this {CFG['unit']}", variant="primary") | |
| out_md = gr.Markdown(); out_plot = gr.Plot() | |
| preset.change(apply_preset, preset, ordered) | |
| score_btn.click(score_one, ordered, [out_md, out_plot]) | |
| with gr.Tab("2 路 Batch upload"): | |
| gr.Markdown(f"Upload a CSV of {CFG['units']} (the same columns as the working file). {CFG['system']} scores every row, ranks the file, and marks the top {int(SHARE*100)} percent, the operating point the {CFG['memo_ref']} set. Top-N is how many of the ranked list the team can work. `sample_month.csv` in this repository is a file to try.") | |
| with gr.Row(): | |
| up = gr.File(label=f"{CFG['units'].capitalize()} CSV", file_types=[".csv"]) | |
| with gr.Column(): | |
| topn = gr.Slider(5, 100, value=20, step=5, label="Top-N by capacity (how many the team can work)") | |
| batch_btn = gr.Button("Score & rank", variant="primary") | |
| batch_msg = gr.Markdown(); batch_tbl = gr.Dataframe(interactive=False, wrap=True); batch_dl = gr.File(label="Download the full ranked list") | |
| batch_btn.click(score_batch, [up, topn], [batch_tbl, batch_msg, batch_dl]) | |
| with gr.Tab("3 路 Monitoring"): | |
| gr.Markdown(f"**One month at a time.** Upload the month's file. The page reads it against the training file and the tripwires the {CFG['memo_ref']} set: the inputs' distributions, the scores, the groups at the cut, " + ("and the calibration table." if CFG.get("calibration") else "and the ruled group check.") + f" `sample_month.csv` in this repository is a stand-in drawn from the Week 6 test set, so the page can be tried before a monthly file exists.") | |
| mf = gr.File(label="Monthly file (CSV)", file_types=[".csv"]) | |
| mon_btn = gr.Button("Read this month", variant="primary") | |
| m_msg = gr.Markdown() | |
| m_drift = gr.Dataframe(interactive=False, label="Inputs against the training file", wrap=True) | |
| m_groups = gr.Dataframe(interactive=False, label="Groups at the cut, against the whole file and the line", wrap=True) | |
| m_cal = gr.Dataframe(interactive=False, label="Calibration by score band", visible=bool(CFG.get("calibration"))) | |
| mon_btn.click(monitor, mf, [m_msg, m_drift, m_groups, m_cal]) | |
| with gr.Tab("4 路 The dossier"): | |
| gr.Markdown(dossier_markdown()) | |
| gr.Markdown(f"---\n*{CFG['system']} is a fictional model built on a public dataset for the AI Governance Practicum (AI Governance Collective). The organization, the people, and the program are fictional. Data: [{CFG['data_file']}]({CFG['data_url']}).*") | |
| if __name__ == "__main__": | |
| demo.launch() | |