Hybrid-Zonotope Reachability Loss for Neural Closed Loops / reachability_mvp.py

Failed on benchmark

Raw ⬇ ZIP
  1import json
  2import math
  3import random
  4from dataclasses import dataclass
  5
  6import numpy as np
  7import torch
  8
  9SEED = 2686
 10random.seed(SEED)
 11np.random.seed(SEED)
 12torch.manual_seed(SEED)
 13torch.set_num_threads(4)
 14
 15
 16@dataclass
 17class HybridZonotope:
 18    """Small hybrid-zonotope container: c + G xi + H beta, with optional coupling."""
 19    c: np.ndarray
 20    G: np.ndarray
 21    H: np.ndarray
 22    E: np.ndarray | None = None
 23    F: np.ndarray | None = None
 24    b: np.ndarray | None = None
 25
 26    def affine(self, A, B=None, offset=None):
 27        A = np.asarray(A)
 28        c = A @ self.c
 29        G = A @ self.G
 30        H = A @ self.H
 31        if B is not None:
 32            c = c + np.asarray(B) @ np.asarray(offset)
 33        return HybridZonotope(c, G, H, self.E, self.F, self.b)
 34
 35    def interval(self):
 36        rad = np.abs(self.G).sum(axis=1) + np.abs(self.H).sum(axis=1)
 37        return self.c - rad, self.c + rad
 38
 39
 40def relu_interval_zonotope(z: HybridZonotope):
 41    """Sound interval relaxation of ReLU, retaining a zonotope approximation."""
 42    lo, hi = z.interval()
 43    out_lo, out_hi = np.maximum(lo, 0), np.maximum(hi, 0)
 44    c = (out_lo + out_hi) / 2
 45    G = np.diag((out_hi - out_lo) / 2)
 46    H = np.zeros((len(c), 0))
 47    return HybridZonotope(c, G, H)
 48
 49
 50def scalar_interval_policy(lo, hi, w1, b1, w2, b2):
 51    """Interval bound for a one-hidden-layer scalar ReLU policy."""
 52    # Each hidden preactivation is affine in scalar x.
 53    p0, p1 = w1[:, 0] * lo + b1, w1[:, 0] * hi + b1
 54    pl, ph = np.minimum(p0, p1), np.maximum(p0, p1)
 55    rl, rh = np.maximum(pl, 0), np.maximum(ph, 0)
 56    terms_l = np.where(w2[0] >= 0, w2[0] * rl, w2[0] * rh)
 57    terms_h = np.where(w2[0] >= 0, w2[0] * rh, w2[0] * rl)
 58    return float(terms_l.sum() + b2[0]), float(terms_h.sum() + b2[0])
 59
 60
 61def exact_radius(lam, r0, gamma, n):
 62    """Exact reachable radius for x+=lam*x+w, |w|<=gamma."""
 63    if abs(lam - 1.0) < 1e-12:
 64        return r0 + n * gamma
 65    return abs(lam) ** n * r0 + gamma * sum(abs(lam) ** i for i in range(n))
 66
 67
 68def mechanism_sweeps():
 69    r0, gamma, n = 0.2, 0.03, 8
 70    lam_values = [0.0, 0.2, 0.5, 0.8, 1.0, 1.2]
 71    rows = []
 72    for lam in lam_values:
 73        observed = exact_radius(lam, r0, gamma, n)
 74        predicted = (abs(lam) ** n * r0 + gamma * sum(abs(lam) ** i for i in range(n)))
 75        rows.append({"lambda": lam, "observed_rN": observed, "predicted_rN": predicted,
 76                     "abs_error": abs(observed - predicted)})
 77
 78    # At fixed lambda<1, the terminal radius is affine in gamma with known slope.
 79    lam, r0, n = 0.6, 0.2, 7
 80    gammas = [0.0, 0.01, 0.02, 0.04, 0.08]
 81    slope = sum(lam ** i for i in range(n))
 82    scaling = [{"gamma": g, "observed_rN": exact_radius(lam, r0, g, n),
 83                "predicted_rN": lam ** n * r0 + g * slope} for g in gammas]
 84
 85    # Constraint boundary: r_N <= R gives gamma* exactly.
 86    lam, r0, n, R = 0.7, 0.2, 6, 0.30
 87    geom = sum(lam ** i for i in range(n))
 88    gamma_star = (R - lam ** n * r0) / geom
 89    boundary = [{"gamma": g, "violation": max(0.0, exact_radius(lam, r0, g, n) - R)}
 90                for g in [0.0, gamma_star * 0.99, gamma_star, gamma_star * 1.01]]
 91
 92    # Contraction boundary: disturbance-free radius decays iff |lambda|<1.
 93    contraction = [{"lambda": lam, "ratio_rN_r0": exact_radius(lam, 0.2, 0.0, 10) / 0.2,
 94                    "predicted_contracts": abs(lam) < 1.0}
 95                   for lam in [0.8, 0.99, 1.0, 1.01, 1.2]]
 96    return {"radius_recurrence": rows, "disturbance_linear_scaling": scaling,
 97            "constraint_boundary": {"predicted_gamma_star": gamma_star, "samples": boundary},
 98            "contraction_boundary": contraction}
 99
100
101def rollout_loss(k, disturbances, x0=0.8, horizon=8):
102    x = torch.tensor(x0, dtype=torch.float32)
103    loss = 0.0
104    for t in range(horizon):
105        loss = loss + x * x
106        x = (1.0 - k) * x + disturbances[t]
107    return loss + 2.0 * x * x
108
109
110def reach_loss(k, gamma, x0=0.8, radius0=0.08, horizon=8, limit=0.45):
111    """Differentiable interval certificate for x+=x+u+w, u=-k*x."""
112    center = torch.tensor(x0, dtype=torch.float32)
113    radius = torch.tensor(radius0, dtype=torch.float32)
114    total = 0.0
115    for _ in range(horizon):
116        total = total + torch.relu(torch.abs(center) + radius - limit) ** 2
117        center = (1.0 - k) * center
118        radius = torch.abs(1.0 - k) * radius + gamma
119        total = total + 0.5 * torch.relu(radius - limit) ** 2
120    total = total + 5.0 * torch.relu(torch.abs(center) + radius - 0.25) ** 2
121    # contraction deficit, active when the closed-loop gain is not below one
122    total = total + 0.2 * torch.relu(torch.abs(1.0 - k) - 0.9) ** 2
123    return total
124
125
126def train_controllers(gamma, steps=500):
127    # Same scalar controller and initialization; baseline sees nominal trajectories only.
128    torch.manual_seed(SEED)
129    raw_b = torch.nn.Parameter(torch.tensor(-0.2))
130    opt_b = torch.optim.Adam([raw_b], lr=0.035)
131    nominal = torch.zeros(8)
132    for _ in range(steps):
133        opt_b.zero_grad()
134        k = torch.sigmoid(raw_b) * 2.0
135        loss = rollout_loss(k, nominal)
136        loss.backward(); opt_b.step()
137    kb = float((torch.sigmoid(raw_b) * 2).detach())
138
139    torch.manual_seed(SEED)
140    raw_r = torch.nn.Parameter(torch.tensor(-0.2))
141    opt_r = torch.optim.Adam([raw_r], lr=0.035)
142    for _ in range(steps):
143        opt_r.zero_grad()
144        k = torch.sigmoid(raw_r) * 2.0
145        loss = reach_loss(k, gamma)
146        loss.backward(); opt_r.step()
147    kr = float((torch.sigmoid(raw_r) * 2).detach())
148
149    def cert(k):
150        radii = [exact_radius(abs(1-k), 0.08, gamma, t) for t in range(9)]
151        return max(radii), radii[-1], max(0.0, radii[-1] - 0.45)
152    return {"gamma": gamma, "baseline_k": kb, "reach_k": kr,
153            "baseline_certificate": cert(kb), "reach_certificate": cert(kr),
154            "baseline_nominal_loss": float(rollout_loss(torch.tensor(kb), nominal)),
155            "reach_training_loss": float(reach_loss(torch.tensor(kr), gamma))}
156
157
158def main():
159    results = {"seed": SEED, "mechanism": mechanism_sweeps(),
160               "training": [train_controllers(g) for g in [0.0, 0.12, 0.30, 0.60]]}
161    print(json.dumps(results, indent=2))
162
163
164if __name__ == "__main__":
165    main()