Spaces:
Running
Running
| """The public page of a governed system: the signed model, its monitoring page, and the links to its repositories. | |
| 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"] | |
| 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 | |
| _ENC = MODEL.named_steps[MODEL.steps[0][0]] | |
| _CAT_KIND = {} | |
| for _name, _trans, _cols in _ENC.transformers_: | |
| if _name == "cat": | |
| for _c, _cats in zip(_cols, _trans.categories_): _CAT_KIND[_c] = "num" if np.issubdtype(np.asarray(_cats).dtype, np.number) else "str" | |
| def coerce(df): | |
| """Read a file the way the working file was read when the model was fitted: numeric inputs as numbers, categorical inputs as text, | |
| except the coded categoricals the encoder learned as numbers (a code column read from a CSV as text would otherwise be unknown to it).""" | |
| 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] = pd.to_numeric(df[c], errors="coerce") if _CAT_KIND.get(c) == "num" else 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 3: monitoring (Evidently) | |
| import html as _html | |
| REFERENCE = pd.read_csv(os.path.join(HERE, f"{CFG['short']}_reference.csv"), low_memory=False, keep_default_na=False, na_values=[]) | |
| MONTHS = {f"Month {i}": f"{CFG['short']}_month_{i}.csv" for i in (1, 2, 3, 4, 5) if os.path.exists(os.path.join(HERE, f"{CFG['short']}_month_{i}.csv"))} | |
| MIN_GROUP = 30 # a group's rate is shown but not judged below this many of the people it is a rate of | |
| SCORE_COL, FLAG_COL = "predicted_score", "flagged" | |
| def monitoring_frame(df): | |
| """The columns Evidently reads: every field in the file except the identifier, plus the model's score and the flag. Numbers as numbers, text as text.""" | |
| df = coerce(df); out = pd.DataFrame(index=df.index) | |
| skip = {CFG.get("id_col")} | {SCORE_COL, FLAG_COL} | |
| num, cat = [], [] | |
| for c in df.columns: | |
| if c in skip or c is None: continue | |
| if c in CFG["numeric"] or (c not in CFG["categorical"] and pd.api.types.is_numeric_dtype(df[c]) and c != CFG["outcome"]): | |
| out[c] = pd.to_numeric(df[c], errors="coerce").astype("float64").to_numpy(); num.append(c) | |
| else: | |
| out[c] = df[c].astype(str).to_numpy().astype(object); cat.append(c) | |
| s = score_frame(df); out[SCORE_COL] = s.astype("float64"); num.append(SCORE_COL) | |
| out[FLAG_COL] = np.where(s >= REF["cut_score"], "flagged", "not flagged").astype(object); cat.append(FLAG_COL) | |
| return out, num, cat | |
| def rates(frame, s): | |
| """Caught rate (the share of {positives} the model flagged) and false alarm rate (the share of the others it flagged), for the file and by group.""" | |
| y = (frame[CFG["outcome"]].astype(str) == str(CFG["positive"])).to_numpy(); flag = s >= REF["cut_score"] | |
| whole = dict(caught=float(flag[y].mean()) if y.any() else None, false_alarm=float(flag[~y].mean()) if (~y).any() else None, positives=int(y.sum())) | |
| G = groups_of(frame); by = {} | |
| for dim in G.columns: | |
| rows = [] | |
| for name in sorted(G[dim].astype(str).unique(), key=str): | |
| m = (G[dim].astype(str) == name).to_numpy(); yp, yn = y & m, (~y) & m | |
| rows.append(dict(group=name, n=int(m.sum()), positives=int(yp.sum()), negatives=int(yn.sum()), | |
| caught=float(flag[yp].mean()) if yp.sum() else None, false_alarm=float(flag[yn].mean()) if yn.sum() else None)) | |
| by[dim] = rows | |
| return whole, by | |
| def gap(rows, key, denom): | |
| use = [r[key] for r in rows if r[key] is not None and r[denom] >= MIN_GROUP] | |
| return (max(use) - min(use)) if len(use) >= 2 else None | |
| def fmt(x, d=1): return "" if x is None else f"{x * 100:.{d}f}" | |
| def group_table(by_ref, by_cur): | |
| out = [] | |
| for dim, rows in by_cur.items(): | |
| ref = {r["group"]: r for r in by_ref.get(dim, [])} | |
| for r in rows: | |
| rr = ref.get(r["group"], {}) | |
| out.append({"group": f"{dim}: {r['group']}", f"{CFG['units']} this month": r["n"], f"of which {CFG['positive_label']}": r["positives"], | |
| "caught rate, reference": fmt(rr.get("caught")), "caught rate, this month": fmt(r["caught"]) + ("" if r["positives"] >= MIN_GROUP else " (too few to judge)"), | |
| "false alarm rate, reference": fmt(rr.get("false_alarm")), "false alarm rate, this month": fmt(r["false_alarm"]) + ("" if r["negatives"] >= MIN_GROUP else " (too few to judge)")}) | |
| return pd.DataFrame(out) | |
| def monitor(choice, file): | |
| """Run Evidently's data-drift report for one month against the reference file, and summarize the signals above it: | |
| the inputs, the scores, the outcomes, the caught rate, and the groups the Week 6 memo watches.""" | |
| if file is not None: | |
| try: cur = pd.read_csv(file.name if hasattr(file, "name") else file, low_memory=False, keep_default_na=False, na_values=[]); label = "your file" | |
| except Exception as e: return f"Could not read the file: {e}", None, "", None | |
| elif choice in MONTHS: | |
| cur = pd.read_csv(os.path.join(HERE, MONTHS[choice]), low_memory=False, keep_default_na=False, na_values=[]); label = choice | |
| else: | |
| return "Pick a month, or upload a file.", None, "", None | |
| missing = [c for c in INPUTS if c not in cur.columns] | |
| if missing: return "The file is missing input columns: " + ", ".join(missing) + ". Nothing was read.", None, "", None | |
| from evidently import Dataset, DataDefinition, Report | |
| from evidently.presets import DataDriftPreset | |
| ref_f, num, cat = monitoring_frame(REFERENCE) | |
| cur_f, num_c, cat_c = monitoring_frame(cur) | |
| cols = [c for c in ref_f.columns if c in cur_f.columns] | |
| num = [c for c in num if c in cols]; cat = [c for c in cat if c in cols] | |
| dd = DataDefinition(numerical_columns=num, categorical_columns=cat) | |
| # two-sided proportion test for the yes/no columns (the flag and the outcome), where a distance score reads a real shift as small | |
| binary = [c for c in (FLAG_COL, CFG["outcome"]) if c in cat] | |
| preset = DataDriftPreset(per_column_method={c: "z" for c in binary}, per_column_threshold={c: 0.05 for c in binary}) | |
| snap = Report([preset]).run(Dataset.from_pandas(cur_f[cols], data_definition=dd), Dataset.from_pandas(ref_f[cols], data_definition=dd)) | |
| ms = snap.dict()["metrics"] | |
| def is_drift(m): | |
| c = m.get("config", {}); col = c.get("column") | |
| if not col: return False | |
| if col in binary: return float(m["value"]) < c.get("threshold", 0.05) # a p-value | |
| return float(m["value"]) > c.get("threshold", 0.1) # a distance | |
| drifted = [m["config"]["column"] for m in ms[1:] if is_drift(m)] | |
| inputs_drifted = [c for c in drifted if c in INPUTS]; other_drifted = [c for c in drifted if c not in INPUTS and c not in (SCORE_COL, FLAG_COL, CFG["outcome"])] | |
| s_ref, s_cur = ref_f[SCORE_COL].values, cur_f[SCORE_COL].values | |
| f_ref, f_cur = float((s_ref >= REF["cut_score"]).mean()), float((s_cur >= REF["cut_score"]).mean()) | |
| L = [f"**{label}:** {len(cur):,} {CFG['units']} read against the reference file ({len(REFERENCE):,} {CFG['units']}, {CFG.get('reference_label', 'the Week 6 test set')}). Every column in the file is compared, plus the model's score and whether the {CFG['unit']} is flagged at the Week 6 cut.", ""] | |
| L.append(f"**The inputs.** {len(inputs_drifted)} of {len(INPUTS)} input columns drifted" + (": " + ", ".join(inputs_drifted) + "." if inputs_drifted else ".") + (f" Other fields in the file that drifted (not model inputs): {', '.join(other_drifted)}." if other_drifted else "")) | |
| L.append(f"**The scores.** Mean score {s_ref.mean():.4f} in the reference, {s_cur.mean():.4f} this month. Share flagged at the Week 6 cut: {pct(f_ref)} in the reference, {pct(f_cur)} this month." + (" The score column drifted." if SCORE_COL in drifted else " The score column did not drift.")) | |
| table = None | |
| if CFG["outcome"] in cur.columns: | |
| o_ref = float((REFERENCE[CFG["outcome"]].astype(str) == str(CFG["positive"])).mean()); o_cur = float((cur[CFG["outcome"]].astype(str) == str(CFG["positive"])).mean()) | |
| L.append(f"**The outcomes.** {CFG['positive_label'].capitalize()}: {pct(o_ref)} of the reference, {pct(o_cur)} this month." + (" The outcome column drifted." if CFG["outcome"] in drifted else " The outcome column did not drift.")) | |
| w_ref, by_ref = rates(REFERENCE, s_ref); w_cur, by_cur = rates(cur, s_cur) | |
| L.append(f"**The caught rate.** Of the {CFG['units']} whose outcome is {CFG['positive_label']}, the share the model flagged at the Week 6 cut: {pct(w_ref['caught'])} in the reference, {pct(w_cur['caught'])} this month. " | |
| f"False alarm rate (the share of the others it flagged): {pct(w_ref['false_alarm'])} in the reference, {pct(w_cur['false_alarm'])} this month.") | |
| parts = [] | |
| for dim in by_cur: | |
| g_ref, g_cur = gap(by_ref[dim], "caught", "positives"), gap(by_cur[dim], "caught", "positives") | |
| f_ref, f_cur = gap(by_ref[dim], "false_alarm", "negatives"), gap(by_cur[dim], "false_alarm", "negatives") | |
| bits = [] | |
| if g_ref is not None and g_cur is not None: bits.append(f"caught-rate gap {fmt(g_ref)} to {fmt(g_cur)} points") | |
| if f_ref is not None and f_cur is not None: bits.append(f"false alarm gap {fmt(f_ref)} to {fmt(f_cur)} points") | |
| if bits: parts.append(f"{dim}: " + ", ".join(bits)) | |
| L.append("**The groups.** The widest gap between any two groups (groups with at least 30 of the people the rate is about), reference to this month. " + "; ".join(parts) + ". The table below has every group.") | |
| table = group_table(by_ref, by_cur) | |
| else: | |
| L.append("**The outcomes.** This file carries no outcome column, so the outcome rate, the caught rate, and the group gaps cannot be read yet.") | |
| L.append("\nThe full report is below (Evidently, data drift preset: Wasserstein distance for numbers, Jensen-Shannon distance for categories, drift called at 0.1, and a proportion test for the flag and the outcome). Download it as a page to keep with your monitoring read.") | |
| report_html = snap.get_html_str(as_iframe=False) | |
| path = os.path.join("/tmp", f"{CFG['short']}_{label.replace(' ', '_')}_evidently_report.html"); open(path, "w", encoding="utf-8").write(report_html) | |
| frame = f'<iframe srcdoc="{_html.escape(report_html, quote=True)}" style="width:100%;height:1400px;border:1px solid #ddd;border-radius:8px;background:white"></iframe>' | |
| return "\n".join(L), table, frame, path | |
| # ----------------------------------------------------------------------------- tab 4: the repositories | |
| def repositories_markdown(): | |
| short = CFG["short"] | |
| L = [f"## The repositories: {CFG['system']} at {CFG['org']}", "", | |
| "A governed system leaves its record in several places, and this page is only one of them. Each link opens on Hugging Face.", "", | |
| f"- **Model repository**, [drdavidprivacy/{short}-model](https://huggingface.co/drdavidprivacy/{short}-model): the model card, the signed file and its SHA-256, and the fixed weights as a table. The card on the page is the AI Governance Lead's exemplar.", | |
| f"- **Dataset repository**, [drdavidprivacy/{short}-data](https://huggingface.co/datasets/drdavidprivacy/{short}-data): the datasheet and the remediation log for the working file. The data itself is linked, not copied.", | |
| f"- **Portfolio page**, [drdavidprivacy/{CFG['portfolio']}](https://huggingface.co/spaces/drdavidprivacy/{CFG['portfolio']}): the six deliverables of the practicum as PDFs, in course order, on a page with no code behind it. Each learner builds one of these under their own name.", | |
| f"- **Collection**, [{CFG['collection_name']}]({CFG['collection_url']}): all of the above in one place.", ""] | |
| 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']} | |
| **Kept by: {CFG['lead']}** 路 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. `{CFG['short']}_month_1.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.** Pick one of the five monthly files, or upload a file of your own, and read it against the reference file ({CFG.get('reference_label', 'the Week 6 test set')}). The page scores every row with the signed version, runs Evidently's data-drift report (every column in the file, the model's score, and the flag at the Week 6 cut), and, where the file carries the outcome, reads the caught rate and the groups the Week 6 memo watches. The five files and the reference are in this repository's Files tab (`{CFG['short']}_reference.csv`, `{CFG['short']}_month_1.csv` to `_5.csv`).") | |
| if CFG.get("reference_note"): gr.Markdown(CFG["reference_note"]) | |
| with gr.Row(): | |
| m_choice = gr.Dropdown(choices=list(MONTHS), value=(list(MONTHS)[0] if MONTHS else None), label="Monthly file") | |
| mf = gr.File(label="Or upload a file of your own (CSV)", file_types=[".csv"]) | |
| mon_btn = gr.Button("Run the monitoring report", variant="primary") | |
| m_msg = gr.Markdown() | |
| m_tbl = gr.Dataframe(label="The groups: caught rate and false alarm rate, reference and this month (percent)", interactive=False, wrap=True) | |
| m_dl = gr.File(label="Download the report (HTML)") | |
| m_html = gr.HTML() | |
| mon_btn.click(monitor, [m_choice, mf], [m_msg, m_tbl, m_html, m_dl]) | |
| with gr.Tab("4 路 The repositories"): | |
| gr.Markdown(repositories_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() | |