Update app.py

app.py

@@ -1,120 +1,76 @@
+# app.py
+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("data/asset/rowsquared-logo-large.png")
+with open(LOGO_PATH, "rb") as f:
+    logo_b64 = base64.b64encode(f.read()).decode()
 
-PROCESSED_DATA_DIR = Path(".")
-DATA_DIR = Path("./data")
-# 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
-# ----------------------------
+# --- load your ISCO hierarchy and build nested dict (as before) ---
+PROCESSED_DATA_DIR = Path("./data/processed")
 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)
 )
-
-# Build nested hierarchy dict: {major: {sub: {minor: [units]}}}
-hierarchy: Dict[str, Dict[str, Dict[str, List[str]]]] = {}
+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
+# dedupe & sort units
 for maj in hierarchy:
     for sub in hierarchy[maj]:
         for mn in hierarchy[maj][sub]:
-            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"]
+            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 (same as before) ---
+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 (copy in your code) ---
+def clamp_path(maj,sub,mn,un):
+    maj_c = majors()
+    if maj not in maj_c: maj = maj_c[0] if maj_c else ""
+    sub_c = submajors(maj)
+    if sub not in sub_c: sub = sub_c[0] if sub_c else ""
+    mn_c  = minors(maj,sub)
+    if mn  not in mn_c:  mn  = mn_c[0]  if mn_c  else ""
+    un_c  = units(maj,sub,mn)
+    if un  not in un_c:  un  = 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
     )
-    # 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']}"
+    save_record(i, maj,sub,mn,un)
+    md = f"## Occupation: {r.occupation_title_main}\n## Industry: {r.industry_title_main}"
     return (
-        record_md,
+        md,
         status_text(i),
         gr.update(choices=maj_c, value=maj),
         gr.update(choices=sub_c, value=sub),
@@ -122,161 +78,150 @@ def load_record(i: int):
         gr.update(choices=un_c, value=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)
+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)
     return (
-        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),
+        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),
         status_text(i),
     )
 
-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
+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
     return (
-        gr.update(choices=submajors(major), value=new_sub),
-        gr.update(choices=mn_c, value=mn),
-        gr.update(choices=un_c, value=un),
+        gr.update(choices=submajors(maj),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_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
+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
     return (
-        gr.update(choices=minors(major, sub), value=new_minor),
-        gr.update(choices=un_c, value=un),
+        gr.update(choices=minors(maj,sub),value=new_mn),
+        gr.update(choices=un_c,value=un),
         status_text(i),
     )
 
-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)
+def on_unit_change(new_un,i,maj,sub,mn):
+    un_c=units(maj,sub,mn)
+    if new_un not in un_c: new_un = 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,dir):
+    # clamp and save first
+    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 dir=="next" else go_prev(i)
+    return (j,)+load_record(j)
+
+def download_csv():
+    path = PROCESSED_DATA_DIR/"annotated_output.csv"
+    records.to_csv(path,index=False)
     return str(path)
 
-# --------------
-# Build the UI
-# --------------
+# --- now build the Blocks with a login gate ---
 def build_gradio_app():
-    with gr.Blocks() as demo:
-        # a flag in State for whether the user is logged in
-        is_auth = gr.State(False)
+    USER = os.getenv("APP_USER","")
+    PWD  = os.getenv("APP_PASS","")
 
-        # ————————
-        # LOGIN SCREEN
-        # ————————
-        with gr.Column(visible=not is_auth.value) as login_box:
+    with gr.Blocks() as demo:
+        gr.HTML(f"""<style>
+            footer, .share-link, .api-link {{display:none!important}}
+            #logo {{margin-bottom:1em}}
+            #title{{text-align:center;margin-bottom:1em}}
+        </style>""")
+
+        # login panel
+        with gr.Column(elem_id="login_panel"):
+            gr.HTML(f'<img id="logo" src="data:image/png;base64,{logo_b64}" width="180" '
+                    '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 me in")
+            login_btn= gr.Button("🔒 Log in")
             login_msg= gr.Markdown("", visible=False)
 
-        def check_creds(u, p):
-            # load your secrets from env
-            USER = os.environ.get("APP_USER", "")
-            PWD  = os.environ.get("APP_PASS", "")
-            if u == USER and p == PWD:
-                return gr.update(visible=False), gr.update(visible=True), gr.update(value=True, visible=False)
+        # main app panel (hidden until auth)
+        with gr.Column(elem_id="app_panel", visible=False):
+            gr.HTML(f'<img id="logo" src="data:image/png;base64,{logo_b64}" width="180" '
+                    'style="pointer-events:none;user-select:none;display:block;margin:auto;" />')
+            gr.Markdown("# ISCO Annotation", elem_id="title")
+            idx_state = gr.State(0)
+            record_md = gr.Markdown()
+            status_md = gr.Markdown()
+            with gr.Row():
+                prev_btn = gr.Button("⬅ Previous")
+                next_btn = gr.Button("✅ Next")
+            with gr.Row():
+                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 logic
+        def check_creds(u,p):
+            if u==USER and p==PWD:
+                return gr.update(visible=False), gr.update(visible=True), gr.update(visible=False), ""
             else:
-                return gr.update(visible=True, value="❌ Bad credentials!"), None, None
+                return None, None, gr.update(visible=True,value="❌ Bad credentials"), ""
 
         login_btn.click(
             check_creds,
-            inputs=[user_in, pass_in],
-            outputs=[login_msg, login_box, is_auth]
+            inputs=[user_in,pass_in],
+            outputs=[login_panel, app_panel, login_msg, user_in]
         )
 
-        # ————————
-        # MAIN APP (hidden until auth)
-        # ————————
-        with gr.Column(visible=is_auth.value) as app_box:
-            # your logo + title + everything else goes here…
-            with gr.Row():
-                # embed your base64 logo exactly where you want
-                gr.HTML(
-                    f'<img src="data:image/png;base64,{logo_b64}" '
-                    'style="pointer-events:none; user-select:none; width:200px; display:block;" />'
-                )
-            gr.Markdown("# ISCO Annotation", elem_id="isco-title")
-
-            # then all your States, buttons, radios, callbacks…
-            idx_state = gr.State(0)
-            record_md = gr.Markdown()
-            status_md = gr.Markdown()
-            prev_btn = gr.Button("⬅ Previous")
-            next_btn = gr.Button("✅ Next")
-            major_radio = gr.Radio(label="Major", choices=[], interactive=True)
-            sub_radio   = gr.Radio(label="Sub-major", choices=[], interactive=True)
-            minor_radio = gr.Radio(label="Minor", choices=[], interactive=True)
-            unit_radio  = gr.Radio(label="Unit", choices=[], interactive=True)
-            download_btn = gr.Button("📥 Download")
-            download_file= gr.File(visible=False)
-
-            # wire up your existing load_record, on_change, save_and_jump, etc.
-            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])
-            # … and so on for prev_btn, radio.change handlers, download_btn …
-
-        # hide Gradio footer & share links:
-        gr.HTML("""
-        <style>
-          footer { display: none !important; }
-          .gradio-container .api-link,
-          .gradio-container .share-link { display: none !important; }
-          #isco-title { text-align: center; margin-top: 1em; }
-        </style>
-        """)
+        # app logic wiring
+        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"),
+                       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(
+            lambda: gr.update(visible=True), None, [download_file]
+        )
 
     return demo
 
-if __name__ == "__main__":
+if __name__=="__main__":
     demo = build_gradio_app()
-    demo.queue().launch(show_api=False, share=True, server_name="0.0.0.0")
+    demo.queue().launch(show_api=False, share=True)