Skip to content

BUG: make every step_simulation call advance the flight #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
thc1006 merged 1 commit into from
Aug 2, 2026
+135 −91
Merged

BUG: make every step_simulation call advance the flight #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
188 changes: 97 additions & 91 deletions rocketpy/simulation/flight.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -879,97 +879,78 @@ def step_simulation(self):
if state["finished"]:
return

phase_index = state["phase_index"]
if phase_index >= len(self.flight_phases) - 1:
state["finished"] = True
# One call has to leave the flight further along than it found it, so a
# call that lands on a phase boundary carries on into the new phase
# instead of returning. Only the finish path below returns without
# advancing, and by then there is nothing left to advance.
while True:
phase_index = state["phase_index"]
if phase_index >= len(self.flight_phases) - 1:
state["finished"] = True

self.post_process_simulation()
self.initialize_prints_plots()
return

phase = self.flight_phases[phase_index]

# Determine maximum time for this flight phase
phase.time_bound = self.flight_phases[phase_index + 1].t

# Initialize phase only once
if not state["phase_initialized"]:
# Evaluate callbacks
for callback in phase.callbacks:
callback(self)
self.post_process_simulation()
self.initialize_prints_plots()
return

# Create solver for this flight phase
self.function_evaluations.append(0)
phase = self.flight_phases[phase_index]

phase.solver = self._solver(
phase.derivative,
t0=phase.t,
y0=self.y_sol,
t_bound=phase.time_bound,
rtol=self.rtol,
atol=self.atol,
max_step=self.max_time_step,
min_step=self.min_time_step,
)

# Initialize phase time nodes
self.__setup_phase_time_nodes(phase)
# Determine maximum time for this flight phase
phase.time_bound = self.flight_phases[phase_index + 1].t

state["phase_initialized"] = True
state["node_index"] = 0
# Initialize phase only once
if not state["phase_initialized"]:
# Evaluate callbacks
for callback in phase.callbacks:
callback(self)

# Check if current phase is fully processed
if state["node_index"] >= len(phase.time_nodes) - 1:
state["phase_index"] += 1
state["phase_initialized"] = False
state["node_index"] = 0
return # Move to next phase on next call
# Create solver for this flight phase
self.function_evaluations.append(0)

node_index = state["node_index"]
node = phase.time_nodes[node_index]
phase.solver = self._solver(
phase.derivative,
t0=phase.t,
y0=self.y_sol,
t_bound=phase.time_bound,
rtol=self.rtol,
atol=self.atol,
max_step=self.max_time_step,
min_step=self.min_time_step,
)

# Determine time bound for this time node
node.time_bound = phase.time_nodes[node_index + 1].t
phase.solver.t_bound = node.time_bound
# Initialize phase time nodes
self.__setup_phase_time_nodes(phase)

if self.__is_lsoda:
phase.solver._lsoda_solver._integrator.rwork[0] = phase.solver.t_bound
phase.solver._lsoda_solver._integrator.call_args[4] = (
phase.solver._lsoda_solver._integrator.rwork
)
state["phase_initialized"] = True
state["node_index"] = 0

phase.solver.status = "running"
# Check if current phase is fully processed
if state["node_index"] >= len(phase.time_nodes) - 1:
state["phase_index"] += 1
state["phase_initialized"] = False
state["node_index"] = 0
continue # the new phase is initialised below, in this same call

# Feed required parachute and discrete controller triggers
# TODO: parachutes should be moved to controllers
for callback in node.callbacks:
callback(self)
node_index = state["node_index"]
node = phase.time_nodes[node_index]

for controller in node._controllers:
controller(
self.t,
self.y_sol,
self.solution,
self.sensors,
self.env,
)
# Determine time bound for this time node
node.time_bound = phase.time_nodes[node_index + 1].t
phase.solver.t_bound = node.time_bound

# Placeholder for parachute triggers in step simulation, which is currently not migrated
if self.__is_lsoda:
phase.solver._lsoda_solver._integrator.rwork[0] = phase.solver.t_bound
phase.solver._lsoda_solver._integrator.call_args[4] = (
phase.solver._lsoda_solver._integrator.rwork
)

while phase.solver.status == "running":
# Execute solver step, log solution and function evaluations
phase.solver.step()
self.solution += [[phase.solver.t, *phase.solver.y]]
self.function_evaluations.append(phase.solver.nfev)
phase.solver.status = "running"

# Update time and state
self.t = phase.solver.t
self.y_sol = phase.solver.y
if self.verbose:
print(f"Current Simulation Time: {self.t:3.4f} s", end="\r")
logger.debug("Current Simulation Time: %3.4f s", self.t)
# Feed required parachute and discrete controller triggers
# TODO: parachutes should be moved to controllers
for callback in node.callbacks:
callback(self)

for controller in self._continuous_controllers:
for controller in node._controllers:
controller(
self.t,
self.y_sol,
Expand All @@ -978,25 +959,50 @@ def step_simulation(self):
self.env,
)

if self.__check_simulation_events(phase, phase_index, node_index):
break # Stop if simulation termination event occurred
# Placeholder for parachute triggers in step simulation, which is currently not migrated

# Process overshootable time nodes if enabled
if self.time_overshoot and self.__process_overshootable_nodes(
phase, phase_index, node_index
):
break
while phase.solver.status == "running":
# Execute solver step, log solution and function evaluations
phase.solver.step()
self.solution += [[phase.solver.t, *phase.solver.y]]
self.function_evaluations.append(phase.solver.nfev)

# If controlled flight, post process must be done on sim time
# Post-process controllers if needed
if self._controllers:
phase.derivative(self.t, self.y_sol, post_processing=True)
# Update time and state
self.t = phase.solver.t
self.y_sol = phase.solver.y
if self.verbose:
print(f"Current Simulation Time: {self.t:3.4f} s", end="\r")
logger.debug("Current Simulation Time: %3.4f s", self.t)

if node._component_sensors:
u_dot = phase.derivative(self.t, self.y_sol)
self.__measure_sensors(node._component_sensors, u_dot)
for controller in self._continuous_controllers:
controller(
self.t,
self.y_sol,
self.solution,
self.sensors,
self.env,
)

state["node_index"] += 1
if self.__check_simulation_events(phase, phase_index, node_index):
break # Stop if simulation termination event occurred

# Process overshootable time nodes if enabled
if self.time_overshoot and self.__process_overshootable_nodes(
phase, phase_index, node_index
):
break

# If controlled flight, post process must be done on sim time
# Post-process controllers if needed
if self._controllers:
phase.derivative(self.t, self.y_sol, post_processing=True)

if node._component_sensors:
u_dot = phase.derivative(self.t, self.y_sol)
self.__measure_sensors(node._component_sensors, u_dot)

state["node_index"] += 1
return

def __setup_phase_time_nodes(self, phase):
"""Set up time nodes for the current phase.
Expand Down
38 changes: 38 additions & 0 deletions tests/unit/simulation/test_step_simulation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,44 @@ def _step_with_roll(env, rocket, command, max_steps=100000):
return flight, steps


class TestEveryCallAdvances:
"""A call has to leave the flight further along than it found it.

The phase transition used to return without touching ``t`` or ``y_sol``,
leaving the new phase to be initialised on the call after. A caller that
counts a step per call, which is what the Balloon Popping Challenge
environment does, then has its own clock ahead of the flight's.
"""

def test_no_call_returns_without_advancing(self, flight_calisto):
stepped = _stepped_twin(flight_calisto)
stalled = []
calls = 0
while not stepped._step_state["finished"]:
before = stepped.t
stepped.step_simulation()
calls += 1
if not stepped._step_state["finished"] and stepped.t <= before:
stalled.append(calls)

assert not stalled, f"calls {stalled} of {calls} did not advance"

def test_a_transition_is_absorbed_rather_than_costing_a_call(self, flight_calisto):
"""The control for the test above, which returning early on every call
would also pass. More than one phase has to actually be visited."""
stepped = _stepped_twin(flight_calisto)
_, phases_seen = _run_stepped(stepped)

assert len(phases_seen) > 1

def test_the_flight_still_ends_where_simulate_ends(self, flight_calisto):
"""Absorbing the transition must not skip the node it was standing on."""
stepped = _stepped_twin(flight_calisto)
_run_stepped(stepped)

np.testing.assert_allclose(stepped.t, flight_calisto.t, rtol=1e-8, atol=1e-10)


class TestControlledStepSimulation:
"""Injecting an actuator command between steps must move the trajectory.

Expand Down
Loading