ENH: Controller (AirBrakes) and Sensors Encoding (RocketPy-Team#849)

* ENH: add an option to discretize callable sources encoding.

* ENH: allow for disallowing pickle on encoding.

* MNT: Update CHANGELOG.

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

* STY: solve linting and style remarks.

* BUG: parachute callbacks attribute naming.

* GIT: test agg backend for matplotlib workflows.

* TST: include sensors and controllers encoding tests as non slow.

* MNT: change recursive to iterative approach on hash search.

Author: uniKorqi422

.github/workflows/test-pytest-slow.yaml

@@ -5,7 +5,7 @@ on:
     - cron: "0 17 * * 5"  # at 05:00 PM, only on Friday
   push:
     branches:
-      - main
+      - master
     paths:
       - "**.py"
       - ".github/**"
@@ -24,6 +24,7 @@ jobs:
         python-version: [3.9, 3.13]
     env:
       PYTHON: ${{ matrix.python-version }}
+      MPLBACKEND: Agg
     steps:
       - uses: actions/checkout@main
       - name: Set up Python

.github/workflows/test_pytest.yaml

@@ -23,6 +23,7 @@ jobs:
     env:
       OS: ${{ matrix.os }}
       PYTHON: ${{ matrix.python-version }}
+      MPLBACKEND: Agg
     steps:
       - uses: actions/checkout@main
       - name: Set up Python

CHANGELOG.md

@@ -31,7 +31,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 Attention: The newest changes should be on top -->
 
 ### Added
-
+- ENH: Controller (AirBrakes) and Sensors Encoding [#849] (https://github.com/RocketPy-Team/RocketPy/pull/849)
 - EHN: Addition of ensemble variable to ECMWF dictionaries [#842] (https://github.com/RocketPy-Team/RocketPy/pull/842) 
 - ENH: Added Crop and Clip Methods to Function Class [#817](https://github.com/RocketPy-Team/RocketPy/pull/817)
 - DOC: Add Flight class usage documentation and update index [#841](https://github.com/RocketPy-Team/RocketPy/pull/841)

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,7 @@
 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 +184,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/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):  # pylint: disable=unused-argument
+        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,4 +1,6 @@
 import math
+import warnings
+from typing import Iterable
 
 import numpy as np
 
@@ -21,7 +23,11 @@
 from rocketpy.rocket.aero_surface.generic_surface import GenericSurface
 from rocketpy.rocket.components import Components
 from rocketpy.rocket.parachute import Parachute
-from rocketpy.tools import deprecated, parallel_axis_theorem_from_com
+from rocketpy.tools import (
+    deprecated,
+    find_obj_from_hash,
+    parallel_axis_theorem_from_com,
+)
 
 
 # pylint: disable=too-many-instance-attributes, too-many-public-methods, too-many-instance-attributes
@@ -2070,17 +2076,29 @@ 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"]:
+            interactive_objects_hash = getattr(controller, "_interactive_objects_hash")
+            if 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"],
+        )