"""
app.py — SpectraQual Streamlit Dashboard (v3.0)
Updated to use the new SpectraQualEnv class with OpenEnv interface.
Features:
- Real-time stacked reward component charts
- Per-step accuracy / throughput display
- Action confidence from reward components
- Anomaly flag indicators
- Explainability: "Why this decision?"
"""
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
import streamlit as st
import matplotlib.pyplot as plt
import time
from env import SpectraQualEnv
from models import PCBAction
from config import (
COLOR_PRIMARY, COLOR_SUCCESS, COLOR_WARNING,
COLOR_DANGER, COLOR_BG, COLOR_CARD, COLOR_MUTED,
TASKS,
)
# ---------------------------
# PAGE CONFIG
# ---------------------------
st.set_page_config(
page_title="SpectraQual",
page_icon="⚔️",
layout="wide",
initial_sidebar_state="collapsed",
)
# ---------------------------
# GLOBAL STYLES
# ---------------------------
st.markdown("""
""", unsafe_allow_html=True)
# ---------------------------
# SESSION STATE
# ---------------------------
def _init_state():
if "env" not in st.session_state:
st.session_state.env = None
if "score" not in st.session_state:
st.session_state.score = 0.0
if "history" not in st.session_state:
st.session_state.history = [] # cumulative reward over time
if "running" not in st.session_state:
st.session_state.running = False
if "log" not in st.session_state:
st.session_state.log = [] # list of (pcb_obs, action, rc)
if "task_id" not in st.session_state:
st.session_state.task_id = "task_easy"
if "last_result" not in st.session_state:
st.session_state.last_result = None
if "episode_done" not in st.session_state:
st.session_state.episode_done = False
_init_state()
# ---------------------------
# HELPERS
# ---------------------------
def defect_badge(d):
m = {
"none": ("b-none", "✓ NONE"),
"missing_component": ("b-missing", "⚠ MISSING COMPONENT"),
"solder_bridge": ("b-solder", "⚡ SOLDER BRIDGE"),
"short_circuit": ("b-short", "✗ SHORT CIRCUIT"),
}
cls, label = m.get(d, ("b-none", d.upper()))
return f'{label}'
def reward_bar_html(label, score, color="#00e5ff"):
pct = int(score * 100)
return (
f''
f'
{label}'
f'
'
f'
{score:.2f}'
f'
'
'REAL-TIME QUALITY INTELLIGENCE ENGINE · v3.0 · OpenEnv Compliant
',
unsafe_allow_html=True,
)
# ---------------------------
# SIDEBAR TASK SELECTOR
# ---------------------------
with st.sidebar:
st.markdown("### 🎯 Task Selection")
task_choice = st.selectbox(
"Select Task",
options=list(TASKS.keys()),
format_func=lambda t: f"{t} ({TASKS[t]['difficulty'].upper()})",
index=list(TASKS.keys()).index(st.session_state.task_id),
)
if task_choice != st.session_state.task_id:
st.session_state.task_id = task_choice
st.session_state.env = None
st.session_state.score = 0.0
st.session_state.history = []
st.session_state.log = []
st.session_state.last_result = None
st.session_state.episode_done = False
cfg = TASKS[st.session_state.task_id]
st.markdown(f"""
**Boards:** {cfg['n_boards']}
**Slots:** {cfg['n_slots']}
**Seed:** {cfg['seed']}
**Anomaly Rate:** {cfg['anomaly_rate']:.0%}
**Difficulty:** {cfg['difficulty'].upper()}
""")
st.markdown("---")
speed = st.slider("⚡ Speed (s/step)", 0.2, 2.0, 0.8, step=0.1)
# ---------------------------
# SPEED (fallback if sidebar collapsed)
# ---------------------------
if "speed" not in dir():
speed = 0.8
st.markdown("
", unsafe_allow_html=True)
# ---------------------------
# METRICS BAR
# ---------------------------
env_obj = get_env()
state = env_obj.state()
m1, m2, m3, m4, m5 = st.columns(5)
m1.metric("💰 Cumul. Reward", f"{state['cumulative_reward']:.3f}")
m2.metric("🎯 Accuracy", f"{state['rolling_accuracy']:.1%}")
m3.metric("⚙️ Active Slots", sum(1 for s in state['slots'] if 0 < s < 9999))
m4.metric("🧠 Decisions", state['total_count'])
m5.metric("⚠️ Bottlenecks", state['bottleneck_count'])
last_r = round(st.session_state.log[-1][2].normalized, 3) if st.session_state.log else "N/A"
status_color = "#00e5ff" if st.session_state.log else "#1e4a5a"
st.markdown(f"""
🟢 TASK: {st.session_state.task_id.upper()} · LAST REWARD: {last_r} · STEPS: {state['step']}
""", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
# ---------------------------
# CONTROL BUTTONS
# ---------------------------
c1, c2, c3, c4, c5 = st.columns(5)
with c1:
if st.button("▶ RUN STEP"):
st.session_state.running = False
st.session_state.run_once = True
with c2:
if st.button("⚡ AUTO RUN"):
st.session_state.running = True
with c3:
if st.button("⛔ STOP"):
st.session_state.running = False
with c4:
if st.button("🔄 RESET"):
env_obj.reset()
st.session_state.score = 0.0
st.session_state.history = []
st.session_state.log = []
st.session_state.last_result = None
st.session_state.episode_done = False
with c5:
if st.button("🆕 NEW TASK"):
st.session_state.env = None
st.session_state.score = 0.0
st.session_state.history = []
st.session_state.log = []
st.session_state.last_result = None
st.session_state.episode_done = False
# ---------------------------
# CORE STEP
# ---------------------------
def run_step():
env = get_env()
# Initialize if needed
if env._done or env._current_pcb is None:
result = env.reset()
if result.done:
st.session_state.episode_done = True
return None
# Get current obs to determine action
obs = env._build_observation(*__import__("reward").detect_anomaly(env._current_pcb))
# Use rule-based decision (greedy heuristic)
from env import decide_action
pcb_dict = {
"defect_type": obs.defect_type,
"component_cost": obs.component_cost,
"criticality": obs.criticality,
}
action_str = decide_action(pcb_dict)
result = env.step(PCBAction(action=action_str))
rc = result.reward_components
st.session_state.score = env.state()["cumulative_reward"]
st.session_state.history.append(st.session_state.score)
st.session_state.log.append((result.observation, action_str, rc))
st.session_state.last_result = result
if result.done:
st.session_state.episode_done = True
return result
# ---------------------------
# DISPLAY
# ---------------------------
def display(result):
from collections import Counter
obs = result.observation
rc = result.reward_components
col1, col2 = st.columns(2, gap="large")
# ── LEFT ──
with col1:
st.subheader("PCB Info")
anomaly_html = ""
if obs.is_anomaly:
anomaly_html = f'⚠️ ANOMALY {obs.anomaly_score:.2f}'
st.markdown(f"""
Board ID
{obs.board_id}
Defect Type
{defect_badge(obs.defect_type)}
Component Cost
₹{obs.component_cost:.2f}
Criticality
{obs.criticality:.2f}
Anomaly
{anomaly_html if anomaly_html else 'Normal'}
""", unsafe_allow_html=True)
st.subheader("Decision")
action = st.session_state.log[-1][1] if st.session_state.log else "N/A"
if action == "PASS":
st.success(f"✅ {action}")
elif "ROUTE" in action:
st.warning(f"🛠️ {action}")
elif action == "WAIT":
st.warning("⏳ WAITING FOR SLOT AVAILABILITY")
else:
st.error(f"❌ {action}")
if rc:
st.subheader("🧠 Why this decision?")
explanation_parts = rc.explanation.split(" | ")
for part in explanation_parts[:3]:
st.info(part)
st.subheader("Step Reward")
r = result.reward
if r >= 0.6:
st.markdown(f'▲ {r:.4f}', unsafe_allow_html=True)
elif r >= 0.35:
st.markdown(f'● {r:.4f}', unsafe_allow_html=True)
else:
st.markdown(f'▼ {r:.4f}', unsafe_allow_html=True)
if rc:
st.subheader("📊 Reward Component Breakdown")
components = [
("Defect Handling", rc.defect_reward, "#00e5ff"),
("Cost Efficiency", rc.cost_efficiency, "#00e676"),
("Queue Mgmt", rc.queue_penalty, "#ffb700"),
("Risk Factor", rc.criticality_factor, "#ff7800"),
("Anomaly Bonus", rc.anomaly_bonus, "#ff00c8"),
]
bars_html = ""
for label, val, color in components:
bars_html += reward_bar_html(label, val, color)
st.markdown(bars_html, unsafe_allow_html=True)
st.subheader("Rolling Metrics")
sub1, sub2 = st.columns(2)
with sub1:
st.metric("🎯 Accuracy", f"{obs.rolling_accuracy:.1%}")
with sub2:
st.metric("⚡ Throughput", f"{obs.throughput:.2f}")
# ── RIGHT ──
with col2:
st.subheader("Factory Slots")
slot_html = ''
for i, slot in enumerate(obs.slots_state):
if slot == -1:
slot_html += (f'
'
f'
SLOT {i:02d} · LOCKED')
elif slot > 0:
slot_html += (f'
'
f'
SLOT {i:02d} · {slot}t')
else:
slot_html += (f'
')
slot_html += '
'
f'{st.session_state.score:.4f}
',
unsafe_allow_html=True,
)
st.subheader("📈 Reward Trend")
fig, ax = plt.subplots(figsize=(5.5, 3))
fig.patch.set_facecolor("#080c12")
ax.set_facecolor("#0a1420")
history = st.session_state.history
if history:
ax.plot(history, color="#00e5ff", linewidth=1.8,
marker='o', markersize=3.5,
markerfacecolor="#00e5ff", markeredgewidth=0)
ax.fill_between(range(len(history)), history, alpha=0.10, color="#00e5ff")
ax.axhline(y=0.6, color="#00e676", linewidth=0.8, linestyle="--", alpha=0.5, label="Success threshold")
ax.set_title("Cumulative Reward", color="#2e6a80", fontsize=9, pad=8)
ax.set_xlabel("Steps", color="#2e6a80", fontsize=8)
ax.set_ylabel("Score", color="#2e6a80", fontsize=8)
ax.set_ylim(0, max(max(history, default=1.0) * 1.1, 1.0))
ax.tick_params(colors="#2e6a80", labelsize=7)
ax.grid(color="#0d2535", linewidth=0.7, linestyle="--")
for spine in ax.spines.values():
spine.set_edgecolor("#0d2535")
fig.tight_layout(pad=1.2)
st.pyplot(fig)
plt.close(fig)
# Stacked Reward Components Over Time
if len(st.session_state.log) >= 2:
st.subheader("📊 Component Breakdown Over Time")
steps_data = st.session_state.log[-20:] # last 20 steps
comp_labels = ["Defect", "Cost", "Queue", "Risk", "Anomaly"]
comp_colors = ["#00e5ff", "#00e676", "#ffb700", "#ff7800", "#ff00c8"]
comp_data = {l: [] for l in comp_labels}
for _, _, rc_entry in steps_data:
if rc_entry:
comp_data["Defect"].append(rc_entry.defect_reward)
comp_data["Cost"].append(rc_entry.cost_efficiency)
comp_data["Queue"].append(rc_entry.queue_penalty)
comp_data["Risk"].append(rc_entry.criticality_factor)
comp_data["Anomaly"].append(rc_entry.anomaly_bonus)
if any(comp_data.values()):
fig2, ax2 = plt.subplots(figsize=(5.5, 2.8))
fig2.patch.set_facecolor("#080c12")
ax2.set_facecolor("#0a1420")
x = list(range(len(next(iter(comp_data.values())))))
bottom = [0.0] * len(x)
for label, color in zip(comp_labels, comp_colors):
vals = comp_data[label]
if vals and len(vals) == len(x):
# Normalize each component's contribution by weight
ax2.fill_between(x, bottom,
[b + v * 0.2 for b, v in zip(bottom, vals)],
alpha=0.6, color=color, label=label)
bottom = [b + v * 0.2 for b, v in zip(bottom, vals)]
ax2.set_title("Reward Components (last 20 steps)", color="#2e6a80", fontsize=8, pad=6)
ax2.set_xlabel("Steps", color="#2e6a80", fontsize=7)
ax2.tick_params(colors="#2e6a80", labelsize=6)
ax2.grid(color="#0d2535", linewidth=0.5, linestyle="--")
for spine in ax2.spines.values():
spine.set_edgecolor("#0d2535")
ax2.legend(loc="upper right", fontsize=6,
facecolor="#080c12", edgecolor="#2e6a80", labelcolor="#c9d4e0")
fig2.tight_layout(pad=1.0)
st.pyplot(fig2)
plt.close(fig2)
# Decision Distribution
if st.session_state.log:
st.subheader("📊 Decision Distribution")
decisions = [entry[1] for entry in st.session_state.log]
from collections import Counter
counts = dict(Counter(decisions))
st.bar_chart(counts)
# Episode Done banner
if st.session_state.episode_done:
final = st.session_state.score
if final >= 0.6:
st.success(f"🏆 EPISODE COMPLETE — Score: {final:.4f} — SUCCESS!")
else:
st.warning(f"⚠️ EPISODE COMPLETE — Score: {final:.4f} — Below success threshold (0.60)")
# ---------------------------
# EXECUTION
# ---------------------------
if "run_once" in st.session_state and st.session_state.run_once:
result = run_step()
if result:
display(result)
st.session_state.run_once = False
elif st.session_state.running:
placeholder = st.empty()
for _ in range(1000):
if not st.session_state.running:
break
if st.session_state.episode_done:
st.session_state.running = False
break
result = run_step()
if result:
with placeholder.container():
display(result)
time.sleep(speed)
elif st.session_state.last_result:
display(st.session_state.last_result)
else:
st.markdown("""
[ SYSTEM IDLE ]
SELECT A TASK IN THE SIDEBAR · PRESS ▶ RUN STEP OR ⚡ AUTO RUN TO BEGIN
""", unsafe_allow_html=True)