"""
game/ui.py — "Beat the Machine" duel tab.
A card from the curated deck is dealt; you call its safety tier (SAFE / CAUTION /
DEADLY) and the same on-device pipeline calls it at the same instant. Score
accrues per session; post it to the leaderboard. The machine's edge is that it
refuses when unsure — so the deadly cards punish overconfidence in both players.
Stay sharper than the machine you built.
"""
from functools import partial
import gradio as gr
import pandas as pd
from PIL import Image
from pipeline.convergence import build_result
from . import datastore
from . import deck as deck_mod
TIER_COLOR = {"SAFE": "#2f6b2b", "CAUTION": "#87671c", "DEADLY": "#8c1d14", "UNKNOWN": "#57544c"}
_LB_COLS = ["Player", "You %", "Machine %", "Rounds"]
MIN_RANK_ROUNDS = 5 # ranked only after 5 rounds, so a 1-round 100% can't top the board
def _machine_call(pipe, path: str):
res = build_result(pipe.identify(Image.open(path)))
tier = "UNKNOWN" if res.abstained else res.safety
return tier, res
def _scoreboard(score: dict) -> str:
return (
f""
f"YOU {score['you']}"
f"/ {score['total']} · vs ·"
f"MACHINE {score['machine']}"
f"
"
)
def _pretty(species: str) -> str:
return species.replace("_toxic", "").replace("_deadly", "").replace("_", " ").title()
def _chip(label: str, tier: str, right: bool) -> str:
c = TIER_COLOR.get(tier, "#57544c")
mark = "✓" if right else "✕"
# color set inline with !important on the spans: Gradio's theme paints spans in
# the display panel a light color that overrides inherited/class color (the big
# truth div is fine, but these spans aren't) — inline !important wins the cascade.
return (
f""
f"{label}"
f"{tier} {mark}
"
)
def _reveal_html(card: dict, you_tier: str, you_right: bool, mtier: str, mach_right: bool) -> str:
tc = TIER_COLOR[card["tier"]]
if you_right and not mach_right:
flav = "🏆 You beat the machine on this one."
elif mach_right and not you_right:
flav = "The machine got you. Study the look-alike."
elif you_right and mach_right:
flav = "Dead heat — both correct."
else:
flav = "Neither nailed it. The woods don't grade on a curve."
abst = ""
if mtier == "UNKNOWN":
abst = ("The machine refused to commit — that's its whole "
"point. A refusal beats a confident wrong call.
")
return (
f""
f"
TRUTH
"
f"
{card['tier']}
"
f"
{_pretty(card['species'])} · {card['scientific']}
"
f"
{_chip('YOUR CALL', you_tier, you_right)}"
f"{_chip('MACHINE', mtier, mach_right)}
"
f"
{flav}
{abst}
"
"
CAN YOU BEAT THE MACHINE?
"
"
We built the machine. Now try to out-forage it. "
"A card is dealt — call its tier before it does. Remember: the machine's "
"trick is knowing when not to guess. Every round you play sharpens you "
"and sharpens the open dataset behind it.
Deal a card to start the duel.
")
score_html = gr.HTML(_scoreboard({"you": 0, "machine": 0, "total": 0}))
gr.HTML("▌ LEADERBOARD · ranked at 5+ rounds
")
post_btn = gr.Button("POST MY SCORE", elem_classes="eink-scan")
lb_status = gr.HTML("")
lb = gr.Dataframe(value=_lb_df(), headers=_LB_COLS, interactive=False,
elem_classes="gm-lb", wrap=True)
refresh_btn = gr.Button("⟳ refresh", elem_classes="gm-refresh")
# ── handlers (closures capture `pipe`) ───────────────────────────────────
def _deal(score, card):
prev = card["file"] if card else None
c = deck_mod.random_card(exclude_file=prev)
new = {**c, "answered": False}
return c["path"], new, "Your call?
", _scoreboard(score)
def _guess(tier, card, score):
if not card or card.get("answered"):
return gr.update(), card, score # ignore double-taps / no card yet
mtier, _ = _machine_call(pipe, card["path"])
you_right = tier == card["tier"]
mach_right = mtier == card["tier"]
score = {
"you": score["you"] + int(you_right),
"machine": score["machine"] + int(mach_right),
"total": score["total"] + 1,
}
card = {**card, "answered": True}
return _reveal_html(card, tier, you_right, mtier, mach_right), card, score
def _post(score, profile: gr.OAuthProfile | None = None):
if profile is None:
return "Log in (top of the page) to post your score.
", gr.update()
if score["total"] == 0:
return "Play at least one round first.
", gr.update()
datastore.post_score(profile.username, score["you"], score["total"], score["machine"])
note = "" if datastore.persistence_enabled() else \
" (session-only until the dataset token is set)"
return f"Posted as {profile.username}.{note}
", _lb_df()
deal_btn.click(_deal, [score_state, card_state], [card_img, card_state, reveal, score_html])
for b, t in [(safe_b, "SAFE"), (caut_b, "CAUTION"), (dead_b, "DEADLY")]:
b.click(partial(_guess, t), [card_state, score_state], [reveal, card_state, score_state]) \
.then(_scoreboard, [score_state], [score_html])
post_btn.click(_post, [score_state], [lb_status, lb])
refresh_btn.click(lambda: _lb_df(), None, [lb])
# ── Stump the Machine: upload your own find -> the data flywheel ──────────────
_CONTRIB_COLS = ["Contributor", "Sightings"]
def _contrib_df() -> pd.DataFrame:
rows = datastore.load_contributors()
return pd.DataFrame([[r["contributor"], r["count"]] for r in rows], columns=_CONTRIB_COLS)
def _stump_unrouted_html(status: str) -> str:
body = {
"stored": "Saved for review. If it's a real find, you just handed us a "
"training example the model is missing.
",
"duplicate": "Already flagged for review — this one's logged.
",
"disabled": "Couldn't save for review (review-queue token scope not set "
"on the Space).
",
}.get(status, "")
return (
""
"
YOU FOUND A GAP
"
"
OFF THE MAP
"
"
The machine couldn't place this in berry, mushroom, or plant — "
"the ultimate stump.
"
f"{body}"
"
Genuinely off-topic shots (not a wild plant, mushroom, or berry) get "
"filtered out in review.
"
"
✓ Added to the open dataset — thank you. This is exactly "
"the data that makes the next model better.
",
"duplicate": "Already in the dataset — this photo (or a near-match) "
"was logged before, so it wasn't added again.
",
"disabled": "Couldn't persist (dataset token not set on the Space).
",
}.get(status, "")
return (
f""
f"
YOUR FIND
"
f"
{head}
"
f"
you said: {label}
"
f"
"
f"{store_line}"
f"
An ID here is never permission to eat — verify with an expert.
"
f"
"
"
STUMP THE MACHINE
"
"
Upload your own find. The machine calls it on the spot — and "
"when it can't, you've stumped it. Every accepted photo joins the open "
"CC-BY-4.0 dataset that trains the next model. You're not just playing; you're "
"building the thing.
Upload a find, add your guess, and submit.
")
gr.HTML("▌ TOP CONTRIBUTORS
")
contrib = gr.Dataframe(value=_contrib_df(), headers=_CONTRIB_COLS, interactive=False,
elem_classes="gm-lb", wrap=True)
crefresh = gr.Button("⟳ refresh", elem_classes="gm-refresh")
def _submit(image, user_label, consented, profile: gr.OAuthProfile | None = None):
if profile is None:
return "Log in (top of the page) to contribute.
", gr.update()
if image is None:
return "Upload a photo first.
", gr.update()
if not (user_label or "").strip():
return "Add your guess first.
", gr.update()
if not consented:
return "Check the consent box to contribute.
", gr.update()
call = pipe.identify(image)
res = build_result(call)
in_domain = (not res.abstained) or call.get("reason", "") == "low_confidence"
if not in_domain:
router = {"domain": call.get("domain", "unknown"),
"domain_confidence": call.get("domain_confidence", 0.0),
"reason": call.get("reason", "")}
status = datastore.append_unrouted(image, user_label.strip(), router, profile.username)
return _stump_unrouted_html(status), gr.update() # contributor board unchanged
machine = {"species": res.species, "confidence": res.confidence,
"abstained": res.abstained, "safety": res.safety, "domain": res.domain}
status = datastore.append_sighting(image, user_label.strip(), machine, profile.username)
board = _contrib_df() if status == "stored" else gr.update()
return _stump_result_html(user_label.strip(), res, status), board
submit.click(_submit, [up, guess, consent], [sresult, contrib])
crefresh.click(lambda: _contrib_df(), None, [contrib])