"""
PLRS — Logic Engine
HuggingFace Space entry point.
Loads SAKT model weights from HF Hub (Clementio/PLRS).
Bundles the plrs package inline (until PyPI release).
"""
import json
import sys
from pathlib import Path
import numpy as np
import streamlit as st
import torch
ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))
from plrs.curriculum.loader import load_dag
from plrs.pipeline import PLRSPipeline
# ── Page config ───────────────────────────────────────────────────────────────
st.set_page_config(
page_title="PLRS · Logic Engine",
page_icon="🧠",
layout="wide",
initial_sidebar_state="expanded",
)
# ── Styling ───────────────────────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Model + pipeline loading ──────────────────────────────────────────────────
@st.cache_resource(show_spinner="Loading curriculum & model from HuggingFace...")
def load_pipelines():
from plrs.model.model_loader import load_model_from_hub
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
maps = ROOT / "data" / "knowledge_maps"
# Load model (tries decay, vanilla, then base)
model, model_type = load_model_from_hub(device=str(device))
pipelines = {}
for domain, fname in [("math", "math_dag.json"), ("cs", "cs_dag.json")]:
path = maps / fname
if path.exists():
curriculum = load_dag(path)
pipeline = PLRSPipeline(curriculum)
if model:
pipeline._model = model
pipelines[domain] = pipeline
return pipelines, model is not None, model_type
@st.cache_data
def load_skill_encoder():
import pandas as pd
path = ROOT / "data" / "skill_encoder_v2.csv"
if path.exists():
return pd.read_csv(path)
return None
pipelines, has_model, model_type = load_pipelines()
skill_encoder = load_skill_encoder()
# ── Sidebar ───────────────────────────────────────────────────────────────────
with st.sidebar:
st.markdown("### 🧠 PLRS")
st.markdown('LOGIC ENGINE v0.2.0
', unsafe_allow_html=True)
if has_model:
st.markdown(f'● {model_type} LOADED
', unsafe_allow_html=True)
else:
st.markdown('● MANUAL MODE
', unsafe_allow_html=True)
st.markdown("---")
domain_label = st.selectbox("Curriculum", ["Nigerian SS Mathematics", "CS Fundamentals"])
domain_key = "math" if "Mathematics" in domain_label else "cs"
pipeline = pipelines[domain_key]
curriculum = pipeline.curriculum
st.markdown("---")
threshold = st.slider("Mastery threshold", 0.50, 0.90, 0.70, 0.05)
soft_threshold = st.slider("Challenging threshold", 0.20, 0.65, 0.50, 0.05)
top_n = st.slider("Top N recommendations", 3, 10, 5)
pipeline.threshold = threshold
pipeline.soft_threshold = soft_threshold
pipeline.top_n = top_n
st.markdown("---")
st.markdown(f'NODES: {curriculum.num_nodes}
', unsafe_allow_html=True)
st.markdown(f'EDGES: {curriculum.num_edges}
', unsafe_allow_html=True)
st.markdown(f'MODEL: {model_type}
', unsafe_allow_html=True)
st.markdown(f'VIOLATION RATE: 0.0%
', unsafe_allow_html=True)
st.markdown("---")
st.markdown('github.com/clementina-tom/plrs
', unsafe_allow_html=True)
# ── Header ────────────────────────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Tabs ──────────────────────────────────────────────────────────────────────
tab1, tab2, tab3 = st.tabs(["RECOMMENDATIONS", "WHAT-IF SIMULATOR", "CURRICULUM MAP"])
ACTIVITY_TO_DOMAIN = {
"math": {
"oucontent": "algebraic_expressions", "forumng": "statistics_basic",
"homepage": "whole_numbers", "subpage": "plane_shapes",
"resource": "indices", "url": "number_bases",
"ouwiki": "proportion_variation", "glossary": "algebraic_factorization",
"quiz": "quadratic_equations",
},
"cs": {
"oucontent": "programming_concepts", "forumng": "ethics_technology",
"homepage": "computer_basics", "subpage": "html_basics",
"resource": "networking_fundamentals", "url": "internet_basics",
"ouwiki": "cloud_basics", "glossary": "intro_databases",
"quiz": "python_basics",
},
}
# ══════════════════════════════════════════════════════════════════════════════
# TAB 1 — RECOMMENDATIONS
# ══════════════════════════════════════════════════════════════════════════════
with tab1:
col_left, col_right = st.columns([1, 1.4], gap="large")
with col_left:
st.markdown('Learner Profile
', unsafe_allow_html=True)
mode = st.radio("Input mode", ["Manual sliders", "Simulate student"], horizontal=True, label_visibility="collapsed")
mastery_scores = {}
if mode == "Manual sliders":
for node in curriculum.nodes:
label = curriculum.label(node)
level = curriculum.level(node)
val = st.slider(
f"{label}",
0.0, 1.0, 0.0, 0.05,
key=f"mastery_{node}",
help=f"Level: {level}"
)
mastery_scores[node] = val
else:
seq_len = st.slider("Sequence length", 10, 200, 50)
seed = st.number_input("Student seed", 1, 9999, 42)
np.random.seed(int(seed))
activity_types = list(ACTIVITY_TO_DOMAIN[domain_key].keys())
activity_probs = [0.38, 0.20, 0.15, 0.10, 0.06, 0.04, 0.03, 0.02, 0.02]
mapping = ACTIVITY_TO_DOMAIN[domain_key]
# Use skill_encoder to simulate skills that actually exist in the mapping
if skill_encoder is not None:
available_skills = skill_encoder["skill_id"].tolist()
sim_skills = np.random.choice(available_skills, seq_len).tolist()
else:
n_skills = 5736
sim_skills = np.random.randint(0, n_skills, seq_len).tolist()
sim_corrects = np.random.randint(0, 2, seq_len).tolist()
topic_scores: dict = {}
# Map simulated skills back to topics using the CSV work you did
for skill_id in sim_skills:
# If we have the encoder, find the activity type
if skill_encoder is not None:
row = skill_encoder[skill_encoder["skill_id"] == skill_id]
if not row.empty:
act = row["activity_type"].values[0]
topic_id = mapping.get(act)
if topic_id and topic_id in curriculum.nodes:
# Generate a mastery signal based on frequency/success
score = topic_scores.get(topic_id, 0.1)
# Every time they see this topic, increase mastery slightly
topic_scores[topic_id] = min(1.0, score + np.random.random() * 0.2)
else:
# Fallback to the old simple mapping if CSV is missing
act_idx = skill_id % 100
cumulative = 0
thresholds = [int(p * 100) for p in activity_probs]
thresholds[-1] += 100 - sum(thresholds)
act = activity_types[-1]
for a, thresh in zip(activity_types, thresholds):
cumulative += thresh
if act_idx < cumulative:
act = a
break
topic_id = mapping.get(act)
if topic_id and topic_id in curriculum.nodes:
topic_scores[topic_id] = 0.5 + np.random.random() * 0.4
mastery_scores = {n: 0.0 for n in curriculum.nodes}
mastery_scores.update(topic_scores)
st.success(f"Simulated {seq_len} interactions → {len(topic_scores)} topics mapped")
if topic_scores:
st.markdown('Mapped Mastery Signal
', unsafe_allow_html=True)
for tid, score in sorted(topic_scores.items(), key=lambda x: -x[1]):
pct = int(score * 100)
color = "#22c55e" if score >= threshold else "#f59e0b" if score >= soft_threshold else "#ef4444"
st.markdown(f"""
{curriculum.label(tid)}
{pct}%
""", unsafe_allow_html=True)
run = st.button("⚡ Generate Recommendations", type="primary", use_container_width=True)
with col_right:
if run or mode == "Simulate student":
# Enable cascading for simulation to ensure prerequisites are also "mastered"
is_sim = (mode == "Simulate student")
results = pipeline.recommend_from_mastery(mastery_scores, cascade=is_sim)
summary = results["mastery_summary"]
stats = results["stats"]
mastery_pct = int(summary["mastery_rate"] * 100)
vrate_pct = int(stats["prerequisite_violation_rate"] * 100)
st.markdown(f"""
Mastered
{summary['mastered']}/{summary['total_topics']}
{mastery_pct}% rate
Approved
{stats['approved_count']}
ready to learn
Challenging
{stats['challenging_count']}
partial prereqs
Violation rate
{vrate_pct}%
blocked topics
""", unsafe_allow_html=True)
if results["approved"]:
st.markdown('✅ Approved Recommendations
', unsafe_allow_html=True)
for i, rec in enumerate(results["approved"]):
score_pct = int(rec["score"] * 100)
st.markdown(f"""
{i+1}. {rec['topic_label']}
score: {rec['score']:.3f} · mastery: {int(rec['mastery']*100)}% · unlocks: {rec['downstream_count']}
{rec['reasoning']}
""", unsafe_allow_html=True)
else:
st.info("No approved topics — lower the mastery threshold or set some mastery levels.")
if results["challenging"]:
st.markdown('⚠️ Challenging
', unsafe_allow_html=True)
for rec in results["challenging"]:
score_pct = int(rec["score"] * 100)
unmet = ", ".join(rec["unmet_prerequisites"]) or "—"
st.markdown(f"""
{rec['topic_label']}
score: {rec['score']:.3f} · strengthen: {unmet}
{rec['reasoning']}
""", unsafe_allow_html=True)
if results["vetoed"]:
with st.expander(f"❌ Vetoed topics ({stats['vetoed_count']} total — prerequisite check failed)"):
for rec in results["vetoed"]:
unmet = ", ".join(rec["unmet_prerequisites"]) or "—"
st.markdown(f"""
{rec['topic_label']}
blocked by: {unmet}
""", unsafe_allow_html=True)
else:
st.markdown("""
⚡
SET MASTERY LEVELS · THEN GENERATE
""", unsafe_allow_html=True)
# ══════════════════════════════════════════════════════════════════════════════
# TAB 2 — WHAT-IF SIMULATOR
# ══════════════════════════════════════════════════════════════════════════════
with tab2:
st.markdown('Prerequisite Impact Simulator
', unsafe_allow_html=True)
st.markdown('Select any topic to see what it unlocks and what currently blocks it.
', unsafe_allow_html=True)
node_options = {curriculum.label(n): n for n in curriculum.nodes}
selected_label = st.selectbox("Select topic", list(node_options.keys()))
selected_id = node_options[selected_label]
wi = pipeline.what_if(selected_id)
col_a, col_b = st.columns(2, gap="large")
with col_a:
st.markdown('🔓 What This Unlocks
', unsafe_allow_html=True)
if wi["direct_unlocks"]:
st.markdown("**Directly unlocks:**")
st.markdown("".join(f'{u["label"]}' for u in wi["direct_unlocks"]), unsafe_allow_html=True)
else:
st.markdown('Leaf node — no further topics.', unsafe_allow_html=True)
if wi["all_unlocks"]:
st.markdown(f"**All downstream ({wi['total_unlocked']}):**")
st.markdown("".join(f'{u["label"]}' for u in wi["all_unlocks"]), unsafe_allow_html=True)
st.markdown(f"""
Total Unlocked
{wi['total_unlocked']}
""", unsafe_allow_html=True)
with col_b:
st.markdown('🔒 What Blocks This
', unsafe_allow_html=True)
if wi["blocked_by"]:
st.markdown("**Prerequisites:**")
st.markdown("".join(f'{b["label"]}' for b in wi["blocked_by"]), unsafe_allow_html=True)
else:
st.markdown('Root topic — no prerequisites.', unsafe_allow_html=True)
# ══════════════════════════════════════════════════════════════════════════════
# TAB 3 — CURRICULUM MAP
# ══════════════════════════════════════════════════════════════════════════════
with tab3:
st.markdown('Curriculum Knowledge Graph
', unsafe_allow_html=True)
col_info, col_table = st.columns([1, 2], gap="large")
with col_info:
roots = [n for n in curriculum.nodes if not curriculum.prerequisites(n)]
leaves = [n for n in curriculum.nodes if not curriculum.successors(n)]
st.markdown(f"""
Domain
{curriculum.domain}
Topics
{curriculum.num_nodes}
Prerequisite Edges
{curriculum.num_edges}
""", unsafe_allow_html=True)
st.markdown('Root Topics
', unsafe_allow_html=True)
st.markdown("".join(f'{curriculum.label(r)}' for r in roots), unsafe_allow_html=True)
st.markdown('Leaf Topics
', unsafe_allow_html=True)
st.markdown("".join(f'{curriculum.label(l)}' for l in leaves), unsafe_allow_html=True)
with col_table:
import pandas as pd
st.markdown('All Topics
', unsafe_allow_html=True)
rows = []
for node in curriculum.nodes:
rows.append({
"Topic": curriculum.label(node),
"Level": curriculum.level(node),
"Prerequisites": len(curriculum.prerequisites(node)),
"Unlocks (direct)": len(curriculum.successors(node)),
"Total Downstream": len(curriculum.descendants(node)),
})
df = pd.DataFrame(rows).sort_values("Total Downstream", ascending=False)
st.dataframe(df, use_container_width=True, height=480, hide_index=True)