Added test for serialization and deserialization of online server

bioptim/gui/online_callback_server.py

@@ -521,7 +521,7 @@ def eval(self, arg: list | tuple, enforce: bool = False) -> list[int]:
 
         self._socket.sendall(f"{_ServerMessages.NEW_DATA.value}\n{[len(header), len(data_serialized)]}".encode())
         # If send_confirmation is True, we should wait for the server to acknowledge the data here (sends OK)
-        self._socket.sendall(header.encode())
+        self._socket.sendall(header)
         self._socket.sendall(data_serialized)
         # Again, if send_confirmation is True, we should wait for the server to acknowledge the data here (sends OK)
         return [0]
@@ -565,6 +565,7 @@ def _serialize_xydata(xdata: list, ydata: list) -> tuple:
             y_steps_tp = y_steps.tolist()
             data_serialized += struct.pack("d" * len(y_steps_tp), *y_steps_tp)
 
+    header = header.encode()
     return header, data_serialized
 
 

tests/shard4/test_solver_options.py

@@ -1,7 +1,5 @@
-import pytest
-
 from bioptim import Solver
-from bioptim.misc.enums import SolverType, OnlineOptim
+from bioptim.misc.enums import SolverType
 
 
 class FakeSolver:

tests/shard5/test_plot_server.py

@@ -0,0 +1,42 @@
+import os
+
+from bioptim.gui.online_callback_server import _serialize_xydata, _deserialize_xydata
+from bioptim.gui.plot import PlotOcp
+from bioptim.optimization.optimization_vector import OptimizationVectorHelper
+from casadi import DM, Function
+import numpy as np
+
+
+def test_serialize_deserialize():
+    # Prepare a set of data to serialize and deserialize
+    from bioptim.examples.getting_started import pendulum as ocp_module
+
+    bioptim_folder = os.path.dirname(ocp_module.__file__)
+
+    ocp = ocp_module.prepare_ocp(
+        biorbd_model_path=bioptim_folder + "/models/pendulum.bioMod",
+        final_time=1,
+        n_shooting=40,
+    )
+
+    dummy_phase_times = OptimizationVectorHelper.extract_step_times(ocp, DM(np.ones(ocp.n_phases)))
+    plotter = PlotOcp(ocp, dummy_phase_times=dummy_phase_times, show_bounds=True, only_initialize_variables=True)
+
+    np.random.seed(42)
+    xdata, ydata = plotter.parse_data(**{"x": np.random.rand(ocp.variables_vector.shape[0])[:, None]})
+
+    # Serialize and deserialize the data
+    serialized_data = _serialize_xydata(xdata, ydata)
+    deserialized_xdata, deserialized_ydata = _deserialize_xydata(serialized_data)
+
+    # Compare the outputs
+    for x_phase, deserialized_x_phase in zip(xdata, deserialized_xdata):
+        for x_node, deserialized_x_node in zip(x_phase, deserialized_x_phase):
+            assert np.allclose(x_node, DM(deserialized_x_node))
+
+    for y_variable, deserialized_y_variable in zip(ydata, deserialized_ydata):
+        if isinstance(y_variable, np.ndarray):
+            assert np.allclose(y_variable, deserialized_y_variable[0], equal_nan=True)
+        else:
+            for y_phase, deserialized_y_phase in zip(y_variable, deserialized_y_variable):
+                assert np.allclose(y_phase, deserialized_y_phase)