Planning

Planning problems: the deterministic task, its stochastic class, and their parameters.

class minilink.planning.problems.PlanningProblem(sys, x_start=None, x_goal=None, cost=None, X=None, U=None, X0=None, Xf=None, tf=None, params=None, metadata=None, infeasible_cost=None)

Deterministic planning problem for a Minilink system.

The generic mathematical form is x(t) in X(t), u(t) in U(x(t), t), with boundary sets X0 and optional Xf. A cost can be attached for optimization and policy-planning solvers, but feasibility/search planners may leave it unset.

The problem is the continuous task: the system, the admissible and boundary sets, the continuous horizon tf, the optional cost and parameter bundle. The knot count and the other discretization choices belong to the solvers, not to the problem.

Parameters:

• sys (System) – Minilink system or diagram to plan for.

• x_start (array_like, optional) – Representative initial state. When X0 is omitted, this creates a singleton X0. When X0 is provided, it must belong to X0 and is used by solvers or guesses that need one concrete start point. Defaults to singleton X0 when available, otherwise sys.x0.

• x_goal (array_like, optional) – Representative terminal target. When Xf is omitted, this creates a singleton Xf. When Xf is provided, it must belong to Xf and is used by initial guesses or reports that need one concrete target.

• cost (CostFunction, optional) – Planning cost. Required by solvers that optimize an objective.

• X (Set, optional) – State constraint x(t) in X: a hard constraint, leaving it is a failure. Unconstrained by default (the whole state space); pass X=sys.state.box to constrain the plant to its declared state range, or any set (sys.state.box & free).

• U (InputSet, optional) – Input constraint u(t) in U(x, t). Defaults to the input ports’ box: an actuator limit is part of the plant model.

• X0 (Set, optional) – Initial and terminal boundary sets. These are the authoritative feasibility constraints; x_start and x_goal are shortcuts or representative points.

• Xf (Set, optional) – Initial and terminal boundary sets. These are the authoritative feasibility constraints; x_start and x_goal are shortcuts or representative points.

• tf (float, optional) – Continuous planning horizon length in seconds. None means unset (RRT / some DP tasks). +inf means infinite-horizon. Finite trajopt / MPC grids require a finite tf — demos should always set it explicitly. Knot count N lives on transcription options.

• params (ProblemParameters, optional) – Explicit parameter bundle for system, cost, and set evaluation.

• infeasible_cost (float or callable, optional) – The finite stand-in for the infinite cost of an infeasible trajectory (one that leaves X): a scalar, or infeasible_cost(x, t) evaluated at the exit state. Value iteration reads it as its default out_of_bound_cost; the learners and the Monte Carlo score charge it when a trajectory leaves X. Unset, the sampling tools derive a bound of any feasible cost and announce it; a declared price should exceed the cost-to-go of finishing the task from a typical start.

• metadata (Mapping[str, object] | None)

property has_goal: bool

Return True when a terminal goal or boundary set is available.

property has_cost: bool

Return True when the problem has a cost function.

property is_stochastic: bool

True when the problem draws starts, parameters or disturbances.

infeasible_penalty(x, t=0.0)

The declared price of infeasibility at (x, t) (scalar or callable); None when unset.

horizon_kind()

"finite" when tf is a finite length (h is charged there), "infinite" for +inf or unset.

Return type:

str

require_cost()

Refuse a problem without a cost; return the cost.

Return type:

CostFunction

require_goal()

Refuse a problem without a terminal set; return it.

Return type:

Set

require_finite_tf()

Refuse an unset or infinite horizon; return tf as a float.

Return type:

float

class minilink.planning.problems.StochasticPlanningProblem(sys, x_start=None, x_goal=None, cost=None, X=None, U=None, X0=None, Xf=None, tf=None, params=None, metadata=None, infeasible_cost=None, x0_distribution=None, params_distribution=None, disturbances=None, criterion='expectation')

Planning problem with probabilistic uncertainty (the stochastic class).

Same spine as PlanningProblem — the deterministic problem is its base class, so every planner that accepts one accepts this — plus the uncertainty channels and the criterion:

Parameters:

• x0_distribution (Distribution) – Law of the initial state x(0) ~ p(x0). Its mean is the default x_start and its support (when it has one) the default X0.

• params_distribution (mapping, optional) – Mapping of parameter names to Distribution over entries of sys.params (domain randomization, robustness sweeps); each sample overrides those entries. A dotted name reaches into a diagram’s subsystem params (sys.mass for the plant inside a closed loop).

• disturbances (mapping, optional) – Mapping of input port ids to Distribution on sys: a fresh draw per step held on that port (a seeded disturbance signal).

• criterion ({"expectation", "worst_case"}) – What “optimal” means over the draws. Reinforcement learning optimizes the expectation; Monte Carlo evaluation reports both.

• sys (System)

• x_start (ndarray | None)

• x_goal (ndarray | None)

• cost (CostFunction | None)

• X (Set | None)

• U (InputSet | None)

• X0 (Set | None)

• Xf (Set | None)

• tf (float | None)

• params (ProblemParameters | None)

• metadata (Mapping[str, object] | None)

• infeasible_cost (object)

Notes

Solve verbs (RL, gain search) and the evaluate verb (Monte Carlo) both read this description; nominal() returns the certainty-equivalent PlanningProblem for trajectory optimization and LQR.

property is_stochastic: bool

True when the problem draws starts, parameters or disturbances.

sample_x0(key, n=None)

Draw initial states: (n,) for one, (n_samples, n) with n.

sample_params(key)

Draw one {name: value} override of sys.params (empty if none), nominal shapes.

sample_disturbances(key)

Draw one {port_id: value} of held disturbance inputs (empty if none).

nominal()

Certainty-equivalent deterministic problem: the mean start, no draws.

The start law’s mean is the one start (a singleton X0): the support X0 of the start law and an explicit x_start are dropped, and the parameter and disturbance draws with them.

Return type:

PlanningProblem

class minilink.planning.problems.ProblemParameters(system=None, cost=None, sets=None, scene=None)

Optional scenario-level parameter bundle for planning.

Parameters:

• system (object, optional) – Parameters for the system dynamics; a solver may freeze them at compile time.

• cost (object, optional) – Parameters passed to CostFunction.

• sets (object, optional) – Parameters passed to allowable set objects.

• scene (object, optional) – Spatial overrides of the scene bind (ObstacleBank / SceneParameters), not implemented: always None, and an online params={"scene": …} raises NotImplementedError.

minilink.planning.problems.as_stochastic(problem)

The same task as a stochastic problem, the inverse of StochasticPlanningProblem.nominal().

A stochastic problem is returned unchanged. A deterministic one keeps every field and starts from its single x_start, with no parameter or disturbance draws: it trains and evaluates from one start, so coverage of other starts comes from giving a stochastic problem a start distribution.

Parameters:

problem (PlanningProblem)

Return type:

StochasticPlanningProblem

The planner base: a planner solves a declarative problem into a stored solution.

class minilink.planning.planner.Planner(problem)

Base class for planners.

Concrete planners live in family subpackages: trajectory_optimization/, search/ (RRT family), policy_synthesis/ (dynamic programming) and reinforcement_learning/. Every solve returns a PlanningSolution and stores it as last_solution; the PlanningProblem stays declarative and does not solve itself.

Parameters:

problem (PlanningProblem) – Declarative planning problem consumed by this planner.

Notes

Offline entry is solve(), which returns and stores the PlanningSolution; get_controller() is its policy, plot_solution() and animate_solution() its trajectory, plot_control_law() and plot_cost_to_go() its law and cost-to-go.

accepts_stochastic = False

Planners that read the uncertainty of a stochastic problem set this; the others read its start set X0 and its x_start, and warn.

solve(**kwargs)

Offline solve entry: the planner’s verb, each planner’s own solve.

A trajectory planner’s solve runs solve_trajectory(). A policy planner’s solve is the solve itself, and solve_policy(), the family name every policy planner answers, calls it.

solve_trajectory(**kwargs)

Fixed (offline) traj-family solve.

Return type:

PlanningSolution

solve_trajectory_from(x0, **kwargs)

Online traj-family solve from measured x0.

Return type:

PlanningSolution

solve_policy(**kwargs)

Fixed (offline) policy-family solve.

Return type:

PlanningSolution

require_solution()

Refuse a planner with no solution yet; return the latest one.

Return type:

PlanningSolution

get_controller()

The solution’s policy: a feedback block, or the open-loop source of a trajectory planner.

nominal_trajectory(tf=None)

The solution’s policy from the problem’s start on the planner’s grid (feedback planners).

Return type:

Trajectory

require_cost()

Refuse a problem without a cost; return problem.cost.

Return type:

CostFunction

require_goal()

Refuse a problem without a goal set; return problem.Xf.

Return type:

Set

plot_solution(*, signals=('x', 'u'), backend='matplotlib')

Plot the latest solution’s trajectory with the problem system.

animate_solution(**kwargs)

Animate the latest solution’s trajectory with the problem system.

plot_control_law(**kwargs)

The latest solution’s law over the problem’s box (PlanningSolution.plot_control_law()).

plot_cost_to_go(**kwargs)

The latest solution’s cost-to-go field (PlanningSolution.plot_cost_to_go()).

The planning solution: the policy, its cost-to-go, and the evidence for them.

class minilink.planning.results.PlanningSolution(problem, policy, solver, trajectory, evaluation, cost_to_go=None)

What every planner returns: the optimal-control pair and the evidence for it.

Everything here is the solver’s own claim under its own model — the table on its grid, the Riccati form on the linear model, the critic’s fit. A measurement of the law comes from the evaluators (MonteCarloEvaluator, PolicyEvaluator) as its own object and is never written back here.

Parameters:

• problem (PlanningProblem) – What was solved: the system, its sets, the cost and the horizon. A frozen record; its sys is a live reference, not a snapshot.

• policy (System) – The law. Feedback u = pi(x) is a controller block, policy @ plant closes the loop; open loop u = pi(t) is a TrajectorySource, policy >> plant drives the plant; a finite-horizon design is feedback in time, u = pi(x, t).

• solver (record) – The planner’s account of the solve, one dataclass per planner (TrajectoryOptimizationRecord, TreeSearchRecord, ValueIterationRecord, TabularLearningRecord, ReinforcementLearningRecord, RiccatiRecord), each with a success flag and a one-line str.

• trajectory (Trajectory or None) – What the policy produces from the problem’s start on the planner’s grid with nominal parameters: the schedule of an open-loop planner, the rollout of a feedback law when the solve was asked to evaluate. It is the planner’s evidence, not a closed-loop simulation: close the loop in the script, policy @ plant, and simulate that.

• evaluation (Evaluation or None) – The cost over the problem’s draws under the one scoring contract (one trial on a deterministic problem), when the solve was asked to evaluate; None otherwise, or when the problem declares no cost.

• cost_to_go (callable or None) – J(x), the method’s own estimate under its own model, where the method produces one: the interpolated table of dynamic programming and tabular learning, the Riccati form of a linear-quadratic design, the critic of a learned law when it estimates the problem’s own discount. On a stochastic problem it is the expected cost-to-go when the method takes the distributions.

property success: bool

feasible, converged, goal reached, finite weights, stable.

Type:

The solver record’s own success

property open_loop: bool

True for a time-based policy (a source with no input port).

property method: str

The planning method, read from the solver record ("value iteration", "riccati", …).

plot_control_law(**kwargs)

Draw the feedback law over the problem’s state box.

The policy’s own plot_control_law(), with bounds filled from the problem’s constraint set (or the plant’s state box) and the colour limits from its input set, so that laws of one problem share one scale. Every keyword passes through (x_axis, y_axis, u_axis, ax, show, …).

plot_cost_to_go(**kwargs)

Draw the method’s cost-to-go over two state axes, the others pinned at the goal.

A heatmap of cost_to_go sampled on the problem’s state box; see plot_cost_to_go() for axes, anchor, bounds, jmax, ax and show.

plot_trajectory(*, signals=('x', 'u'), backend='matplotlib', show=True)

Draw trajectory on the problem’s system: the plan, or the nominal rollout.

The figure says which. A closed-loop simulation is not this: wire policy @ plant and plot that diagram’s trajectory.

plot_cost(*, backend='matplotlib', show=True)

Draw the running cost and J(t) along trajectory under the problem’s cost.

Several solutions of one problem side by side: a table, the laws, the cost-to-go fields, the plans.

minilink.planning.comparison.compare(*args, **solutions)

Name solutions of one problem for a side-by-side reading.

compare(VI=vi, LQR=lqr) or compare({"VI": vi, "LQR": lqr}).

Return type:

Comparison

class minilink.planning.comparison.Comparison(solutions)

Named solutions of one problem and the overlays that read them together.

solutions maps a name to a PlanningSolution (compare(VI=vi, LQR=lqr, PPO=ppo) builds one). print is the table of their records, and each plot_* draws every solution on one figure with one scale. The verbs read what the planners produced and nothing else: closing a loop and simulating it stays in the script, solution.policy @ plant. The first solution’s system labels the axes.

Parameters:

solutions (dict)

evaluate(evaluator)

Score every policy with one evaluator, the same draws for all.

Returns a new comparison whose solutions carry that evaluation, so print(race.evaluate(evaluator)) is the table with one yardstick. A Monte Carlo evaluator’s backend="auto" is chosen once for the whole comparison: NumPy for every policy when one of them does not trace on JAX, JAX otherwise.

Return type:

Comparison

plot_control_law(*, show=True, **kwargs)

One panel per feedback law, side by side, one colour scale from the problem’s inputs.

plot_cost_to_go(*, jmax=None, show=True, **kwargs)

One panel per cost-to-go field, side by side, one colour scale (jmax or the largest finite value).

plot_trajectory(*, signals=('x', 'u'), show=True)

The plans and nominal rollouts overlaid, one row per signal component, legend by name.

minilink.planning.comparison.plot_cost_to_go(solution, *, axes=(0, 1), anchor=None, bounds=None, grid_shape=(51, 51), jmax=None, t=0.0, cmap='YlOrRd', ax=None, title=None, show=True)

Heatmap of a solution’s cost_to_go over two state axes, the others pinned at anchor.

bounds (((xlo, xhi), (ylo, yhi))) defaults to the problem’s constraint box, else the plant’s state box; anchor to the goal, else the start; jmax clips the colour scale.

minilink.planning.comparison.sample_cost_to_go(solution, *, axes=(0, 1), anchor=None, bounds=None, grid_shape=(51, 51), t=0.0)

(x_level, y_level, J): the cost-to-go sampled on a mesh over two state axes.

Monte Carlo evaluation: a policy’s cost on a planning problem, over one trajectory or the problem’s draws.

class minilink.planning.evaluation.MonteCarloEvaluator(problem, *, dt=0.05, n_trials=100, episode_length=None, backend='auto', seed=0, record=False)

Score a policy over the draws of a planning problem.

Parameters:

• problem (StochasticPlanningProblem) – Task (plant, cost, constraint set and its price, horizon, distributions). A deterministic PlanningProblem is scored from its single start.

• dt (float) – Control period (the law is held over it) and sampling step.

• n_trials (int) – Number of draws.

• episode_length (float, optional) – Duration of each trial for an infinite-horizon problem (default 10 s).

• backend ({"auto", "jax", "numpy", "simulator"}) – "jax" vmaps the held-input rollout over the trials (static laws), "numpy" runs the same trials one at a time, "simulator" integrates the continuous-time loop (any controller, no parameter or disturbance draws). "auto" (default) is JAX when it is installed and the plant, the law, X and the cost trace on it, NumPy otherwise (a lookup table’s law, say); that fallback warns when it changes the draws.

• seed (int) – Seed of the draws (the same seed gives the same draws on the JAX and NumPy backends only through a Particles start distribution; the two random streams differ otherwise).

• record (bool) – Keep the trial trajectories (NumPy and simulator backends).

property tf: float

the finite horizon, else episode_length (10 s by default).

Type:

Duration of a trial

evaluate(controller)

Return the Evaluation of a policy block (or of a solution’s policy) on the problem.

Return type:

Evaluation

class minilink.planning.evaluation.Evaluation(J, failed, x0, trajectories=None)

The cost of a policy on the problem’s draws: per-trial J and its summary.

failed marks the trials that left the allowed box; x0 holds the starts; trajectories the recorded trials when asked for. One trial is a deterministic problem’s whole evaluation.

Parameters:

• J (ndarray)

• failed (ndarray)

• x0 (ndarray)

• trajectories (list | None)

classmethod of_trajectory(problem, traj)

One trial: a sampled trajectory scored under the problem’s contract.

Parameters:

traj (Trajectory)

Return type:

Evaluation

property n_trials: int

Number of trials.

property mean: float

Mean cost over the trials.

property std: float

Standard deviation of the cost over the trials.

property worst: float

Largest cost over the trials.

property failure_rate: float

Fraction of the trials that left the allowed box.

value(criterion='expectation')

The problem’s criterion: "expectation" (mean) or "worst_case" (max).

Return type:

float

minilink.planning.evaluation.score_trajectory(problem, traj, params=None, infeasible_cost=inf)

Return (J, failed) of a sampled closed-loop trajectory under the problem’s contract.

The one scoring contract every tool reports with: the discounted trapezoid of g, cut at the first exit from X, plus h at a reached finite horizon.

traj carries the states and the applied inputs on a time grid; params are the cost parameters. failed is True when the trajectory left X (read at each sample time on the problem’s set parameters) before the end of the grid; the trial is then charged the problem’s infeasible_cost, else infeasible_cost given here (an evaluator passes the bound its environment derived; +inf otherwise).

Parameters:

traj (Trajectory)

minilink.planning.evaluation.nominal_trajectory(problem, policy, *, dt, tf=None)

The policy from the problem’s start on a control grid, nominal parameters and disturbances.

One recorded trial of the NumPy evaluator on the certainty-equivalent problem: the trajectory a feedback planner reports beside its policy.

Return type:

Trajectory

Initial guesses: the first trajectory a planning problem’s solver starts from.

minilink.planning.initial_guess.default_initial_trajectory(problem, t)

Return the default planning-domain trajectory guess for problem.

Mechanical systems get a cubic position/velocity guess; other systems get a linear state interpolation.

Parameters:

• problem (PlanningProblem)

• t (ndarray)

Return type:

Trajectory

minilink.planning.initial_guess.mechanical_cubic_initial_trajectory(problem, t)

Return a cubic generalized-coordinate guess for mechanical systems.

The expected state convention is x = [q; dq]. The guess is a cubic Hermite curve matching boundary positions and velocities.

Parameters:

• problem (PlanningProblem)

• t (ndarray)

Return type:

Trajectory

minilink.planning.initial_guess.linear_initial_trajectory(problem, t)

Return a linear state interpolation with nominal input.

This generic guess works for any state-space system. It uses singleton boundary sets when present and falls back to problem.x_start / problem.x_goal shortcuts.

Parameters:

• problem (PlanningProblem)

• t (ndarray)

Return type:

Trajectory

Trajectory optimization: a planning problem transcribed into a nonlinear program and solved.

class minilink.planning.trajectory_optimization.planner.TrajectoryOptimizationPlanner(problem, *, transcription=None, n_steps=None, options=None, compile_backend=<object object>, initial_guess=<object object>, warm_start=<object object>, optimizer_method=<object object>, optimizer_options=<object object>, use_hessian=<object object>, record_history=<object object>, callback=<object object>, record_solve_time=<object object>, verbose=<object object>, feasibility_tol=<object object>, live_plot=<object object>)

Generic trajectory-optimization planner.

A transcription owns the decision-vector layout and method-specific constraints. This planner owns the domain workflow: problem -> transcription -> mathematical program -> optimizer -> trajectory.

Offline solve_trajectory() always rebuilds a fresh MathematicalProgram. Online solve_trajectory_from() rebuilds with the measured x0 unless compile_parametric_program() has been called, in which case each from-solve is bind + solve only.

Parameters:

• problem (PlanningProblem)

• transcription (Transcription | str | None)

• n_steps (int | None)

• options (TrajectoryOptimizationOptions | None)

property has_parametric_program: bool

True after a successful compile_parametric_program().

solve(*, initial_guess=None, warm_start=None, evaluate=False, n_trials=50)

Offline trajopt entry (solve_trajectory).

The solution’s policy is the optimized input as a source block, policy >> plant; evaluate=True also replays it over the problem’s draws (one trial when nothing is random).

Parameters:

• initial_guess (ndarray | Trajectory | None)

• warm_start (bool | None)

• evaluate (bool)

• n_trials (int)

Return type:

PlanningSolution

solve_trajectory(*, initial_guess=None, warm_start=None, evaluate=False, n_trials=50)

Compute and store a trajectory-optimization solution (always rebuild).

Parameters:

• initial_guess (ndarray | Trajectory | None)

• warm_start (bool | None)

• evaluate (bool)

• n_trials (int)

Return type:

PlanningSolution

compile_parametric_program()

Build and JIT-compile a parametric NLP once (idempotent).

Requires options.compile_backend='jax' and a transcription that implements transcribe_parametric() (direct collocation).

Return type:

None

solve_trajectory_from(x0, *, params=None, initial_guess=None, evaluate=False, n_trials=50)

Online traj-family solve from measured x0.

Parameters:

• x0 (array_like) – Measured initial state.

• params (mapping, optional) – Scene / bind façade. None or {} binds x0 only. Non-empty keys are refused (scene is reserved for the scene bind).

• initial_guess (ndarray or Trajectory, optional) – NLP seed (schedule and/or packed z). Warm-start policy (reuse last latch) is owned by ModelPredictiveController, not this method.

• evaluate (bool)

• n_trials (int)

Return type:

PlanningSolution

Notes

Fast bind path when has_parametric_program; otherwise rebuilds the NLP with x_start replaced (one-time warning suggesting compile_parametric_program()).

step(x0, *, initial_guess=None)

Solve one from-tick and return the bare schedule (hand loops).

Parameters:

initial_guess (ndarray | Trajectory | None)

Return type:

Trajectory

class minilink.planning.trajectory_optimization.planner.TrajectoryOptimizationOptions(compile_backend='numpy', initial_guess=None, warm_start=False, optimizer_method='scipy_slsqp', optimizer_options=<factory>, use_hessian=False, record_history=False, callback=None, record_solve_time=False, verbose=False, feasibility_tol=1e-06, live_plot=False)

Generic trajectory-optimization workflow options.

verbose prints the Minilink trajectory-optimization preamble and report on TrajectoryOptimizationPlanner.solve(), and one timing line per tick on the online TrajectoryOptimizationPlanner.solve_trajectory_from() path. It is separate from SciPy’s per-iteration options['disp'].

record_history=True reconstructs a full Trajectory per optimizer iterate — convenient for live plots and teaching, but it adds one rollout/unpack per iteration; leave it off for production solves.

Parametric compile (TrajectoryOptimizationPlanner.compile_parametric_program()) requires compile_backend='jax'.

Parameters:

• compile_backend (str)

• initial_guess (ndarray | Trajectory | None)

• warm_start (bool)

• optimizer_method (str)

• optimizer_options (dict[str, object])

• use_hessian (bool)

• record_history (bool)

• callback (Callable[[TrajectoryOptimizationIteration], None] | None)

• record_solve_time (bool)

• verbose (bool)

• feasibility_tol (float)

• live_plot (bool)

feasibility_tol: float = 1e-06

success is granted when the returned plan satisfies the program constraints (equality residuals, inequality margins, bounds) to this tolerance, even if the solver stopped on an iteration limit.

live_plot: bool = False

Redraw the iterate trajectory during the solve (matplotlib); a custom callback takes precedence.

class minilink.planning.trajectory_optimization.planner.TrajectoryOptimizationRecord(cost, solver_success, message, iterations, solve_time_s, feasibility_tol, max_equality_violation, min_inequality_margin, max_bound_violation, feasible, stats)

The optimizer’s account of one solve, and the plan’s constraint check.

success is the returned plan satisfies the program constraints to ``feasibility_tol``, nothing else: an iteration-limit stop on a feasible plan is not a failure, a solver that reports convergence on an infeasible plan is. The solver’s own flag and message stay beside it. An online tick (solve_trajectory_from on a compiled program) runs no residual check, feasible is None and success is the solver’s flag.

Parameters:

• cost (float | None)

• solver_success (bool)

• message (str)

• iterations (int | None)

• solve_time_s (float | None)

• feasibility_tol (float)

• max_equality_violation (float | None)

• min_inequality_margin (float | None)

• max_bound_violation (float | None)

• feasible (bool | None)

• stats (dict)

property success: bool

The plan is feasible to feasibility_tol; on an online tick, the solver’s flag.

class minilink.planning.trajectory_optimization.planner.TrajectoryOptimizationIteration(iteration, z, trajectory, cost, max_eq, min_ineq)

One optimizer iterate as the planner reports it: z, its plan, its cost and its residuals.

Parameters:

• iteration (int)

• z (ndarray)

• trajectory (Trajectory)

• cost (float)

• max_eq (float)

• min_ineq (float | None)

minilink.planning.trajectory_optimization.planner.require_online_params(params)

Refuse online params that solve_trajectory_from / MPC cannot bind.

None or {} → bind x0 only. The reserved key scene raises NotImplementedError. Any other key raises ValueError.

Return type:

None

Transcriptions: a planning problem as a finite-dimensional program, and back.

class minilink.planning.trajectory_optimization.transcription.Transcription

Base class for finite-dimensional trajectory transcriptions.

A transcription maps a continuous planning problem to a finite-dimensional mathematical program, then reconstructs a trajectory from the optimizer output.

supports_parametric = False

True when transcribe_parametric() builds a correct parametric program (the receding-horizon / MPC path). Checked by the planner instead of hasattr so a subclass cannot inherit a wrong one.

abstract transcribe(problem, *, compile_backend='numpy')

Convert problem into a finite-dimensional mathematical program.

Parameters:

• problem (PlanningProblem)

• compile_backend (str)

Return type:

MathematicalProgram

abstract pack_initial_guess(problem, guess)

Pack a trajectory or array guess into the transcription decision vector.

Parameters:

• problem (PlanningProblem)

• guess (ndarray | Trajectory | None)

Return type:

ndarray

abstract reconstruct_result(result, *, problem, compile_backend='numpy')

Convert an optimizer result back into a trajectory.

Parameters:

• result (OptimizationResult)

• problem (PlanningProblem)

• compile_backend (str)

Return type:

Trajectory

abstract initial_guess_time_grid(problem)

Return the time grid used by generic trajectory guesses.

Parameters:

problem (PlanningProblem)

Return type:

ndarray

class minilink.planning.trajectory_optimization.transcription.FixedGridOptions(n_steps)

Uniform fixed-grid options — knot count only; horizon is problem.tf.

Parameters:

n_steps (int)

t(problem)

Time grid from finite problem.tf and n_steps.

Parameters:

problem (PlanningProblem)

Return type:

ndarray

dt(problem)

Step size from finite problem.tf and n_steps.

Parameters:

problem (PlanningProblem)

Return type:

float

minilink.planning.trajectory_optimization.transcription.trajectory_cost(cost, x, u, t, dt, params)

Sampled Bolza objective on a uniform grid.

x and u are knot matrices with shape (dim, N); the running cost is integrated with the trapezoid rule.

minilink.planning.trajectory_optimization.transcription.running_cost_samples(cost, x, u, t, params)

Sample the running cost g(x_k, u_k, t_k) at every knot.

NumPy inputs use a plain loop; JAX inputs vectorize with vmap so the whole-program jit stays compact.

minilink.planning.trajectory_optimization.transcription.trapezoid_integral(values, dt)

Trapezoid rule on a uniform grid.

minilink.planning.trajectory_optimization.transcription.trapezoidal_defect(x, dx, dt)

Trapezoidal collocation defect between neighboring knots.

x and dx have shape (n, N); the result has shape (n, N-1).

minilink.planning.trajectory_optimization.transcription.rk4_step_between_knots(f, x, u0, u1, t, dt)

One RK4 step from knot k to k+1 with linear input interpolation.

Direct collocation: a trajectory transcribed as knots joined by trapezoidal defects.

class minilink.planning.trajectory_optimization.direct_collocation.DirectCollocationOptions(n_steps)

Grid options for direct-collocation transcriptions.

Parameters:

n_steps (int)

class minilink.planning.trajectory_optimization.direct_collocation.DirectCollocationTranscription(options)

Fixed-time-grid trapezoidal direct-collocation transcription.

The decision vector packs every sampled state followed by every sampled input, both in (dim, N) row-major order:

z = [x[0, :], ..., x[n-1, :], u[0, :], ..., u[m-1, :]].

Parameters:

options (DirectCollocationOptions)

supports_parametric = True

True when transcribe_parametric() builds a correct parametric program (the receding-horizon / MPC path). Checked by the planner instead of hasattr so a subclass cannot inherit a wrong one.

transcribe(problem, *, compile_backend='numpy')

Build the trapezoidal collocation nonlinear program.

Parameters:

• problem (PlanningProblem)

• compile_backend (str)

Return type:

MathematicalProgram

transcribe_parametric(problem, *, compile_backend='jax')

Build a JAX parametric collocation program with runtime x0.

Equality constraints are h(z, x0): dynamics / terminal (fixed) plus the initial-state residual x[:, 0] - x0. Requires singleton X0.

Parameters:

• problem (PlanningProblem)

• compile_backend (str)

reconstruct_result(result, *, problem, compile_backend='numpy', dynamics=None)

Read (x, u) from the optimizer result.

Prefer a cached dynamics callable (parametric compile path); otherwise build one via compile_backend.

Parameters:

• result (OptimizationResult)

• problem (PlanningProblem)

• compile_backend (str)

Return type:

Trajectory

decision_dimension(problem)

Return the packed decision-vector dimension for problem.

Parameters:

problem (PlanningProblem)

Return type:

int

pack(x, u, problem)

Pack sampled state and input matrices into one decision vector.

Parameters:

• x (ndarray)

• u (ndarray)

• problem (PlanningProblem)

Return type:

ndarray

unpack(z, problem)

Unpack a decision vector into sampled (x, u) matrices.

Parameters:

• z (ndarray)

• problem (PlanningProblem)

Return type:

tuple[ndarray, ndarray]

pack_initial_guess(problem, guess)

Return a packed initial guess for the mathematical program.

Parameters:

• problem (PlanningProblem)

• guess (ndarray | Trajectory | None)

Return type:

ndarray

initial_guess_time_grid(problem)

Return the collocation time grid.

Parameters:

problem (PlanningProblem)

Return type:

ndarray

decision_bounds(problem)

Build box bounds on the packed decision vector when sets expose boxes.

Parameters:

problem (PlanningProblem)

Return type:

tuple[ndarray, ndarray]

Multiple shooting: a trajectory transcribed as knots joined by RK4 shooting defects.

class minilink.planning.trajectory_optimization.multiple_shooting.MultipleShootingOptions(n_steps)

Grid options for fixed-step multiple shooting.

Parameters:

n_steps (int)

class minilink.planning.trajectory_optimization.multiple_shooting.MultipleShootingTranscription(options)

Fixed-grid multiple-shooting transcription.

The decision vector is the same as direct collocation: z = [x[0, :], ..., x[n-1, :], u[0, :], ..., u[m-1, :]]. Dynamics are enforced by RK4 shooting defects between neighboring knots.

Parameters:

options (MultipleShootingOptions)

supports_parametric = False

True when transcribe_parametric() builds a correct parametric program (the receding-horizon / MPC path). Checked by the planner instead of hasattr so a subclass cannot inherit a wrong one.

transcribe(problem, *, compile_backend='numpy')

Build the multiple-shooting nonlinear program.

Parameters:

• problem (PlanningProblem)

• compile_backend (str)

Return type:

MathematicalProgram

transcribe_parametric(*args, **kwargs)

Refuse: multiple shooting has no parametric (MPC) form.

Spatial

Spatial scenes: hard obstacles and soft workspace fields on the workspace.

class minilink.planning.spatial.scene.Scene(obstacles=(), workspace_fields=())

Hard Shape obstacles and soft WorkspaceField sources on the workspace.

A scene composes into a PlanningProblem through its state fields:

X = bounds & scene.clearance_field(body).as_constraint()
cost = base + w * scene.cost_field(body).as_cost()

Parameters:

• obstacles (tuple)

• workspace_fields (tuple)

clearance(p, t=0.0, params=None)

Signed distance from p to the nearest obstacle.

cost_density(p, t=0.0, params=None)

Sum of workspace penalty densities at p.

clearance_field(body)

The clearance of body’s probes to the obstacles, as a field over the state.

Parameters:

body (CollisionBody)

Return type:

StateField

cost_field(body)

The largest cost density over body’s probes, as a field over the state.

Parameters:

body (CollisionBody)

Return type:

StateField

plot(**kwargs)

Plot obstacles and workspace fields (lazy matplotlib import).

as_visualizer(**kwargs)

Return a time-only overlay that draws this scene’s obstacle skin.

Collision bodies: body-frame geometry placed in the world by one pose per rigid part.

class minilink.planning.spatial.collision.CollisionBody

Collision model: per-part body-frame geometry placed by world poses.

The moving counterpart of a workspace Shape: it carries body-frame collision geometry and places it in the world with one pose transform per rigid part.

Subclasses expose shapes (static, one per rigid part) and body_poses() (one homogeneous transform per part, aligned with shapes), or override iter_placed_parts() directly.

abstract property shapes: tuple[Shape, ...]

Body-frame collision geometry, one Shape per part.

abstract body_poses(x, u=None, t=0.0, params=None)

Return one homogeneous transform per part, aligned with shapes.

iter_placed_parts(x, u=None, t=0.0, params=None)

Yield (world_transform, body_frame_shape) for each collision part.

Return type:

Iterator

minilink.planning.spatial.collision.bind(sys, spec, *, frame='body')

Bind frameless collision geometry to a planner plant’s tf dict.

The world poses then come from sys.tf: one forward-kinematics pass, shared with rendering.

Single-frame (default frame="body"):

body = bind(sys_mpc, disc(0.25))
body = bind(sys_mpc, car_outline(2.4, 0.2, margin=0.05))

Multi-frame — pass a list of (frame_key, geometry) pairs:

body = bind(sys, [("body", car_outline(2.4, 0.2)), ("link3", disc(0.15))])

Parameters:

frame (str)

Return type:

CollisionBody

minilink.planning.spatial.collision.car_outline(length, width, *, margin=0.0)

Body-frame rectangular car footprint: corners and edge midpoints.

Each probe is a Sphere offset in the body frame. Attach to a planner plant with bind() and frame="body" (default).

Parameters:

• length (float)

• width (float)

• margin (float)

Return type:

tuple[Sphere, …]

minilink.planning.spatial.collision.collision_spheres(shape)

Return body-frame (center, radius) probes for a body part.

This is the only place the sphere restriction lives: a Sphere yields one probe, and any other shape is refused.

Parameters:

shape (Shape)

Return type:

tuple

minilink.planning.spatial.collision.disc(radius)

Body-frame disc/sphere probe at the origin (frameless geometry).

Parameters:

radius (float)

Return type:

Sphere

minilink.planning.spatial.collision.iter_probes(body, x, u=None, t=0.0, params=None)

Yield (world_center, radius) for each body-frame collision sphere.

The spheres come from collision_spheres(), placed in the world by apply().

minilink.planning.spatial.collision.point_probe()

Body-frame radius-zero point probe at the origin.

Return type:

Sphere

State fields: a scalar value of the state, exported as a set or a cost.

class minilink.planning.spatial.state_fields.StateField

Scalar value(x); export with as_constraint() or as_cost().

abstract value(x, u=None, t=0.0, params=None)

The field’s scalar value at the state x.

as_constraint(*, lower=0.0, upper=None)

The set of states where lower ≤ value(x) ≤ upper.

Parameters:

• lower (float | None)

• upper (float | None)

Return type:

Set

as_cost(*, weight=1.0, shaping=None)

The running cost g = weight · shaping(value(x)), no terminal cost.

Parameters:

• weight (float)

• shaping (Callable | None)

Return type:

CostFunction

class minilink.planning.spatial.state_fields.ClearanceField(scene, body)

value(x) = minᵢ (clearance(pᵢ) − rᵢ) over the body’s probes; nonnegative when free.

Parameters:

• scene (Scene)

• body (CollisionBody)

value(x, u=None, t=0.0, params=None)

The field’s scalar value at the state x.

class minilink.planning.spatial.state_fields.CostDensityField(scene, body)

value(x) = maxᵢ density(pᵢ) over the body’s probes.

Parameters:

• scene (Scene)

• body (CollisionBody)

value(x, u=None, t=0.0, params=None)

The field’s scalar value at the state x.

class minilink.planning.spatial.state_fields.PathDistanceField(track, body)

value(x) = minᵢ (distance(pᵢ) − rᵢ) to the reference centerline.

Parameters:

• track (ReferenceTrack)

• body (CollisionBody)

value(x, u=None, t=0.0, params=None)

The field’s scalar value at the state x.

class minilink.planning.spatial.state_fields.CorridorMarginField(track, body)

value(x) = minᵢ (w − distance(pᵢ) − rᵢ); nonnegative in the tube.

Parameters:

• track (ReferenceTrack)

• body (CollisionBody)

value(x, u=None, t=0.0, params=None)

The field’s scalar value at the state x.

class minilink.planning.spatial.state_fields.FieldSet(field, lower=0.0, upper=None)

Feasible where lower <= value <= upper.

Parameters:

• field (StateField)

• lower (float | None)

• upper (float | None)

margin(z, t=0.0, params=None)

Return nonnegative feasibility margins for z, shape (k,).

class minilink.planning.spatial.state_fields.FieldCost(field, weight=1.0, shaping=None)

g = weight * shaping(value); shaping=None uses value directly.

Parameters:

• field (StateField)

• weight (float)

• shaping (Callable | None)

g(x, u, t=0.0, params=None)

Return the native scalar running cost density g(x, u, t).

h(x, t=0.0, params=None)

Return the native scalar terminal cost h(x, t).

Reference tracks: a centerline path with a constant-width corridor around it.

class minilink.planning.spatial.track.ReferenceTrack(path, half_width=1.0)

Target path with a constant half-width corridor in workspace units.

A track composes into a PlanningProblem through its state fields:

track = ReferenceTrack(from_waypoints(pts), half_width=1.0)
X = bounds & track.corridor_field(body).as_constraint()
cost = base + track.distance_field(body).as_cost(weight=5.0)

Parameters:

• path (ReferencePath) – Workspace centerline.

• half_width (float) – Corridor radius around the path (same units as path.distance).

distance(p, t=0.0, params=None)

Distance from p to the centerline.

corridor_margin(p, t=0.0, params=None)

Positive when p lies inside the corridor: half_width - distance.

distance_field(body)

The least distance from body’s probes to the centerline, as a field.

Parameters:

body (CollisionBody)

Return type:

StateField

corridor_field(body)

The smallest corridor margin over body’s probes, as a field.

Parameters:

body (CollisionBody)

Return type:

StateField

plot(**kwargs)

Plot the centerline and corridor (lazy matplotlib import).

Reference paths: workspace curves for path tracking and corridor constraints.

class minilink.planning.spatial.paths.ReferencePath

Curve in workspace; distance is always nonnegative length units.

Build one from waypoints with from_waypoints() (default kind="polyline") and pass it to ReferenceTrack.

abstract property total_length: float

Arc length of the path.

abstract distance(p, t=0.0, params=None)

Shortest distance from workspace point p to the path.

abstract project(p, t=0.0, params=None)

Return (arc_length, closest_point) on the path.

abstract sample(s, t=0.0, params=None)

Point at arc length s along the path.

abstract tangent(s, t=0.0, params=None)

Unit tangent at arc length s.

class minilink.planning.spatial.paths.PolylinePath(waypoints)

Piecewise-linear path through waypoints (C⁰).

Parameters:

waypoints (array_like, shape (N, d)) – Vertices with N >= 2.

property total_length: float

Arc length of the path.

distance(p, t=0.0, params=None)

Shortest distance from workspace point p to the path.

project(p, t=0.0, params=None)

Return (arc_length, closest_point) on the path.

sample(s, t=0.0, params=None)

Point at arc length s along the path.

tangent(s, t=0.0, params=None)

Unit tangent at arc length s.

minilink.planning.spatial.paths.from_waypoints(waypoints, *, kind='polyline')

Build a reference path from a waypoint polyline.

Parameters:

• waypoints (array_like, shape (N, d)) – Path vertices with N >= 2.

• kind ({"polyline"}) – Path primitive; default is the robust piecewise-linear polyline.

Return type:

ReferencePath

minilink.planning.spatial.paths.circuit_waypoints(length=14.0, width=9.0, radius=2.5, spacing=0.25)

Closed test circuit: a rounded rectangle, sampled counter-clockwise.

Four quarter turns joined by four straights, then resampled at a constant arc-length spacing so that a tracker’s lookahead always lands between two waypoints.

Parameters:

• length (float) – Outer dimensions of the loop [m], corner to corner.

• width (float) – Outer dimensions of the loop [m], corner to corner.

• radius (float) – Corner radius [m].

• spacing (float) – Distance between consecutive waypoints [m].

Returns:

waypoints – Polyline [m] going once around the loop. The first point is not repeated at the end: a zero-length closing segment has no tangent, and the path objects that consume waypoints divide by it.

Return type:

array of shape (N, 2)

Cost shaping: the penalty of a scalar field value, before a state field’s cost weights it.

minilink.planning.spatial.shaping.quadratic_hinge(threshold=0.0)

One-sided quadratic penalty below threshold: max(threshold - v, 0)**2.

Zero in free space and quadratic once the value drops past threshold — the same scale as a quadratic tracking cost, so their weights are directly comparable, and the gradient keeps growing with penetration.

Parameters:

threshold (float)

minilink.planning.spatial.shaping.quadratic_excess(threshold=0.0)

One-sided quadratic penalty above threshold: max(v - threshold, 0)**2.

Use with PathDistanceField to penalize deviation from a reference path (mirror of quadratic_hinge() used for clearance margins).

Parameters:

threshold (float)

minilink.planning.spatial.shaping.inverse_barrier(epsilon=0.01)

Repulsive barrier 1 / max(v, epsilon)**2 that blows up at contact.

Strong, unbounded repulsion near the boundary; epsilon caps the value once the clearance is exhausted so the penalty stays finite.

Parameters:

epsilon (float)

minilink.planning.spatial.shaping.occupancy(scale=1.0)

Bounded occupancy score in [0, 1]: sigmoid(-v / scale).

~0 well outside, 0.5 on the boundary, ~1 deep inside. Bounded, so weighting an obstacle term against an unbounded cost is interpretable — at the price of a gradient that saturates far from the boundary.

Parameters:

scale (float)

Workspace fields: smooth penalty densities over a workspace point p in ℝ² or ℝ³.

class minilink.planning.spatial.workspace_fields.WorkspaceField

Base class for workspace penalty densities.

A point carries penalty density(p) >= 0. The field may depend on time and optional parameters, so the generic signature is density(p, t, params).

A field models a smooth penalty over the workspace — mud, slope preference, heat maps — not a solid region. Densities sum at the scene level into cost_density().

Construction is a NumPy/Python boundary; density() is a native-array math path that stays NumPy under NumPy input and traces under JAX.

abstract density(p, t=0.0, params=None)

Return the nonnegative penalty density at workspace point p.

class minilink.planning.spatial.workspace_fields.GaussianField(center, amplitude, sigma)

Smooth Gaussian penalty peak centered at center.

Parameters:

• center (array_like) – Peak location with shape (dim,).

• amplitude (float) – Peak density at center.

• sigma (float) – Standard deviation of the Gaussian envelope.

density(p, t=0.0, params=None)

Return the nonnegative penalty density at workspace point p.

Track overlays: a reference track’s corridor drawn into a 2-D animation.

class minilink.planning.spatial.overlays.TrackCorridorOverlay(track, *, n_samples=200, corridor_edge='#9aa3af', centerline='#5c6570')

Corridor edge and dashed centerline polylines for 2-D animation.

Parameters:

• n_samples (int)

• corridor_edge (str)

• centerline (str)

Spatial plots: a scene, a reference track and a workspace cost field in Matplotlib.

minilink.planning.spatial.plotting.plot_scene(scene, *, bounds=None, grid=(120, 120), show_obstacles=True, show_density=True, show_clearance_contour=False, body=None, x=None, states=None, show=True, ax=None, figsize=(6.0, 6.0), t=0.0, params=None, obstacle_color='#c44e52', obstacle_alpha=0.45, density_cmap='Oranges', density_alpha=0.55, body_color='#4c72b0', body_alpha=0.35, point_marker_size=9.0, title=None, equal_aspect=True)

Plot hard obstacles and soft workspace fields for a 2-D scene.

Pass body with x (one state) or states (several) to overlay collision discs at the placed body poses.

Returns (fig, ax). Matplotlib is imported on first call.

Parameters:

• show_obstacles (bool)

• show_density (bool)

• show_clearance_contour (bool)

• show (bool)

• t (float)

• obstacle_color (str)

• obstacle_alpha (float)

• density_cmap (str)

• density_alpha (float)

• body_color (str)

• body_alpha (float)

• point_marker_size (float)

• title (str | None)

• equal_aspect (bool)

minilink.planning.spatial.plotting.plot_track(track, *, bounds=None, n_samples=200, show=True, ax=None, figsize=(6.0, 6.0), centerline_color='#5c6570', corridor_edge_color='#9aa3af', waypoint_color='#1f77b4', title=None, equal_aspect=True)

Plot a ReferenceTrack in 2-D.

Returns (fig, ax). Matplotlib is imported on first call.

Parameters:

• n_samples (int)

• show (bool)

• centerline_color (str)

• corridor_edge_color (str)

• waypoint_color (str)

• title (str | None)

• equal_aspect (bool)

minilink.planning.spatial.plotting.plot_cost_field(grid, *, ax=None, cmap=None, log_scale=False, vmax=None, vmax_percentile=99.0, gamma=0.55, normalize_range=True, facecolor='white', colorbar=True, colorbar_label=None, show=True, figsize=(6.0, 6.0), title='Spatial cost', equal_aspect=True)

Plot a workspace cost heatmap. Returns (fig, ax).

Parameters:

• grid (FieldGrid)

• log_scale (bool)

• vmax (float | None)

• vmax_percentile (float)

• gamma (float)

• normalize_range (bool)

• facecolor (str)

• colorbar (bool)

• colorbar_label (str | None)

• show (bool)

• title (str | None)

• equal_aspect (bool)

minilink.planning.spatial.plotting.plot_cost_field_3d(grid, *, ax=None, cmap=None, log_scale=False, vmax=None, vmax_percentile=99.0, gamma=0.55, normalize_range=True, facecolor='white', colorbar=True, colorbar_label=None, show=True, figsize=(8.0, 6.0), title='Spatial cost', elev=35.0, azim=-60.0, alpha=0.92, rstride=1, cstride=1)

Plot workspace cost as a 3-D surface z = cost(x, y). Returns (fig, ax).

Parameters:

• grid (FieldGrid)

• log_scale (bool)

• vmax (float | None)

• vmax_percentile (float)

• gamma (float)

• normalize_range (bool)

• facecolor (str)

• colorbar (bool)

• colorbar_label (str | None)

• show (bool)

• title (str | None)

• elev (float)

• azim (float)

• alpha (float)

• rstride (int)

• cstride (int)

minilink.planning.spatial.plotting.plot_cost_field_exports(layers, *, track=None, scene=None, overlay_bounds=None, traj=None, log_scale=True, cmap=None, save_dir=None, show=True, figsize_2d=(9.0, 7.0), figsize_3d=(9.0, 7.0), surface_stride=2, **plot_kwargs)

Plot each named cost layer in 2-D and 3-D with the same colormap.

Parameters:

• layers (dict[str, FieldGrid]) – e.g. {"path": g, "corridor": g, "obstacle": g, "combined": g}.

• track – Optional overlays drawn on the 2-D figures (overlay_bounds for limits).

• scene – Optional overlays drawn on the 2-D figures (overlay_bounds for limits).

• traj – Optional executed path (x, y) polyline on 2-D figures.

• save_dir – If set, write {name}_2d.png and {name}_3d.png for each layer.

• **plot_kwargs – Forwarded to plot_cost_field() / plot_cost_field_3d() (vmax_percentile, gamma, etc.).

• log_scale (bool)

• show (bool)

• surface_stride (int)

Returns:

{name: {"2d": (fig, ax), "3d": (fig, ax)}}.

Return type:

dict

minilink.planning.spatial.plotting.cost_field_cmap()

Low cost (blue) to high cost (red) — matplotlib turbo.

Search

The RRT planner: a tree grown from the start until it reaches the goal region.

class minilink.planning.search.rrt.RRTPlanner(problem, extender=None, *, metric=<function euclidean>, options=None, max_nodes=<object object>, goal_bias=<object object>, goal_tolerance=<object object>, seed=<object object>, edge_resolution=<object object>, max_sample_attempts=<object object>, return_best_effort=<object object>, callback=<object object>, live_plot=<object object>, live_plot_every=<object object>, live_plot_pause=<object object>, live_plot_after_goal_only=<object object>, live_plot_ax=<object object>, nearest_backend=<object object>)

Rapidly-exploring random tree over a PlanningProblem.

The planner owns the loop (sample, nearest, propose, select, add, goal test, extract) and reads every concern from the problem: collision from problem.X, the goal region from problem.Xf / x_goal, sampling from problem.X (by rejection when needed), dynamics from problem.sys. Only the extender (how two states connect) and the metric (the nearest-neighbour distance) are swappable.

Parameters:

• problem (PlanningProblem) – Supplies dynamics, free space (X), goal (Xf/x_goal), and bounds.

• extender (TrajectoryExtender) – Proposes candidate edges; the orchestrator selects the best collision-free one.

• metric (callable) – Pairwise nearest-neighbour distance metric(a, b) -> float.

• options (RRTOptions) – Tier-2 workflow bag. Flat kwargs below overlay matching fields.

Notes

Flat keyword arguments overlay RRTOptions: max_nodes, goal_bias, goal_tolerance, seed, edge_resolution, max_sample_attempts, return_best_effort, callback, live_plot, live_plot_every, live_plot_pause, live_plot_after_goal_only, live_plot_ax, nearest_backend.

Provisional tool: maintainer review before assigning for coursework.

reached_goal

Set by solve(); True only when the goal region is reached.

Type:

bool

solution_node

Tree node whose parent chain defines the returned trajectory.

Type:

Node or None

solve(*, evaluate=False, n_trials=50)

Offline traj-family entry.

The solution’s policy is the path’s input held by a source block, policy >> plant; evaluate=True replays it over the problem’s draws when the problem declares a cost.

Return type:

PlanningSolution

solve_trajectory(*, evaluate=False, n_trials=50)

Grow the tree until the goal region is reached or the budget is spent.

Return type:

PlanningSolution

search_record()

The search’s account: goal reached or not, the budget it took, the path’s cost-to-come.

Return type:

TreeSearchRecord

plot_tree(**kwargs)

Plot the final tree and solution path (lazy matplotlib import).

x_axis / y_axis choose the state projection; pass ax to overlay on a scene.plot heatmap.

animate_search(**kwargs)

Animate the tree growth in a state projection (lazy matplotlib import).

show_live_search(**kwargs)

Build a LiveSearchPlotCallback.

class minilink.planning.search.rrt.RRTOptions(max_nodes=5000, goal_bias=0.1, goal_tolerance=0.5, seed=None, edge_resolution=None, max_sample_attempts=100, return_best_effort=True, callback=None, live_plot=False, live_plot_every=5, live_plot_pause=0.001, live_plot_after_goal_only=False, live_plot_ax=None, nearest_backend='brute_force')

Workflow options for RRTPlanner (step/resolution are extender-owned).

Parameters:

• max_nodes (int)

• goal_bias (float)

• goal_tolerance (float)

• seed (int | None)

• edge_resolution (float | None)

• max_sample_attempts (int)

• return_best_effort (bool)

• callback (Callable | None)

• live_plot (bool)

• live_plot_every (int)

• live_plot_pause (float)

• live_plot_after_goal_only (bool)

• live_plot_ax (object | None)

• nearest_backend (str)

class minilink.planning.search.rrt.TreeSearchRecord(reached_goal, iterations, nodes, cost, converged, history)

Whether the goal was reached, how much of the budget the search took, and the path’s cost-to-come.

Parameters:

• reached_goal (bool)

• iterations (int)

• nodes (int)

• cost (float)

• converged (bool)

• history (list)

property success: bool

The search reached the goal region.

RRT*: the RRT with a near-neighbour parent search and cost-based rewiring.

class minilink.planning.search.rrt_star.RRTStarPlanner(problem, extender=None, *, metric=<function euclidean>, options=None, max_nodes=<object object>, goal_bias=<object object>, goal_tolerance=<object object>, seed=<object object>, edge_resolution=<object object>, max_sample_attempts=<object object>, return_best_effort=<object object>, callback=<object object>, live_plot=<object object>, live_plot_every=<object object>, live_plot_pause=<object object>, live_plot_after_goal_only=<object object>, live_plot_ax=<object object>, nearest_backend=<object object>, gamma=<object object>, rewire_eta=<object object>, rewire=<object object>, optimize_after_goal=<object object>, cost_tol=<object object>, convergence_patience=<object object>, record_history=<object object>, history_stride=<object object>)

RRT* over a PlanningProblem.

Extends RRTPlanner with a near-neighbour parent search and cost-based rewiring; it reuses the same TrajectoryExtender and metric seams, and only the attach step changes. Same constructor as RRTPlanner, plus RRT* flat kwargs / RRTStarOptions. A plain RRTOptions still works with rewiring defaults.

Parameters:

options (RRTOptions | None)

converged

True when post-goal optimization stopped on the patience criterion.

Type:

bool

iterations

Number of successful tree extensions performed.

Type:

int

best_goal_cost

Best cost-to-come among nodes in the goal region.

Type:

float or None

history

Recorded frames when record_history is enabled.

Type:

list of RRTStarSnapshot

solve(*, evaluate=False, n_trials=50)

Offline traj-family entry (see RRTPlanner.solve()).

Return type:

PlanningSolution

solve_trajectory(*, evaluate=False, n_trials=50)

Grow an RRT* until goal convergence, patience, or max_nodes.

Return type:

PlanningSolution

search_record()

The RRT* record: the best goal cost, and whether patience declared convergence.

animate_convergence(**kwargs)

Animate recorded RRT* history (requires record_history=True).

class minilink.planning.search.rrt_star.RRTStarOptions(max_nodes=5000, goal_bias=0.1, goal_tolerance=0.5, seed=None, edge_resolution=None, max_sample_attempts=100, return_best_effort=True, callback=None, live_plot=False, live_plot_every=5, live_plot_pause=0.001, live_plot_after_goal_only=False, live_plot_ax=None, nearest_backend='brute_force', gamma=None, rewire_eta=None, rewire=True, optimize_after_goal=False, cost_tol=0.05, convergence_patience=500, record_history=False, history_stride=1)

Workflow options for RRTStarPlanner.

When optimize_after_goal is set, the search keeps running after the first feasible goal connection and stops once the best goal cost-to-come has not improved by at least cost_tol for convergence_patience consecutive successful extensions.

Parameters:

• gamma (float, optional) – Scale in the rewiring radius r_n = min(gamma * (log(n)/n)^(1/d), eta). When omitted, gamma = 2 (1 + 1/d)^(1/d) volume(X)^(1/d) from the problem state bounding box.

• rewire_eta (float, optional) – Maximum connection length eta. Defaults to the extender step (SteeringExtender.max_distance or KinodynamicExtender.horizon).

• rewire (bool) – When False, skip the rewiring pass (parent selection only).

• optimize_after_goal (bool) – When True, continue searching after the first goal connection until the best goal cost stops improving (see cost_tol and convergence_patience). When False, return at the first goal hit.

• cost_tol (float) – Minimum goal cost improvement that resets the convergence counter.

• convergence_patience (int) – Stop after this many successful extensions without a goal-cost improvement of at least cost_tol once a goal has been reached.

• record_history (bool) – Record RRTStarSnapshot frames for animation (every history_stride extensions).

• history_stride (int) – Snapshot interval in successful extension steps.

• max_nodes (int)

• goal_bias (float)

• goal_tolerance (float)

• seed (int | None)

• edge_resolution (float | None)

• max_sample_attempts (int)

• return_best_effort (bool)

• callback (Callable | None)

• live_plot (bool)

• live_plot_every (int)

• live_plot_pause (float)

• live_plot_after_goal_only (bool)

• live_plot_ax (object | None)

• nearest_backend (str)

class minilink.planning.search.rrt_star.RRTStarSnapshot(iteration, best_cost, reached_goal, tree_edges, path_states=None)

One recorded frame of an RRT* search for convergence animation.

Parameters:

• iteration (int) – Extension index when the snapshot was taken.

• best_cost (float or None) – Best goal cost-to-come at this frame (None before any goal node).

• reached_goal (bool) – Whether a goal-region node exists at this frame.

• tree_edges (list of np.ndarray) – Edge state samples (k+1, n) for every branch in the tree.

• path_states (np.ndarray or None) – State samples (k+1, n) along the best goal path at this frame.

Trajectory extenders: the candidate edges the RRT family grows toward a sample.

class minilink.planning.search.extenders.TrajectoryExtender

Propose candidate trajectory segments toward a target state.

propose(from_state, toward_state, problem, rng) returns candidate Edge segments grown from from_state. Extenders are metric-free and collision-free: they only propose, and the planner selects the best collision-free candidate. KinodynamicExtender forward-integrates problem.sys.f under each proposed control (any system); SteeringExtender makes one exact candidate through a SteeringFunction.

abstract propose(from_state, toward_state, problem, rng)

Return candidate edges grown from from_state toward toward_state.

Return type:

Iterable[Edge]

class minilink.planning.search.extenders.KinodynamicExtender(controls='bang-bang', *, horizon=0.3, n_substeps=6)

Forward-integration extender: one fixed-time rollout per control.

Works with any system: each candidate integrates problem.sys.f under one held control.

Parameters:

• controls (sequence of array_like, int, or "bang-bang") – An explicit list of control inputs (motion primitives); an int n meaning “n random samples of problem.U per extension”; or "bang-bang" (default): the corners of the box problem.U plus its centre, so every planner call works from the input bounds alone.

• horizon (float) – Edge duration; the control is held over n_substeps of dt = horizon / n_substeps.

• n_substeps (int) – Integration steps per edge.

propose(from_state, toward_state, problem, rng)

Return candidate edges grown from from_state toward toward_state.

Return type:

Iterable[Edge]

rollout(evaluator, from_state, u)

Hold u over the horizon: n_substeps RK4 steps of dt.

Return type:

Edge

class minilink.planning.search.extenders.SteeringExtender(steering, *, max_distance=0.5, resolution=0.05)

Exact-connection extender: one candidate via a SteeringFunction.

Parameters:

• steering (SteeringFunction) – Local connector whose connect returns a feasible (states, inputs, times, cost) for the model — must match problem.sys.

• max_distance (float) – Arc length the edge is truncated to.

• resolution (float) – Arc-length spacing of the sampled edge (collision-check density).

propose(from_state, toward_state, problem, rng)

Return candidate edges grown from from_state toward toward_state.

Return type:

Iterable[Edge]

minilink.planning.search.extenders.bang_bang_controls(U)

Corners of the box input set U plus its centre (axis extremes above 3 inputs).

Steering functions: exact two-point connectors with inverse dynamics.

class minilink.planning.search.steering.SteeringFunction

Exact local connector for a kinematic model.

connect() returns the states, inputs and times that realise a dynamically-feasible segment under that model’s f, so the result must match problem.sys (re-simulating inputs reproduces states). distance() is the natural nearest-neighbour metric for that model.

abstract connect(x0, x1, *, max_distance, resolution)

Return (states, inputs, times, cost) from x0 toward x1.

The segment is truncated to arc length max_distance and sampled at spacing resolution. Returns None when x0 and x1 coincide.

distance(a, b)

Nearest-neighbour distance under this model (default Euclidean).

Return type:

float

class minilink.planning.search.steering.StraightLineSteering(speed=1.0)

Constant-velocity straight line for a single integrator dx = u.

The input is the velocity, so a unit-speed line from x0 toward x1 is exactly feasible: u = speed * direction reproduces the sampled states.

Parameters:

speed (float)

connect(x0, x1, *, max_distance, resolution)

Return (states, inputs, times, cost) from x0 toward x1.

The segment is truncated to arc length max_distance and sampled at spacing resolution. Returns None when x0 and x1 coincide.

class minilink.planning.search.steering.DubinsSteering(*, wheelbase, max_steering, speed=1.0)

Forward Dubins-car connector for a kinematic bicycle.

Connects (x, y, theta) poses along the shortest forward Dubins curve of minimum turning radius wheelbase / tan(max_steering), at constant speed. The inputs are (speed, steering) with steering in {+max_steering, 0, -max_steering} — exactly the bicycle controls that trace each circular arc, so the segment is feasible for the matching KinematicBicycle/KinematicCar.

Parameters:

• wheelbase (float)

• max_steering (float)

• speed (float)

connect(x0, x1, *, max_distance, resolution)

Return (states, inputs, times, cost) from x0 toward x1.

The segment is truncated to arc length max_distance and sampled at spacing resolution. Returns None when x0 and x1 coincide.

distance(a, b)

The length of the shortest Dubins path from a to b, inf when there is none.

Return type:

float

The search tree: nodes, nearest-neighbour queries and path extraction.

class minilink.planning.search.tree.Node(x, parent, edge, cost, children=<factory>)

A tree node: a state, its parent, the reaching edge, and cost-to-come.

Parameters:

• x (ndarray)

• parent (Node | None)

• edge (Edge | None)

• cost (float)

• children (list)

class minilink.planning.search.tree.Tree(root, *, nearest_backend='brute_force')

A search tree rooted at a start state.

Nearest and near queries run by brute force over a caller-supplied metric, or on a SciPy cKDTree index (Euclidean L2 only). Path extraction concatenates the edges along a parent chain into a single Trajectory.

Parameters:

• root (Node)

• nearest_backend (str)

add(node)

Append a node and return it.

Parameters:

node (Node)

Return type:

Node

nearest(x, metric)

Return the node minimising metric(node.x, x).

Parameters:

metric (Callable)

Return type:

Node

near(x, radius, metric)

Return all nodes within radius of x (for RRT*).

Parameters:

• radius (float)

• metric (Callable)

Return type:

list[Node]

rewire(node, new_parent, new_edge)

Reparent node through new_edge and refresh its cost-to-come.

Parameters:

• node (Node)

• new_parent (Node)

• new_edge (Edge)

Return type:

None

propagate_cost(node)

Refresh cost-to-come for node and all descendants.

Parameters:

node (Node)

Return type:

None

extract_trajectory(node)

Concatenate edges from the root to node into a Trajectory.

Parameters:

node (Node)

Return type:

Trajectory

Policy synthesis

Value iteration: the Bellman equation solved backward on a discretized state space.

minilink.planning.policy_synthesis.dp.BACKEND_LOOP = 'loop'

Per-node Python reference engine (pyro’s base DynamicProgramming).

class minilink.planning.policy_synthesis.dp.DynamicProgrammingPlanner(problem, *, grid=None, x_grid=None, u_grid=None, dt=None, options=None, backend=<object object>, alpha=<object object>, tol=<object object>, max_iterations=<object object>, interpolation=<object object>, out_of_bound_cost=<object object>, final_time=<object object>, record_history=<object object>, verbose=<object object>, clean_infeasible=<object object>)

Value-iteration planner over a discretized state space.

Sweeps the Bellman equation backward in time on a StateSpaceGrid, from the terminal cost h at the horizon, and keeps the minimizing action of every node as the greedy policy pi:

DynamicProgrammingPlanner(problem, x_grid=(201, 201), u_grid=(21,), dt=0.05)

builds the grid itself; pass grid= for a custom one. The backend picks how the backward step runs: "loop" is the textbook double loop, "numpy" (default) the same step over lookup tables, and "jax" the table step jitted on device (dp_jax).

Parameters:

• problem (PlanningProblem) – Planning problem (system, sets, and cost). A cost is required.

• grid (StateSpaceGrid, optional) – Discretization of problem’s state and input spaces. Give x_grid, u_grid and dt instead and the planner builds it.

• options (DynamicProgrammingOptions, optional) – Workflow options. The flat keyword arguments below overlay matching fields: backend, alpha, tol, max_iterations, interpolation, out_of_bound_cost, final_time, record_history, verbose, clean_infeasible.

solve(*, evaluate=False, n_trials=50)

Infinite horizon: sweep backward until the cost-to-go stops changing.

Stops once the largest change of J falls under tol, or after max_iterations sweeps. The solution’s policy is the greedy lookup table and its cost_to_go the interpolated J. evaluate=True also rolls the law out from the problem’s start and scores it over the problem’s draws.

Return type:

PlanningSolution

solve_steps(n, *, evaluate=False, n_trials=50)

Finite horizon: exactly n backward sweeps from the terminal cost.

Parameters:

n (int)

Return type:

PlanningSolution

solve_policy(*, evaluate=False, n_trials=50)

The planner-family name of solve(), answered by every policy planner.

Return type:

PlanningSolution

value_iteration(max_sweeps, *, stop_on_tol=True)

Sweep the Bellman equation backward from the terminal cost.

Each sweep steps time back by dt and replaces the cost-to-go with the best one-step backup. Stops after max_sweeps or, with stop_on_tol, once the largest change falls under tol. Returns the raw tables; solve() and solve_steps() wrap them in a solution.

Parameters:

• max_sweeps (int)

• stop_on_tol (bool)

Return type:

DynamicProgrammingResult

backward_step_loop(J_next, t)

One backward Bellman step as an explicit loop over nodes and actions.

The readable reference (pyro’s DynamicProgramming): for every state node and every control action, take a forward-Euler step, look up the arrival cost-to-go, and keep the cheapest action.

backward_step_table(J_next, t)

The same backward step, vectorized over the successor and running-cost tables.

terminal_cost(t)

The terminal cost h at time t of every grid node, by node id.

Return type:

ndarray

running_cost_table(t, admissible=None, x_next=None)

Running cost of every (node, action) pair over one step, shape (nodes_n, actions_n).

Pairs that are not admissible (action outside the input set, or a successor outside the state set or the grid) cost out_of_bound_cost: the scalar, or out_of_bound_cost(x_next, t) at the pair’s successor. The mask and the successors are read from the grid’s transition tables when not given.

Warning: on a precomputed grid the table is built once, at the first sweep’s time, and reused for every sweep. A running cost g, or a price out_of_bound_cost(x_next, t), that depends on t is then frozen at that time. Use a grid with precompute=False for a time-varying cost.

Return type:

ndarray

get_controller(**kwargs)

The solution’s greedy lookup law; interpolation= builds a variant of it.

value_at(x)

Interpolate the cost-to-go at the latest solve, by the options’ interpolation.

Return type:

float

nominal_trajectory(tf=None)

The greedy law from the problem’s start on the grid’s control period.

clean_infeasible_set(tol=1.0)

Flag states whose cost-to-go has saturated at out_of_bound_cost.

Their value is pinned to the penalty and their policy to the action nearest the system’s nominal input, mirroring pyro’s cleanup pass. A callable out_of_bound_cost(x_next, t) has no single saturation level: the tables are then left as solved.

Parameters:

tol (float)

Return type:

DynamicProgrammingResult

class minilink.planning.policy_synthesis.dp.DynamicProgrammingOptions(backend='numpy', alpha=1.0, tol=0.1, max_iterations=1000, interpolation='linear', out_of_bound_cost=1000000.0, final_time=0.0, record_history=False, verbose=False, clean_infeasible=True)

Workflow options for DynamicProgrammingPlanner.

Parameters:

• backend ({"numpy", "loop", "jax"}) – Backward-step engine. "numpy" (default) is the vectorized lookup table, "loop" the per-node Python reference, "jax" the jitted device version (linear/nearest interpolation only; traceable cost/sets; its tables are built once, so f and g must not depend on time).

• alpha (float) – Discount (exponential forgetting) factor; use alpha < 1 for guaranteed contraction in infinite-horizon value iteration. When omitted, the cost’s discount_factor(dt) is used if the cost declares a positive discount_rate; otherwise 1 (undiscounted).

• tol (float) – Stopping tolerance on the largest cost-to-go change per sweep.

• max_iterations (int) – Maximum number of backward sweeps.

• interpolation ({"linear", "nearest", "cubic", "quintic"}) – Cost-to-go interpolation. "linear" (default) and "nearest" are robust; spline methods ("cubic", the "spline" alias, "quintic") are smoother but can ring across the infeasibility penalty.

• out_of_bound_cost (float or callable) – Finite penalty charged to inadmissible inputs or out-of-domain successors: a scalar, or out_of_bound_cost(x_next, t) evaluated at each inadmissible pair’s successor (the "jax" backend needs it traceable and time-invariant). The planner reads the problem’s infeasible_cost as its default.

• final_time (float) – Terminal time tf; sweeps step backward as t = tf - k dt. Left at 0.0, the planner reads problem.tf when the problem sets one.

• record_history (bool) – Keep (t, J, pi) per sweep for animation.

• verbose (bool) – Print build progress (50k-item counts with elapsed time and ETA) for lookup-table precompute and backward Bellman sweeps. Grid mesh and x_next progress use StateSpaceGrid.verbose; G, J0, and sweeps use this flag. Progress lines update in place about every 2% until each step completes. Set both to False for silent runs.

• clean_infeasible (bool) – After each solve, pin saturated cost-to-go cells to out_of_bound_cost and their policy to the nominal action (DynamicProgrammingPlanner.clean_infeasible_set()); a callable out_of_bound_cost leaves the tables as solved.

class minilink.planning.policy_synthesis.dp.DynamicProgrammingResult(grid, J, pi, iterations, delta, history=None)

Cost-to-go field and greedy policy from value iteration.

Parameters:

• grid (StateSpaceGrid) – Grid the result is defined on.

• J (np.ndarray) – Cost-to-go by node id, shape (nodes_n,).

• pi (np.ndarray) – Greedy policy as action ids by node id, shape (nodes_n,).

• iterations (int) – Number of backward sweeps performed.

• delta (float) – Largest cost-to-go change in the final sweep.

• history (list, optional) – (t, J, pi) tuples per sweep when recorded.

value_at(x, method='linear')

Interpolate the cost-to-go at an arbitrary state x (method: the grid’s interpolation).

Return type:

float

save(path)

Save J and pi to path (single .npz).

Parameters:

path (str)

Return type:

None

classmethod load(path, grid)

Load a result saved by save(), bound to grid.

Parameters:

• path (str)

• grid (StateSpaceGrid)

Return type:

DynamicProgrammingResult

class minilink.planning.policy_synthesis.dp.ValueIterationRecord(iterations, delta, tol, converged, fixed_horizon)

Sweeps run, the last cost-to-go change, and whether the sweeps converged to tol.

Parameters:

• iterations (int)

• delta (float)

• tol (float)

• converged (bool)

• fixed_horizon (bool)

property success: bool

The sweeps converged (a fixed-horizon solve always completes its sweeps).

State-space grid for value iteration: states, inputs, Euler successors and interpolation.

minilink.planning.policy_synthesis.discretizer.PAIR_CHUNK_SIZE = 50000

Node-action pairs evaluated per JAX chunk when building tables on device.

class minilink.planning.policy_synthesis.discretizer.StateSpaceGrid(problem, *, x_grid_shape, u_grid_shape, dt, precompute=True, precompute_backend='numpy', verbose=False)

Regular grid over the state and input spaces of a planning problem.

Parameters:

• problem (PlanningProblem) – Source of the system, the state set X (grid extent and state validity), and the input set U (grid extent and input validity).

• x_grid_shape (sequence of int) – Number of levels per state axis, length n.

• u_grid_shape (sequence of int) – Number of levels per input axis, length m.

• dt (float) – Forward-Euler time step used by the successor map.

• precompute (bool, optional) – When True (default) the successor and validity tables are built once at construction (t = 0) — fast sweeps, but only valid when neither the dynamics f nor the running cost g depends on time: the value-iteration planner then also builds its running-cost table once. When False both are recomputed at each sweep’s time, which costs less memory and supports a time-varying f or g.

• precompute_backend ({"numpy", "jax"}, optional) – Engine for building the successor lookup table. "numpy" (default) uses nested Python loops (pyro’s reference). "jax" vectorizes the forward dynamics with vmap and is much faster on large grids when the plant is JAX-traceable and the state/input sets are boxes.

• verbose (bool, optional) – Print mesh size and lookup-table build progress: one in-place line about every 2% with elapsed time and ETA until each step completes.

states

Grid states, shape (nodes_n, n).

Type:

np.ndarray

inputs

Grid inputs, shape (actions_n, m).

Type:

np.ndarray

x_next, action_ok, x_next_ok

Successor and validity tables, built at construction when precompute is set (or by ensure_jax_transition() and load()); None until then.

Type:

np.ndarray or None

property X: BoxSet

The state box the grid spans, x_lb ≤ x ≤ x_ub.

property U: BoxInputSet

The input box the grid’s actions span, u_lb ≤ u ≤ u_ub.

transition(t=0.0)

Return (x_next, action_ok, x_next_ok) at time t.

Cached tables are returned when precompute is set; otherwise the successor map and validity masks are recomputed for time t.

Parameters:

t (float)

successors(t=0.0)

Return the successor states x_next of shape (nodes_n, actions_n, n).

Parameters:

t (float)

Return type:

ndarray

on_grid(x_next)

Mask of successors inside the grid’s box, the table’s domain ((N, A) for an (N, A, n) table).

ensure_jax_transition(t=0.0)

Build or rebuild transition tables with the JAX backend.

Called by DynamicProgrammingPlanner when backend="jax". Use precompute=False on the grid in JAX demos so the slow NumPy build is skipped at construction time.

Parameters:

t (float)

Return type:

None

compute_transition_numpy(t)

Build the forward-Euler successors and the input/state validity masks.

compute_transition_jax(t)

Vectorized successor map and box validity masks on device.

build_interpolator(values, method='linear')

Return a regular-grid interpolator callable for a node-indexed field.

method is one of {"linear", "nearest", "cubic", "quintic"}; "spline" is accepted as an alias for "cubic". Out-of-bounds queries return 0. Spline methods are smoother but ring across the infeasibility penalty, so they suit fields without large cliffs.

Parameters:

method (str)

interpolate(values, queries, method='linear')

Interpolate a node-indexed field values at states queries (M, n).

Parameters:

method (str)

Return type:

ndarray

grid_from_array(values)

Reshape a node-indexed array (nodes_n,) into the state grid shape.

Return type:

ndarray

input_from_policy(pi)

Map a policy of action ids (nodes_n,) to inputs (nodes_n, m).

Return type:

ndarray

nearest_node(x)

Return the node id of the grid state closest to x.

Return type:

int

nearest_action(u)

Return the action id of the grid input closest to u.

Return type:

int

slice_2d(grid_nd, axis_x=0, axis_y=1, anchor=None)

Take a 2-D (axis_x, axis_y) slice of an n-D grid array.

Other axes are pinned to the grid node nearest anchor (defaulting to the problem goal when available, otherwise the grid centre).

Parameters:

• axis_x (int)

• axis_y (int)

plot_value(values, *, show_3d=False, **kwargs)

Draw a node-indexed field on this grid: a heatmap, or a surface with show_3d.

values has shape (nodes_n,) — a cost-to-go, a policy’s value, a fitted approximation. Options (axes, anchor, vmin, vmax, title, show) are those of plot_value().

save(path)

Save the successor and validity tables to path (single .npz).

Parameters:

path (str)

Return type:

None

load(path)

Load successor and validity tables saved by save().

Parameters:

path (str)

Return type:

None

Lookup-table feedback: a dynamic-programming policy as a state-feedback block.

class minilink.planning.policy_synthesis.lookup_policy.LookupTableController(grid, pi, *, interpolation='linear')

State-feedback controller backed by a policy lookup table.

Wraps a discrete policy pi (action ids over a StateSpaceGrid) as a state-feedback block u = pi(x). Its input port is named x, so it composes with a plant through controller @ plant and simulates like any other block. Each input axis is interpolated independently, so the discrete table yields a smooth command.

Parameters:

• grid (StateSpaceGrid) – Grid the policy is defined on.

• pi (array of int, shape (nodes_n,)) – Greedy policy as action ids by node id.

• interpolation ({"linear", "nearest"}, optional) – Per-axis input interpolation method.

action(x)

Return the interpolated command u at a single state x.

Return type:

ndarray

ctl(x, u, t=0, params=None)

State feedback; the x input port carries the plant state in u.

The linear-quadratic regulator as a planner: the Riccati solution of a quadratic problem at its goal.

class minilink.planning.policy_synthesis.lqr.LQRPlanner(problem, *, x_bar=None, u_bar=None, dt=0.01, n_steps=1001, t=0.0, method='auto', eps=1e-06)

Linear-quadratic regulator design as a planner.

Lowers a PlanningProblem whose cost is a QuadraticCost to the matrices (A, B, Q, R) of the plant linearized at (x̄, ū) about its command port (a disturbance d or a noise n is never decided) and solves the Riccati equation with the factories of minilink.control.design. Its solution carries the law, u = ū − K (x − x̄), and what the method itself knows about it: the cost-to-go J(x) = (x − x̄)ᵀ P (x − x̄) of the linear model and the closed-loop poles. The problem’s horizon picks the equation: infinite, the algebraic one; finite, the differential one swept backward from the terminal weight S, and the law is then u = ū − K(t) (x − x̄).

Parameters:

• problem (PlanningProblem) – Its cost must be a QuadraticCost; the operating point is the cost’s (xbar, ubar) unless given here. A cost over every input stacked weighs the command by its block of R. A finite tf selects the finite-horizon design with the cost’s terminal weight S; an infinite or unset tf the stationary one.

• x_bar (array-like, optional) – Operating point of the linearization; default the cost’s target.

• u_bar (array-like, optional) – Operating point of the linearization; default the cost’s target.

• dt (float, optional) – Sampling period of the nominal rollout and the Monte Carlo score (solve(evaluate=True)); a reporting choice, the law is continuous.

• n_steps (int, optional) – Samples of the finite-horizon gain schedule K(t).

• t (float, optional) – Time at which the Jacobians are evaluated.

• method ({"auto", "fd", "jax"}, optional) – Differentiation backend of jacobian().

• eps (float, optional) – Central-difference step.

solve(*, evaluate=False, n_trials=50)

Solve the Riccati equation of the linearized problem.

The solution’s policy is the state-feedback block, its cost_to_go the quadratic form of the linear model. evaluate=True also rolls the law out from the problem’s start on the nonlinear plant and scores it over the problem’s draws.

Return type:

PlanningSolution

solve_policy(*, evaluate=False, n_trials=50)

The planner-family name of solve(), answered by every policy planner.

Return type:

PlanningSolution

nominal_trajectory(tf=None)

The law from the problem’s start on the nonlinear plant, sampled every dt.

linear_model()

(A, B): the plant’s Jacobians at the operating point, B over its command port.

class minilink.planning.policy_synthesis.lqr.RiccatiRecord(K, P, poles, horizon, schedule=None)

The Riccati solution: the gain, the cost-to-go matrix, the closed-loop poles.

K and P are the values at t = 0; on a finite horizon schedule holds the whole sweep (t, K(t), S(t)). success is the closed loop of the linear model being stable (every pole in the left half-plane) on the infinite horizon, a finite sweep otherwise.

Parameters:

• K (ndarray)

• P (ndarray)

• poles (ndarray)

• horizon (float)

• schedule (tuple | None)

property success: bool

A finite sweep on a finite horizon; a stable linear closed loop otherwise.

minilink.planning.policy_synthesis.lqr.quadratic_cost_to_go(x_bar, P, t_samples=None)

J(x) = (x − x̄)ᵀ P (x − x̄); with a schedule P = S(t) on t_samples, J(x, t).

Policy evaluation: the cost-to-go of a fixed control law on a discretized state space.

class minilink.planning.policy_synthesis.policy_eval.PolicyEvaluator(problem, *, grid, policy, options=None)

Evaluate the cost-to-go of a fixed feedback policy.

Solves the Bellman expectation equation for a given policy: the dynamic-programming sweep with the action fixed by a control law instead of minimized over the action set. It reuses the same StateSpaceGrid and interpolation as DynamicProgrammingPlanner and returns the policy’s cost-to-go field J.

Parameters:

• problem (PlanningProblem) – Planning problem (system, sets, and cost). A cost is required.

• grid (StateSpaceGrid) – Discretization of the state and input spaces.

• policy (callable, System or PlanningSolution) – State-feedback law u = policy(x), a solution (its policy), or a controller block that reads the state on an input named x (or y) and commands u (minilink.core.feedback.block_roles()): the grid state feeds that port, the reference stays pinned at its nominal value, and the control port supplies u.

• options (DynamicProgrammingOptions, optional) – Shared workflow options (alpha, tol, interpolation, …).

solve()

Iterate the fixed-policy Bellman update to tolerance and return J.

Return type:

ndarray

value_at(x)

Interpolate the policy’s cost-to-go at an arbitrary state x, by the options’ interpolation.

Return type:

float

plot_cost2go(**kwargs)

Draw the policy’s cost-to-go field; options as in StateSpaceGrid.plot_value().

Linear-in-the-weights function approximation f_hat(x | w) = wᵀ φ(x).

class minilink.planning.policy_synthesis.approximation.LinearApproximator(features, w=None)

f_hat(x | w) = wᵀ φ(x): fixed features, learned weights.

The features φ(x) are fixed basis functions of the state (a quadratic form about a point, Gaussian bumps on a mesh) and the weights w are the only parameters. fit() solves the least-squares problem on a batch of samples and sgd_step() applies one stochastic-gradient update on a single one: the two updates every approximate dynamic-programming and value-learning method is built on. Calling the approximator evaluates f_hat at a state or a batch.

fit(X, y)

Least squares: set w to minimize Σ_i (y_i - wᵀφ(x_i))² and return it.

sgd_step(x, y, eta)

One stochastic-gradient step w ← w + η (y - wᵀφ(x)) φ(x); returns w.

class minilink.planning.policy_synthesis.approximation.Features

Basis functions φ(x); + stacks the bases of two feature sets.

phi(x)

Return the feature vector φ(x) of one state, shape (n_features,).

matrix(X)

Return the regression matrix Φ, one row φ(x_i)ᵀ per state of X.

class minilink.planning.policy_synthesis.approximation.QuadraticFeatures(xbar)

Quadratic form about xbar: φ(x) = [1, δx, δx_i δx_j (i ≤ j)], δx = x - xbar.

wᵀφ(x) spans every function c + bᵀδx + δxᵀ S δx, the shape of an LQR cost-to-go; quadratic_form() reads (c, b, S) back from w.

phi(x)

Return the feature vector φ(x) of one state, shape (n_features,).

quadratic_form(w)

Return (c, b, S) such that wᵀφ(x) = c + bᵀδx + δxᵀ S δx.

class minilink.planning.policy_synthesis.approximation.RadialBasisFeatures(centers, sigma)

Gaussian bumps φ_i(x) = exp(-‖x - μ_i‖² / (2σ²)) centred on centers.

classmethod on_grid(lower, upper, shape, *, sigma=None)

Centres on a regular mesh of shape over the box [lower, upper].

sigma defaults to the largest mesh spacing, so neighbouring bumps overlap and the sum can represent a smooth field.

phi(x)

Return the feature vector φ(x) of one state, shape (n_features,).

class minilink.planning.policy_synthesis.approximation.ConcatenatedFeatures(first, second)

The bases of two feature sets stacked: φ(x) = [φ_1(x); φ_2(x)].

phi(x)

Return the feature vector φ(x) of one state, shape (n_features,).

Reinforcement learning

Reinforcement learning as a planner: a neural law trained on a stochastic planning problem.

minilink.planning.reinforcement_learning.planner.ALGORITHMS = {'actor_critic': <class 'minilink.planning.reinforcement_learning.algorithms.actor_critic.ActorCritic'>, 'ppo': <class 'minilink.planning.reinforcement_learning.algorithms.ppo.PPO'>, 'reinforce': <class 'minilink.planning.reinforcement_learning.algorithms.reinforce.REINFORCE'>, 'sac': <class 'minilink.planning.reinforcement_learning.algorithms.sac.SAC'>}

The update rule of each name algorithm= accepts.

minilink.planning.reinforcement_learning.planner.DEFAULT_GAMMA = 0.99

Discount per control period when neither the algorithm, the planner nor the cost sets one.

class minilink.planning.reinforcement_learning.planner.ReinforcementLearningPlanner(problem, *, dt=0.05, policy=None, features=None, hidden=(64, 64), activation='tanh', normalize=True, algorithm='ppo', n_envs=16, n_steps=128, episode_length=None, gamma=None, log_std_init=0.0, integrator='rk4', training_zone=None, seed=0, verbose=True, **algorithm_kwargs)

Policy-family planner that learns u = pi(x) by reinforcement learning.

The training objective is the expected discounted return on the control grid, E[-sum_k gamma^k g(x_k, u_k) dt]: a failure (leaving X) pays the problem’s price of infeasibility, leaving the training zone truncates the episode. solve returns the learned law as a PlanningSolution; asked to evaluate, it also rolls the law out and scores it by Monte Carlo on the problem’s own cost, the yardstick shared with every planner. The loop is generic, collect experience then update, and the update rule is an Algorithm, so a new method is one file, not a new planner. The law being trained is a StochasticPolicy: the controller block’s mean action with the algorithm family’s exploration head around it.

Parameters:

• problem (StochasticPlanningProblem) – Task with a start distribution, cost and constraint set. A plain PlanningProblem is accepted and trained from its single x_start.

• dt (float) – Control period of the learned law (one RK4 step per period).

• policy (NeuralPolicyController, optional) – The law to train; default built from features, hidden, activation and normalize on the problem’s plant.

• algorithm (str or Algorithm) – "reinforce", "actor_critic", "ppo" (on-policy) or "sac" (off-policy), or an Algorithm instance; keyword arguments not listed here go to the algorithm (learning_rate, n_epochs, tau, …).

• n_envs (int) – Plants simulated in parallel and control periods per collection (an episodic method collects one episode length instead of n_steps).

• n_steps (int) – Plants simulated in parallel and control periods per collection (an episodic method collects one episode length instead of n_steps).

• training_zone (Set, optional) – Where episodes run (the plant’s state box by default); leaving it truncates the episode. A training choice, not a constraint: the constraint is problem.X.

• episode_length (float, optional) – Episode duration for an infinite-horizon problem (default 10 s).

• gamma (float, optional) – Discount per control period, used when the algorithm does not set its own. Default: the cost’s discount_factor(dt) when the cost declares a rate, else 0.99 with a warning that states the effective horizon.

• log_std_init (float) – Initial exploration log-std on the normalized action (Gaussian head).

• seed (int) – Seed of the weights, exploration and starts.

• verbose (bool) – Print the environment’s semantics and one line per learning iteration (default True). Pass False to silence training.

controller

The law, updated in place after every learning iteration.

Type:

NeuralPolicyController

policy

The controller with its exploration head, what the algorithm trains.

Type:

StochasticPolicy

critic

The critic of the algorithm’s family.

Type:

ValueFunction or QFunction or None

history

One record per iteration (timesteps, mean episode return, losses, fps).

Type:

list of dict

env

The compiled environment (its describe() states the semantics).

Type:

RolloutEnvironment

accepts_stochastic = True

Planners that read the uncertainty of a stochastic problem set this; the others read its start set X0 and its x_start, and warn.

property gamma

The one discount per control period the algorithm trains with.

solve(timesteps=100000, *, evaluate=False, n_trials=50, verbose=None)

Train for timesteps plant steps; return the learned law as a PlanningSolution.

The solver record carries the training facts. evaluate=True also rolls the deterministic law out from the problem’s start and scores it by Monte Carlo over n_trials draws of the problem. verbose overrides verbose for this call only.

Return type:

PlanningSolution

solve_policy(timesteps=100000, *, evaluate=False, n_trials=50, verbose=None)

The planner-family name of solve(), answered by every policy planner.

Return type:

PlanningSolution

learn(timesteps, *, verbose=None)

Train for timesteps plant steps (rounded up to whole collections); returns self.

verbose overrides verbose for this call only.

sync_controller()

Copy the current policy network weights into the controller block.

get_controller()

The learned law as a controller block (controller @ plant).

predict(x, deterministic=True)

The learned law u = pi(x) for one state or a batch of states (rows).

deterministic=False draws the action from the exploration policy instead, a ~ pi_theta(. | x) on a fresh split of the planner’s key, and maps it onto the input port. Returns (u, None), the shape of Stable-Baselines3’s predict, so a script written for one trains and evaluates with the other.

solve_trajectory_from(x0, tf=None, *, evaluate=False)

Roll the learned law out from x0 on the training environment, disturbances drawn.

The solution’s trajectory is that rollout; evaluate=True scores it under the problem’s contract. The planner’s stored solution is left as it is.

Return type:

PlanningSolution

nominal_trajectory(tf=None)

The learned law from the problem’s start, with nominal parameters and disturbances.

Return type:

Trajectory

plot_learning_curve(ax=None)

Mean return of the exploration episodes against training steps.

class minilink.planning.reinforcement_learning.planner.ReinforcementLearningRecord(algorithm, timesteps, train_time_s, cost, weights_finite, history)

The training facts of a learned law: plant steps, training time, and the last mean episode cost.

cost is the mean episode cost J of the last collection, positive and falling. history holds one dict per learning iteration (timesteps, cost, n_episodes, the algorithm’s losses); returns stay internal to the algorithms.

Parameters:

• algorithm (str)

• timesteps (int)

• train_time_s (float)

• cost (float)

• weights_finite (bool)

• history (list)

property success: bool

The training finished with finite weights.

Tabular reinforcement learning on a state-space grid: Q-learning, SARSA and Monte Carlo control.

class minilink.planning.reinforcement_learning.tabular.TabularLearningPlanner(problem, *, grid=None, x_grid=None, u_grid=None, dt=None, algorithm='q_learning', exploration=None, eta=0.1, alpha=None, episode_length=None, exploring_starts=True, rounding='stochastic', integrator='rk4', seed=0)

Model-free learning of the cost-to-go table Q(x, u) on a state-space grid.

The learner never reads f. Its world is the grid: an episode moves from node to node, each control period stepping the plant from the current node under one of the grid’s input levels and rounding the state it reaches to a node, so the learner sees only transitions (x_k, u_k, c_k, x_{k+1}) between nodes. Every algorithm updates one cell of the table by a moving average toward its own sample q of the cost-to-go,

Q(x, u) <- Q(x, u) + eta (q - Q(x, u)),

Q-learning with the Bellman target q = c + alpha min_u' Q(x', u'), SARSA with the action actually taken next, Monte Carlo control with the cost observed to the end of the episode. The result is the cost-to-go field and greedy policy that value iteration returns, so plot_cost2go, plot_policy and get_controller (a lookup-table block) read the same way, and the same grid solved by DynamicProgrammingPlanner is the exact answer the learner approaches.

Parameters:

• problem (PlanningProblem or StochasticPlanningProblem) – The task: plant, cost, box and exit rule; a stochastic problem adds its start, parameter and disturbance draws.

• x_grid – Grid levels per state and input axis and the control period, as for value iteration; or pass grid= instead, never both.

• u_grid – Grid levels per state and input axis and the control period, as for value iteration; or pass grid= instead, never both.

• dt – Grid levels per state and input axis and the control period, as for value iteration; or pass grid= instead, never both.

• algorithm ({"q_learning", "sarsa", "monte_carlo"} or object) – The target of the moving average, or an instance of QLearning, SARSA or MonteCarloControl.

• exploration (EpsilonGreedy or UCB, optional) – How actions are chosen while learning (default EpsilonGreedy(), annealed from 1 to 0.05 over each call to learn()).

• eta (float, optional) – Learning rate of the moving average; None averages the samples of each cell (the step 1/N(x, u)).

• alpha (float, optional) – Discount per control period; default the cost’s discount_factor(dt), which is 1 for an undiscounted cost, as for value iteration.

• episode_length (float, optional) – Duration of an episode for an infinite-horizon problem (default 10 s).

• exploring_starts (bool) – Draw episode starts uniformly over the grid (default), or from the problem’s start distribution.

• rounding ({"stochastic", "nearest"}) – How a state becomes a node. "nearest" rounds each coordinate. "stochastic" (default) picks, on each axis, the lower or the upper level with the probability of the state’s fractional position, so a value read at the node is, in expectation, the multilinear interpolation value iteration uses: the learner then samples the same Markov decision process value iteration solves on the grid.

• integrator ({"rk4", "euler"}) – Integration scheme of one control period; value iteration’s grid steps with Euler.

• seed (int) – Seed of the starts, the exploration and the rounding.

Q

The cost-to-go table, updated in place.

Type:

ndarray, shape (nodes_n, actions_n)

visits

How often each cell was updated.

Type:

ndarray of int, shape (nodes_n, actions_n)

history

One record per episode: its cost, length and exploration rate.

Type:

list of dict

accepts_stochastic = True

Planners that read the uncertainty of a stochastic problem set this; the others read its start set X0 and its x_start, and warn.

property result: DynamicProgrammingResult

The table as a cost-to-go field J = min_u Q and greedy policy pi = argmin_u Q.

A cell never visited keeps its initial value, zero: the optimistic start that makes untried actions worth trying (visits tells which).

solve(episodes=1000, *, evaluate=False, n_trials=50)

Learn for episodes episodes; return the greedy law and its cost-to-go as a PlanningSolution.

evaluate=True also rolls the law out from the problem’s start and scores it over the problem’s draws.

Return type:

PlanningSolution

solve_policy(episodes=1000, *, evaluate=False, n_trials=50)

The planner-family name of solve(), answered by every policy planner.

Return type:

PlanningSolution

learn(episodes)

Run episodes episodes of interaction, learning as they go; returns self.

learning_record()

The solver record: episodes, coverage of the grid, and the cost of the last episodes.

Return type:

TabularLearningRecord

get_controller(**kwargs)

The greedy policy as a LookupTableController.

value_at(x)

Interpolate the learned cost-to-go at a state x.

Return type:

float

nominal_trajectory(tf=None)

The greedy law from the problem’s start on the grid’s control period.

plot_learning_curve(ax=None, window=20)

Episode cost against episodes, smoothed by a moving average of window episodes.

class minilink.planning.reinforcement_learning.tabular.TabularLearningRecord(algorithm, episodes, visited_fraction, cost, table_finite, history)

Episodes run, how much of the grid was visited, and the mean episode cost at the end of learning.

Parameters:

• algorithm (str)

• episodes (int)

• visited_fraction (float)

• cost (float)

• table_finite (bool)

• history (list)

property success: bool

The table stayed finite.

class minilink.planning.reinforcement_learning.tabular.TabularAlgorithm

The target of the moving average: one of two hooks, per step or per episode.

update(Q, s, a, c, s_next, a_next, terminated, eta, alpha)

Update after one transition (temporal-difference methods).

update_episode(Q, visited, alpha)

Update once the episode ended, from its (s, a, c, eta) steps (Monte Carlo methods).

class minilink.planning.reinforcement_learning.tabular.QLearning

Off-policy temporal-difference control: the target takes the best next action, whatever the agent does next.

update(Q, s, a, c, s_next, a_next, terminated, eta, alpha)

Update after one transition (temporal-difference methods).

class minilink.planning.reinforcement_learning.tabular.SARSA

On-policy temporal-difference control: the target takes the action the agent really takes next.

update(Q, s, a, c, s_next, a_next, terminated, eta, alpha)

Update after one transition (temporal-difference methods).

class minilink.planning.reinforcement_learning.tabular.MonteCarloControl

On-policy Monte Carlo control: the target is the cost actually observed to the end of the episode.

update_episode(Q, visited, alpha)

Update once the episode ended, from its (s, a, c, eta) steps (Monte Carlo methods).

class minilink.planning.reinforcement_learning.tabular.EpsilonGreedy(epsilon=1.0, final=0.05)

With probability epsilon a random action, otherwise the greedy one.

epsilon decreases linearly to final over a run: exploration first, exploitation at the end. Ties among greedy actions break at random.

epsilon(progress)

The exploration rate at progress in [0, 1] of the run.

choose(Q_s, N_s, rng, progress)

Action id at a state with cost-to-go row Q_s and visit counts N_s.

class minilink.planning.reinforcement_learning.tabular.UCB(c=1.0)

Optimism in the face of uncertainty: the action with the lowest confidence bound on its cost.

Q(x, u) - c sqrt(ln N(x) / N(x, u)): an action tried few times carries a large bonus and gets tried; a bad one tried often sees its bonus fade. Untried actions come first.

epsilon(progress)

No random action: the bonus is the exploration.

choose(Q_s, N_s, rng, progress)

Action id at a state with cost-to-go row Q_s and visit counts N_s.