diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ef335ed5..a120386e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,11 @@ Attention: The newest changes should be on top --> ### Fixed +- BUG: actuator argument checks survive `python -O`, a non-finite command is + refused at the setter rather than reaching the integrator, and an + `add_*_control` call now hands the controller the same sampling rate the + actuator kept [#19](https://github.com/ARRC-Rocket/ActiveRocketPy/pull/19) + ## [v1.13.0] - 2026-07-21 ### Added diff --git a/rocketpy/rocket/actuator/actuator.py b/rocketpy/rocket/actuator/actuator.py index 3fd81581d..3f4185028 100644 --- a/rocketpy/rocket/actuator/actuator.py +++ b/rocketpy/rocket/actuator/actuator.py @@ -4,6 +4,71 @@ import numpy as np +def _finite_or_raise(value, description): + """Return ``value`` as a float, refusing anything that is not finite. + + A positive test on the value rather than a negated comparison. NaN fails + every ordered comparison, so a check written as ``value <= 0`` accepts it + where the assert it replaced rejected it; and the self-comparison that + caught it, ``not value == value``, reads as a redundant comparison to a + reader and to pylint alike. + + This also settles two cases the comparisons never named. Infinity is + refused, since an actuator cannot start at or be driven to one, and a value + that is not a number at all is refused here rather than a few lines later + inside ``np.clip``. + + ``bool`` is refused with them. YAML reads ``yes`` and ``on`` as ``True``, and + ``float(True)`` is 1.0, so a sampling rate written that way became 1 Hz with + nothing said. + """ + if isinstance(value, bool): + raise ValueError(f"{description} must be a number, not a boolean.") + try: + number = float(value) + except (TypeError, ValueError, OverflowError) as error: + raise ValueError(f"{description} must be a number.") from error + if not np.isfinite(number): + raise ValueError(f"{description} must be a finite number.") + return number + + +def _positive_or_none(value, description): + """An optional number that has to be finite and greater than zero.""" + if value is None: + return None + number = _finite_or_raise(value, description) + if number <= 0: + raise ValueError(f"{description} must be positive or None.") + return number + + +def _non_negative_or_none(value, description): + """An optional number that has to be finite and not below zero.""" + if value is None: + return None + number = _finite_or_raise(value, description) + if number < 0: + raise ValueError(f"{description} must be non-negative or None.") + return number + + +def _two_numbers_or_raise(value, description): + """A pair of numbers. Infinite is allowed here: it means unbounded.""" + try: + lower, upper = value + except (TypeError, ValueError) as error: + raise ValueError(f"{description} must be exactly two numbers.") from error + numbers = [] + for endpoint, where in ((lower, 0), (upper, 1)): + # Not float(), which takes "0" and would leave a YAML string range half + # converted while the rest of the config kept its strings. + if isinstance(endpoint, bool) or not isinstance(endpoint, (int, float)): + raise ValueError(f"{description}[{where}] must be a number.") + numbers.append(float(endpoint)) + return numbers[0], numbers[1] + + class Actuator(ABC): """Abstract class used to define actuators. @@ -36,7 +101,9 @@ def __init__( clamp : bool, optional Whether to clamp the actuator output. Default is True. actuator_initial_output : float, optional - Initial output of the actuator. Default is 0.0. + Initial output of the actuator. Default is 0.0. A value outside + actuator_range is treated the way the output setter treats one: it is + clamped when clamp is True, and warned about otherwise. actuator_time_constant : float, optional Time constant of the actuator, implemented as a discrete IIR filter. Default is None. @@ -47,29 +114,75 @@ def __init__( self.name = name - assert demand_rate > 0 or demand_rate is None, ( - "demand_rate must be positive or None." - ) - self.demand_rate = demand_rate + # These are argument checks rather than internal invariants, so they raise + # instead of asserting: python -O drops assert statements, and a negative + # time constant or rate limit would then be accepted in silence. + # Finite first, then the bound, so each check says one thing. Written + # as a negated comparison these accepted NaN, which fails every ordered + # comparison, and infinity, which passes them: a demand rate of infinity + # makes the sampling period zero, which drives the IIR coefficient to + # zero and freezes the output. + # + # The range endpoints are deliberately not put through this. The base + # default really is (-inf, inf), meaning an actuator with no range. + self.demand_rate = _positive_or_none(demand_rate, "demand_rate") - assert actuator_range[0] <= actuator_range[1], ( - "actuator_range[0] must be <= actuator_range[1]." - ) - self.actuator_range = actuator_range + # A range read out of a config file arrives as anything. Without the + # shape and number checks, three endpoints drop the third in silence and + # string endpoints fail inside np.clip with a ufunc loop error. + lower, upper = _two_numbers_or_raise(actuator_range, "actuator_range") + if not lower <= upper: + raise ValueError("actuator_range[0] must be <= actuator_range[1].") + self.actuator_range = (lower, upper) + actuator_range = self.actuator_range - assert actuator_rate_limit is None or actuator_rate_limit >= 0, ( - "actuator_rate_limit must be non-negative or None." + self.actuator_rate_limit = _non_negative_or_none( + actuator_rate_limit, "actuator_rate_limit" ) - self.actuator_rate_limit = actuator_rate_limit self.clamp = clamp - assert actuator_time_constant is None or actuator_time_constant >= 0, ( - "actuator_time_constant must be non-negative or None." + self.actuator_time_constant = _non_negative_or_none( + actuator_time_constant, "actuator_time_constant" ) - self.actuator_time_constant = actuator_time_constant + + # Neither is implemented for a continuous actuator, and both were + # accepted with a warning. Asking for a 0.1 s lag and getting an + # instant response is worth stopping for rather than mentioning. + if self.demand_rate is None: + for option, given in ( + ("actuator_rate_limit", self.actuator_rate_limit), + ("actuator_time_constant", self.actuator_time_constant), + ): + if given: + raise ValueError( + f"{option} needs a demand_rate: it is not applied to a " + f"continuous actuator." + ) + self._update_iir_coefficients() + # An initial output outside the range used to survive here and come back + # on every _reset(), even though the output setter would never let the + # actuator reach such a value afterwards. Treat it the way the setter + # treats any other out-of-range value, so the two agree. + # Refused before clamping: np.clip propagates NaN, so clamping would + # store it and _reset() would restore it, and there is no direction to + # clamp it towards in any case. + actuator_initial_output = _finite_or_raise( + actuator_initial_output, f"Actuator '{name}' initial output" + ) + if self.clamp: + actuator_initial_output = float( + np.clip(actuator_initial_output, actuator_range[0], actuator_range[1]) + ) + elif not actuator_range[0] <= actuator_initial_output <= actuator_range[1]: + warnings.warn( + f"Actuator '{name}' initial output {actuator_initial_output} " + f"is outside its range {actuator_range}.", + stacklevel=2, + ) + self.actuator_initial_output = actuator_initial_output self._actuator_output = actuator_initial_output @@ -82,9 +195,12 @@ def _update_iir_coefficients(self): if self.actuator_time_constant is not None and self.actuator_time_constant > 0: if self.demand_rate is not None: - demand_period = 1.0 / self.demand_rate - self._alpha = demand_period / ( - self.actuator_time_constant + demand_period + # Algebraically Ts / (tau + Ts) with Ts = 1 / demand_rate, + # written without forming Ts. One expression instead of two, + # and the two agree to 2.2e-16 over the range of time + # constants and rates a real actuator uses. + self._alpha = 1.0 / ( + 1.0 + self.actuator_time_constant * self.demand_rate ) else: warnings.warn( @@ -111,6 +227,18 @@ def actuator_output(self, value): ------- None """ + # The same rule as the initial output, so the two agree. It used to + # reject a NaN handed to the constructor and accept one handed to the + # setter on the next timestep: np.clip returns NaN for NaN, and both + # range comparisons are false for NaN, so neither branch noticed and it + # was stored. + # + # This is the path an agent writes to. BalloonPoppingChallenge assigns + # its actions straight into these setters, and a policy that goes + # unstable emits NaN, which from here reaches the forces, the moments + # and the integrator state. + value = _finite_or_raise(value, f"Actuator '{self.name}' output") + # Apply first-order IIR actuator dynamics value = self._alpha * value + (1 - self._alpha) * self._actuator_output diff --git a/rocketpy/rocket/rocket.py b/rocketpy/rocket/rocket.py index ec06b85bc..46d11cc08 100644 --- a/rocketpy/rocket/rocket.py +++ b/rocketpy/rocket/rocket.py @@ -2017,26 +2017,37 @@ def add_thrust_vector_control( sampling_rate : float The sampling rate of the controller function in Hertz (Hz). This means that the controller function will be called every - `1/sampling_rate` seconds. + `1/sampling_rate` seconds. Must be positive, or None for a + continuous-time controller called at every integration step. max_gimbal_angle : int, float Maximum gimbal angle in degrees. Both x and y gimbal angles are clamped to this range if clamp is True. Must be non-negative. gimbal_rate_limit : int, float Maximum gimbal rate in degrees per second. Both x and y gimbal - angles are limited to this rate of change. Default is None, no rate limit. + angles are limited to this rate of change. Must be non-negative. + Default is None, no rate limit. clamp : bool, optional If True, the simulation will clamp gimbal angles to the range [-max_gimbal_angle, max_gimbal_angle]. If False, a warning is issued when gimbal angles exceed the range. Default is True. - initial_gimbal_angle : int, float, tuple, list - The initial gimbal angle in degrees. If a single value is provided, - it is used for both x and y gimbal angles. If a tuple or list is - provided, the first element is used for the x-axis and the second - for the y-axis. Default is 0.0. + initial_gimbal_angle : int, float + The initial gimbal angle in degrees, used for both the x and y + axes. A value outside the range is clamped into it when clamp is + True, and warned about otherwise. Default is 0.0. + + A per-axis tuple or list was described here and has never been + implemented: ``ThrustVectorActuator2D`` hands the value it is given + to both single-axis actuators unchanged, and ``to_dict`` records + only the x-axis value, so an asymmetric pair could not survive a + round trip either. It used to be accepted and silently applied to + both axes; it now raises, because the initial output is converted + to a float. Corrected here rather than implemented, since the + feature belongs upstream rather than in this fork. gimbal_time_constant : float, optional - Time constant for the gimbal dynamics in seconds. If None, no - gimbal dynamics are applied. Default is None. + Time constant for the gimbal dynamics in seconds. Must be + non-negative. If None, no gimbal dynamics are applied. Default is + None. initial_observed_variables : list, optional A list of the initial values of the variables that the controller function manages. This list is used to initialize the @@ -2064,13 +2075,6 @@ def add_thrust_vector_control( "Only one thrust_vector_control per rocket is currently supported. " + "Overwriting previous thrust_vector_control and controllers." ) - self._controllers = [ - controller - for controller in self._controllers - if not isinstance( - controller.interactive_objects, ThrustVectorActuator2D - ) - ] thrust_vector_control = ThrustVectorActuator2D( name=name, @@ -2084,10 +2088,31 @@ def add_thrust_vector_control( _controller = _Controller( interactive_objects=thrust_vector_control, controller_function=controller_function, - sampling_rate=sampling_rate, + # The actuator's normalized rate, not the argument. The actuator + # runs its own validation and stores a float, so handing the + # controller the original leaves the two holding different types + # for one quantity: sampling_rate="100" gives the actuator 100.0 + # and the controller "100", this call returns successfully, and the + # failure surfaces later in Flight at 1 / controller.sampling_rate, + # by which point the rocket is already half built. + # + # Read off the x axis because ThrustVectorActuator2D holds no + # demand_rate of its own. Its class docstring lists one, along with + # six other attributes it also never assigns, but __init__ only + # forwards the argument to the two axes. Both are built from that + # one argument, so either axis gives the same number, and reaching + # into .x is what Flight already does to reset this class. + sampling_rate=thrust_vector_control.x.demand_rate, initial_observed_variables=initial_observed_variables, name=controller_name, ) + # Removed only once both halves are built, so a rejected argument on a + # second call leaves the rocket as it was. + self._controllers = [ + controller + for controller in self._controllers + if not isinstance(controller.interactive_objects, ThrustVectorActuator2D) + ] self.thrust_vector_control = thrust_vector_control self._add_controllers(_controller) if return_controller: @@ -2152,7 +2177,8 @@ def add_roll_control( sampling_rate : float The sampling rate of the controller function in Hertz (Hz). This means that the controller function will be called every - `1/sampling_rate` seconds. + `1/sampling_rate` seconds. Must be positive, or None for a + continuous-time controller called at every integration step. max_roll_torque : int, float Maximum roll torque magnitude in N·m. Must be non-negative. torque_rate_limit : int, float @@ -2163,9 +2189,12 @@ def add_roll_control( [-max_roll_torque, max_roll_torque]. If False, a warning is issued when roll torque exceeds the range. Default is True. initial_roll_torque : int, float - Initial roll torque in N·m. Default is 0.0. + Initial roll torque in N·m. A value outside the range is clamped + into it when clamp is True, and warned about otherwise. Default is + 0.0. roll_torque_time_constant : float, optional - Time constant for the roll torque dynamics in seconds. Default is None, no dynamics are applied. + Time constant for the roll torque dynamics in seconds. Must be + non-negative. Default is None, no dynamics are applied. initial_observed_variables : list, optional A list of the initial values of the variables that the controller function manages. This list is used to initialize the @@ -2195,11 +2224,6 @@ def add_roll_control( "Only one roll control per rocket is currently supported. " + "Overwriting previous roll control and controllers." ) - self._controllers = [ - controller - for controller in self._controllers - if not isinstance(controller.interactive_objects, RollActuator) - ] roll_control = RollActuator( name=name, @@ -2213,10 +2237,18 @@ def add_roll_control( _controller = _Controller( interactive_objects=roll_control, controller_function=controller_function, - sampling_rate=sampling_rate, + # The actuator's normalized rate. See add_thrust_vector_control. + sampling_rate=roll_control.demand_rate, initial_observed_variables=initial_observed_variables, name=controller_name, ) + # Removed only once both halves are built, so a rejected argument on a + # second call leaves the rocket as it was. + self._controllers = [ + controller + for controller in self._controllers + if not isinstance(controller.interactive_objects, RollActuator) + ] self.roll_control = roll_control self._add_controllers(_controller) if return_controller: @@ -2281,7 +2313,8 @@ def add_throttle_control( sampling_rate : float The sampling rate of the controller function in Hertz (Hz). This means that the controller function will be called every - `1/sampling_rate` seconds. + `1/sampling_rate` seconds. Must be positive, or None for a + continuous-time controller called at every integration step. throttle_range : tuple, optional A tuple containing the minimum and maximum throttle values. Must be in the range [0, 1]. Default is (0.0, 1.0). throttle_rate_limit : float, optional @@ -2292,11 +2325,13 @@ def add_throttle_control( [throttle_range[0], throttle_range[1]]. If False, a warning is issued when throttle values exceed the range. Default is True. initial_throttle : float, optional - Initial throttle value at the start of the simulation. Must be within - the range [throttle_range[0], throttle_range[1]]. Default is 1.0. + Initial throttle value at the start of the simulation. A value + outside [throttle_range[0], throttle_range[1]] is clamped into the + range when clamp is True, and warned about otherwise. Default is + 1.0. throttle_time_constant : float, optional - Time constant for the throttle actuator dynamics in seconds. - If None, no actuator dynamics are applied. + Time constant for the throttle actuator dynamics in seconds. Must be + non-negative. If None, no actuator dynamics are applied. initial_observed_variables : list, optional A list of the initial values of the variables that the controller function manages. This list is used to initialize the @@ -2327,11 +2362,6 @@ def add_throttle_control( "Only one throttle control per rocket is currently supported. " + "Overwriting previous throttle control and controllers." ) - self._controllers = [ - controller - for controller in self._controllers - if not isinstance(controller.interactive_objects, ThrottleActuator) - ] throttle_control = ThrottleActuator( name=name, @@ -2346,11 +2376,19 @@ def add_throttle_control( _controller = _Controller( interactive_objects=throttle_control, controller_function=controller_function, - sampling_rate=sampling_rate, + # The actuator's normalized rate. See add_thrust_vector_control. + sampling_rate=throttle_control.demand_rate, initial_observed_variables=initial_observed_variables, name=controller_name, ) + # Removed only once both halves are built, so a rejected argument on a + # second call leaves the rocket as it was. + self._controllers = [ + controller + for controller in self._controllers + if not isinstance(controller.interactive_objects, ThrottleActuator) + ] self.throttle_control = throttle_control self._add_controllers(_controller) diff --git a/tests/unit/rocket/test_actuators.py b/tests/unit/rocket/test_actuators.py index 94d077cb8..09a7b84d9 100644 --- a/tests/unit/rocket/test_actuators.py +++ b/tests/unit/rocket/test_actuators.py @@ -1,3 +1,7 @@ +import math +import subprocess +import sys + import pytest from rocketpy.rocket.actuator.roll import RollActuator @@ -452,30 +456,225 @@ def test_time_constant_iir_filter(self): assert initial_output < filtered_output < 1.0 +NAN = float("nan") +INF = float("inf") + + +def _as_source(value): + """Render a value as source the ``-O`` subprocess can evaluate. + + ``repr`` is almost enough, except that it renders NaN as the bare name + ``nan`` and infinity as ``inf``, neither of which the subprocess has bound. + Left as ``repr`` those cases died on NameError, and a test that only checked + the return code would have called that a pass. + """ + if isinstance(value, float) and not math.isfinite(value): + return f'float("{value}")' + if isinstance(value, tuple): + return "(" + ", ".join(_as_source(item) for item in value) + ",)" + return repr(value) + + +# One table, walked twice: once in-process for the message, once under ``-O``. +# Keeping them in step is the point. An argument that is only rejected in the +# default interpreter is not rejected, because the checks these replaced were +# asserts and asserts are what ``-O`` removes. +# +# Each entry names the message it expects, so a case cannot pass on some other +# argument's check. Every NaN case is here because NaN fails every ordered +# comparison: `nan <= 0` is false just as `nan > 0` is, so a check written as the +# inverted comparison accepts it while the assert it replaces rejected it. +INVALID_ARGUMENTS = [ + (RollActuator, {"demand_rate": -1}, "demand_rate"), + (RollActuator, {"demand_rate": 0}, "demand_rate"), + (RollActuator, {"demand_rate": NAN}, "demand_rate"), + (RollActuator, {"max_roll_torque": -5}, "actuator_range"), + (ThrottleActuator, {"throttle_range": (NAN, 1.0)}, "actuator_range"), + (ThrottleActuator, {"throttle_range": (0.0, NAN)}, "actuator_range"), + (ThrustVectorActuator, {"gimbal_rate_limit": -1.0}, "rate_limit"), + (ThrustVectorActuator, {"gimbal_rate_limit": NAN}, "rate_limit"), + (ThrottleActuator, {"throttle_time_constant": -0.1}, "time_constant"), + (ThrottleActuator, {"throttle_time_constant": NAN}, "time_constant"), + (ThrottleActuator, {"initial_throttle": NAN}, "initial output"), + # clamp is what would otherwise absorb an out-of-range initial value, and + # np.clip returns NaN for NaN, so both settings have to refuse it. + (ThrottleActuator, {"initial_throttle": NAN, "clamp": False}, "initial output"), + # Infinity was never named by the comparisons. An actuator cannot start at + # one, and clamping would quietly turn it into a range endpoint. + (ThrottleActuator, {"initial_throttle": INF}, "initial output"), + (RollActuator, {"demand_rate": INF}, "demand_rate"), +] +INVALID_IDS = [ + f"{cls.__name__}-{'-'.join(kwargs)}-{'clamped' if kwargs.get('clamp', True) else 'unclamped'}" + for cls, kwargs, _ in INVALID_ARGUMENTS +] + + +class TestTheOutputSetterRefusesWhatTheConstructorDoes: + """The two used to disagree, and the setter is the one an agent writes to. + + A NaN handed to the constructor was rejected; the same NaN handed to the + setter on the next timestep was stored. ``np.clip`` returns NaN for NaN, and + both range comparisons are false for NaN, so neither the clamped nor the + unclamped branch noticed. + + BalloonPoppingChallenge assigns agent actions straight into these setters, + and a policy that goes unstable emits NaN, which from there reaches the + forces, the moments and the integrator state. + """ + + @pytest.mark.parametrize("clamp", [True, False], ids=["clamped", "unclamped"]) + @pytest.mark.parametrize("value", [NAN, INF, -INF], ids=["nan", "inf", "-inf"]) + def test_a_non_finite_command_is_refused(self, clamp, value): + actuator = ThrottleActuator(clamp=clamp) + + with pytest.raises(ValueError, match="output"): + actuator.actuator_output = value + + @pytest.mark.parametrize("clamp", [True, False], ids=["clamped", "unclamped"]) + def test_an_ordinary_command_still_goes_through(self, clamp): + """Or refusing everything would satisfy the test above.""" + actuator = ThrottleActuator(clamp=clamp) + + actuator.actuator_output = 0.5 + + assert actuator.actuator_output == pytest.approx(0.5) + + def test_the_stored_output_is_untouched_by_a_refused_command(self): + actuator = ThrottleActuator() + actuator.actuator_output = 0.5 + + with pytest.raises(ValueError): + actuator.actuator_output = NAN + + assert actuator.actuator_output == pytest.approx(0.5) + + class TestActuatorValidation: """Test suite for actuator parameter validation.""" - def test_invalid_demand_rate_negative(self): - """Test that negative demand rate raises assertion error.""" - with pytest.raises(AssertionError): - RollActuator(demand_rate=-1) + @pytest.mark.parametrize( + "actuator_class, kwargs, message", INVALID_ARGUMENTS, ids=INVALID_IDS + ) + def test_invalid_arguments_are_rejected(self, actuator_class, kwargs, message): + with pytest.raises(ValueError, match=message): + actuator_class(**kwargs) + + @pytest.mark.parametrize( + "actuator_class, kwargs, message", INVALID_ARGUMENTS, ids=INVALID_IDS + ) + def test_validation_survives_optimized_mode(self, actuator_class, kwargs, message): + """``python -O`` drops assert statements, so these must not be asserts. + + Run in a subprocess because the flag is set at interpreter startup. Under + the old bare asserts every one of these was accepted in silence. + """ + arguments = ", ".join(f"{k}={_as_source(v)}" for k, v in kwargs.items()) + source = ( + "from rocketpy.rocket.actuator import " + f"{actuator_class.__name__} as A; A({arguments})" + ) + result = subprocess.run( + [sys.executable, "-O", "-c", source], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode != 0, "invalid arguments were accepted under -O" + assert "ValueError" in result.stderr + assert message in result.stderr + + def test_demand_rate_none_builds_a_continuous_actuator(self): + """None is the documented continuous-time mode and must be accepted. + + The check used to read ``demand_rate > 0 or demand_rate is None``, and + Python evaluates the left operand first, so this raised TypeError and the + mode could not be constructed at all. + """ + actuator = RollActuator(demand_rate=None) + + assert actuator.demand_rate is None + + @pytest.mark.parametrize( + "actuator_class, kwargs", + [ + (RollActuator, {"torque_rate_limit": 0.0}), + (ThrottleActuator, {"throttle_time_constant": 0.0}), + (ThrustVectorActuator, {"gimbal_rate_limit": 0.0}), + ], + ) + def test_zero_is_accepted_where_the_bound_is_non_negative( + self, actuator_class, kwargs + ): + """Zero is on the legal side of every non-negative bound. + + Worth its own case because the fix moved these from ``x < 0`` to + ``not x >= 0``, and an off-by-one there would turn the documented "no + dynamics" and "no rate limit" settings into errors. A zero rate limit + does freeze the actuator, but that is the caller's business, and the + constructor is not where that is decided. + """ + actuator = actuator_class(**kwargs) + + assert actuator is not None + + +class TestActuatorInitialOutput: + """An actuator must not start outside its own range. + + The output setter already refuses to leave the range, but the initial value + bypassed it and ``_reset()`` restored that same value, so a reset actuator + ended up somewhere the setter would never have put it. + """ + + def test_an_out_of_range_initial_value_is_clamped(self): + actuator = ThrottleActuator(throttle_range=(0.0, 1.0), initial_throttle=2.0) + + assert actuator.actuator_output == 1.0 + assert actuator.actuator_initial_output == 1.0 + + def test_the_clamped_value_survives_a_reset(self): + actuator = ThrottleActuator(throttle_range=(0.0, 1.0), initial_throttle=-3.0) + actuator.actuator_output = 0.5 + actuator._reset() + + assert actuator.actuator_output == 0.0 + + def test_a_value_inside_the_range_is_untouched(self): + actuator = ThrottleActuator(throttle_range=(0.0, 1.0), initial_throttle=0.25) - def test_invalid_range(self): - """Test that invalid range raises assertion error.""" - with pytest.raises(AssertionError): - RollActuator( - max_roll_torque=-5 - ) # This creates range (5, -5) which is invalid + assert actuator.actuator_initial_output == 0.25 - def test_invalid_time_constant_negative(self): - """Test that negative time constant raises assertion error.""" - with pytest.raises(AssertionError): - ThrottleActuator(throttle_time_constant=-0.1) + def test_without_clamping_it_warns_instead(self): + # clamp=False is the documented way to let an actuator report outside its + # range, so the initial value follows the setter and only warns. + with pytest.warns(UserWarning, match="outside its range"): + actuator = ThrottleActuator( + throttle_range=(0.0, 1.0), initial_throttle=2.0, clamp=False + ) - def test_invalid_rate_limit_negative(self): - """Test that negative rate limit raises assertion error.""" - with pytest.raises(AssertionError): - ThrustVectorActuator(gimbal_rate_limit=-1.0) + assert actuator.actuator_initial_output == 2.0 + + def test_the_warning_is_not_blamed_on_the_actuator_module(self): + """``pytest.warns`` reads the message, and the message is not the whole + warning. Without a stacklevel the report points at the ``warnings.warn`` + line inside ``actuator.py``, which is the same line for every caller, so + ``-W`` filters keyed on a module and the printed location are both + useless. + + Asserting the negative rather than a specific file because the warning is + raised in a base ``__init__`` reached through ``super()``, so no single + stacklevel lands on user code for every actuator: 2 reaches the concrete + subclass, and the dual-axis actuator adds another frame on top of that. + Not blaming the base module is the part that holds for all of them. + """ + with pytest.warns(UserWarning, match="outside its range") as record: + ThrottleActuator( + throttle_range=(0.0, 1.0), initial_throttle=2.0, clamp=False + ) + + assert not record[0].filename.endswith("actuator.py") class TestActuatorWarnings: @@ -494,3 +693,196 @@ def test_clamping_applied(self): actuator.actuator_output = 15.0 # Output should be clamped to range assert actuator.roll_torque == 10.0 + + +class TestTheFilterCoefficient: + """The coefficient is one expression rather than two. + + Same value either way, so this is what says the rewrite is a rewrite. + """ + + @pytest.mark.parametrize( + "demand_rate, time_constant", + [(1e-3, 1e-6), (1.0, 1.0), (100.0, 0.05), (1e4, 1e3)], + ) + def test_the_two_forms_agree_where_both_work(self, demand_rate, time_constant): + """So the rewrite is a rewrite and not a change of behaviour. Measured + over 2000 random pairs across these ranges: the largest difference is + 2.2e-16.""" + actuator = ThrottleActuator( + demand_rate=demand_rate, + throttle_time_constant=time_constant, + throttle_range=(0.0, 1.0), + ) + demand_period = 1.0 / demand_rate + + assert actuator._alpha == pytest.approx( + demand_period / (time_constant + demand_period), rel=1e-12 + ) + + +class TestAMisspeltRange: + """A range read out of a config file arrives as anything. + + Three endpoints dropped the third in silence, one raised IndexError, and + string endpoints failed inside np.clip with a ufunc loop error. + """ + + @pytest.mark.parametrize( + "limits", [(0.0,), (0.0, 1.0, 2.0), ("0", "1"), 1.0, None, (True, False)] + ) + def test_it_is_refused_at_the_boundary(self, limits): + with pytest.raises(ValueError, match="actuator_range"): + ThrottleActuator(throttle_range=limits) + + @pytest.mark.parametrize("limits", [(0.0, 1.0), (0, 1), [0.0, 1.0]]) + def test_the_spellings_that_work_still_work(self, limits): + assert ThrottleActuator(throttle_range=limits).actuator_range == (0.0, 1.0) + + +class TestDynamicsNeedADemandRate: + """Neither is implemented for a continuous actuator. + + Both were accepted with a warning, so asking for a 0.1 s lag gave an instant + response and said so in a line that a simulation log buries. + """ + + @pytest.mark.parametrize( + "option", [{"throttle_time_constant": 0.1}, {"throttle_rate_limit": 0.5}] + ) + def test_asking_for_one_without_a_rate_is_refused(self, option): + with pytest.raises(ValueError, match="needs a demand_rate"): + ThrottleActuator(demand_rate=None, **option) + + def test_a_continuous_actuator_on_its_own_still_builds(self): + assert ThrottleActuator(demand_rate=None).demand_rate is None + + +def _no_op_controller(time, sampling_rate, state, state_history, observed, interactive): # pylint: disable=unused-argument + """A controller that commands nothing, so these tests are about the wiring.""" + return None + + +class TestTheControllerAndTheActuatorShareOneSamplingRate: + """``add_*_control`` built the two halves from the same argument. + + The actuator normalizes what it is given and stores a float. The controller + was handed the original object, so one quantity ended up held twice, in two + types. Nothing complained: the call returned a rocket that looked complete, + and the failure surfaced later inside Flight at + ``controller_time_step = 1 / controller.sampling_rate``, by which point the + actuator and the controller were already attached. + + Passing the actuator's normalized value is the smaller of the two available + fixes. Refusing strings and bools outright is the other, and it is a wider + behaviour change than this needs. + """ + + # The third entry reaches through .x because ThrustVectorActuator2D keeps + # no demand_rate of its own, only the two axes it builds from the argument. + ADDERS = [ + ("add_roll_control", "roll_control", {"max_roll_torque": 10.0}, False), + ("add_throttle_control", "throttle_control", {}, False), + ( + "add_thrust_vector_control", + "thrust_vector_control", + {"max_gimbal_angle": 5.0}, + True, + ), + ] + + @staticmethod + def _rate_of(rocket, attribute, per_axis): + actuator = getattr(rocket, attribute) + return (actuator.x if per_axis else actuator).demand_rate + + @pytest.mark.parametrize("adder, attribute, extra, per_axis", ADDERS) + @pytest.mark.parametrize("given", ["100", 100, 100.0]) + def test_both_halves_hold_the_same_normalized_value( + self, calisto, adder, attribute, extra, per_axis, given + ): + getattr(calisto, adder)( + controller_function=_no_op_controller, sampling_rate=given, **extra + ) + rate = self._rate_of(calisto, attribute, per_axis) + controller = calisto._controllers[-1] + + assert controller.sampling_rate == rate + assert type(controller.sampling_rate) is type(rate) + + @pytest.mark.parametrize( + "adder, extra", [(adder, extra) for adder, _, extra, _ in ADDERS] + ) + def test_the_controller_rate_is_usable_as_a_time_step(self, calisto, adder, extra): + """The exact expression Flight evaluates, which is where a string blew + up with a TypeError after the rocket had already been assembled.""" + getattr(calisto, adder)( + controller_function=_no_op_controller, sampling_rate="100", **extra + ) + controller = calisto._controllers[-1] + + assert 1 / controller.sampling_rate == pytest.approx(0.01) + + +class TestAFailedReplacementLeavesTheRocketAlone: + """The old controller used to go before the new one was built. + + A rejected argument on a second call then left the rocket carrying an + actuator that simulation would never call, with nothing said. + """ + + ADDERS = [ + ("add_roll_control", {"max_roll_torque": 10.0}), + ("add_throttle_control", {}), + ("add_thrust_vector_control", {"max_gimbal_angle": 5.0}), + ] + + @pytest.mark.parametrize("adder, extra", ADDERS) + def test_a_rejected_second_call_keeps_the_first_controller( + self, calisto, adder, extra + ): + getattr(calisto, adder)( + controller_function=_no_op_controller, sampling_rate=100, **extra + ) + before = list(calisto._controllers) + + with pytest.raises(ValueError): + getattr(calisto, adder)( + controller_function=_no_op_controller, sampling_rate=-1, **extra + ) + + assert calisto._controllers == before + + @pytest.mark.parametrize("adder, extra", ADDERS) + def test_an_accepted_second_call_still_replaces(self, calisto, adder, extra): + getattr(calisto, adder)( + controller_function=_no_op_controller, sampling_rate=100, **extra + ) + getattr(calisto, adder)( + controller_function=_no_op_controller, sampling_rate=50, **extra + ) + + assert len(calisto._controllers) == 1 + assert calisto._controllers[0].sampling_rate == 50.0 + + +class TestABooleanIsNotARate: + """YAML reads `yes` and `on` as True, and float(True) is 1.0. + + A sampling rate written that way became 1 Hz with nothing said, and an + earlier revision of this file pinned that as the contract. + """ + + @pytest.mark.parametrize( + "option", + ["demand_rate", "throttle_rate_limit", "throttle_time_constant"], + ) + def test_a_boolean_is_refused(self, option): + with pytest.raises(ValueError, match="boolean"): + ThrottleActuator(**{option: True}) + + def test_an_integer_too_large_for_a_float_is_a_value_error(self): + """`float(10**400)` raises OverflowError, so the validator leaked a + different exception type than everything beside it.""" + with pytest.raises(ValueError): + ThrottleActuator(demand_rate=10**400)