File size: 5,204 Bytes
6ba1ba5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import numpy as np
import pandas as pd
import json
import os
import random
from sklearn.metrics import roc_auc_score
# --- CONFIGURATION ---
SEEDS = [42, 100, 2024]
BATCH_SIZE = 64
EPOCHS = 10
SP_TARGET = 0.85
MAX_STEPS = 25
PATIENCE = 3
TEMPERATURE = 0.5
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# --- DATASET: PHYSICS-BASED BEARINGS ---
class SyntheticBearings(Dataset):
"""
Simulates rotating machinery.
Class 0: Healthy (Sine Wave + Noise)
Class 1: Inner Race Fault (Impulses at specific frequencies)
"""
def __init__(self, n_samples):
self.data = []
self.targets = []
t = np.linspace(0, 1, 200) # 200 time steps (0.2s at 1kHz)
for _ in range(n_samples):
# Base Carrier (Shaft Rotation 30Hz)
signal = 0.5 * np.sin(2 * np.pi * 30 * t)
label = 0
# Fault Injection (50% chance)
if np.random.rand() > 0.5:
label = 1
# Fault: High freq impulses (120Hz) decaying exponentially
fault_sig = 0.8 * np.sin(2 * np.pi * 120 * t) * np.exp(-5*t)
signal += fault_sig
# Industrial Noise
signal += np.random.normal(0, 0.4, size=len(t))
self.data.append(torch.tensor(signal, dtype=torch.float32).unsqueeze(0))
self.targets.append(label)
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
return self.data[idx], self.targets[idx]
# --- MODEL ---
class BearingCNN(nn.Module):
def __init__(self):
super().__init__()
self.conv1 = nn.Conv1d(1, 16, 5)
self.conv2 = nn.Conv1d(16, 32, 5)
self.dropout = nn.Dropout(0.3)
self.pool = nn.MaxPool1d(2)
self.fc = nn.Linear(32 * 47, 2)
def forward(self, x):
x = self.pool(F.relu(self.conv1(x)))
x = self.pool(F.relu(self.conv2(x)))
x = self.dropout(x)
x = x.view(x.size(0), -1)
x = self.fc(x)
return x
# --- UTILS ---
def compute_sp(probs):
probs = torch.clamp(probs, min=1e-9)
entropy = -torch.sum(probs * torch.log(probs), dim=1)
sp = 1.0 - (entropy / np.log(2))
return sp
# --- RUNNER ---
def run_experiment(seed):
print(f"Running Bearings Seed {seed}...")
torch.manual_seed(seed)
np.random.seed(seed)
train_ds = SyntheticBearings(2000)
test_ds = SyntheticBearings(500)
train_loader = DataLoader(train_ds, batch_size=64, shuffle=True)
test_loader = DataLoader(test_ds, batch_size=1, shuffle=False)
model = BearingCNN().to(DEVICE)
opt = optim.Adam(model.parameters(), lr=0.001)
model.train()
for _ in range(EPOCHS):
for x, y in train_loader:
x, y = x.to(DEVICE), y.to(DEVICE)
opt.zero_grad()
out = model(x)
loss = F.cross_entropy(out, y)
loss.backward()
opt.step()
# Eval SCI
logs = []
with torch.no_grad():
for x, y in test_loader:
x = x.to(DEVICE)
accum = model(x)
steps = 1
sp_hist = []
while steps < MAX_STEPS:
new_logits = model(x)
accum += new_logits
steps += 1
curr_prob = F.softmax(accum/steps, dim=1)
curr_sp = compute_sp(curr_prob).item()
sp_hist.append(curr_sp)
# Convergence Check
if len(sp_hist) >= PATIENCE:
if abs(sp_hist[-1] - sp_hist[-PATIENCE]) < 0.005 and curr_sp > 0.8:
break
final_prob = F.softmax(accum/steps, dim=1)
pred = final_prob.argmax().item()
correct = (pred == y.item())
delta = abs(SP_TARGET - curr_sp)
logs.append({
"correct": int(correct),
"delta": delta,
"steps": steps
})
return logs
def analyze(logs):
df = pd.DataFrame(logs)
errors = 1 - df['correct']
# Safety Analysis
auc = roc_auc_score(errors, df['delta'])
steps_correct = df[df['correct']==1]['steps'].mean()
steps_wrong = df[df['correct']==0]['steps'].mean()
print("\n" + "="*40)
print("BEARINGS (INDUSTRIAL) RESULTS")
print("="*40)
print(f"Error Rate: {errors.mean()*100:.2f}%")
print(f"Safety AUROC: {auc:.4f}")
print("-" * 40)
print("Metacognition (Avg Steps):")
print(f"Correct: {steps_correct:.2f}")
print(f"Wrong: {steps_wrong:.2f}")
print("="*40)
if __name__ == "__main__":
all_logs = []
for s in SEEDS:
all_logs.extend(run_experiment(s))
analyze(all_logs) |