Update app.py

app.py

@@ -1,49 +1,30 @@
 import ast
 import logging
 import re
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List
 
 import gradio as gr
 import matplotlib.pyplot as plt
 import numpy as np
 import pandas as pd
 from datasets import load_dataset
-from sklearn.ensemble import ExtraTreesClassifier
-from sklearn.impute import SimpleImputer
-from sklearn.metrics import accuracy_score, classification_report, confusion_matrix, f1_score
 from sklearn.model_selection import train_test_split
-from sklearn.pipeline import Pipeline
-from sklearn.preprocessing import StandardScaler
+from sklearn.ensemble import HistGradientBoostingClassifier
+from sklearn.metrics import classification_report, confusion_matrix
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 APP_TITLE = "Noise Detection"
-APP_SUBTITLE = (
-    "Classify quantum circuits into clean, depolarizing, amplitude_damping, or hardware-aware noise conditions."
-)
+APP_SUBTITLE = "Classify circuits by noise type: clean, depolarizing, amplitude_damping, hardware_aware."
 
 REPO_CONFIG = {
-    "clean": {
-        "label": "clean",
-        "repo": "QSBench/QSBench-Core-v1.0.0-demo",
-    },
-    "depolarizing": {
-        "label": "depolarizing",
-        "repo": "QSBench/QSBench-Depolarizing-Demo-v1.0.0",
-    },
-    "amplitude_damping": {
-        "label": "amplitude_damping",
-        "repo": "QSBench/QSBench-Amplitude-v1.0.0-demo",
-    },
-    "hardware_aware": {
-        "label": "hardware_aware",
-        "repo": "QSBench/QSBench-Transpilation-v1.0.0-demo",
-    },
+    "Core (Clean)": "QSBench/QSBench-Core-v1.0.0-demo",
+    "Depolarizing Noise": "QSBench/QSBench-Depolarizing-Demo-v1.0.0",
+    "Amplitude Damping": "QSBench/QSBench-Amplitude-v1.0.0-demo",
+    "Hardware-Aware Noise": "QSBench/QSBench-Transpilation-v1.0.0-demo",
 }
 
-CLASS_ORDER = ["clean", "depolarizing", "amplitude_damping", "hardware_aware"]
-
 NON_FEATURE_COLS = {
     "sample_id",
     "sample_seed",
@@ -61,20 +42,24 @@ NON_FEATURE_COLS = {
     "backend_device",
     "precision_mode",
     "circuit_signature",
-    "entanglement",
-    "meyer_wallach",
-    "cx_count",
-    "noise_label",
 }
 
-SOFT_EXCLUDE_PATTERNS = ["ideal_", "noisy_", "error_", "sign_ideal_", "sign_noisy_"]
+_SOFT_EXCLUDE_PATTERNS = ["ideal_", "noisy_", "error_", "sign_ideal_", "sign_noisy_"]
 
 _ASSET_CACHE: Dict[str, pd.DataFrame] = {}
-_COMBINED_CACHE: Optional[pd.DataFrame] = None
+
+
+def load_dataset_df(dataset_key: str) -> pd.DataFrame:
+    if dataset_key not in _ASSET_CACHE:
+        ds = load_dataset(REPO_CONFIG[dataset_key])
+        df = pd.DataFrame(ds["train"])
+        df = enrich_dataframe(df)
+        df["noise_label"] = dataset_key
+        _ASSET_CACHE[dataset_key] = df
+    return _ASSET_CACHE[dataset_key]
 
 
 def safe_parse(value):
-    """Safely parse stringified Python literals."""
     if isinstance(value, str):
         try:
             return ast.literal_eval(value)
@@ -84,15 +69,9 @@ def safe_parse(value):
 
 
 def adjacency_features(adj_value) -> Dict[str, float]:
-    """Derive graph statistics from an adjacency matrix."""
     parsed = safe_parse(adj_value)
     if not isinstance(parsed, list) or len(parsed) == 0:
-        return {
-            "adj_edge_count": np.nan,
-            "adj_density": np.nan,
-            "adj_degree_mean": np.nan,
-            "adj_degree_std": np.nan,
-        }
+        return {"adj_edge_count": np.nan, "adj_density": np.nan, "adj_degree_mean": np.nan, "adj_degree_std": np.nan}
 
     try:
         arr = np.array(parsed, dtype=float)
@@ -108,32 +87,17 @@ def adjacency_features(adj_value) -> Dict[str, float]:
             "adj_degree_std": float(np.std(degrees)),
         }
     except Exception:
-        return {
-            "adj_edge_count": np.nan,
-            "adj_density": np.nan,
-            "adj_degree_mean": np.nan,
-            "adj_degree_std": np.nan,
-        }
+        return {"adj_edge_count": np.nan, "adj_density": np.nan, "adj_degree_mean": np.nan, "adj_degree_std": np.nan}
 
 
 def qasm_features(qasm_value) -> Dict[str, float]:
-    """Extract lightweight text statistics from QASM."""
     if not isinstance(qasm_value, str) or not qasm_value.strip():
-        return {
-            "qasm_length": np.nan,
-            "qasm_line_count": np.nan,
-            "qasm_gate_keyword_count": np.nan,
-            "qasm_measure_count": np.nan,
-            "qasm_comment_count": np.nan,
-        }
+        return {"qasm_length": np.nan, "qasm_line_count": np.nan, "qasm_gate_keyword_count": np.nan,
+                "qasm_measure_count": np.nan, "qasm_comment_count": np.nan}
 
     text = qasm_value
     lines = [line for line in text.splitlines() if line.strip()]
-    gate_keywords = re.findall(
-        r"\b(cx|h|x|y|z|rx|ry|rz|u1|u2|u3|u|swap|cz|ccx|rxx|ryy|rzz)\b",
-        text,
-        flags=re.IGNORECASE,
-    )
+    gate_keywords = re.findall(r"\b(cx|h|x|y|z|rx|ry|rz|u1|u2|u3|u|swap|cz|ccx|rxx|ryy|rzz)\b", text, flags=re.IGNORECASE)
     measure_count = len(re.findall(r"\bmeasure\b", text, flags=re.IGNORECASE))
     comment_count = sum(1 for line in lines if line.strip().startswith("//"))
 
@@ -147,9 +111,7 @@ def qasm_features(qasm_value) -> Dict[str, float]:
 
 
 def enrich_dataframe(df: pd.DataFrame) -> pd.DataFrame:
-    """Add derived numeric features for classification."""
     df = df.copy()
-
     if "adjacency" in df.columns:
         adj_df = df["adjacency"].apply(adjacency_features).apply(pd.Series)
         df = pd.concat([df, adj_df], axis=1)
@@ -158,274 +120,60 @@ def enrich_dataframe(df: pd.DataFrame) -> pd.DataFrame:
     if qasm_source in df.columns:
         qasm_df = df[qasm_source].apply(qasm_features).apply(pd.Series)
         df = pd.concat([df, qasm_df], axis=1)
-
     return df
 
 
-def load_single_dataset(dataset_key: str) -> pd.DataFrame:
-    """Load a dataset shard from Hugging Face and cache it in memory."""
-    if dataset_key not in _ASSET_CACHE:
-        logger.info("Loading dataset: %s", dataset_key)
-        ds = load_dataset(REPO_CONFIG[dataset_key]["repo"])
-        df = pd.DataFrame(ds["train"])
-        df = enrich_dataframe(df)
-        df["noise_label"] = REPO_CONFIG[dataset_key]["label"]
-        _ASSET_CACHE[dataset_key] = df
-    return _ASSET_CACHE[dataset_key]
-
-
-def load_combined_dataset() -> pd.DataFrame:
-    """Load and merge all four noise-condition datasets."""
-    global _COMBINED_CACHE
-    if _COMBINED_CACHE is None:
-        frames = [load_single_dataset(key) for key in REPO_CONFIG.keys()]
-        combined = pd.concat(frames, ignore_index=True)
-        combined = combined[combined["noise_label"].isin(CLASS_ORDER)].copy()
-        _COMBINED_CACHE = combined
-    return _COMBINED_CACHE
-
-
-def load_guide_content() -> str:
-    """Load the markdown guide if it exists."""
-    try:
-        with open("GUIDE.md", "r", encoding="utf-8") as f:
-            return f.read()
-    except FileNotFoundError:
-        return "# Guide\n\nGuide file not found."
-
-
 def get_available_feature_columns(df: pd.DataFrame) -> List[str]:
-    """Return numeric feature columns excluding metadata and target columns."""
     numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
     features = []
     for col in numeric_cols:
         if col in NON_FEATURE_COLS:
             continue
-        if any(pattern in col for pattern in SOFT_EXCLUDE_PATTERNS):
+        if any(pattern in col for pattern in _SOFT_EXCLUDE_PATTERNS):
             continue
         features.append(col)
     return sorted(features)
 
 
 def default_feature_selection(features: List[str]) -> List[str]:
-    """Select a stable default feature subset."""
-    preferred = [
-        "gate_entropy",
-        "adj_density",
-        "adj_degree_mean",
-        "adj_degree_std",
-        "depth",
-        "total_gates",
-        "single_qubit_gates",
-        "two_qubit_gates",
-        "cx_count",
-        "qasm_length",
-        "qasm_line_count",
-        "qasm_gate_keyword_count",
-    ]
-    selected = [feature for feature in preferred if feature in features]
-    return selected[:8] if selected else features[:8]
-
-
-def make_classification_figure(
-    y_true: np.ndarray,
-    y_pred: np.ndarray,
-    class_names: List[str],
-    feature_names: Optional[List[str]] = None,
-    importances: Optional[np.ndarray] = None,
-) -> plt.Figure:
-    """Create a compact classification summary figure."""
-    fig = plt.figure(figsize=(20, 6))
-    gs = fig.add_gridspec(1, 3)
-
-    ax1 = fig.add_subplot(gs[0, 0])
-    ax2 = fig.add_subplot(gs[0, 1])
-    ax3 = fig.add_subplot(gs[0, 2])
-
-    cm = confusion_matrix(y_true, y_pred, labels=class_names)
-    image = ax1.imshow(cm, interpolation="nearest")
-    ax1.set_title("Confusion Matrix")
-    ax1.set_xlabel("Predicted")
-    ax1.set_ylabel("Actual")
-    ax1.set_xticks(np.arange(len(class_names)))
-    ax1.set_yticks(np.arange(len(class_names)))
-    ax1.set_xticklabels(class_names, rotation=45, ha="right")
-    ax1.set_yticklabels(class_names)
-    for i in range(cm.shape[0]):
-        for j in range(cm.shape[1]):
-            ax1.text(j, i, cm[i, j], ha="center", va="center")
-    fig.colorbar(image, ax=ax1, fraction=0.046, pad=0.04)
-
-    incorrect = (y_true != y_pred).astype(int)
-    ax2.hist(incorrect, bins=[-0.5, 0.5, 1.5])
-    ax2.set_title("Correct vs Incorrect")
-    ax2.set_xlabel("0 = Correct, 1 = Incorrect")
-    ax2.set_ylabel("Count")
-
-    if importances is not None and feature_names is not None and len(importances) == len(feature_names):
-        idx = np.argsort(importances)[-10:]
-        ax3.barh([feature_names[i] for i in idx], importances[idx])
-        ax3.set_title("Top-10 Feature Importances")
-        ax3.set_xlabel("Importance")
-    else:
-        ax3.text(0.5, 0.5, "Feature importances are unavailable.", ha="center", va="center")
-        ax3.set_axis_off()
-
-    fig.tight_layout()
-    return fig
-
-
-def build_dataset_profile(df: pd.DataFrame) -> str:
-    """Build a short dataset summary for the explorer tab."""
-    return (
-        f"### Dataset profile\n\n"
-        f"**Rows:** {len(df):,}  \n"
-        f"**Columns:** {len(df.columns):,}  \n"
-        f"**Classes:** {', '.join(CLASS_ORDER)}"
-    )
+    preferred = ["gate_entropy", "adj_density", "adj_degree_mean", "adj_degree_std",
+                 "depth", "total_gates", "cx_count", "qasm_length"]
+    return [f for f in preferred if f in features]
 
 
-def refresh_explorer(dataset_key: str, split_name: str) -> Tuple[gr.update, pd.DataFrame, str, str, str, str]:
-    """Refresh the explorer view for the selected source dataset."""
-    df = load_single_dataset(dataset_key)
-    splits = df["split"].dropna().unique().tolist() if "split" in df.columns else ["train"]
-    if not splits:
-        splits = ["train"]
-
-    if split_name not in splits:
-        split_name = splits[0]
-
-    filtered = df[df["split"] == split_name] if "split" in df.columns else df
-    display_df = filtered.head(12).copy()
-
-    raw_qasm = display_df["qasm_raw"].iloc[0] if "qasm_raw" in display_df.columns and not display_df.empty else "// N/A"
-    transpiled_qasm = display_df["qasm_transpiled"].iloc[0] if "qasm_transpiled" in display_df.columns and not display_df.empty else "// N/A"
-
-    profile_box = build_dataset_profile(df)
-    summary_box = (
-        f"### Split summary\n\n"
-        f"**Dataset:** `{dataset_key}`  \n"
-        f"**Label:** `{REPO_CONFIG[dataset_key]['label']}`  \n"
-        f"**Available splits:** {', '.join(splits)}  \n"
-        f"**Preview rows:** {len(display_df)}"
-    )
-
-    return (
-        gr.update(choices=splits, value=split_name),
-        display_df,
-        raw_qasm,
-        transpiled_qasm,
-        profile_box,
-        summary_box,
-    )
-
-
-def sync_feature_picker(_dataset_key: str) -> gr.update:
-    """Refresh the feature list from the combined dataset."""
-    df = load_combined_dataset()
-    features = get_available_feature_columns(df)
-    defaults = default_feature_selection(features)
-    return gr.update(choices=features, value=defaults)
-
-
-def train_classifier(
-    feature_columns: List[str],
-    test_size: float,
-    n_estimators: int,
-    max_depth: float,
-    random_state: float,
-) -> Tuple[Optional[plt.Figure], str]:
-    """Train a four-class classifier and return metrics plus a plot."""
+def train_classifier(dataset_keys, feature_columns, test_size, seed):
     if not feature_columns:
-        return None, "### ❌ Please select at least one feature."
-
-    df = load_combined_dataset()
-    required_cols = feature_columns + ["noise_label"]
-    train_df = df.dropna(subset=required_cols).copy()
-    train_df = train_df[train_df["noise_label"].isin(CLASS_ORDER)]
+        return None, "No features selected"
 
-    if len(train_df) < 20:
-        return None, "### ❌ Not enough clean rows after filtering missing values."
+    dfs = [load_dataset_df(k) for k in dataset_keys]
+    df = pd.concat(dfs, axis=0, ignore_index=True)
+    df = df.dropna(subset=feature_columns + ["noise_label"])
 
-    X = train_df[feature_columns]
-    y = train_df["noise_label"]
+    X = df[feature_columns]
+    y = df["noise_label"]
 
-    seed = int(random_state)
-    depth = int(max_depth) if max_depth and int(max_depth) > 0 else None
-    trees = int(n_estimators)
+    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=int(seed), stratify=y)
 
-    try:
-        X_train, X_test, y_train, y_test = train_test_split(
-            X,
-            y,
-            test_size=test_size,
-            random_state=seed,
-            stratify=y,
-        )
-    except ValueError:
-        X_train, X_test, y_train, y_test = train_test_split(
-            X,
-            y,
-            test_size=test_size,
-            random_state=seed,
-        )
-
-    model = Pipeline(
-        steps=[
-            ("imputer", SimpleImputer(strategy="median")),
-            ("scaler", StandardScaler()),
-            (
-                "classifier",
-                ExtraTreesClassifier(
-                    n_estimators=trees,
-                    max_depth=depth,
-                    random_state=seed,
-                    n_jobs=-1,
-                    class_weight="balanced",
-                    min_samples_leaf=1,
-                ),
-            ),
-        ]
+    model = HistGradientBoostingClassifier(
+        learning_rate=0.05,
+        max_iter=200,
+        max_depth=5,
+        min_samples_leaf=10,
+        l2_regularization=0.1,
+        class_weight="balanced",
+        random_state=int(seed),
     )
-
     model.fit(X_train, y_train)
-    y_pred = model.predict(X_test)
+    preds = model.predict(X_test)
 
-    accuracy = float(accuracy_score(y_test, y_pred))
-    macro_f1 = float(f1_score(y_test, y_pred, average="macro", zero_division=0))
-    weighted_f1 = float(f1_score(y_test, y_pred, average="weighted", zero_division=0))
+    report = classification_report(y_test, preds, output_dict=False)
+    cm = confusion_matrix(y_test, preds)
 
-    classifier = model.named_steps["classifier"]
-    importances = getattr(classifier, "feature_importances_", None)
-    fig = make_classification_figure(y_test.to_numpy(), y_pred, CLASS_ORDER, list(feature_columns), importances)
-
-    report = classification_report(
-        y_test,
-        y_pred,
-        labels=CLASS_ORDER,
-        zero_division=0,
-    )
-    results = (
-        "### Classification results\n\n"
-        f"**Rows used:** {len(train_df):,}  \n"
-        f"**Test size:** {test_size:.0%}  \n"
-        f"**Accuracy:** {accuracy:.4f}  \n"
-        f"**Macro F1:** {macro_f1:.4f}  \n"
-        f"**Weighted F1:** {weighted_f1:.4f}\n\n"
-        "```text\n"
-        f"{report}"
-        "```"
-    )
-    return fig, results
+    return report, cm.tolist()
 
 
 CUSTOM_CSS = """
-.gradio-container {
-    max-width: 1400px !important;
-}
-footer {
-    margin-top: 1rem;
-}
+.gradio-container {max-width: 1400px !important;}
 """
 
 with gr.Blocks(title=APP_TITLE) as demo:
@@ -433,38 +181,16 @@ with gr.Blocks(title=APP_TITLE) as demo:
     gr.Markdown(APP_SUBTITLE)
 
     with gr.Tabs():
-        with gr.TabItem("🔎 Explorer"):
-            dataset_dropdown = gr.Dropdown(
-                list(REPO_CONFIG.keys()),
-                value="clean",
-                label="Dataset",
-            )
-            split_dropdown = gr.Dropdown(
-                ["train"],
-                value="train",
-                label="Split",
-            )
-
-            profile_box = gr.Markdown(value="### Loading dataset...")
-            summary_box = gr.Markdown(value="### Loading split summary...")
-            explorer_df = gr.Dataframe(label="Preview", interactive=False)
-
-            with gr.Row():
-                raw_qasm = gr.Code(label="Raw QASM", language=None)
-                transpiled_qasm = gr.Code(label="Transpiled QASM", language=None)
-
         with gr.TabItem("🧠 Classification"):
-            feature_picker = gr.CheckboxGroup(label="Input features", choices=[])
-            test_size = gr.Slider(0.1, 0.4, value=0.2, step=0.05, label="Test split")
-            n_estimators = gr.Slider(50, 400, value=200, step=10, label="Trees")
-            max_depth = gr.Slider(1, 30, value=12, step=1, label="Max depth")
-            seed = gr.Number(value=42, precision=0, label="Random seed")
+            dataset_dropdown = gr.CheckboxGroup(list(REPO_CONFIG.keys()), value=list(REPO_CONFIG.keys()), label="Datasets")
+            feature_picker = gr.CheckboxGroup(label="Input features")
+
+            test_size = gr.Slider(0.1, 0.5, value=0.2, step=0.05, label="Test split")
+            seed = gr.Number(value=42, label="Random seed")
             run_btn = gr.Button("Train & Evaluate", variant="primary")
-            plot = gr.Plot()
-            metrics = gr.Markdown()
 
-        with gr.TabItem("📖 Guide"):
-            gr.Markdown(load_guide_content())
+            metrics = gr.Markdown()
+            cm_plot = gr.Plot()
 
     gr.Markdown("---")
     gr.Markdown(
@@ -475,32 +201,16 @@ with gr.Blocks(title=APP_TITLE) as demo:
     )
 
     dataset_dropdown.change(
-        refresh_explorer,
-        [dataset_dropdown, split_dropdown],
-        [split_dropdown, explorer_df, raw_qasm, transpiled_qasm, profile_box, summary_box],
-    )
-
-    split_dropdown.change(
-        refresh_explorer,
-        [dataset_dropdown, split_dropdown],
-        [split_dropdown, explorer_df, raw_qasm, transpiled_qasm, profile_box, summary_box],
+        lambda datasets: gr.update(choices=get_available_feature_columns(pd.concat([load_dataset_df(k) for k in datasets]))),
+        [dataset_dropdown],
+        [feature_picker]
     )
 
-    dataset_dropdown.change(sync_feature_picker, [dataset_dropdown], [feature_picker])
-
     run_btn.click(
         train_classifier,
-        [feature_picker, test_size, n_estimators, max_depth, seed],
-        [plot, metrics],
+        [dataset_dropdown, feature_picker, test_size, seed],
+        [metrics, cm_plot]
     )
 
-    demo.load(
-        refresh_explorer,
-        [dataset_dropdown, split_dropdown],
-        [split_dropdown, explorer_df, raw_qasm, transpiled_qasm, profile_box, summary_box],
-    )
-    demo.load(sync_feature_picker, [dataset_dropdown], [feature_picker])
-
-
 if __name__ == "__main__":
-    demo.launch(theme=gr.themes.Soft(), css=CUSTOM_CSS)
+    demo.launch(theme=gr.themes.Soft(), css=CUSTOM_CSS)