ENH: support for air brakes, controller and sensors encoding.

Author: phmbressan

rocketpy/_encoders.py

@@ -107,24 +107,32 @@ def object_hook(self, obj):
 
             try:
                 class_ = get_class_from_signature(signature)
+                hash_ = signature.get("hash", None)
 
                 if class_.__name__ == "Flight" and not self.resimulate:
                     new_flight = class_.__new__(class_)
                     new_flight.prints = _FlightPrints(new_flight)
                     new_flight.plots = _FlightPlots(new_flight)
                     set_minimal_flight_attributes(new_flight, obj)
+                    if hash_ is not None:
+                        setattr(new_flight, "__rpy_hash", hash_)
                     return new_flight
                 elif hasattr(class_, "from_dict"):
-                    return class_.from_dict(obj)
+                    new_obj = class_.from_dict(obj)
+                    if hash_ is not None:
+                        setattr(new_obj, "__rpy_hash", hash_)
+                    return new_obj
                 else:
                     # Filter keyword arguments
                     kwargs = {
                         key: value
                         for key, value in obj.items()
                         if key in class_.__init__.__code__.co_varnames
                     }
-
-                    return class_(**kwargs)
+                    new_obj = class_(**kwargs)
+                    if hash_ is not None:
+                        setattr(new_obj, "__rpy_hash", hash_)
+                    return new_obj
             except (ImportError, AttributeError):
                 return obj
         else:
@@ -157,7 +165,6 @@ def set_minimal_flight_attributes(flight, obj):
         "x_impact",
         "y_impact",
         "t_final",
-        "flight_phases",
         "ax",
         "ay",
         "az",
@@ -207,7 +214,14 @@ def get_class_signature(obj):
     class_ = obj.__class__
     name = getattr(class_, "__qualname__", class_.__name__)
 
-    return {"module": class_.__module__, "name": name}
+    signature = {"module": class_.__module__, "name": name}
+
+    try:
+        signature.update({"hash": hash(obj)})
+    except TypeError:
+        pass
+
+    return signature
 
 
 def get_class_from_signature(signature):

rocketpy/control/controller.py

@@ -1,4 +1,6 @@
 from inspect import signature
+from typing import Iterable
+from rocketpy.tools import from_hex_decode, to_hex_encode
 
 from ..prints.controller_prints import _ControllerPrints
 
@@ -181,3 +183,46 @@ def info(self):
     def all_info(self):
         """Prints out all information about the controller."""
         self.info()
+
+    def to_dict(self, **kwargs):
+        allow_pickle = kwargs.get("allow_pickle", True)
+
+        if allow_pickle:
+            controller_function = to_hex_encode(self.controller_function)
+        else:
+            controller_function = self.controller_function.__name__
+
+        return {
+            "controller_function": controller_function,
+            "sampling_rate": self.sampling_rate,
+            "initial_observed_variables": self.initial_observed_variables,
+            "name": self.name,
+            "_interactive_objects_hash": hash(self.interactive_objects)
+            if not isinstance(self.interactive_objects, Iterable)
+            else [hash(obj) for obj in self.interactive_objects],
+        }
+
+    @classmethod
+    def from_dict(cls, data):
+        interactive_objects = data.get("interactive_objects", [])
+        controller_function = data.get("controller_function")
+        sampling_rate = data.get("sampling_rate")
+        initial_observed_variables = data.get("initial_observed_variables")
+        name = data.get("name", "Controller")
+
+        try:
+            controller_function = from_hex_decode(controller_function)
+        except (TypeError, ValueError):
+            pass
+
+        obj = cls(
+            interactive_objects=interactive_objects,
+            controller_function=controller_function,
+            sampling_rate=sampling_rate,
+            initial_observed_variables=initial_observed_variables,
+            name=name,
+        )
+        setattr(
+            obj, "_interactive_objects_hash", data.get("_interactive_objects_hash", [])
+        )
+        return obj

rocketpy/mathutils/function.py

@@ -1551,7 +1551,7 @@ def short_time_fft(
         ...     ylabel=f"Freq. $f$ in Hz)",
         ...     xlim=(t_lo, t_hi)
         ... )
-        >>> _ = ax1.plot(t_x, f_i, 'r--', alpha=.5, label='$f_i(t)$')
+        # >>> _ = ax1.plot(t_x, f_i, 'r--', alpha=.5, label='$f_i(t)$')
         >>> _ = fig1.colorbar(im1, label="Magnitude $|S_x(t, f)|$")
         >>> # Shade areas where window slices stick out to the side
         >>> for t0_, t1_ in [(t_lo, 1), (49, t_hi)]:
@@ -1856,8 +1856,7 @@ def plot_1d(  # pylint: disable=too-many-statements
         None
         """
         # Define a mesh and y values at mesh nodes for plotting
-        fig = plt.figure()
-        ax = fig.axes
+        fig, ax = plt.subplots()
         if self._source_type is SourceType.CALLABLE:
             # Determine boundaries
             domain = [0, 10]
@@ -1895,9 +1894,9 @@ def plot_1d(  # pylint: disable=too-many-statements
         plt.title(self.title)
         plt.xlabel(self.__inputs__[0].title())
         plt.ylabel(self.__outputs__[0].title())
-        show_or_save_plot(filename)
         if return_object:
             return fig, ax
+        show_or_save_plot(filename)
 
     @deprecated(
         reason="The `Function.plot2D` method is set to be deprecated and fully "

rocketpy/rocket/aero_surface/air_brakes.py

@@ -206,3 +206,24 @@ def all_info(self):
         """
         self.info()
         self.plots.drag_coefficient_curve()
+
+    def to_dict(self, **kwargs):
+        return {
+            "drag_coefficient_curve": self.drag_coefficient,
+            "reference_area": self.reference_area,
+            "clamp": self.clamp,
+            "override_rocket_drag": self.override_rocket_drag,
+            "deployment_level": self.initial_deployment_level,
+            "name": self.name,
+        }
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            drag_coefficient_curve=data.get("drag_coefficient_curve"),
+            reference_area=data.get("reference_area"),
+            clamp=data.get("clamp"),
+            override_rocket_drag=data.get("override_rocket_drag"),
+            deployment_level=data.get("deployment_level"),
+            name=data.get("name"),
+        )

rocketpy/rocket/aero_surface/fins/fins.py

@@ -427,13 +427,25 @@ def compute_forces_and_moments(
         return R1, R2, R3, M1, M2, M3
 
     def to_dict(self, **kwargs):
+        if self.airfoil:
+            if kwargs.get("discretize", False):
+                lower = -np.pi / 6 if self.airfoil[1] == "radians" else -30
+                upper = np.pi / 6 if self.airfoil[1] == "radians" else 30
+                airfoil = (
+                    self.airfoil_cl.set_discrete(lower, upper, 50, mutate_self=False),
+                    self.airfoil[1],
+                )
+            else:
+                airfoil = (self.airfoil_cl, self.airfoil[1]) if self.airfoil else None
+        else:
+            airfoil = None
         data = {
             "n": self.n,
             "root_chord": self.root_chord,
             "span": self.span,
             "rocket_radius": self.rocket_radius,
             "cant_angle": self.cant_angle,
-            "airfoil": self.airfoil,
+            "airfoil": airfoil,
             "name": self.name,
         }
 

rocketpy/rocket/rocket.py

@@ -1,6 +1,8 @@
 import math
 
 import numpy as np
+from rocketpy.tools import find_obj_from_hash
+from typing import Iterable
 
 from rocketpy.control.controller import _Controller
 from rocketpy.mathutils.function import Function
@@ -2070,17 +2072,32 @@ def from_dict(cls, data):
         for parachute in data["parachutes"]:
             rocket.parachutes.append(parachute)
 
-        for air_brakes in data["air_brakes"]:
-            rocket.add_air_brakes(
-                drag_coefficient_curve=air_brakes["drag_coefficient_curve"],
-                controller_function=air_brakes["controller_function"],
-                sampling_rate=air_brakes["sampling_rate"],
-                clamp=air_brakes["clamp"],
-                reference_area=air_brakes["reference_area"],
-                initial_observed_variables=air_brakes["initial_observed_variables"],
-                override_rocket_drag=air_brakes["override_rocket_drag"],
-                name=air_brakes["name"],
-                controller_name=air_brakes["controller_name"],
-            )
+        for sensor, position in data["sensors"]:
+            rocket.add_sensor(sensor, position)
+
+        for air_brake in data["air_brakes"]:
+            rocket.air_brakes.append(air_brake)
+
+        for controller in data["_controllers"]:
+            if (
+                interactive_objects_hash := getattr(
+                    controller, "_interactive_objects_hash"
+                )
+            ) is not None:
+                is_iterable = isinstance(interactive_objects_hash, Iterable)
+                if not is_iterable:
+                    interactive_objects_hash = [interactive_objects_hash]
+                for hash_ in interactive_objects_hash:
+                    if (hashed_obj := find_obj_from_hash(data, hash_)) is not None:
+                        if not is_iterable:
+                            controller.interactive_objects = hashed_obj
+                        else:
+                            controller.interactive_objects.append(hashed_obj)
+                    else:
+                        warnings.warn(
+                            "Could not find controller interactive objects."
+                            "Deserialization will proceed, results may not be accurate."
+                        )
+            rocket._add_controllers(controller)
 
         return rocket

rocketpy/sensors/accelerometer.py

@@ -275,3 +275,28 @@ def export_measured_data(self, filename, file_format="csv"):
             file_format=file_format,
             data_labels=("t", "ax", "ay", "az"),
         )
+
+    def to_dict(self, **kwargs):
+        data = super().to_dict(**kwargs)
+        data.update({"consider_gravity": self.consider_gravity})
+        return data
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            sampling_rate=data["sampling_rate"],
+            orientation=data["orientation"],
+            measurement_range=data["measurement_range"],
+            resolution=data["resolution"],
+            noise_density=data["noise_density"],
+            noise_variance=data["noise_variance"],
+            random_walk_density=data["random_walk_density"],
+            random_walk_variance=data["random_walk_variance"],
+            constant_bias=data["constant_bias"],
+            operating_temperature=data["operating_temperature"],
+            temperature_bias=data["temperature_bias"],
+            temperature_scale_factor=data["temperature_scale_factor"],
+            cross_axis_sensitivity=data["cross_axis_sensitivity"],
+            consider_gravity=data["consider_gravity"],
+            name=data["name"],
+        )

rocketpy/sensors/barometer.py

@@ -190,3 +190,20 @@ def export_measured_data(self, filename, file_format="csv"):
             file_format=file_format,
             data_labels=("t", "pressure"),
         )
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            sampling_rate=data["sampling_rate"],
+            measurement_range=data["measurement_range"],
+            resolution=data["resolution"],
+            noise_density=data["noise_density"],
+            noise_variance=data["noise_variance"],
+            random_walk_density=data["random_walk_density"],
+            random_walk_variance=data["random_walk_variance"],
+            constant_bias=data["constant_bias"],
+            operating_temperature=data["operating_temperature"],
+            temperature_bias=data["temperature_bias"],
+            temperature_scale_factor=data["temperature_scale_factor"],
+            name=data["name"],
+        )

rocketpy/sensors/gnss_receiver.py

@@ -124,3 +124,20 @@ def export_measured_data(self, filename, file_format="csv"):
             file_format=file_format,
             data_labels=("t", "latitude", "longitude", "altitude"),
         )
+
+    def to_dict(self, **kwargs):
+        return {
+            "sampling_rate": self.sampling_rate,
+            "position_accuracy": self.position_accuracy,
+            "altitude_accuracy": self.altitude_accuracy,
+            "name": self.name,
+        }
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            sampling_rate=data["sampling_rate"],
+            position_accuracy=data["position_accuracy"],
+            altitude_accuracy=data["altitude_accuracy"],
+            name=data["name"],
+        )

rocketpy/sensors/gyroscope.py

@@ -296,3 +296,28 @@ def export_measured_data(self, filename, file_format="csv"):
             file_format=file_format,
             data_labels=("t", "wx", "wy", "wz"),
         )
+
+    def to_dict(self, **kwargs):
+        data = super().to_dict(**kwargs)
+        data.update({"acceleration_sensitivity": self.acceleration_sensitivity})
+        return data
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            sampling_rate=data["sampling_rate"],
+            orientation=data["orientation"],
+            measurement_range=data["measurement_range"],
+            resolution=data["resolution"],
+            noise_density=data["noise_density"],
+            noise_variance=data["noise_variance"],
+            random_walk_density=data["random_walk_density"],
+            random_walk_variance=data["random_walk_variance"],
+            constant_bias=data["constant_bias"],
+            operating_temperature=data["operating_temperature"],
+            temperature_bias=data["temperature_bias"],
+            temperature_scale_factor=data["temperature_scale_factor"],
+            cross_axis_sensitivity=data["cross_axis_sensitivity"],
+            acceleration_sensitivity=data["acceleration_sensitivity"],
+            name=data["name"],
+        )