-
Notifications
You must be signed in to change notification settings - Fork 7
Expand file tree
/
Copy pathtesseract_api.py
More file actions
157 lines (140 loc) · 5.14 KB
/
Copy pathtesseract_api.py
File metadata and controls
157 lines (140 loc) · 5.14 KB
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
# Copyright 2025 Pasteur Labs. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
# Tesseract API module for lp_solve
# Generated by tesseract 0.9.0 on 2025-06-04T13:11:17.208977
import functools
from typing import Self
import jax
import jax.numpy as jnp
import qpax
from pydantic import BaseModel, Field, model_validator
from tesseract_core.runtime import Array, Differentiable, Float32
#
# Schemas
#
class InputSchema(BaseModel):
Q: Differentiable[Array[(None, None), Float32]] = Field(
description="Quadratic cost matrix Q for the quadratic program."
)
q: Differentiable[Array[(None,), Float32]] | None = Field(
description="Linear cost vector q for the quadratic program.", default=None
)
A: Differentiable[Array[(None, None), Float32]] | None = Field(
default=None,
description="Linear equality constraint matrix A for the quadratic program.",
)
b: Differentiable[Array[(None,), Float32]] | None = Field(
default=None,
description="Linear equality constraint rhs b for the quadratic program.",
)
G: Differentiable[Array[(None, None), Float32]] | None = Field(
description="Linear inequality constraint matrix G for the quadratic program.",
default=None,
)
h: Differentiable[Array[(None,), Float32]] | None = Field(
description="Linear inequality constraint rhs h for the quadratic program.",
default=None,
)
solver_tol: Float32 | None = Field(
description="Tolerance for the solver convergence.", default=1e-4
)
target_kappa: Float32 | None = Field(
description="QPAX parameter for QP relaxation.", default=1e-3
)
@model_validator(mode="after")
def validate_shape_inputs(self) -> Self:
if self.Q.shape[0] != self.Q.shape[1]:
raise ValueError("Q must be a square matrix.")
n = self.Q.shape[0]
# set missing arguments to 'empty' arrays expected by qpax
if self.q is None:
self.q = jnp.zeros((n,), dtype=self.Q.dtype)
if self.A is None:
self.A = jnp.zeros((0, n), dtype=self.Q.dtype)
if self.b is None:
self.b = jnp.zeros((0,), dtype=self.Q.dtype)
if self.G is None:
self.G = jnp.zeros((0, n), dtype=self.Q.dtype)
if self.h is None:
self.h = jnp.zeros((0,), dtype=self.Q.dtype)
# check shapes of inputs
if self.Q.shape[0] != self.q.shape[0]:
raise ValueError("Q and q must have compatible shapes.")
if self.A.shape[1] != self.q.shape[0]:
raise ValueError("A.shape[1] must be equal to number of vars.")
if self.A.shape[0] != self.b.shape[0]:
raise ValueError("A and b must have compatible shapes.")
if self.A.shape[0] >= self.A.shape[1]:
raise ValueError("A must have fewer rows than columns (m < n).")
if self.G.shape[1] != self.q.shape[0]:
raise ValueError("G.shape[1] must be equal to number of vars.")
if self.G.shape[0] != self.h.shape[0]:
raise ValueError("G and h must have compatible shapes.")
return self
class OutputSchema(BaseModel):
x: Differentiable[Array[(None,), Float32]] = Field(
description="Optimal solution vector x for the quadratic program."
)
objective_value: Float32 = Field(
description="Optimal objective value for the quadratic program."
)
jit_solve_qp_primal = jax.jit(
qpax.solve_qp_primal, static_argnames=("solver_tol", "target_kappa")
)
#
# Required endpoints
#
def apply(inputs: InputSchema) -> OutputSchema:
x = jit_solve_qp_primal(
Q=inputs.Q,
q=inputs.q,
A=inputs.A,
b=inputs.b,
G=inputs.G,
h=inputs.h,
solver_tol=inputs.solver_tol,
target_kappa=inputs.target_kappa,
)
obj = jnp.dot(inputs.q, x) + 0.5 * jnp.dot(x, inputs.Q @ x)
return OutputSchema(
x=x,
objective_value=obj,
)
def vector_jacobian_product(
inputs: InputSchema,
vjp_inputs: set[str],
vjp_outputs: set[str],
cotangent_vector,
):
qpsolve_signature = ["Q", "q", "A", "b", "G", "h"]
# only one output - x
tangent = cotangent_vector[vjp_outputs.pop()]
success = False
kappa = inputs.target_kappa
max_kappa = 1
while (not success) and kappa < max_kappa:
success = True
solve_fn = functools.partial(
jit_solve_qp_primal,
solver_tol=inputs.solver_tol,
target_kappa=kappa,
)
_, f_vjp = jax.vjp(
solve_fn, inputs.Q, inputs.q, inputs.A, inputs.b, inputs.G, inputs.h
)
vjps = f_vjp(tangent)
out = {}
for dx in vjp_inputs:
val = vjps[qpsolve_signature.index(dx)]
# heuristic attempts to adjust kappa if NaN values are encountered
# not sure how well this works in practice
if jnp.any(jnp.isnan(val)):
kappa *= 2
success = False
break
out[dx] = val
if not success:
raise ValueError(
"Failed to compute VJP with the given inputs. Try increasing target_kappa."
)
return out