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