atakan
feat: Initial open-source release of ControlAI
9936912
Raw
History Blame Contribute Delete
28.7 kB
"""Safety, aerospace, stochastic, fault-tolerant, and learning-control families with Chain of Thought (CoT)."""
from __future__ import annotations
from collections.abc import Callable
import numpy as np
from controlai_data.schema import make_record
OPENERS = (
"Provide a detailed step-by-step engineering calculation with intermediate derivations.",
"Solve this numerical control problem step-by-step without omitting intermediate math.",
"State the governing equations, compute the requested values, and verify the conclusion.",
"Give a rigorous control engineering derivation with verified numbers.",
)
def number(value: float) -> str:
if abs(value) < 5e-12:
value = 0.0
return f"{value:.10g}"
def record(
*,
index: int,
prefix: str,
domain: str,
family: str,
difficulty: str,
prompt: str,
answer: str,
ground_truth: dict,
source_refs: list[str],
verifier: str,
) -> dict:
return make_record(
record_id=f"{prefix}_{index:05d}",
domain=domain,
family=family,
task_type="numerical_reasoning",
difficulty=difficulty,
template_id=f"{prefix}_prompt_{index % 4}",
prompt=prompt,
answer=answer,
ground_truth=ground_truth,
source_refs=source_refs,
verifier=verifier,
)
def nonlinear_dynamic_inversion(_: np.random.Generator, index: int) -> dict:
a = -0.7 + 0.018 * index
b = 0.8 + 0.012 * index
x = -1.2 + 0.041 * index
reference = 0.35 - 0.006 * index
gain = 1.1 + 0.017 * index
virtual = -gain * (x - reference)
control = (virtual - a * x) / b
prompt = (
f"{OPENERS[index % 4]} For scalar affine system x_dot={number(a)}x+{number(b)}u, "
f"use exact nonlinear dynamic inversion so that x_dot=v=-{number(gain)}(x-r). "
f"At x={number(x)}, r={number(reference)}, compute the virtual control v and the required control input u."
)
answer = (
f"### 1. Virtual Control Law Definition\n"
f"To enforce the target error dynamics $\\dot{{e}} = \\dot{{x}} - \\dot{{r}} = -k(x - r)$ with setpoint $r = {number(reference)}$:\n"
f"$$v = -k (x - r) = -{number(gain)} \\times ({number(x)} - {number(reference)}) = -{number(gain)} \\times ({number(x - reference)}) = {number(virtual)}$$\n\n"
f"### 2. Exact Dynamic Inversion Algebraic Solution\n"
f"Equating the open-loop dynamics $\\dot{{x}} = a x + b u$ to the synthetic virtual input $v$:\n"
f"$$a x + b u = v \\implies u = \\frac{{v - a x}}{{b}}$$\n"
f"$$u = \\frac{{{number(virtual)} - ({number(a)})({number(x)})}}{{{number(b)}}} = \\frac{{{number(virtual)} - ({number(a * x)})}}{{{number(b)}}} = {number(control)}$$\n\n"
f"### 3. Direct Closed-Loop Verification\n"
f"$$\\dot{{x}} = ({number(a)})({number(x)}) + ({number(b)})({number(control)}) = {number(a * x)} + {number(b * control)} = {number(a * x + b * control)} = v$$\n"
f"**Summary:** Virtual control v = {number(virtual)}, control input u = {number(control)}."
)
return record(
index=index,
prefix="ndi_scalar",
domain="aerospace_control",
family="scalar_affine_nonlinear_dynamic_inversion",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "scalar_dynamic_inversion",
"a": a,
"b": b,
"x": x,
"reference": reference,
"gain": gain,
"virtual": virtual,
"u": control,
},
source_refs=["slotine_li_applied_nonlinear_control", "ndi_control_literature"],
verifier="analytic_scalar_dynamic_inversion",
)
def gain_schedule(_: np.random.Generator, index: int) -> dict:
low = np.array([0.8 + 0.01 * index, 0.25 + 0.004 * index])
high = np.array([2.0 + 0.014 * index, 0.75 + 0.006 * index])
sigma_low = 10.0
sigma_high = 30.0 + 0.1 * index
sigma = 11.0 + (index % 17) * 0.9
weight = (sigma - sigma_low) / (sigma_high - sigma_low)
scheduled = (1 - weight) * low + weight * high
prompt = (
f"{OPENERS[index % 4]} Linearly schedule gain vector K between K_low={low.tolist()} at "
f"sigma={number(sigma_low)} and K_high={high.tolist()} at sigma={number(sigma_high)}. "
f"Find K at operating point sigma={number(sigma)}; do not extrapolate or claim global stability."
)
answer = (
f"### 1. Interpolation Weight Calculation\n"
f"$$\\mu = \\frac{{\\sigma - \\sigma_{{low}}}}{{\\sigma_{{high}} - \\sigma_{{low}}}} = \\frac{{{number(sigma)} - {number(sigma_low)}}}{{{number(sigma_high)} - {number(sigma_low)}}} = {number(weight)}$$\n\n"
f"### 2. Gain Vector Linear Interpolation\n"
f"$$K(\\sigma) = (1 - \\mu) K_{{low}} + \\mu K_{{high}} = (1 - {number(weight)}) {low.tolist()} + {number(weight)} {high.tolist()} = {scheduled.tolist()}$$\n\n"
f"### 3. Engineering Context\n"
f"The scheduled gain is $K = {scheduled.tolist()}$. This gain scheduling is valid for slow parameter variation; it does not constitute a global nonlinear stability certificate."
)
return record(
index=index,
prefix="gain_schedule",
domain="aerospace_control",
family="linear_gain_schedule_interpolation",
difficulty="intermediate",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "gain_schedule_interpolation",
"low": low.tolist(),
"high": high.tolist(),
"sigma_low": sigma_low,
"sigma_high": sigma_high,
"sigma": sigma,
"weight": weight,
"scheduled": scheduled.tolist(),
},
source_refs=["gain_scheduling_control_literature"],
verifier="analytic_linear_interpolation",
)
def cbf_filter(_: np.random.Generator, index: int) -> dict:
x_min = -0.4 + 0.005 * index
x = x_min + 0.08 + 0.012 * (index % 9)
alpha = 0.8 + 0.025 * index
u_nom = -1.1 + 0.033 * index
h = x - x_min
lower = -alpha * h
safe = max(u_nom, lower)
active = u_nom < lower
prompt = (
f"{OPENERS[index % 4]} A single integrator x_dot=u must satisfy safety set x>={number(x_min)}. "
f"With barrier function h(x)=x-x_min, enforce the forward-invariance CBF condition h_dot+{number(alpha)}h>=0 "
f"while minimally modifying nominal command u_nom={number(u_nom)}. At current state x={number(x)}, compute the filtered control input u."
)
answer = (
f"### 1. Control Barrier Function (CBF) Condition\n"
f"For safety barrier $h(x) = x - x_{{min}} = {number(x)} - ({number(x_min)}) = {number(h)}$:\n"
f"$$\\dot{{h}}(x, u) + \\alpha h(x) \\ge 0 \\implies u + {number(alpha)} ({number(h)}) \\ge 0 \\implies u \\ge -{number(alpha)} \\times {number(h)} = {number(lower)}$$\n\n"
f"### 2. Quadratic Program Safety Filter Projection\n"
f"$$\\min_u \\frac{{1}}{{2}} (u - u_{{nom}})^2 \\quad \\text{{s.t.}} \\quad u \\ge {number(lower)}$$\n"
f"The analytical scalar projection is:\n"
f"$$u^* = \\max(u_{{nom}}, -\\alpha h) = \\max({number(u_nom)}, {number(lower)}) = {number(safe)}$$\n\n"
f"### 3. Verification\n"
f"- Constraint status: **{'Active (intervened to enforce safety)' if active else 'Inactive (nominal command is already safe)'}**\n"
f"- Residual safety derivative: $\\dot{{h}} + \\alpha h = {number(safe)} + {number(alpha * h)} = {number(safe + alpha * h)} \\ge 0$\n\n"
f"**Summary:** Filtered control u = {number(safe)}."
)
return record(
index=index,
prefix="cbf_scalar",
domain="safety_critical_control",
family="scalar_control_barrier_safety_filter",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "scalar_cbf_filter",
"x_min": x_min,
"x": x,
"alpha": alpha,
"u_nom": u_nom,
"h": h,
"lower_bound": lower,
"u_safe": safe,
"active": active,
},
source_refs=["ames_control_barrier_functions"],
verifier="analytic_scalar_cbf_projection",
)
def scalar_stationary_covariance(_: np.random.Generator, index: int) -> dict:
a = 0.2 + 0.009 * index
q = 0.05 + 0.003 * index
p = q / (1 - a * a)
prompt = (
f"{OPENERS[index % 4]} For discrete stochastic system x_(k+1)={number(a)}x_k+w_k with E[w_k^2]="
f"{number(q)}, compute the stationary variance P if it exists and verify the Lyapunov equation."
)
answer = (
f"### 1. Existence of Stationary Second Moment\n"
f"For $x_{{k+1}} = a x_k + w_k$, a stationary variance exists if and only if $|a| < 1$.\n"
f"Here $|a| = {number(abs(a))} < 1$, so the state process is stationary.\n\n"
f"### 2. Discrete Lyapunov Equation for Covariance\n"
f"$$P = a^2 P + Q \\implies (1 - a^2) P = Q \\implies P = \\frac{{Q}}{{1 - a^2}}$$\n"
f"$$P = \\frac{{{number(q)}}}{{1 - ({number(a)})^2}} = \\frac{{{number(q)}}}{{1 - {number(a * a)}}} = \\frac{{{number(q)}}}{{{number(1 - a * a)}}} = {number(p)}$$\n\n"
f"### 3. Verification\n"
f"$$a^2 P + Q = ({number(a * a)})({number(p)}) + {number(q)} = {number(a * a * p + q)} = P$$\n"
f"**Summary:** Stationary variance P = {number(p)}."
)
return record(
index=index,
prefix="stationary_covariance",
domain="stochastic_control",
family="scalar_stationary_process_covariance",
difficulty="intermediate",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "scalar_stationary_covariance",
"a": a,
"q": q,
"P": p,
"exists": abs(a) < 1,
},
source_refs=["anderson_moore_optimal_filtering"],
verifier="scalar_discrete_lyapunov",
)
def fault_allocation(_: np.random.Generator, index: int) -> dict:
effectiveness = np.array([1.0, 0.75 + 0.003 * index, 0.45 + 0.002 * index])
desired = -1.0 + 0.047 * index
b_norm_sq = float(effectiveness @ effectiveness)
allocation = effectiveness * desired / b_norm_sq
achieved = float(effectiveness @ allocation)
prompt = (
f"{OPENERS[index % 4]} Three redundant actuators have effectiveness row "
f"B={effectiveness.tolist()}. Find the unconstrained minimum-2-norm control command u "
f"that produces desired virtual control tau={number(desired)}, and verify Bu=tau."
)
answer = (
f"### 1. Minimum 2-Norm Control Allocation Formulation\n"
f"$$\\min_u \\frac{{1}}{{2}} \\|u\\|_2^2 \\quad \\text{{s.t.}} \\quad B u = \\tau$$\n"
f"For full-row-rank matrix $B$, the unique minimum-norm solution via Moore-Penrose pseudoinverse is:\n"
f"$$u^* = B^T (B B^T)^{{-1}} \\tau = \\frac{{\\tau}}{{B B^T}} B^T$$\n\n"
f"### 2. Evaluation\n"
f"- Denominator: $B B^T = \\|B\\|_2^2 = 1.0^2 + {number(effectiveness[1])}^2 + {number(effectiveness[2])}^2 = {number(b_norm_sq)}$\n"
f"- Multiplier: $\\frac{{\\tau}}{{\\|B\\|_2^2}} = \\frac{{{number(desired)}}}{{{number(b_norm_sq)}}} = {number(desired / b_norm_sq)}$\n"
f"- Allocated actuator commands:\n"
f" $$u^* = {allocation.tolist()}$$\n\n"
f"### 3. Virtual Control Verification\n"
f"$$B u^* = {effectiveness.tolist()} \\cdot {allocation.tolist()} = {number(achieved)}$$\n"
f"Residual: $|B u^* - \\tau| = {number(abs(achieved - desired))} \\approx 0$.\n\n"
f"**Summary:** Allocated command u = {allocation.tolist()}."
)
return record(
index=index,
prefix="fault_allocation",
domain="fault_tolerant_control",
family="minimum_norm_redundant_control_allocation",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "minimum_norm_allocation",
"B": effectiveness.tolist(),
"desired": desired,
"u": allocation.tolist(),
"achieved": achieved,
},
source_refs=["fault_tolerant_control_allocation_literature"],
verifier="pseudoinverse_allocation",
)
def jump_scalar(_: np.random.Generator, index: int) -> dict:
probability = 0.2 + 0.006 * (index % 30)
a1 = 0.45 + 0.003 * index
a2 = 1.02 + 0.002 * index
factor = probability * a1 * a1 + (1 - probability) * a2 * a2
stable = factor < 1
prompt = (
f"{OPENERS[index % 4]} An i.i.d. scalar jump linear system x_(k+1)=a_k x_k uses multiplier a1={number(a1)} "
f"with probability p={number(probability)} and a2={number(a2)} with probability 1-p={number(1 - probability)}. "
"Compute the expected squared multiplier E[a^2] and decide mean-square stability."
)
answer = (
f"### 1. Second-Moment Dynamics\n"
f"For the i.i.d. scalar jump system $x_{{k+1}} = a_k x_k$:\n"
f"$$\\mathbb{{E}}[x_{{k+1}}^2] = \\mathbb{{E}}[a_k^2] \\mathbb{{E}}[x_k^2]$$\n"
f"The system is mean-square stable if and only if $\\mathbb{{E}}[a^2] < 1$.\n\n"
f"### 2. Expected Squared Multiplier Calculation\n"
f"$$\\mathbb{{E}}[a^2] = p a_1^2 + (1 - p) a_2^2$$\n"
f"$$\\mathbb{{E}}[a^2] = ({number(probability)}) ({number(a1)})^2 + ({number(1 - probability)}) ({number(a2)})^2$$\n"
f"$$\\mathbb{{E}}[a^2] = ({number(probability)}) ({number(a1 * a1)}) + ({number(1 - probability)}) ({number(a2 * a2)}) = {number(factor)}$$\n\n"
f"### 3. Stability Conclusion\n"
f"Since $\\mathbb{{E}}[a^2] = {number(factor)}$ {'< 1' if stable else '>= 1'}, the system is **{'mean-square stable' if stable else 'not certified mean-square stable'}**."
)
return record(
index=index,
prefix="jump_scalar",
domain="stochastic_control",
family="iid_scalar_jump_mean_square_test",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "iid_jump_mean_square",
"p": probability,
"a1": a1,
"a2": a2,
"factor": factor,
"stable": stable,
},
source_refs=["markov_jump_linear_systems_literature"],
verifier="expected_square_multiplier",
)
def event_trigger(_: np.random.Generator, index: int) -> dict:
state = -1.4 + 0.05 * index
held = state + (-0.3 + 0.011 * index)
sigma = 0.12 + 0.002 * index
error = held - state
threshold = sigma * abs(state)
trigger = abs(error) >= threshold
prompt = (
f"{OPENERS[index % 4]} An event-triggered control mechanism broadcasts an update when |x_hat-x|>=sigma|x|. "
f"Given current state x={number(state)}, last transmitted state x_hat={number(held)}, and threshold parameter sigma={number(sigma)}, "
"compute both sides of the condition and decide whether an event is triggered."
)
answer = (
f"### 1. State Error and Threshold Evaluation\n"
f"- State discrepancy: $|\\hat{{x}} - x| = |{number(held)} - ({number(state)})| = |{number(error)}| = {number(abs(error))}$\n"
f"- State-dependent threshold: $\\sigma |x| = {number(sigma)} \\times |{number(state)}| = {number(threshold)}$\n\n"
f"### 2. Trigger Condition Check\n"
f"Condition $|\\hat{{x}} - x| \\ge \\sigma |x|$ evaluates to ${number(abs(error))} \\ge {number(threshold)}$, which is **{trigger}**.\n\n"
f"**Summary:** Trigger is {str(trigger).lower()}."
)
return record(
index=index,
prefix="event_trigger",
domain="networked_control",
family="relative_state_error_event_trigger",
difficulty="intermediate",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "relative_event_trigger",
"x": state,
"x_hat": held,
"sigma": sigma,
"error_abs": abs(error),
"threshold": threshold,
"trigger": trigger,
},
source_refs=["event_triggered_control_literature"],
verifier="analytic_event_threshold",
)
def tube_tightening(_: np.random.Generator, index: int) -> dict:
bound = 2.0 + 0.025 * index
error = 0.15 + 0.004 * index
tightened = bound - error
prompt = (
f"{OPENERS[index % 4]} In scalar tube MPC with state decomposition x=z+e, the true state constraint is "
f"|x|<={number(bound)}, and the robustly invariant error tube is |e|<={number(error)}. "
"Compute the Pontryagin-tightened nominal state constraint interval for z."
)
answer = (
f"### 1. Pontryagin Set Difference $\\mathbb{{Z}} = \\mathbb{{X}} \\ominus \\mathbb{{E}}$\n"
f"For state constraint set $\\mathbb{{X}} = [-{number(bound)}, {number(bound)}]$ and error tube $\\mathbb{{E}} = [-{number(error)}, {number(error)}]$:\n"
f"$$z \\in \\mathbb{{X}} \\ominus \\mathbb{{E}} \\iff |z| \\le X_{{\\max}} - E_{{\\max}}$$\n\n"
f"### 2. Tightened Bound Calculation\n"
f"$$z_{{\\max}} = {number(bound)} - {number(error)} = {number(tightened)}$$\n"
f"The nominal state $z$ must satisfy $|z| \\le {number(tightened)}$, i.e., $z \\in [{number(-tightened)}, {number(tightened)}]$.\n\n"
f"**Summary:** Tightened interval is [{number(-tightened)}, {number(tightened)}]."
)
return record(
index=index,
prefix="tube_tightening",
domain="mpc",
family="symmetric_tube_mpc_constraint_tightening",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "symmetric_tube_tightening",
"bound": bound,
"error": error,
"tightened": tightened,
"interval": [-tightened, tightened],
"nonempty": tightened >= 0,
},
source_refs=["rawlings_mayne_diehl_mpc_2e"],
verifier="interval_pontryagin_difference",
)
def discounted_return(_: np.random.Generator, index: int) -> dict:
rewards = np.array([1.0 + 0.01 * index, -0.25 + 0.004 * index, 0.6 - 0.003 * index, 0.2 + 0.002 * index])
gamma = 0.82 + 0.0015 * index
terms = rewards * gamma ** np.arange(len(rewards))
total = float(terms.sum())
prompt = (
f"{OPENERS[index % 4]} For reward sequence r={rewards.tolist()} and discount factor gamma="
f"{number(gamma)}, compute the finite discounted return G_0 = sum_(k=0)^3 gamma^k r_k."
)
answer = (
f"### 1. Discounted Return Formula\n"
f"$$G_0 = r_0 + \\gamma r_1 + \\gamma^2 r_2 + \\gamma^3 r_3$$\n\n"
f"### 2. Step-by-Step Discounted Terms\n"
f"- $k=0$: $r_0 = {number(terms[0])}$\n"
f"- $k=1$: $\\gamma r_1 = ({number(gamma)})({number(rewards[1])}) = {number(terms[1])}$\n"
f"- $k=2$: $\\gamma^2 r_2 = ({number(gamma**2)})({number(rewards[2])}) = {number(terms[2])}$\n"
f"- $k=3$: $\\gamma^3 r_3 = ({number(gamma**3)})({number(rewards[3])}) = {number(terms[3])}$\n\n"
f"### 3. Total Return\n"
f"$$G_0 = {number(terms[0])} + ({number(terms[1])}) + {number(terms[2])} + {number(terms[3])} = {number(total)}$$\n\n"
f"**Summary:** G_0 = {number(total)}."
)
return record(
index=index,
prefix="rl_return",
domain="reinforcement_learning_control",
family="finite_discounted_control_return",
difficulty="foundation",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "finite_discounted_return",
"rewards": rewards.tolist(),
"gamma": gamma,
"terms": terms.tolist(),
"return": total,
},
source_refs=["sutton_barto_reinforcement_learning_2e"],
verifier="finite_discounted_sum",
)
def nonlinear_feedforward(_: np.random.Generator, index: int) -> dict:
a = 0.5 + 0.012 * index
b = 0.9 + 0.008 * index
reference = -0.8 + 0.026 * index
equilibrium_input = a * reference ** 3 / b
residual = -a * reference ** 3 + b * equilibrium_input
prompt = (
f"{OPENERS[index % 4]} For nonlinear system x_dot=-{number(a)}x^3+{number(b)}u, find the "
f"constant feedforward control input u_eq that makes state x=r={number(reference)} an equilibrium. "
"Verify the residual; do not claim stability."
)
answer = (
f"### 1. Equilibrium Condition\n"
f"At equilibrium $x = r = {number(reference)}$, $\\dot{{x}} = 0$:\n"
f"$$\\dot{{x}} = -{number(a)} r^3 + {number(b)} u_{{eq}} = 0 \\implies u_{{eq}} = \\frac{{{number(a)} r^3}}{{{number(b)}}}$$\n\n"
f"### 2. Numerical Computation\n"
f"$$u_{{eq}} = \\frac{{{number(a)} \\times ({number(reference)})^3}}{{{number(b)}}} = \\frac{{{number(a * reference**3)}}}{{{number(b)}}} = {number(equilibrium_input)}$$\n\n"
f"### 3. Residual Verification\n"
f"$$\\dot{{x}} = -{number(a)}({number(reference)})^3 + {number(b)}({number(equilibrium_input)}) = {number(residual)}$$\n"
f"**Summary:** u_eq = {number(equilibrium_input)}."
)
return record(
index=index,
prefix="nonlinear_feedforward",
domain="nonlinear_control",
family="nonlinear_equilibrium_feedforward_input",
difficulty="intermediate",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "nonlinear_equilibrium_feedforward",
"a": a,
"b": b,
"reference": reference,
"u_eq": equilibrium_input,
"residual": residual,
},
source_refs=["khalil_nonlinear_systems"],
verifier="analytic_equilibrium_substitution",
)
def robust_safe_interval(_: np.random.Generator, index: int) -> dict:
a = 0.82 + 0.002 * index
b = 0.65 + 0.004 * index
state = -1.0 + 0.031 * index
lower_state = -2.2 - 0.005 * index
upper_state = 2.0 + 0.006 * index
disturbance = 0.08 + 0.0015 * index
nominal = -1.1 + 0.037 * index
lower_u = (lower_state + disturbance - a * state) / b
upper_u = (upper_state - disturbance - a * state) / b
safe_u = float(np.clip(nominal, lower_u, upper_u))
prompt = (
f"{OPENERS[index % 4]} For uncertain discrete system x_next={number(a)}x+{number(b)}u+w with bounded disturbance "
f"|w|<={number(disturbance)}, require x_next in [{number(lower_state)}, {number(upper_state)}] for all possible w. "
f"At current state x={number(state)}, derive the robust admissible u interval and project nominal command u_nom={number(nominal)} onto it."
)
answer = (
f"### 1. Robust Constraint Inversion under Worst-Case Disturbance\n"
f"Since $x_{{next}} = a x + b u + w$, the safety condition $x_{{min}} \\le x_{{next}} \\le x_{{max}}$ requires:\n"
f"- Worst-case lower bound ($w = -w_{{max}}$): $a x + b u - w_{{max}} \\ge x_{{min}} \\implies u \\ge \\frac{{x_{{min}} + w_{{max}} - a x}}{{b}} = {number(lower_u)}$\n"
f"- Worst-case upper bound ($w = +w_{{max}}$): $a x + b u + w_{{max}} \\le x_{{max}} \\implies u \\le \\frac{{x_{{max}} - w_{{max}} - a x}}{{b}} = {number(upper_u)}$\n\n"
f"### 2. Admissible Control Interval and Projection\n"
f"Admissible interval: $u \\in [{number(lower_u)}, {number(upper_u)}]$.\n"
f"Projecting $u_{{nom}} = {number(nominal)}$ onto this interval yields:\n"
f"$$u^* = \\text{{clip}}({number(nominal)}, {number(lower_u)}, {number(upper_u)}) = {number(safe_u)}$$\n\n"
f"**Summary:** Admissible interval = [{number(lower_u)}, {number(upper_u)}], safe control u = {number(safe_u)}."
)
return record(
index=index,
prefix="robust_safe_interval",
domain="safety_critical_control",
family="robust_one_step_safe_input_interval",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "robust_safe_input_interval",
"a": a,
"b": b,
"x": state,
"x_min": lower_state,
"x_max": upper_state,
"w_max": disturbance,
"u_nom": nominal,
"u_lower": lower_u,
"u_upper": upper_u,
"u_safe": safe_u,
},
source_refs=["ames_control_barrier_functions", "rawlings_mayne_diehl_mpc_2e"],
verifier="robust_interval_projection",
)
def actuator_loss_isolation(_: np.random.Generator, index: int) -> dict:
effectiveness = np.array([1.0, 0.8 + 0.002 * index, 0.55 + 0.0015 * index])
command = np.array([0.4 + 0.011 * index, -0.7 + 0.008 * index, 0.9 - 0.006 * index])
loss_factor = 0.35 + 0.004 * (index % 20)
failed = (index - 1) % 3
candidate_outputs = []
for candidate in range(3):
candidate_effectiveness = effectiveness.copy()
candidate_effectiveness[candidate] *= loss_factor
candidate_outputs.append(float(candidate_effectiveness @ command))
measured = candidate_outputs[failed]
residuals = np.abs(np.asarray(candidate_outputs) - measured)
identified = int(np.argmin(residuals))
prompt = (
f"{OPENERS[index % 4]} Nominal actuator effectiveness is B={effectiveness.tolist()} "
f"and commanded input is u={command.tolist()}. Exactly one actuator retains effectiveness factor "
f"lambda={number(loss_factor)} while the others remain nominal. The measured virtual control is tau={number(measured)}. "
"Compute each single-fault hypothesis prediction/residual and isolate the failed actuator."
)
answer = (
f"### 1. Single-Fault Hypothesis Testing\n"
f"For each actuator $i \\in \\{{1, 2, 3\\}}$, compute predicted virtual control $\\tau_i = B^{{(i)}} u$ with $B^{{(i)}}_i = \\lambda B_i$:\n"
f"- Hypothesis 1 (Actuator 1 degraded): $\\tau_1 = {number(candidate_outputs[0])}$, residual $r_1 = |\\tau_1 - \\tau| = {number(residuals[0])}$\n"
f"- Hypothesis 2 (Actuator 2 degraded): $\\tau_2 = {number(candidate_outputs[1])}$, residual $r_2 = |\\tau_2 - \\tau| = {number(residuals[1])}$\n"
f"- Hypothesis 3 (Actuator 3 degraded): $\\tau_3 = {number(candidate_outputs[2])}$, residual $r_3 = |\\tau_3 - \\tau| = {number(residuals[2])}$\n\n"
f"### 2. Fault Isolation Decision\n"
f"The minimum residual is $r_{{{identified + 1}}} = {number(residuals[identified])} \\approx 0$.\n\n"
f"**Summary:** Actuator {identified + 1} is isolated as the degraded unit."
)
return record(
index=index,
prefix="actuator_loss",
domain="fault_tolerant_control",
family="single_actuator_loss_residual_isolation",
difficulty="advanced",
prompt=prompt,
answer=answer,
ground_truth={
"kind": "single_actuator_loss_isolation",
"B": effectiveness.tolist(),
"u": command.tolist(),
"loss_factor": loss_factor,
"measured": measured,
"candidate_outputs": candidate_outputs,
"residuals": residuals.tolist(),
"failed_index": failed,
"identified_index": identified,
},
source_refs=["fault_tolerant_control_allocation_literature"],
verifier="single_fault_hypothesis_residuals",
)
FAMILIES: tuple[tuple[str, Callable[[np.random.Generator, int], dict]], ...] = (
("scalar_affine_nonlinear_dynamic_inversion", nonlinear_dynamic_inversion),
("linear_gain_schedule_interpolation", gain_schedule),
("scalar_control_barrier_safety_filter", cbf_filter),
("scalar_stationary_process_covariance", scalar_stationary_covariance),
("minimum_norm_redundant_control_allocation", fault_allocation),
("iid_scalar_jump_mean_square_test", jump_scalar),
("relative_state_error_event_trigger", event_trigger),
("symmetric_tube_mpc_constraint_tightening", tube_tightening),
("finite_discounted_control_return", discounted_return),
("nonlinear_equilibrium_feedforward_input", nonlinear_feedforward),
("robust_one_step_safe_input_interval", robust_safe_interval),
("single_actuator_loss_residual_isolation", actuator_loss_isolation),
)
def generate_extended_v1(count_per_family: int, seed: int) -> list[dict]:
rng = np.random.default_rng(seed)
return [
generator(rng, index)
for _, generator in FAMILIES
for index in range(1, count_per_family + 1)
]