File size: 5,048 Bytes
b195a41 | 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 | import torch
import torch.nn as nn
import torch.nn.functional as F
import math
class BayesianLinear(nn.Module):
"""
Variational Bayesian Linear Layer using the Reparameterization Trick.
"""
def __init__(self, in_features: int, out_features: int, prior_sigma: float = 1.0):
super(BayesianLinear, self).__init__()
self.in_features = in_features
self.out_features = out_features
self.prior_sigma = prior_sigma
# Weight variational parameters (Mean and Rho)
self.weight_mu = nn.Parameter(torch.Tensor(out_features, in_features).normal_(0, 0.1))
self.weight_rho = nn.Parameter(torch.Tensor(out_features, in_features).fill_(-3.0))
# Bias variational parameters
self.bias_mu = nn.Parameter(torch.Tensor(out_features).normal_(0, 0.1))
self.bias_rho = nn.Parameter(torch.Tensor(out_features).fill_(-3.0))
def forward(self, x: torch.Tensor) -> torch.Tensor:
# Reparameterization trick: std = log(1 + exp(rho))
weight_sigma = torch.log1p(torch.exp(self.weight_rho))
bias_sigma = torch.log1p(torch.exp(self.bias_rho))
epsilon_w = torch.randn_like(self.weight_mu)
epsilon_b = torch.randn_like(self.bias_mu)
weight = self.weight_mu + weight_sigma * epsilon_w
bias = self.bias_mu + bias_sigma * epsilon_b
return F.linear(x, weight, bias)
def kl_divergence(self) -> torch.Tensor:
"""
Calculates KL Divergence between variational posterior q(w) and Gaussian prior p(w).
"""
weight_sigma = torch.log1p(torch.exp(self.weight_rho))
bias_sigma = torch.log1p(torch.exp(self.bias_rho))
kl_w = torch.sum(
torch.log(self.prior_sigma / weight_sigma) +
(weight_sigma**2 + self.weight_mu**2) / (2 * self.prior_sigma**2) - 0.5
)
kl_b = torch.sum(
torch.log(self.prior_sigma / bias_sigma) +
(bias_sigma**2 + self.bias_mu**2) / (2 * self.prior_sigma**2) - 0.5
)
return kl_w + kl_b
class PhysicsGuidedBNN(nn.Module):
"""
Physics-Guided Bayesian Neural Network for Wind Turbine Diagnostics.
"""
def __init__(self, config: dict):
super(PhysicsGuidedBNN, self).__init__()
self.config = config
in_dim = config["in_features"]
hidden_dims = config["hidden_dims"]
num_classes = config["num_classes"]
layers = []
curr_dim = in_dim
for h_dim in hidden_dims:
layers.append(BayesianLinear(curr_dim, h_dim, prior_sigma=config.get("prior_sigma", 1.0)))
layers.append(nn.ReLU())
layers.append(nn.BatchNorm1d(h_dim))
curr_dim = h_dim
self.backbone = nn.Sequential(*layers)
self.classifier = BayesianLinear(curr_dim, num_classes, prior_sigma=config.get("prior_sigma", 1.0))
def forward(self, x: torch.Tensor) -> torch.Tensor:
features = self.backbone(x)
logits = self.classifier(features)
return logits
def total_kl_divergence(self) -> torch.Tensor:
kl = torch.tensor(0.0, device=next(self.parameters()).device)
for module in self.modules():
if isinstance(module, BayesianLinear):
kl = kl + module.kl_divergence()
return kl
def compute_physics_loss(self, x: torch.Tensor, logits: torch.Tensor) -> torch.Tensor:
"""
Computes physical consistency violation penalty.
Assumptions on feature indices:
x[:, 0] -> Wind Speed (m/s)
x[:, 1] -> Generated Power (kW)
x[:, 2] -> Gearbox Oil Temperature (°C)
x[:, 3] -> Generator Speed (RPM)
"""
probs = F.softmax(logits, dim=-1)[:, 1] # Failure probability
wind_speed = x[:, 0]
power = x[:, 1]
gearbox_temp = x[:, 2]
gen_speed = x[:, 3]
# 1. Aerodynamic Power Limit (Betz Law Constraint): P <= 0.5 * rho * A * v^3 * Cp_max
rho = self.config["physics_params"]["air_density"]
radius = self.config["physics_params"]["rotor_radius"]
cp_max = self.config["physics_params"]["cp_max"]
area = math.pi * (radius ** 2)
max_theoretical_power = (0.5 * rho * area * (torch.clamp(wind_speed, min=0) ** 3) * cp_max) / 1000.0 # kW
# Power exceeding physical limit violates physics
power_violation = F.relu(power - max_theoretical_power)
# 2. Thermal Energy Dissipation Residual: dT/dt ~ alpha * Power - beta * (T_gearbox - T_ambient)
alpha = self.config["physics_params"]["thermal_alpha"]
beta = self.config["physics_params"]["thermal_beta"]
expected_thermal_load = alpha * power - beta * gearbox_temp
# Unrealistic low temp under high power/speed triggers high physical residual
thermal_residual = F.relu(expected_thermal_load - gearbox_temp)
# Total Physics Regularization Penalty
physics_loss = torch.mean(power_violation + thermal_residual)
return physics_loss |