diff --git a/apps/studio/tests/test_force_output_pipeline.py b/apps/studio/tests/test_force_output_pipeline.py new file mode 100644 index 0000000..e0aa2e5 --- /dev/null +++ b/apps/studio/tests/test_force_output_pipeline.py @@ -0,0 +1,52 @@ +"""One thresholded record fans out without changing calibration inputs.""" + +from types import SimpleNamespace + +import numpy as np + +from orisys_app_shared.force.session import SessionStore +from orisys_studio.recording import RecordingCoordinator + + +class _Capture: + is_recording = False + + def set_alarm_handler(self, handler): + self.alarm_handler = handler + + def on_packet(self, packet, *, sample): + self.sample = sample + + +class _Calibration: + def feed(self, sample): + self.sample = sample + + +def test_output_and_originals_share_frame_across_consumers(): + store = SessionStore() + calibration = _Calibration() + recording = _Capture() + coordinator = RecordingCoordinator( + session_store=store, cal_recorder=calibration, recorder=recording + ) + coordinator.set_force_thresholds(raw=50, newton=0.5) + packet = SimpleNamespace( + frame_id=7, mono_t=7.0, wall_t=1007.0, valid=True, + raw_available=True, force_calibrated=True, + calibrated_axes=("fn", "fx", "fy"), + normal_force_raw=49.0, shear_x_raw=-51.0, shear_y_raw=20.0, + normal_force=0.4, shear_x=-0.6, shear_y=0.1, + ) + output = coordinator.on_packet(packet) + + assert output is recording.sample + assert output.raw_triplet() == (49.0, -51.0, 20.0) + assert output.processed_triplet("raw") == (0.0, -51.0, 0.0) + assert output.processed_triplet("N") == (0.0, -0.6, 0.0) + assert calibration.sample.raw_triplet() == output.raw_triplet() + stored = store.data() + assert stored.frame_ids.tolist() == [7] + assert stored.raw[0].tolist() == [49.0, -51.0, 20.0] + assert stored.output_raw[0].tolist() == [0.0, -51.0, 0.0] + assert np.allclose(stored.output_newtons[0], [0.0, -0.6, 0.0])