(mpc_problem)
| 84 | |
| 85 | |
| 86 | def test_parallel_mpc(mpc_problem): |
| 87 | problem = mpc_problem |
| 88 | |
| 89 | TOL = 1e-5 |
| 90 | mu_init = 1e-8 |
| 91 | verbose = aligator.VerboseLevel.QUIET |
| 92 | solver = aligator.SolverProxDDP(TOL, mu_init, verbose=verbose) |
| 93 | solver.rollout_type = aligator.ROLLOUT_LINEAR |
| 94 | solver.max_iters = 100 |
| 95 | solver.sa_strategy = aligator.SA_FILTER |
| 96 | solver.force_initial_condition = True |
| 97 | solver.linear_solver_choice = aligator.LQ_SOLVER_PARALLEL |
| 98 | solver.setNumThreads(2) |
| 99 | solver.filter.beta = 1e-5 |
| 100 | solver.setup(problem) |
| 101 | |
| 102 | solver.run(problem, [], []) |
| 103 | |
| 104 | xs = solver.results.xs.tolist().copy() |
| 105 | us = solver.results.us.tolist().copy() |
| 106 | |
| 107 | # Launch MPC |
| 108 | for t in range(NUM_MPC_CYCLES): |
| 109 | print("Time " + str(t)) |
| 110 | |
| 111 | xs = xs[1:] + [xs[-1]] |
| 112 | us = us[1:] + [us[-1]] |
| 113 | |
| 114 | problem.x0_init = xs[0] |
| 115 | solver.setup(problem) |
| 116 | solver.run(problem, xs, us) |
| 117 | |
| 118 | xs = solver.results.xs.tolist().copy() |
| 119 | us = solver.results.us.tolist().copy() |
| 120 | |
| 121 | |
| 122 | def test_serial_mpc(mpc_problem): |
nothing calls this directly
no test coverage detected