import gradio as gr import pandas as pd import folium import numpy as np import os import re import json BASE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in dir() else os.getcwd() STAY_POINTS = os.path.join(BASE, "data", "stay_points_inference_sample.csv") POI_PATH = os.path.join(BASE, "data", "poi_inference_sample.csv") DEMO_PATH = os.path.join(BASE, "data", "demographics_inference_sample.csv") COT_PATH = os.path.join(BASE, "data", "inference_results_sample.json") SEX_MAP = {1:"Male", 2:"Female", -8:"Unknown", -7:"Prefer not to answer"} EDU_MAP = {1:"Less than HS", 2:"HS Graduate/GED", 3:"Some College/Associate", 4:"Bachelor's Degree", 5:"Graduate/Professional Degree", -1:"N/A", -7:"Prefer not to answer", -8:"Unknown"} INC_MAP = {1:"<$10,000", 2:"$10,000–$14,999", 3:"$15,000–$24,999", 4:"$25,000–$34,999", 5:"$35,000–$49,999", 6:"$50,000–$74,999", 7:"$75,000–$99,999", 8:"$100,000–$124,999", 9:"$125,000–$149,999", 10:"$150,000–$199,999", 11:"$200,000+", -7:"Prefer not to answer", -8:"Unknown", -9:"Not ascertained"} RACE_MAP = {1:"White", 2:"Black or African American", 3:"Asian", 4:"American Indian or Alaska Native", 5:"Native Hawaiian or Other Pacific Islander", 6:"Multiple races", 97:"Other", -7:"Prefer not to answer", -8:"Unknown"} ACT_MAP = {0:"Transportation", 1:"Home", 2:"Work", 3:"School", 4:"ChildCare", 5:"BuyGoods", 6:"Services", 7:"EatOut", 8:"Errands", 9:"Recreation", 10:"Exercise", 11:"Visit", 12:"HealthCare", 13:"Religious", 14:"SomethingElse", 15:"DropOff"} print("Loading data...") sp = pd.read_csv(STAY_POINTS) poi = pd.read_csv(POI_PATH) demo = pd.read_csv(DEMO_PATH) sp = sp.merge(poi, on="poi_id", how="left") sp["start_datetime"] = pd.to_datetime(sp["start_datetime"], utc=True) sp["end_datetime"] = pd.to_datetime(sp["end_datetime"], utc=True) sp["duration_min"] = ((sp["end_datetime"] - sp["start_datetime"]).dt.total_seconds() / 60).round(1) def parse_act_types(x): try: codes = list(map(int, str(x).strip("[]").split())) return ", ".join(ACT_MAP.get(c, str(c)) for c in codes) except: return str(x) sp["act_label"] = sp["act_types"].apply(parse_act_types) # Load CoT JSON (optional) cot_by_agent = {} if os.path.exists(COT_PATH): with open(COT_PATH, "r") as f: cot_raw = json.load(f) records = cot_raw if isinstance(cot_raw, list) else cot_raw.get("inference_results", []) for result in records: cot_by_agent[int(result["agent_id"])] = result print(f"Loaded CoT for {len(cot_by_agent)} agents.") sample_agents = sorted(sp["agent_id"].unique().tolist()) print(f"Ready. {len(sample_agents)} agents loaded.") def get_cot(agent_id): result = cot_by_agent.get(int(agent_id), {}) s1 = result.get("step1_response", "") s2 = result.get("step2_response", "") s3 = result.get("step3_response", "") p1 = result.get("step1_prompt", "") p2 = result.get("step2_prompt", "") p3 = result.get("step3_prompt", "") return s1, s2, s3, p1, p2, p3 # ── Mobility text builders ──────────────────────────────────────────────────── def build_mobility_summary(agent_sp): top5 = (agent_sp.groupby("name")["duration_min"] .agg(visits="count", avg_dur="mean") .sort_values("visits", ascending=False) .head(5)) obs_start = agent_sp["start_datetime"].min().strftime("%Y-%m-%d") obs_end = agent_sp["end_datetime"].max().strftime("%Y-%m-%d") days = (agent_sp["end_datetime"].max() - agent_sp["start_datetime"].min()).days act_counts = agent_sp["act_label"].value_counts().head(3) top_acts = ", ".join(f"{a} ({n})" for a, n in act_counts.items()) agent_sp2 = agent_sp.copy() agent_sp2["hour"] = agent_sp2["start_datetime"].dt.hour def tod(h): if 5 <= h < 12: return "Morning" if 12 <= h < 17: return "Afternoon" if 17 <= h < 21: return "Evening" return "Night" agent_sp2["tod"] = agent_sp2["hour"].apply(tod) peak_tod = agent_sp2["tod"].value_counts().idxmax() agent_sp2["is_weekend"] = agent_sp2["start_datetime"].dt.dayofweek >= 5 wd_pct = int((~agent_sp2["is_weekend"]).mean() * 100) lines = [ f"Period: {obs_start} ~ {obs_end} ({days} days)", f"Stay points: {len(agent_sp)} | Unique locations: {agent_sp['name'].nunique()}", f"Weekday/Weekend: {wd_pct}% / {100-wd_pct}% | Peak time: {peak_tod}", f"Top activities: {top_acts}", "", "Top Locations:", ] for i, (name, row) in enumerate(top5.iterrows(), 1): lines.append(f" {i}. {name} — {int(row['visits'])} visits, avg {int(row['avg_dur'])} min") return "\n".join(lines) def build_weekly_checkin(agent_sp, max_days=None): agent_sp2 = agent_sp.copy() agent_sp2["date"] = agent_sp2["start_datetime"].dt.date all_dates = sorted(agent_sp2["date"].unique()) dates_to_show = all_dates[:max_days] if max_days else all_dates total_days = len(all_dates) lines = ["WEEKLY CHECK-IN SUMMARY", "======================="] for date in dates_to_show: grp = agent_sp2[agent_sp2["date"] == date] dow = grp["start_datetime"].iloc[0].strftime("%A") label = "Weekend" if grp["start_datetime"].iloc[0].dayofweek >= 5 else "Weekday" lines.append(f"\n--- {dow}, {date} ({label}) ---") lines.append(f"Total activities: {len(grp)}") for _, row in grp.iterrows(): lines.append( f"- {row['start_datetime'].strftime('%H:%M')}-" f"{row['end_datetime'].strftime('%H:%M')} " f"({int(row['duration_min'])} mins): " f"{row['name']} - {row['act_label']}" ) if max_days and total_days > max_days: lines.append(f"\n... ({total_days - max_days} more days)") return "\n".join(lines) # ── HTML reasoning chain ────────────────────────────────────────────────────── CHAIN_CSS = """ """ def _loading(msg): return (f'
' f'' f'{msg}
') # ── Parsers ─────────────────────────────────────────────────────────────────── def _parse_s1(text): locations, dur_map, tod, wk, dist = [], {}, {}, {}, None for line in text.splitlines(): s = line.strip() # Locations: "- Name: N visits/times/time/times each" m = re.match(r'-\s+(.+?):\s+(\d+)\s+(?:visit|time)', s, re.IGNORECASE) if m: locations.append((m.group(1).strip(), int(m.group(2)))) continue # Duration — 4 formats m2 = re.match(r'-?\s*(.+?):\s+(?:Average duration of\s*)?([\d.]+)\s+min(?:utes?)?\s+on average', s, re.IGNORECASE) if not m2: m2 = re.match(r'-?\s*(.+?):\s+Average duration of ([\d.]+)\s+min', s, re.IGNORECASE) if not m2: m2 = re.match(r'-?\s*Average duration at (.+?):\s+([\d.]+)\s+min', s, re.IGNORECASE) if not m2: m2 = re.search(r'\bat ([A-Za-z][^(,]+?)\s*\(average ([\d.]+)\s*min', s, re.IGNORECASE) if m2: dur_map[m2.group(1).strip()] = float(m2.group(2)) # TOD format A: "65% morning, 23% afternoon, 6% evening, 5% night" if not tod: mA = re.search(r'(\d+)%\s*morning.*?(\d+)%\s*afternoon.*?(\d+)%\s*evening.*?(\d+)%\s*night', s, re.IGNORECASE) if mA: tod = {'Morning': int(mA.group(1)), 'Afternoon': int(mA.group(2)), 'Evening': int(mA.group(3)), 'Night': int(mA.group(4))} # TOD format B: "morning: 40%, afternoon: 36%, ..." if not tod: mB = re.search(r'morning[:\s]+(\d+)%.*?afternoon[:\s]+(\d+)%.*?evening[:\s]+(\d+)%.*?night[:\s]+(\d+)%', s, re.IGNORECASE) if mB: tod = {'Morning': int(mB.group(1)), 'Afternoon': int(mB.group(2)), 'Evening': int(mB.group(3)), 'Night': int(mB.group(4))} # TOD format C: "Afternoon (43%), morning (27%), ..." if not tod: parts = re.findall(r'(morning|afternoon|evening|night)\s*\(?(\d+)%\)?', s, re.IGNORECASE) if len(parts) >= 3: d = {k.capitalize(): int(v) for k, v in parts} if all(k in d for k in ['Morning', 'Afternoon', 'Evening']): d.setdefault('Night', 0) tod = d # Weekday/weekend if not wk: m4 = re.search(r'(\d+)%\s*weekday.*?(\d+)%\s*weekend', s, re.IGNORECASE) if m4: wk = {'Weekday': int(m4.group(1)), 'Weekend': int(m4.group(2))} # Distance if not dist: m5 = re.search(r'average distance of approximately ([\d.]+)\s*(?:km|miles?)', s, re.IGNORECASE) if m5: dist = float(m5.group(1)) return [(n, v, dur_map.get(n)) for n, v in locations[:7]], tod, wk, dist def _parse_s2(text): DIMS = { 'ROUTINE': ['ROUTINE', 'SCHEDULE'], 'ECONOMIC': ['ECONOMIC', 'SPENDING'], 'SOCIAL': ['SOCIAL', 'LIFESTYLE'], 'STABILITY': ['STABILITY', 'REGULARITY', 'CONSISTENCY', 'URBAN'], } sections, current_key, current_lines = {}, None, [] for line in text.splitlines(): s = line.strip() mA = re.match(r'^\d+\.\s+([A-Z][A-Z\s&]+?)(?:\s+ANALYSIS|\s+PATTERNS|\s+INDICATORS|\s+CHARACTERISTICS|\s+STABILITY)?:\s*$', s, re.IGNORECASE) mB = re.match(r'^STEP\s+\d+:\s+([A-Z][A-Z\s&]+?)(?:\s+ANALYSIS|\s+PATTERNS|\s+INDICATORS|\s+CHARACTERISTICS|\s+STABILITY)?\s*$', s, re.IGNORECASE) mm = mA or mB if mm: if current_key and current_lines: sections[current_key] = ' '.join(current_lines) current_key = mm.group(1).upper().strip() current_lines = [] elif current_key and s: if re.match(r'^\d+\.\d+', s): sub = re.sub(r'^\d+\.\d+[^:]*:\s*', '', s) if sub: current_lines.append(sub) elif s.startswith('-'): current_lines.append(s.lstrip('-').strip()) elif not re.match(r'^\d+\.', s): current_lines.append(s) if current_key and current_lines: sections[current_key] = ' '.join(current_lines) result = {} for dim, keywords in DIMS.items(): for k, txt in sections.items(): if any(kw in k for kw in keywords) and txt: sents = re.split(r'(?<=[.!?])\s+', txt.strip()) summary = ' '.join(sents[:2]) result[dim] = summary[:157] + '…' if len(summary) > 160 else summary break return result def _parse_s3(text): pred, conf, r_lines, in_r = '', 0, [], False for line in text.splitlines(): s = line.strip() if s.startswith('INCOME_PREDICTION:'): pred = s.replace('INCOME_PREDICTION:', '').strip() elif s.startswith('INCOME_CONFIDENCE:'): try: conf = int(re.search(r'\d+', s).group()) except: pass elif s.startswith('INCOME_REASONING:'): in_r = True r_lines.append(s.replace('INCOME_REASONING:', '').strip()) elif in_r: if re.match(r'^2\.', s) or s.startswith('INCOME_'): break if s: r_lines.append(s) reasoning = ' '.join(r_lines).strip() sents = re.split(r'(?<=[.!?])\s+', reasoning) reasoning = ' '.join(sents[:3]) return pred, conf, (reasoning[:277] + '…' if len(reasoning) > 280 else reasoning) PROMPT_BULLETS = { 1: [ "Extract objective factual features from the agent's mobility trajectory without any interpretation", "Location inventory: list all visited POIs with visit counts and apparent price tier (budget / mid-range / high-end)", "Temporal patterns: time-of-day distribution, weekday vs. weekend split, and regularity of routines", "Spatial characteristics: activity radius, average movement distance between locations", "Sequence observations: common location transitions and typical daily activity chains", ], 2: [ "Perform behavioral abstraction across four dimensions based on Step 1 features", "Routine & Schedule: infer work schedule type (fixed hours, flexible, shift work, etc.) and daily structure", "Economic Behavior: assess spending tier from venue choices, transportation costs, and lifestyle signals", "Social & Lifestyle: identify social engagement patterns, leisure activities, and community involvement", "Routine Stability: evaluate consistency and regularity of movement patterns over the observation period", ], 3: [ "Synthesize factual features (Step 1) and behavioral patterns (Step 2) to infer household income bracket", "Score location economic indicators: luxury / mid-range / budget venue distribution", "Consider transportation mode signals, activity diversity, and temporal flexibility as income proxies", "Output: INCOME_PREDICTION — a single income range with confidence rating (1–5)", "Output: INCOME_REASONING — evidence-grounded justification referencing specific mobility observations", ], } PROMPT_INPUTS = { 1: "② Activity Chronicles + ③ Visiting Summaries — detailed daily visit logs and weekly behavioral statistics generated from raw stay points", 2: "Stage 1 response — factual features extracted from Activity Chronicles", 3: "Stage 1 + Stage 2 responses — feature extraction and behavioral abstraction combined", } _INPUT_TAG = ('Input') def _extract_prompt_instruction(prompt_text, stage): bullets = PROMPT_BULLETS.get(stage, []) if not bullets: return '' inp = PROMPT_INPUTS.get(stage, '') input_block = ('
' + _INPUT_TAG + inp + '
') items = ''.join('
  • ' + b + '
    • ' for b in bullets) return input_block + '' # ── Body renderers ──────────────────────────────────────────────────────────── def _s1_body(text, active): if not active: return '
    Press ▶ to start
    ' if not text: return _loading('Extracting features') locs, tod, wk, dist = _parse_s1(text) max_v = max((v for _, v, _ in locs), default=1) rows = '' for name, visits, dur in locs: bar_w = int(60 * visits / max_v) dur_str = f'{int(dur)}m' if dur else '—' rows += (f'' f'{name}' f'
    ' f'
    {visits}
    ' f'{dur_str}') table = (f'' f'' f'{rows}
    LocationVisitsAvg Stay
    ') if rows else '' def seg_bar(data, seg_classes): total = sum(data.values()) or 1 segs = ''.join( f'
    ' for (label, v), cls in zip(data.items(), seg_classes)) legend = ''.join( f'
    {label} {v}%
    ' for (label, v), cls in zip(data.items(), seg_classes)) return f'
    {segs}
    {legend}
    ' tod_panel = (f'
    Time of Day
    ' f'{seg_bar(tod, ["seg-morning","seg-afternoon","seg-evening","seg-night"])}
    ') if tod else '' wk_panel = (f'
    Weekday / Weekend
    ' f'{seg_bar(wk, ["seg-weekday","seg-weekend"])}
    ') if wk else '' temporal = f'
    {tod_panel}{wk_panel}
    ' if (tod_panel or wk_panel) else '' dist_line = f'
    📍 Avg trip distance  {dist} mi
    ' if dist else '' return table + temporal + dist_line def _s2_body(text, active): if not active: return '
    Waiting…
    ' if not text: return _loading('Analyzing behavior') dims = _parse_s2(text) DIM_META = [('ROUTINE','🕐','Schedule'), ('ECONOMIC','💰','Economic'), ('SOCIAL','👥','Social'), ('STABILITY','🔄','Stability')] cards = '' for key, icon, label in DIM_META: txt = dims.get(key, '') content = (f'
    {txt}
    ' if txt else '
    —
    ') cards += (f'
    ' f'
    ' f'{icon}' f'{label}
    ' f'{content}
    ') return f'
    {cards}
    ' def _s3_body(text, active): if not active: return '
    Waiting…
    ' if not text: return _loading('Inferring demographics') pred, conf, reasoning = _parse_s3(text) return (f'
    ' f'
    ' f'
    {pred or "—"}
    ' f'
    Income
    ' f'
    ' f'
    {reasoning}
    ') # ── Main renderer ───────────────────────────────────────────────────────────── def render_chain(s1_text, s2_text, s3_text, status="idle", s1_prompt="", s2_prompt="", s3_prompt=""): s1_on = status in ("running1", "running2", "running3", "done") s2_on = status in ("running2", "running3", "done") s3_on = status in ("running3", "done") s1_body = _s1_body(s1_text if s1_on else '', s1_on) s2_body = _s2_body(s2_text if s2_on else '', s2_on) s3_body = _s3_body(s3_text if s3_on else '', s3_on) def prompt_pill(stage_num): bullets_html = _extract_prompt_instruction('', stage_num) if not bullets_html: return '' return (f'
    ' f'' f'
    {bullets_html}
    ' f'
    ') def stage(cls, num, title, body, on, stage_num): dim_cls = 'active' if on else 'dim' pill = prompt_pill(stage_num) if on else '' return (f'
    ' f'
    ' f'{num}' f'{title}' f'
    ' f'{pill}' f'
    {body}
    ' f'
    ') def arrow(label, on): op = '1' if on else '0.2' return (f'
    ' f'
    ' f'
    {label}
    ' f'
    ') flow_banner = ( '
    ' '
    Data Pipeline
    ' '
    ' '
    ' '
    Raw Data
    ' '
    Stay points + POI metadata
    ① Raw Stay Points tab
    ' '
    ' '
    →
    ' '
    ' '
    Activity Chronicles
    ' '
    Detailed Chronicles + Visiting Summaries
    ② ③ tabs · micro + macro level
    ' '
    ' '
    →
    ' '
    ' '
    Prompt 1
    ' '
    Factual feature extraction
    no interpretation · pattern identification
    ' '
    ' '
    ' '
    ' ) html = CHAIN_CSS + '
    ' html += flow_banner html += stage('s1', 'Stage 01', 'Feature Extraction', s1_body, s1_on, 1) html += arrow('behavioral abstraction', s2_on) html += stage('s2', 'Stage 02', 'Behavioral Analysis', s2_body, s2_on, 2) html += arrow('demographic inference', s3_on) html += stage('s3', 'Stage 03', 'Demographic Inference', s3_body, s3_on, 3) html += '
    ' return html def build_map(agent_sp): agent_sp = agent_sp.reset_index(drop=True).copy() agent_sp["latitude"] += np.random.uniform(-0.0003, 0.0003, len(agent_sp)) agent_sp["longitude"] += np.random.uniform(-0.0003, 0.0003, len(agent_sp)) lat = agent_sp["latitude"].mean() lon = agent_sp["longitude"].mean() m = folium.Map(location=[lat, lon], zoom_start=12, tiles="CartoDB positron") coords = list(zip(agent_sp["latitude"], agent_sp["longitude"])) if len(coords) > 1: folium.PolyLine(coords, color="#cc000055", weight=1.5, opacity=0.4).add_to(m) n = len(agent_sp) for i, row in agent_sp.iterrows(): ratio = i / max(n - 1, 1) r = int(255 - ratio * (255 - 139)) g = int(204 * (1 - ratio) ** 2) b = 0 color = f"#{r:02x}{g:02x}{b:02x}" folium.CircleMarker( location=[row["latitude"], row["longitude"]], radius=7, color=color, fill=True, fill_color=color, fill_opacity=0.9, popup=folium.Popup( f"#{i+1} {row['name']}
    " f"{row['start_datetime'].strftime('%a %m/%d %H:%M')}
    " f"{int(row['duration_min'])} min
    {row['act_label']}", max_width=220 ) ).add_to(m) legend_html = """
    Stay Point Legend
    Earlier → Later
    Movement path
    Click dot for details
    """ m.get_root().html.add_child(folium.Element(legend_html)) m.get_root().width = "100%" m.get_root().height = "420px" return m._repr_html_() def build_demo_text(row): age = int(row["age"]) if row["age"] > 0 else "Unknown" return ( f"Age: {age} | " f"Sex: {SEX_MAP.get(int(row['sex']), row['sex'])} | " f"Race: {RACE_MAP.get(int(row['race']), row['race'])} | " f"Education: {EDU_MAP.get(int(row['education']), row['education'])} | " f"Income: {INC_MAP.get(int(row['hh_income']), row['hh_income'])}" ) def build_raw_staypoints(agent_sp, n=12): cols = ["start_datetime", "end_datetime", "duration_min", "latitude", "longitude", "name", "act_label"] df = agent_sp[cols].head(n).copy() df["start_datetime"] = df["start_datetime"].dt.strftime("%m/%d %H:%M") df["end_datetime"] = df["end_datetime"].dt.strftime("%H:%M") df["duration_min"] = df["duration_min"].astype(int).astype(str) + " min" df["latitude"] = df["latitude"].round(5).astype(str) df["longitude"] = df["longitude"].round(5).astype(str) df.columns = ["Start", "End", "Duration", "Lat", "Lon", "Venue", "Activity"] lines = ["Stay Points (raw input — first {} records)".format(n), ""] col_w = {"Start": 11, "End": 7, "Duration": 9, "Lat": 9, "Lon": 10, "Venue": 26, "Activity": 16} header = " ".join(k.ljust(v) for k, v in col_w.items()) lines.append(header) lines.append("-" * len(header)) for _, row in df.iterrows(): line = " ".join(str(row[k]).ljust(v)[:v] for k, v in col_w.items()) lines.append(line) lines.append("") lines.append("→ These records are transformed into Activity Chronicles (Detailed + Visiting Summaries)") lines.append(" and fed into Prompt 1 for factual feature extraction.") return "\n".join(lines) # ── Callbacks ───────────────────────────────────────────────────────────────── def on_select(agent_id): agent_id = int(agent_id) agent_sp = sp[sp["agent_id"] == agent_id].sort_values("start_datetime") agent_demo = demo[demo["agent_id"] == agent_id].iloc[0] map_html = build_map(agent_sp) demo_text = build_demo_text(agent_demo) raw_text = build_mobility_summary(agent_sp) + "\n\n" + build_weekly_checkin(agent_sp) chain_html = render_chain("", "", "", status="idle") return map_html, raw_text, demo_text, chain_html def run_step(agent_id, step): agent_id = int(agent_id) s1, s2, s3, p1, p2, p3 = get_cot(agent_id) next_step = step + 1 if next_step == 1: html = render_chain(s1, "", "", status="running2", s1_prompt=p1) return html, 1, gr.update(value="▶ Stage 2: Behavioral Analysis") elif next_step == 2: html = render_chain(s1, s2, "", status="running3", s1_prompt=p1, s2_prompt=p2) return html, 2, gr.update(value="▶ Stage 3: Demographic Inference") else: html = render_chain(s1, s2, s3, status="done", s1_prompt=p1, s2_prompt=p2, s3_prompt=p3) return html, -1, gr.update(value="↺ Reset") def handle_btn(agent_id, step): if step == -1: html = render_chain("", "", "", status="idle") return html, 0, gr.update(value="▶ Stage 1: Feature Extraction") return run_step(agent_id, step) def on_select_reset(agent_id): agent_id_int = int(agent_id) agent_sp = sp[sp["agent_id"] == agent_id_int].sort_values("start_datetime") agent_demo = demo[demo["agent_id"] == agent_id_int].iloc[0] map_html = build_map(agent_sp) demo_text = build_demo_text(agent_demo) cot_entry = cot_by_agent.get(agent_id_int, {}) summary = build_mobility_summary(agent_sp) raw_full = cot_entry.get("weekly_checkin") or build_weekly_checkin(agent_sp) sep = "\n\n--- " parts = raw_full.split(sep) extra = len(parts) - 1 raw_text = parts[0] + (sep.join([""] + parts[1:2]) + ("\n\n... ({} more days)".format(extra - 1) if extra > 1 else "")) if extra > 0 else raw_full chain_html = render_chain("", "", "", status="idle") raw_sp_text = build_raw_staypoints(agent_sp) return map_html, raw_sp_text, summary, raw_text, demo_text, chain_html, 0, gr.update(value="▶ Stage 1: Feature Extraction") SHOWCASE_AGENTS = sample_agents[:6] def build_agent_cards(selected_id): selected_id = int(selected_id) parts = [] parts.append("
    ") for aid in SHOWCASE_AGENTS: row = demo[demo["agent_id"] == aid].iloc[0] age = int(row["age"]) if row["age"] > 0 else "?" sex = SEX_MAP.get(int(row["sex"]), "?") edu = EDU_MAP.get(int(row["education"]), "?") inc = INC_MAP.get(int(row["hh_income"]), "?") is_sel = (aid == selected_id) sel_style = "border:2px solid #c0392b;background:#fff8f8;box-shadow:0 2px 8px rgba(192,57,43,0.15);" nor_style = "border:1.5px solid #ddd;background:#fafafa;box-shadow:0 1px 3px rgba(0,0,0,0.06);" style = sel_style if is_sel else nor_style dot = "" if is_sel else "" js = "var t=document.querySelector('#agent_hidden_input textarea');t.value='AID';t.dispatchEvent(new Event('input',{bubbles:true}));".replace("AID", str(aid)) parts.append( "
    " "
    " + dot + "Agent #" + str(aid) + "
    " "
    " "Age: " + str(age) + "   Sex: " + sex + "
    " "Edu: " + edu + "
    " "Income: " + inc + "
    " ) parts.append("
    ") return "".join(parts) # ── UI ──────────────────────────────────────────────────────────────────────── with gr.Blocks(title="HiCoTraj Demo") as app: gr.Markdown("## HiCoTraj — Trajectory Visualization & Hierarchical CoT Demo") gr.Markdown("*Zero-Shot Demographic Reasoning via Hierarchical Chain-of-Thought Prompting from Trajectory* · ACM SIGSPATIAL GeoGenAgent 2025") gr.HTML("
    💻 Best experienced on a laptop or desktop — the side-by-side layout requires a wide screen.
    ") gr.HTML("
    ☀️ Use Light Mode — dark mode will hide most UI elements. In your browser: View → Appearance → Light.
    ") gr.Markdown(""" **Dataset:** NUMOSIM[1] > [1]Stanford C, Adari S, Liao X, et al. *NUMoSim: A Synthetic Mobility Dataset with Anomaly Detection Benchmarks.* ACM SIGSPATIAL Workshop on Geospatial Anomaly Detection, 2024. """) gr.Markdown("### Select Agent") agent_cards = gr.HTML(value=build_agent_cards(SHOWCASE_AGENTS[0])) agent_hidden = gr.Textbox( value=str(SHOWCASE_AGENTS[0]), visible=True, elem_id="agent_hidden_input", elem_classes=["hidden-input"] ) gr.HTML("") with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Trajectory Map") map_out = gr.HTML() gr.Markdown("### Mobility Data") with gr.Tabs(): with gr.Tab("① Raw Stay Points"): sp_out = gr.Textbox(lines=10, interactive=False, label="", show_label=False) with gr.Tab("② Activity Chronicles"): raw_out = gr.Textbox(lines=10, interactive=False, label="", show_label=False) show_all_btn = gr.Button("Show All Days", size="sm", variant="secondary") with gr.Tab("③ Visiting Summaries"): summary_out = gr.Textbox(lines=10, interactive=False, label="", show_label=False) with gr.Column(scale=1): gr.Markdown("### Hierarchical Chain-of-Thought Reasoning") step_state = gr.State(value=0) run_btn = gr.Button("▶ Stage 1: Feature Extraction", variant="primary") chain_out = gr.HTML(value=render_chain("", "", "", status="idle")) def on_agent_click(agent_id): cards_html = build_agent_cards(agent_id) map_html, raw_sp, summary, raw_text, _demo_text, chain_html, step, btn = on_select_reset(agent_id) return cards_html, map_html, raw_sp, summary, raw_text, chain_html, step, btn agent_hidden.change( fn=on_agent_click, inputs=agent_hidden, outputs=[agent_cards, map_out, sp_out, summary_out, raw_out, chain_out, step_state, run_btn] ) def on_load(agent_id): map_html, raw_sp, summary, raw_text, _demo_text, chain_html, step, btn = on_select_reset(agent_id) return map_html, raw_sp, summary, raw_text, chain_html, step, btn app.load( fn=on_load, inputs=agent_hidden, outputs=[map_out, sp_out, summary_out, raw_out, chain_out, step_state, run_btn] ) run_btn.click( fn=handle_btn, inputs=[agent_hidden, step_state], outputs=[chain_out, step_state, run_btn] ) def toggle_raw(agent_id, current_text): agent_id_int = int(agent_id) cot_entry = cot_by_agent.get(agent_id_int, {}) agent_sp = sp[sp["agent_id"] == agent_id_int].sort_values("start_datetime") raw_full = cot_entry.get("weekly_checkin") or build_weekly_checkin(agent_sp) if "more days" in current_text: return raw_full, gr.update(value="Show Less") else: sep = "\n\n--- " parts = raw_full.split(sep) extra = len(parts) - 1 short = parts[0] + (sep.join([""] + parts[1:2]) + ("\n\n... ({} more days)".format(extra - 1) if extra > 1 else "")) if extra > 0 else raw_full return short, gr.update(value="Show All Days") show_all_btn.click( fn=toggle_raw, inputs=[agent_hidden, raw_out], outputs=[raw_out, show_all_btn] ) app.launch(show_error=True, theme=gr.themes.Soft(), share=True, js="() => { document.body.classList.remove('dark'); }")