"""
ParticleNet — Graph Neural Network for Particle Collision Event Classification
==============================================================================
Author : Your Name (edit this before pushing to Hugging Face)
Project : AI + Physics Portfolio — Project 7
Dataset : Synthetic CERN-style jet data (self-generated, no download needed)
Model : 3-layer Graph Convolutional Network (GCN) built with PyTorch Geometric
Demo : Hugging Face Spaces · Gradio interface
Physics context
---------------
At colliders like the LHC at CERN, protons smash together millions of times per
second, producing sprays of particles called *jets*. Identifying what kind of
particle initiated a jet — a quark, gluon, W boson, top quark, or Higgs boson —
is a fundamental task in particle physics and a perfect graph learning problem:
each jet is naturally a graph where particles are nodes and their proximity in
momentum space defines the edges.
This demo trains a small GCN on synthetic data that mimics real jet substructure
features (pT, eta, phi, charge, particle ID), then lets you generate a random
event and watch the model classify it in real time, with full visual explanation.
"""
import gradio as gr
import torch
import torch.nn.functional as F
from torch_geometric.nn import GCNConv, global_mean_pool
from torch_geometric.data import Data, Batch
import numpy as np
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
import os, json, time
# ─────────────────────────────────────────────────────────────
# 0. CONSTANTS — physics-inspired class labels
# ─────────────────────────────────────────────────────────────
CLASS_NAMES = ["Gluon jet", "Light-quark jet", "W boson jet",
"Top quark jet", "Higgs boson jet"]
CLASS_COLORS = ["#378ADD", "#1D9E75", "#EF9F27", "#D85A30", "#7F77DD"]
NUM_CLASSES = len(CLASS_NAMES)
# Node feature names — each particle in the jet carries these 5 features
FEATURE_NAMES = ["transverse momentum (pT)",
"pseudorapidity (η)",
"azimuthal angle (φ)",
"electric charge",
"particle type ID"]
# ─────────────────────────────────────────────────────────────
# 1. SYNTHETIC DATA GENERATOR
# Produces CERN-style jet graphs with realistic feature
# distributions for each class. No internet required.
# ─────────────────────────────────────────────────────────────
def generate_jet(label: int, seed: int | None = None) -> Data:
"""
Generate one synthetic jet graph.
Each jet has between 8 and 24 constituent particles (nodes).
Edges connect every particle to its 3 nearest neighbours in
(η, φ) space — this is how real jet algorithms work.
Feature distributions are loosely inspired by particle physics:
- Gluon jets: many soft particles, wide angular spread
- Quark jets: fewer, harder particles
- W jets: two sub-clusters (W → qq decay signature)
- Top jets: three sub-clusters (t → bqq)
- Higgs jets: two sub-clusters with b-quark enrichment
"""
rng = np.random.default_rng(seed)
# --- number of particles varies by jet type ---
n_particles_map = {0: (12, 24), 1: (8, 18), 2: (10, 20),
3: (14, 24), 4: (10, 20)}
lo, hi = n_particles_map[label]
n = rng.integers(lo, hi + 1)
# --- pT spectrum: power-law (harder for quarks/bosons) ---
alpha = {0: 3.5, 1: 2.8, 2: 2.5, 3: 2.2, 4: 2.4}[label]
pt = rng.pareto(alpha, n) * 10 + 1 # GeV, >1
# --- angular spread in (eta, phi) ---
spread = {0: 0.5, 1: 0.3, 2: 0.25, 3: 0.35, 4: 0.28}[label]
if label in (2, 3, 4):
# Multi-prong: split particles into sub-clusters
n_prongs = {2: 2, 3: 3, 4: 2}[label]
# Place cluster centres
centres_eta = rng.uniform(-spread, spread, n_prongs)
centres_phi = rng.uniform(-spread, spread, n_prongs)
assign = rng.integers(0, n_prongs, n)
eta = centres_eta[assign] + rng.normal(0, spread / 3, n)
phi = centres_phi[assign] + rng.normal(0, spread / 3, n)
else:
eta = rng.normal(0, spread, n)
phi = rng.normal(0, spread, n)
# --- charge: mostly neutral for gluons, mix for others ---
charge_prob = {0: 0.2, 1: 0.45, 2: 0.5, 3: 0.55, 4: 0.4}[label]
charge = rng.choice([-1, 0, 1], n,
p=[charge_prob / 2, 1 - charge_prob, charge_prob / 2])
# --- particle type ID: 0=photon,1=neutral hadron,2=charged hadron,3=electron,4=muon ---
pid_probs = {
0: [0.15, 0.40, 0.35, 0.07, 0.03],
1: [0.10, 0.25, 0.50, 0.10, 0.05],
2: [0.08, 0.20, 0.55, 0.12, 0.05],
3: [0.05, 0.15, 0.60, 0.12, 0.08],
4: [0.12, 0.28, 0.48, 0.09, 0.03],
}
pid = rng.choice(5, n, p=pid_probs[label])
# --- build node feature matrix (n × 5) ---
# Normalise to roughly [-1, 1] so the GCN trains easily
pt_norm = np.log1p(pt) / 5.0 # log scale for pT
eta_norm = eta / 1.0
phi_norm = phi / np.pi
chg_norm = charge.astype(float)
pid_norm = pid.astype(float) / 4.0 # [0,1]
x = torch.tensor(
np.stack([pt_norm, eta_norm, phi_norm, chg_norm, pid_norm], axis=1),
dtype=torch.float
)
# --- k-NN graph in (eta, phi) space, k=3 ---
coords = np.stack([eta, phi], axis=1)
from sklearn.neighbors import NearestNeighbors
k = min(3, n - 1)
nbrs = NearestNeighbors(n_neighbors=k + 1).fit(coords)
_, indices = nbrs.kneighbors(coords)
src, dst = [], []
for i, neighbours in enumerate(indices):
for j in neighbours[1:]: # skip self
src.append(i); dst.append(j)
src.append(j); dst.append(i) # undirected
edge_index = torch.tensor([src, dst], dtype=torch.long)
y = torch.tensor([label], dtype=torch.long)
# Store raw coords for visualisation
data = Data(x=x, edge_index=edge_index, y=y)
data.pt = torch.tensor(pt, dtype=torch.float)
data.eta = torch.tensor(eta, dtype=torch.float)
data.phi = torch.tensor(phi, dtype=torch.float)
return data
def generate_dataset(n_per_class: int = 200, seed: int = 42) -> list[Data]:
"""Create a balanced training set with n_per_class jets per category."""
dataset = []
for label in range(NUM_CLASSES):
for i in range(n_per_class):
dataset.append(generate_jet(label, seed=seed * 1000 + label * 100 + i))
rng = np.random.default_rng(seed)
perm = rng.permutation(len(dataset))
return [dataset[i] for i in perm]
# ─────────────────────────────────────────────────────────────
# 2. GCN MODEL
# Three graph convolutional layers followed by global mean
# pooling and a linear classifier head.
# ─────────────────────────────────────────────────────────────
class ParticleGCN(torch.nn.Module):
"""
A 3-layer Graph Convolutional Network for jet classification.
Architecture
------------
Input (5 features per particle node)
→ GCNConv(5 → 64) + ReLU + Dropout(0.2)
→ GCNConv(64 → 128) + ReLU + Dropout(0.2)
→ GCNConv(128 → 64) + ReLU
→ GlobalMeanPool (aggregate all particles into one jet vector)
→ Linear(64 → 32) + ReLU
→ Linear(32 → 5) (one logit per class)
Why GCNs for physics?
---------------------
Unlike CNNs (which need a grid) or RNNs (which need a sequence),
GCNs operate on arbitrary graphs — perfect for jets where the
number of particles varies and their spatial relationships matter.
The message-passing mechanism lets each particle "talk to" its
neighbours, building up a representation of local jet substructure
before the global pool summarises the whole event.
"""
def __init__(self, in_channels: int = 5, hidden: int = 64,
out_channels: int = NUM_CLASSES):
super().__init__()
self.conv1 = GCNConv(in_channels, hidden)
self.conv2 = GCNConv(hidden, hidden * 2)
self.conv3 = GCNConv(hidden * 2, hidden)
self.lin1 = torch.nn.Linear(hidden, 32)
self.lin2 = torch.nn.Linear(32, out_channels)
self.drop = torch.nn.Dropout(p=0.2)
def forward(self, x, edge_index, batch):
# Message passing through the jet graph
x = F.relu(self.conv1(x, edge_index))
x = self.drop(x)
x = F.relu(self.conv2(x, edge_index))
x = self.drop(x)
x = F.relu(self.conv3(x, edge_index))
# Pool all particle embeddings into a single jet embedding
x = global_mean_pool(x, batch)
# Classification head
x = F.relu(self.lin1(x))
x = self.lin2(x)
return x
# ─────────────────────────────────────────────────────────────
# 3. TRAINING
# Runs once at startup; saves model to disk for reuse.
# ─────────────────────────────────────────────────────────────
MODEL_PATH = "particlenet_gcn.pt"
HISTORY_PATH = "training_history.json"
def train_model(n_per_class: int = 300, epochs: int = 60,
lr: float = 1e-3) -> tuple[ParticleGCN, dict]:
"""Train the GCN on synthetic jet data and return model + history."""
print("Generating synthetic jet dataset...")
dataset = generate_dataset(n_per_class=n_per_class)
# 80/20 train-val split
split = int(0.8 * len(dataset))
train_data, val_data = dataset[:split], dataset[split:]
def make_batch(subset):
return Batch.from_data_list(subset)
device = torch.device("cpu") # CPU is fine for this model size
model = ParticleGCN().to(device)
opt = torch.optim.Adam(model.parameters(), lr=lr, weight_decay=1e-4)
sched = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=epochs)
history = {"train_loss": [], "val_loss": [],
"train_acc": [], "val_acc": []}
print(f"Training on {len(train_data)} jets for {epochs} epochs...")
for epoch in range(epochs):
# ── train ──
model.train()
batch = make_batch(train_data)
batch = batch.to(device)
opt.zero_grad()
out = model(batch.x, batch.edge_index, batch.batch)
loss = F.cross_entropy(out, batch.y)
loss.backward()
opt.step()
sched.step()
train_loss = loss.item()
train_acc = (out.argmax(1) == batch.y).float().mean().item()
# ── validate ──
model.eval()
with torch.no_grad():
vbatch = make_batch(val_data).to(device)
vout = model(vbatch.x, vbatch.edge_index, vbatch.batch)
vloss = F.cross_entropy(vout, vbatch.y).item()
vacc = (vout.argmax(1) == vbatch.y).float().mean().item()
history["train_loss"].append(round(train_loss, 4))
history["val_loss"].append(round(vloss, 4))
history["train_acc"].append(round(train_acc, 4))
history["val_acc"].append(round(vacc, 4))
if (epoch + 1) % 10 == 0:
print(f" Epoch {epoch+1:3d}/{epochs} "
f"loss={train_loss:.4f} val_loss={vloss:.4f} "
f"val_acc={vacc:.2%}")
torch.save(model.state_dict(), MODEL_PATH)
with open(HISTORY_PATH, "w") as f:
json.dump(history, f)
print(f"Model saved to {MODEL_PATH}")
return model, history
def load_or_train() -> tuple[ParticleGCN, dict]:
"""Load pretrained model if available, otherwise train from scratch."""
model = ParticleGCN()
if os.path.exists(MODEL_PATH) and os.path.exists(HISTORY_PATH):
print("Loading pre-trained model...")
model.load_state_dict(torch.load(MODEL_PATH, map_location="cpu"))
with open(HISTORY_PATH) as f:
history = json.load(f)
else:
model, history = train_model()
model.eval()
return model, history
# ─────────────────────────────────────────────────────────────
# 4. VISUALISATION HELPERS
# ─────────────────────────────────────────────────────────────
def plot_jet_graph(data: Data, pred_label: int,
true_label: int, probs: np.ndarray) -> go.Figure:
"""
Build an interactive Plotly figure showing:
Left — the jet graph in (η, φ) space, nodes sized by pT
Right — the class probability bar chart
"""
eta = data.eta.numpy()
phi = data.phi.numpy()
pt = data.pt.numpy()
src, dst = data.edge_index.numpy()
fig = make_subplots(
rows=1, cols=2,
column_widths=[0.6, 0.4],
subplot_titles=[
f"Jet graph — {len(eta)} particles in (η, φ) space",
"Class probabilities"
]
)
# ── edge traces ──
ex, ey = [], []
for s, d in zip(src, dst):
ex += [eta[s], eta[d], None]
ey += [phi[s], phi[d], None]
fig.add_trace(
go.Scatter(x=ex, y=ey, mode="lines",
line=dict(color="#B4B2A9", width=0.8),
hoverinfo="skip", name="Edges"),
row=1, col=1
)
# ── node traces, coloured by particle type ──
pid_labels = ["Photon", "Neutral hadron", "Charged hadron",
"Electron", "Muon"]
pid_colors = ["#EF9F27", "#5DCAA5", "#378ADD", "#D85A30", "#7F77DD"]
pid_arr = (data.x[:, 4].numpy() * 4).round().astype(int)
for pid_val in range(5):
mask = pid_arr == pid_val
if not mask.any():
continue
fig.add_trace(
go.Scatter(
x=eta[mask], y=phi[mask],
mode="markers",
marker=dict(
size=np.clip(np.log1p(pt[mask]) * 6, 4, 22),
color=pid_colors[pid_val],
line=dict(width=0.8, color="#2C2C2A"),
opacity=0.85
),
name=pid_labels[pid_val],
hovertemplate=(
f"{pid_labels[pid_val]}
"
"η = %{x:.3f}
φ = %{y:.3f}
"
"pT ≈ %{customdata:.1f} GeV
Graph Neural Networks CERN-style Jets PyTorch Geometric HEP Physics AI + Physics Portfolio
A 3-layer Graph Convolutional Network classifies particle collision events (jets) into 5 categories — the same task performed at the LHC at CERN. Each jet is a graph: particles are nodes, kNN edges in (η, φ) momentum space.