Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 4 additions & 2 deletions rocketpy/control/controller.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import math
from inspect import signature
from typing import Iterable

Expand All @@ -17,7 +18,7 @@ def __init__(
self,
interactive_objects,
controller_function,
sampling_rate,
sampling_rate=math.inf,
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a specific reason for using math.inf here? I think using None to indicate a continuous signal is cleaner and safer. It prevents sampling_rate from being mistakenly evaluated as a number when it doesn't have a meaningful value

initial_observed_variables=None,
name="Controller",
):
Expand Down Expand Up @@ -72,7 +73,8 @@ def __init__(
relevant information in the `observed_variables` list.

.. note:: The function will be called according to the sampling rate
specified.
specified. If unspecified, the default sampling rate is set to infinity, meaning that the
controller function will be called at every step of the simulation.
sampling_rate : float
The sampling rate of the controller function in Hertz (Hz). This
means that the controller function will be called every
Expand Down
44 changes: 44 additions & 0 deletions rocketpy/rocket/rocket.py
Original file line number Diff line number Diff line change
Expand Up @@ -2002,6 +2002,50 @@ def add_thrust_eccentricity(self, x, y):
self.thrust_eccentricity_y = y
return self

def add_discrete_controller(
self,
controller_function,
refresh_rate,
interactive_objects=None,
initial_observed_variables=None,
name="Controller",
):
Comment thread
Malmahrouqi3 marked this conversation as resolved.
"""Creates a new discrete controller, storing its parameters such as
controller function, refresh rate, and interactive objects. The controller
will be called at the specified refresh rate during the simulation."""

controller = _Controller(
controller_function=controller_function,
sampling_rate=refresh_rate,
interactive_objects=interactive_objects,
initial_observed_variables=initial_observed_variables,
name=name,
)
Comment thread
Malmahrouqi3 marked this conversation as resolved.

self._add_controllers(controller)
Comment thread
Malmahrouqi3 marked this conversation as resolved.

def add_continuous_controller(
self,
controller_function,
interactive_objects=None,
initial_observed_variables=None,
name="Controller",
):
"""Creates a new continuous controller, storing its parameters such as
controller function and interactive objects. The controller will
be called at every time step of the simulation."""

controller = _Controller(
controller_function=controller_function,
sampling_rate=math.inf,
interactive_objects=interactive_objects,
initial_observed_variables=initial_observed_variables,
name=name,
)

self._add_controllers(controller)
return controller
Comment on lines +2005 to +2047
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because controllers are still an experimental feature, I´d prefer to only have self._add_controllers for now. This way we keep the interface between Controller and Rocket private and we can make changes as the architecture evolves. For this reason, please remove add_discrete_controller and add_continuos_controller


def draw(self, vis_args=None, plane="xz", *, filename=None):
"""Draws the rocket in a matplotlib figure.

Expand Down
15 changes: 14 additions & 1 deletion rocketpy/simulation/flight.py
Original file line number Diff line number Diff line change
Expand Up @@ -777,7 +777,14 @@ def __simulate(self, verbose):
self.y_sol = phase.solver.y
if verbose:
print(f"Current Simulation Time: {self.t:3.4f} s", end="\r")

for controller in self._continuous_controllers:
controller(
self.t,
self.y_sol,
self.solution[:-1],
self.sensors,
Comment thread
Malmahrouqi3 marked this conversation as resolved.
self.env,
)
Comment thread
Malmahrouqi3 marked this conversation as resolved.
if self.__check_simulation_events(phase, phase_index, node_index):
break # Stop if simulation termination event occurred

Expand Down Expand Up @@ -1576,6 +1583,9 @@ def __init_equations_of_motion(self):
def __init_controllers(self):
"""Initialize controllers and sensors"""
self._controllers = self.rocket._controllers[:]
self._continuous_controllers = [
c for c in self._controllers if math.isinf(c.sampling_rate)
]
self.sensors = self.rocket.sensors.get_components()

# reset controllable object to initial state (only airbrakes for now)
Expand Down Expand Up @@ -4488,6 +4498,9 @@ def add_parachutes(self, parachutes, t_init, t_end):

def add_controllers(self, controllers, t_init, t_end):
for controller in controllers:
# Skip node creation for continuous controllers
if math.isinf(controller.sampling_rate):
continue
# Calculate start of sampling time nodes
controller_time_step = 1 / controller.sampling_rate
controller_node_list = [
Expand Down
Loading