Update app.py

app.py

@@ -1,72 +1,118 @@
-import os, base64
 import gradio as gr
 import pandas as pd
 from pathlib import Path
 from typing import Dict, List, Tuple
+import os
+import base64
 
-# --- load & encode logo so it can't be clicked/downloaded ---
-LOGO_PATH = Path("rowsquared-logo-large.png")
-with open(LOGO_PATH, "rb") as f:
-    logo_b64 = base64.b64encode(f.read()).decode()
-
-# --- load your ISCO hierarchy and build nested dict ---
 PROCESSED_DATA_DIR = Path(".")
+# Embed logo as a base64 data URI to avoid Gradio toolbar interactions
+logo_path = "rowsquared-logo-large.png"
+with open(logo_path, "rb") as f:
+    logo_b64 = base64.b64encode(f.read()).decode("utf-8")
+
+
+
+# ----------------------------
+# Data loading & preprocessing
+# ----------------------------
 df_isco = (
     pd.read_excel(
         PROCESSED_DATA_DIR / "isco_imperfect.xlsx",
         converters={"major": str, "sub_major": str, "minor": str, "unit": str},
-    )[["major_label","sub_major_label","minor_label","unit_label"]]
-    .dropna().drop_duplicates().reset_index(drop=True)
+    )[["major_label", "sub_major_label", "minor_label", "unit_label"]]
+    .dropna()
+    .drop_duplicates()
+    .reset_index(drop=True)
 )
-hierarchy: Dict[str,Dict[str,Dict[str,List[str]]]] = {}
+
+# Build nested hierarchy dict: {major: {sub: {minor: [units]}}}
+hierarchy: Dict[str, Dict[str, Dict[str, List[str]]]] = {}
 for _, r in df_isco.iterrows():
-    hierarchy.setdefault(r.major_label, {})\
-             .setdefault(r.sub_major_label, {})\
-             .setdefault(r.minor_label, [])\
+    hierarchy.setdefault(r.major_label, {}) \
+             .setdefault(r.sub_major_label, {}) \
+             .setdefault(r.minor_label, []) \
              .append(r.unit_label)
+
+# Ensure uniqueness & sorting at leaf lists
 for maj in hierarchy:
     for sub in hierarchy[maj]:
         for mn in hierarchy[maj][sub]:
-            hierarchy[maj][sub][mn] = sorted(dict.fromkeys(hierarchy[maj][sub][mn]))
-
-def majors():     return sorted(hierarchy.keys())
-def submajors(m): return sorted(hierarchy.get(m,{}).keys())
-def minors(m,s):  return sorted(hierarchy.get(m,{}).get(s,{}).keys())
-def units(m,s,n): return hierarchy.get(m,{}).get(s,{}).get(n,[])
-
-# --- load your records ---
-records = pd.read_excel(PROCESSED_DATA_DIR/"isco_predictions.xlsx").copy()
-for c in ["major_label","sub_major_label","minor_label","unit_label"]:
-    if c not in records: records[c] = ""
-if "annotated" not in records: records["annotated"] = False
-records.reset_index(drop=True,inplace=True)
-
-# --- clamp + save + load + handlers ---
-def clamp_path(maj,sub,mn,un):
-    maj_c = majors()
-    maj   = maj if maj in maj_c else (maj_c[0] if maj_c else "")
-    sub_c = submajors(maj)
-    sub   = sub if sub in sub_c else (sub_c[0] if sub_c else "")
-    mn_c  = minors(maj,sub)
-    mn    = mn if mn in mn_c else (mn_c[0] if mn_c else "")
-    un_c  = units(maj,sub,mn)
-    un    = un if un in un_c else (un_c[0] if un_c else "")
-    return maj,sub,mn,un, maj_c,sub_c,mn_c,un_c
-
-def save_record(i,maj,sub,mn,un):
-    records.loc[i,["major_label","sub_major_label","minor_label","unit_label"]] = [maj,sub,mn,un]
-    records.loc[i,"annotated"] = True
-
-def status_text(i):
-    return f"**Status**: {'✅ Annotated' if records.loc[i,'annotated'] else '❌ Not Annotated'}"
-
-def load_record(i):
-    r = records.loc[i]
-    maj,sub,mn,un,maj_c,sub_c,mn_c,un_c = clamp_path(r.major_label, r.sub_major_label, r.minor_label, r.unit_label)
-    save_record(i,maj,sub,mn,un)
-    md = f"## Occupation: {r.occupation_title_main}\n## Industry: {r.industry_title_main}"
+            hierarchy[maj][sub][mn] = sorted(list(dict.fromkeys(hierarchy[maj][sub][mn])))
+
+# Fast helpers for children
+def majors() -> List[str]:
+    return sorted(hierarchy.keys())
+
+def submajors(maj: str) -> List[str]:
+    return sorted(hierarchy.get(maj, {}).keys())
+
+def minors(maj: str, sub: str) -> List[str]:
+    return sorted(hierarchy.get(maj, {}).get(sub, {}).keys())
+
+def units(maj: str, sub: str, mn: str) -> List[str]:
+    return hierarchy.get(maj, {}).get(sub, {}).get(mn, [])
+
+# ----------------------------
+# Records to annotate
+# ----------------------------
+records = pd.read_excel(PROCESSED_DATA_DIR / "isco_predictions.xlsx").copy()
+for col in ["major_label", "sub_major_label", "minor_label", "unit_label"]:
+    if col not in records:
+        records[col] = ""
+
+if "annotated" not in records:
+    records["annotated"] = False
+
+# ensure not views
+for col in ["major_label", "sub_major_label", "minor_label", "unit_label", "annotated"]:
+    records[col] = records[col].copy()
+
+records.reset_index(drop=True, inplace=True)
+
+# -----------------------------------
+# Core logic: clamp & state management
+# -----------------------------------
+def clamp_path(maj: str, sub: str, mn: str, un: str
+               ) -> Tuple[str, str, str, str, List[str], List[str], List[str], List[str]]:
+    """Return a valid (maj, sub, mn, un) tuple + their choices lists.
+       Only replace a level if it's invalid for the hierarchy."""
+    maj_choices = majors()
+    if maj not in maj_choices:
+        maj = maj_choices[0] if maj_choices else ""
+
+    sub_choices = submajors(maj) if maj else []
+    if sub not in sub_choices:
+        sub = sub_choices[0] if sub_choices else ""
+
+    mn_choices = minors(maj, sub) if sub else []
+    if mn not in mn_choices:
+        mn = mn_choices[0] if mn_choices else ""
+
+    un_choices = units(maj, sub, mn) if mn else []
+    if un not in un_choices:
+        un = un_choices[0] if un_choices else ""
+
+    return maj, sub, mn, un, maj_choices, sub_choices, mn_choices, un_choices
+
+def save_record(i: int, maj: str, sub: str, mn: str, un: str) -> None:
+    records.loc[i, ["major_label", "sub_major_label", "minor_label", "unit_label"]] = [maj, sub, mn, un]
+    records.loc[i, "annotated"] = True
+
+def status_text(i: int) -> str:
+    return f"**Status**: {'✅ Annotated' if records.loc[i, 'annotated'] else '❌ Not Annotated'}"
+
+def load_record(i: int):
+    rec = records.loc[i]
+    maj, sub, mn, un, maj_c, sub_c, mn_c, un_c = clamp_path(
+        rec["major_label"], rec["sub_major_label"], rec["minor_label"], rec["unit_label"]
+    )
+    # Persist clamped values back (only if changed)
+    save_record(i, maj, sub, mn, un)
+
+    record_md = f"## Occupation: {rec['occupation_title_main']}\n## Industry: {rec['industry_title_main']}"
     return (
-        md,
+        record_md,
         status_text(i),
         gr.update(choices=maj_c, value=maj),
         gr.update(choices=sub_c, value=sub),
@@ -74,153 +120,176 @@ def load_record(i):
         gr.update(choices=un_c, value=un),
     )
 
-def on_major_change(new_maj,i):
-    sub_c = submajors(new_maj); sub = sub_c[0] if sub_c else ""
-    mn_c  = minors(new_maj,sub); mn  = mn_c[0]  if mn_c  else ""
-    un_c  = units(new_maj,sub,mn); un  = un_c[0]  if un_c  else ""
-    save_record(i,new_maj,sub,mn,un)
+# ---------------------
+# Event handler helpers
+# ---------------------
+def on_major_change(new_major: str, i: int):
+    sub_c = submajors(new_major)
+    sub = sub_c[0] if sub_c else ""
+    mn_c = minors(new_major, sub) if sub else []
+    mn = mn_c[0] if mn_c else ""
+    un_c = units(new_major, sub, mn) if mn else []
+    un = un_c[0] if un_c else ""
+    save_record(i, new_major, sub, mn, un)
     return (
-        gr.update(choices=majors(),  value=new_maj),
-        gr.update(choices=sub_c,     value=sub),
-        gr.update(choices=mn_c,      value=mn),
-        gr.update(choices=un_c,      value=un),
+        gr.update(choices=majors(), value=new_major),
+        gr.update(choices=sub_c, value=sub),
+        gr.update(choices=mn_c, value=mn),
+        gr.update(choices=un_c, value=un),
         status_text(i),
     )
 
-def on_sub_change(new_sub,i,maj):
-    mn_c = minors(maj,new_sub); mn = mn_c[0] if mn_c else ""
-    un_c = units(maj,new_sub,mn); un = un_c[0] if un_c else ""
-    records.loc[i,["sub_major_label","minor_label","unit_label"]] = [new_sub,mn,un]
-    records.loc[i,"annotated"]=True
+def on_sub_change(new_sub: str, i: int, major: str):
+    mn_c = minors(major, new_sub)
+    mn = mn_c[0] if mn_c else ""
+    un_c = units(major, new_sub, mn) if mn else []
+    un = un_c[0] if un_c else ""
+    records.loc[i, ["sub_major_label", "minor_label", "unit_label"]] = [new_sub, mn, un]
+    records.loc[i, "annotated"] = True
     return (
-        gr.update(choices=submajors(maj),value=new_sub),
-        gr.update(choices=mn_c,          value=mn),
-        gr.update(choices=un_c,          value=un),
+        gr.update(choices=submajors(major), value=new_sub),
+        gr.update(choices=mn_c, value=mn),
+        gr.update(choices=un_c, value=un),
         status_text(i),
     )
 
-def on_minor_change(new_mn,i,maj,sub):
-    un_c = units(maj,sub,new_mn); un = un_c[0] if un_c else ""
-    records.loc[i,["minor_label","unit_label"]]=[new_mn,un]
-    records.loc[i,"annotated"]=True
+def on_minor_change(new_minor: str, i: int, major: str, sub: str):
+    un_c = units(major, sub, new_minor)
+    un = un_c[0] if un_c else ""
+    records.loc[i, ["minor_label", "unit_label"]] = [new_minor, un]
+    records.loc[i, "annotated"] = True
     return (
-        gr.update(choices=minors(maj,sub),value=new_mn),
-        gr.update(choices=un_c,           value=un),
+        gr.update(choices=minors(major, sub), value=new_minor),
+        gr.update(choices=un_c, value=un),
         status_text(i),
     )
 
-def on_unit_change(new_un,i,maj,sub,mn):
-    un_c=units(maj,sub,mn)
-    new_un = new_un if new_un in un_c else (un_c[0] if un_c else "")
-    records.loc[i,"unit_label"]=new_un
-    records.loc[i,"annotated"]=True
-    return gr.update(choices=un_c,value=new_un), status_text(i)
-
-def go_next(i): return (i+1)%len(records)
-def go_prev(i): return (i-1)%len(records)
-
-def save_and_jump(i,direction):
-    r=records.loc[i]
-    maj,sub,mn,un,*_ = clamp_path(r.major_label,r.sub_major_label,r.minor_label,r.unit_label)
-    save_record(i,maj,sub,mn,un)
-    j = go_next(i) if direction=="next" else go_prev(i)
-    return (j,)+load_record(j)
-
-def download_csv():
-    out = PROCESSED_DATA_DIR/"annotated_output.csv"
-    records.to_csv(out,index=False)
-    return str(out)
-
-# --- build with login gate ---
-def build_gradio_app():
-    USER = os.getenv("APP_USER","")
-    PWD  = os.getenv("APP_PASS","")
+def on_unit_change(new_unit: str, i: int, major: str, sub: str, mn: str):
+    un_c = units(major, sub, mn)
+    if new_unit not in un_c:
+        new_unit = un_c[0] if un_c else ""
+    records.loc[i, "unit_label"] = new_unit
+    records.loc[i, "annotated"] = True
+    return gr.update(choices=un_c, value=new_unit), status_text(i)
+
+def go_next(i: int) -> int:
+    return (i + 1) % len(records)
+
+def go_prev(i: int) -> int:
+    return (i - 1) % len(records)
+
+# ---- NAVIGATION: save + move + reload in ONE callback ----
 
+def save_and_jump(i: int, direction: str):
+    # Final safety net: clamp and persist whatever is currently stored
+    rec = records.loc[i]
+    maj, sub, mn, un, *_ = clamp_path(
+        rec["major_label"], rec["sub_major_label"], rec["minor_label"], rec["unit_label"]
+    )
+    save_record(i, maj, sub, mn, un)
+    new_i = go_next(i) if direction == "next" else go_prev(i)
+    return (new_i,) + load_record(new_i)
+
+def download_annotations() -> str:
+    path = PROCESSED_DATA_DIR / "annotated_output.csv"
+    records.to_csv(path, index=False)
+    return str(path)
+
+# --------------
+# Build the UI
+# --------------
+def build_gradio_app():
     with gr.Blocks() as demo:
+        with gr.Row():
+            with gr.Column(scale=1):
+                # Static logo, non-interactive
+                gr.HTML(
+                    f'<img src="data:image/png;base64,{logo_b64}" width="200" style="pointer-events:none; user-select:none; display:block;" />'
+                )
+        with gr.Row():
+            gr.Markdown("# ISCO Annotation", elem_id="isco-title")
         gr.HTML("""
         <style>
-          footer, .share-link, .api-link {display:none!important}
-          #title{text-align:center;margin:0.5em}
-        </style>""")
-
-        # === capture the two panels ===
-        with gr.Column(elem_id="login_panel") as login_panel:
-            gr.HTML(f'<img src="data:image/png;base64,{logo_b64}" width="160" '
-                    'style="pointer-events:none;user-select:none;display:block;margin:auto;"/>')
-            user_in   = gr.Textbox(label="Username")
-            pass_in   = gr.Textbox(label="Password", type="password")
-            login_btn = gr.Button("🔒 Log in")
-            login_msg = gr.Markdown("", visible=False)
-
-        with gr.Column(elem_id="app_panel", visible=False) as app_panel:
-            gr.HTML(f'<img src="data:image/png;base64,{logo_b64}" width="160" '
-                    'style="pointer-events:none;user-select:none;display:block;margin:auto;"/>')
-            gr.Markdown("# ISCO Annotation", elem_id="title")
-            idx_state = gr.State(0)
+            #isco-title {
+                text-align: center;
+                width: 100%;
+                margin: 0.5em 0;
+            }
+            footer { display: none !important; }
+            .gradio-container .api-link, .gradio-container .share-link { display: none !important; }
+        </style>
+        """)
+
+        idx_state = gr.State(0)
+
+        with gr.Group():
             record_md = gr.Markdown()
             status_md = gr.Markdown()
-            prev_btn  = gr.Button("⬅ Previous")
-            next_btn  = gr.Button("✅ Next")
-            major_radio = gr.Radio("Level 1 Major",   choices=[], interactive=True)
-            sub_radio   = gr.Radio("Level 2 Sub-major",choices=[], interactive=True)
-            minor_radio = gr.Radio("Level 3 Minor",   choices=[], interactive=True)
-            unit_radio  = gr.Radio("Level 4 Unit",    choices=[], interactive=True)
-            download_btn  = gr.Button("📥 Download CSV")
-            download_file = gr.File(visible=False)
-
-        # --- login handler ---
-        def check_creds(u,p):
-            if u==USER and p==PWD:
-                return (
-                    gr.update(visible=False),  # hide login
-                    gr.update(visible=True),   # show app
-                    gr.update(visible=False),  # hide error
-                    ""                         # clear user_in
-                )
-            else:
-                return (None, None,
-                        gr.update(visible=True,value="❌ Bad credentials"),
-                        "")
-
-        login_btn.click(
-            check_creds,
-            inputs=[user_in,pass_in],
-            outputs=[login_panel,app_panel,login_msg,user_in]
-        )
 
-        # --- main app wiring ---
-        demo.load(lambda: (0,)+load_record(0),
-                  outputs=[idx_state,record_md,status_md,
-                           major_radio,sub_radio,minor_radio,unit_radio])
+        with gr.Row():
+            prev_btn = gr.Button("⬅ Previous")
+            next_btn = gr.Button("✅ Next")
+
+        with gr.Row():
+            with gr.Column():
+                major_radio = gr.Radio(label="Level 1: Major", choices=[], interactive=True)
+            with gr.Column():
+                sub_radio = gr.Radio(label="Level 2: Sub-major", choices=[], interactive=True)
+            with gr.Column():
+                minor_radio = gr.Radio(label="Level 3: Minor", choices=[], interactive=True)
+            with gr.Column():
+                unit_radio = gr.Radio(label="Level 4: Unit", choices=[], interactive=True)
+
+        download_btn = gr.Button("📥 Download Annotations")
+        download_file = gr.File(label="Annotated CSV", visible=False)
 
-        next_btn.click(lambda i: save_and_jump(i,"next"),
+        # Initial load
+        demo.load(
+            lambda: (0,) + load_record(0),
+            outputs=[idx_state, record_md, status_md, major_radio, sub_radio, minor_radio, unit_radio],
+        )
+
+        next_btn.click(lambda i: save_and_jump(i, "next"),
                        inputs=[idx_state],
-                       outputs=[idx_state,record_md,status_md,
-                                major_radio,sub_radio,minor_radio,unit_radio])
-        prev_btn.click(lambda i: save_and_jump(i,"prev"),
+                       outputs=[idx_state, record_md, status_md, major_radio, sub_radio, minor_radio, unit_radio])
+
+        prev_btn.click(lambda i: save_and_jump(i, "prev"),
                        inputs=[idx_state],
-                       outputs=[idx_state,record_md,status_md,
-                                major_radio,sub_radio,minor_radio,unit_radio])
-
-        major_radio.change(on_major_change,
-                           inputs=[major_radio,idx_state],
-                           outputs=[major_radio,sub_radio,minor_radio,unit_radio,status_md])
-        sub_radio.change(on_sub_change,
-                         inputs=[sub_radio,idx_state,major_radio],
-                         outputs=[sub_radio,minor_radio,unit_radio,status_md])
-        minor_radio.change(on_minor_change,
-                           inputs=[minor_radio,idx_state,major_radio,sub_radio],
-                           outputs=[minor_radio,unit_radio,status_md])
-        unit_radio.change(on_unit_change,
-                          inputs=[unit_radio,idx_state,major_radio,sub_radio,minor_radio],
-                          outputs=[unit_radio,status_md])
-
-        download_btn.click(download_csv, outputs=[download_file]).then(
+                       outputs=[idx_state, record_md, status_md, major_radio, sub_radio, minor_radio, unit_radio])
+
+        # Change handlers (also update status)
+        major_radio.change(
+            on_major_change,
+            inputs=[major_radio, idx_state],
+            outputs=[major_radio, sub_radio, minor_radio, unit_radio, status_md],
+        )
+
+        sub_radio.change(
+            on_sub_change,
+            inputs=[sub_radio, idx_state, major_radio],
+            outputs=[sub_radio, minor_radio, unit_radio, status_md],
+        )
+
+        minor_radio.change(
+            on_minor_change,
+            inputs=[minor_radio, idx_state, major_radio, sub_radio],
+            outputs=[minor_radio, unit_radio, status_md],
+        )
+
+        unit_radio.change(
+            on_unit_change,
+            inputs=[unit_radio, idx_state, major_radio, sub_radio, minor_radio],
+            outputs=[unit_radio, status_md],
+        )
+
+        # Download
+        download_btn.click(download_annotations, outputs=[download_file]).then(
             lambda: gr.update(visible=True), None, [download_file]
         )
 
     return demo
 
+
 if __name__=="__main__":
     demo = build_gradio_app()
     demo.queue().launch(
@@ -229,4 +298,5 @@ if __name__=="__main__":
         auth=(os.getenv("APP_USER",""), os.getenv("APP_PASS","")),
         server_name="0.0.0.0",         # optional, but explicit
         server_port=int(os.getenv("PORT", 7860)),
-    )
+    )
+