Neural Network Flight Control
Neural networks can replace traditional controllers (PID, MPC) with learned policies that potentially handle complex scenarios better. However, ensuring stability and safety becomes paramount when neural networks control physical systems.
End-to-End Flight Learning
Sponsored
End-to-end learning maps raw sensor inputs directly to control outputs, eliminating manual feature engineering.
Neural Controller Architecture
Neural Controller Implementation
CNN-LSTM Flight Controller
python
class NeuralFlightController(nn.Module):
def __init__(self,
sensor_dims: dict,
hidden_size: int = 256,
action_dim: int = 4):
super().__init__()
# Sensor encoders
self.imu_encoder = nn.Sequential(
nn.Linear(sensor_dims['imu'], 64),
nn.ReLU(),
nn.Linear(64, 64)
)
self.lidar_encoder = nn.Sequential(
nn.Conv1d(1, 32, kernel_size=5, stride=2),
nn.ReLU(),
nn.Conv1d(32, 64, kernel_size=3, stride=2),
nn.ReLU(),
nn.AdaptiveAvgPool1d(16)
)
self.camera_encoder = nn.Sequential(
nn.Conv2d(3, 32, kernel_size=8, stride=4),
nn.ReLU(),
nn.Conv2d(32, 64, kernel_size=4, stride=2),
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, stride=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d((4, 4))
)
# Temporal model
self.lstm = nn.LSTM(
input_size=64 + 64 * 16 + 64 * 16,
hidden_size=hidden_size,
num_layers=2,
batch_first=True,
dropout=0.1
)
# Attention mechanism
self.attention = nn.MultiheadAttention(
embed_dim=hidden_size,
num_heads=4,
dropout=0.1
)
# Output layers
self.safety_layer = SafetyLayer(action_dim)
self.output_head = nn.Sequential(
nn.Linear(hidden_size, hidden_size // 2),
nn.ReLU(),
nn.Linear(hidden_size // 2, action_dim),
nn.Tanh() # Bound outputs
)
self.hidden_size = hidden_size
def forward(self, imu, lidar, camera, hidden=None):
batch_size = imu.shape[0]
# Encode sensors
imu_features = self.imu_encoder(imu)
lidar_features = self.lidar_encoder(lidar.unsqueeze(1))
lidar_features = lidar_features.view(batch_size, -1)
camera_features = self.camera_encoder(camera)
camera_features = camera_features.view(batch_size, -1)
# Fuse features
combined = torch.cat([imu_features, lidar_features, camera_features], dim=-1)
# Temporal modeling
if hidden is None:
lstm_out, hidden = self.lstm(combined.unsqueeze(1))
else:
lstm_out, hidden = self.lstm(combined.unsqueeze(1), hidden)
# Attention
attn_out, _ = self.attention(lstm_out, lstm_out, lstm_out)
# Output
raw_action = self.output_head(attn_out.squeeze(1))
# Safety constraints
safe_action = self.safety_layer(raw_action)
return safe_action, hidden
def get_action(self, imu, lidar, camera, hidden=None):
self.eval()
with torch.no_grad():
action, hidden = self.forward(imu, lidar, camera, hidden)
return action.numpy(), hidden
class SafetyLayer(nn.Module):
"""Layer that ensures output actions are within safe bounds."""
def __init__(self, action_dim: int):
super().__init__()
self.min_action = torch.tensor([-0.8, -0.8, -0.5, 0.1]) # min [roll, pitch, yaw, throttle]
self.max_action = torch.tensor([0.8, 0.8, 0.5, 1.0])
def forward(self, action):
return torch.clamp(action, self.min_action, self.max_action)
Stability Analysis
Ensuring neural network controllers maintain stability is critical for safety.
Lyapunov-Based Stability
python
class StabilityCertificate(nn.Module):
"""Neural network that learns a Lyapunov function for stability certification."""
def __init__(self, state_dim: int, hidden_size: int = 128):
super().__init__()
self.lyapunov_net = nn.Sequential(
nn.Linear(state_dim, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, 1)
)
def forward(self, state):
return self.lyapunov_net(state)
def check_stability(self, state):
V = self.forward(state)
# V(0) = 0 condition
origin_state = torch.zeros_like(state)
V_origin = self.forward(origin_state)
# Positive definiteness
positive_definite = (V > 0).all() and (V_origin.abs() < 0.01).all()
return {
'lyapunov_value': V.item(),
'is_positive_definite': positive_definite,
'certified': positive_definite
}
def train_lyapunov_certifier(model, controller, state_dim, n_samples=10000):
"""Train a Lyapunov function to certify controller stability."""
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
for epoch in range(100):
# Sample states
states = torch.randn(n_samples, state_dim) * 0.5
# Compute Lyapunov values
V = model(states)
# Loss: V(0) â 0
V_origin = model(torch.zeros(1, state_dim))
origin_loss = V_origin ** 2
# Loss: V(x) > 0 for x â 0
positivity_loss = torch.relu(-V + 0.01).mean()
# Loss: dV/dt < 0 (decreasing along trajectories)
# Use controller to compute next states
with torch.no_grad():
actions = controller(states)
next_states = simple_dynamics(states, actions)
V_next = model(next_states)
decrease_loss = torch.relu(V_next - V + 0.01).mean()
# Total loss
loss = origin_loss + positivity_loss + decrease_loss
optimizer.zero_grad()
loss.backward()
optimizer.step()
return model
Formal Verification
Advertisement
Formal verification provides mathematical guarantees about neural network behavior.
Interval Bound Propagation
python
from typing import List, Tuple
class IntervalBoundPropagation:
"""Propagate interval bounds through neural network layers."""
def __init__(self, layers: List[nn.Module]):
self.layers = layers
def propagate(self, input_bounds: Tuple[np.ndarray, np.ndarray]):
lower, upper = input_bounds
for layer in self.layers:
if isinstance(layer, nn.Linear):
lower, upper = self._propagate_linear(layer, lower, upper)
elif isinstance(layer, nn.ReLU):
lower, upper = self._propagate_relu(lower, upper)
elif isinstance(layer, nn.Tanh):
lower, upper = self._propagate_tanh(lower, upper)
return lower, upper
def _propagate_linear(self, layer, lower, upper):
W = layer.weight.data.numpy()
b = layer.bias.data.numpy() if layer.bias is not None else 0
W_pos = np.maximum(W, 0)
W_neg = np.minimum(W, 0)
new_lower = W_pos @ lower + W_neg @ upper + b
new_upper = W_pos @ upper + W_neg @ lower + b
return new_lower, new_upper
def _propagate_relu(self, lower, upper):
new_lower = np.maximum(lower, 0)
new_upper = np.maximum(upper, 0)
return new_lower, new_upper
def _propagate_tanh(self, lower, upper):
new_lower = np.tanh(lower)
new_upper = np.tanh(upper)
return new_lower, new_upper
def verify_safety(self, input_bounds, safe_region):
lower, upper = self.propagate(input_bounds)
# Check if output is within safe region
safe_lower, safe_upper = safe_region
is_safe = np.all(lower >= safe_lower) and np.all(upper <= safe_upper)
return {
'output_lower': lower,
'output_upper': upper,
'is_safe': is_safe,
'margin_lower': lower - safe_lower,
'margin_upper': safe_upper - upper
}
Certifiable Robustness
python
class RobustNeuralController(nn.Module):
"""Neural controller with certified robustness to input perturbations."""
def __init__(self, base_controller, epsilon=0.1):
super().__init__()
self.base_controller = base_controller
self.epsilon = epsilon # Max perturbation bound
def forward(self, state):
# Base prediction
nominal_action = self.base_controller(state)
# Compute Lipschitz bound for robustness certificate
lipschitz_bound = self.compute_lipschitz_bound()
# Robust output bound
robust_bound = self.epsilon * lipschitz_bound
return nominal_action, robust_bound
def compute_lipschitz_bound(self):
"""Estimate Lipschitz constant of the network."""
lipschitz = 1.0
for module in self.base_controller.modules():
if isinstance(module, nn.Linear):
W = module.weight.data
lipschitz *= torch.linalg.norm(W, ord=2).item()
elif isinstance(module, nn.ReLU):
pass # ReLU is 1-Lipschitz
return lipschitz
def verify_robustness(self, test_states, safe_actions, tolerance=0.2):
"""Verify that controller stays within safe bounds under perturbation."""
violations = 0
for state in test_states:
state_tensor = torch.FloatTensor(state).unsqueeze(0)
action, bound = self.forward(state_tensor)
# Check if all possible actions (within bound) are safe
action_np = action.detach().numpy()[0]
bound_val = bound.item()
safe_min, safe_max = safe_actions
if np.any(action_np - bound_val < safe_min) or \
np.any(action_np + bound_val > safe_max):
violations += 1
return {
'total_tests': len(test_states),
'violations': violations,
'robust_accuracy': 1 - violations / len(test_states)
}
Hands-On Project: Verified Neural Controller
python
from typing import Tuple
class VerifiedNeuralFlightSystem:
def __init__(self, state_dim=12, action_dim=4):
self.state_dim = state_dim
self.action_dim = action_dim
# Main controller
self.controller = self._build_controller()
# Safety monitor
self.safety_monitor = SafetyMonitor(state_dim, action_dim)
# Backup controller (simple PID)
self.backup_controller = SimplePID()
# Verification tools
self.ibp = IntervalBoundPropagation(list(self.controller.modules()))
def _build_controller(self):
return nn.Sequential(
nn.Linear(self.state_dim, 256),
nn.ReLU(),
nn.Linear(256, 256),
nn.ReLU(),
nn.Linear(256, 128),
nn.ReLU(),
nn.Linear(128, self.action_dim),
nn.Tanh()
)
def safe_control(self, state: np.ndarray) -> np.ndarray:
"""Get control action with safety guarantees."""
state_tensor = torch.FloatTensor(state).unsqueeze(0)
# Get neural controller action
with torch.no_grad():
nn_action = self.controller(state_tensor).numpy()[0]
# Safety check
safety_result = self.safety_monitor.check(state, nn_action)
if safety_result['is_safe']:
return nn_action
else:
# Fallback to backup controller
print(f"Safety violation detected! Using backup. "
f"Violation: {safety_result['violation_type']}")
return self.backup_controller.compute(state)
def train_with_safety(self, expert_states, expert_actions, epochs=100):
"""Train controller with safety constraints."""
optimizer = torch.optim.Adam(self.controller.parameters(), lr=1e-3)
states = torch.FloatTensor(expert_states)
actions = torch.FloatTensor(expert_actions)
for epoch in range(epochs):
# Supervised loss
pred_actions = self.controller(states)
supervised_loss = nn.MSELoss()(pred_actions, actions)
# Safety loss (penalize unsafe actions)
safety_loss = self._compute_safety_loss(states)
# Total loss
loss = supervised_loss + 0.1 * safety_loss
optimizer.zero_grad()
loss.backward()
optimizer.step()
if epoch % 20 == 0:
print(f"Epoch {epoch}: Loss={loss.item():.4f}")
def _compute_safety_loss(self, states):
"""Compute loss that encourages safe actions."""
actions = self.controller(states)
# Define safe action bounds
safe_min = torch.tensor([-0.5, -0.5, -0.3, 0.2])
safe_max = torch.tensor([0.5, 0.5, 0.3, 0.9])
# Penalize actions outside safe bounds
violation_min = torch.relu(safe_min - actions).sum()
violation_max = torch.relu(actions - safe_max).sum()
return violation_min + violation_max
def verify_controller(self, test_states, n_samples=1000):
"""Run formal verification on controller."""
results = {
'safe_count': 0,
'total_count': len(test_states),
'violations': []
}
for state in test_states:
state_tensor = torch.FloatTensor(state).unsqueeze(0)
# Add perturbations
for _ in range(n_samples):
perturbed = state_tensor + torch.randn_like(state_tensor) * 0.1
perturbed = torch.clamp(perturbed, -1, 1)
with torch.no_grad():
action = self.controller(perturbed).numpy()[0]
if self.safety_monitor.is_safe(state, action):
results['safe_count'] += 1
else:
results['violations'].append((state, action))
results['robustness'] = results['safe_count'] / (results['total_count'] * n_samples)
return results
class SafetyMonitor:
def __init__(self, state_dim, action_dim):
self.state_bounds = {
'position': (-50, 50),
'velocity': (-10, 10),
'orientation': (-np.pi/4, np.pi/4)
}
self.action_bounds = (-1, 1)
def check(self, state, action):
violations = []
# Check position bounds
if np.any(np.abs(state[:3]) > 50):
violations.append('position_out_of_bounds')
# Check velocity limits
if np.any(np.abs(state[3:6]) > 10):
violations.append('velocity_exceeded')
# Check orientation (prevent flipping)
if np.any(np.abs(state[6:9]) > np.pi/4):
violations.append('orientation_unsafe')
# Check action bounds
if np.any(np.abs(action) > 1):
violations.append('action_out_of_bounds')
return {
'is_safe': len(violations) == 0,
'violations': violations,
'violation_type': violations[0] if violations else None
}
def is_safe(self, state, action):
return self.check(state, action)['is_safe']
class SimplePID:
def __init__(self):
self.kp = 1.0
self.kd = 0.5
def compute(self, state):
target = np.zeros(3)
error = target - state[:3]
velocity = state[3:6]
action = self.kp * error - self.kd * velocity
action = np.clip(action, -0.5, 0.5)
return np.concatenate([action, [0.5]]) # Default throttle
# Usage
def main():
system = VerifiedNeuralFlightSystem()
# Generate training data (from expert)
n_samples = 10000
expert_states = np.random.randn(n_samples, 12) * 0.5
expert_actions = np.random.randn(n_samples, 4) * 0.3
# Train with safety constraints
system.train_with_safety(expert_states, expert_actions, epochs=100)
# Verify
test_states = np.random.randn(100, 12) * 0.3
results = system.verify_controller(test_states)
print(f"\nVerification Results:")
print(f"Robustness: {results['robustness']:.2%}")
print(f"Safe: {results['safe_count']}/{results['total_count'] * 1000}")
if __name__ == "__main__":
main()
Key Takeaways
- Neural controllers can learn complex behaviors from data
- Stability analysis using Lyapunov methods provides guarantees
- Formal verification via IBP proves safety properties
- Always include fallback controllers for safety-critical applications
- Safety layers constrain outputs to safe regions
- Combine learning with classical control for robust systems