id3c.tests.test_flyscan_3idc_analysis ===================================== .. py:module:: id3c.tests.test_flyscan_3idc_analysis .. autoapi-nested-parse:: Pure-Python unit tests for ``id3c.utils.flyscan_3idc_analysis``. No databroker, no tiled, no ophyd, no IOC. Builds duck-typed run objects from synthetic monitor-stream data and asserts the pairing function produces correct ``(image_number, position)`` rows. Run with:: pytest src/id3c/tests/test_flyscan_3idc_analysis.py -v Functions --------- .. autoapisummary:: id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_simple_linear id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_trims_outside_scan_range id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_handles_out_of_order_motor_stream id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_handles_duplicate_timestamps id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_drops_extrapolation id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_empty_hdf_returns_empty_frame id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_requires_two_motor_samples id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_shape_mismatch_raises id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_three_phase_arithmetic_at_constant_velocity id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_in_scan_filter_uses_start_acquire id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_widening_admits_leading_edge_frame id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_widening_admits_trailing_edge_frame id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_widening_rejects_all_when_motor_never_in_range id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_dedups_duplicate_image_numbers id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_end_to_end id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_trims_taxi_frames id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_index_is_timestamp id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_missing_metadata_raises id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_missing_motor_stream_raises id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_against_observed_run_shape id3c.tests.test_flyscan_3idc_analysis.test_array_from_ds_handles_xarray_like id3c.tests.test_flyscan_3idc_analysis.test_array_from_ds_handles_dataframe_like id3c.tests.test_flyscan_3idc_analysis.test_array_from_ds_missing_key_raises id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_end_acquire id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_start_acquire id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_end_period id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_closest_with_noise id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_returns_expected_dict_keys id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_flags_noisy_data id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_flags_sparse_data id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_raises_when_no_in_scan_frames Module Contents --------------- .. py:function:: test_interpolate_positions_simple_linear() At constant velocity, each frame should land at the velocity * (t - t0) position. With t_acquire=1e-9 and hdf_t_phase_offset=0, the three per-phase positions all coincide with the position at hdf_t, so the old-shape assertion translates to position_start_acquire. .. py:function:: test_interpolate_positions_trims_outside_scan_range() Frames whose interpolated positions fall outside [p_start, p_end] are dropped. .. py:function:: test_interpolate_positions_handles_out_of_order_motor_stream() Record-order-shuffled motor stream still produces correct interpolation (mirrors what we observed in real BlueskyRuns: the m1_monitor stream's events are interleaved across CA dispatcher segments). .. py:function:: test_interpolate_positions_handles_duplicate_timestamps() Motor samples with repeated timestamps are deduped (keep first occurrence) — observed in the live m1_monitor stream where the same readback value appeared at the same IOC timestamp on multiple consecutive events. .. py:function:: test_interpolate_positions_drops_extrapolation(caplog) HDF frames with timestamps outside the motor stream's time range are dropped (extrapolation rejected, logged at WARNING). .. py:function:: test_interpolate_positions_empty_hdf_returns_empty_frame() No HDF frames -> empty result, no crash. .. py:function:: test_interpolate_positions_requires_two_motor_samples() Single-sample motor stream cannot be interpolated. .. py:function:: test_interpolate_positions_shape_mismatch_raises() Shape mismatches between paired arrays raise ValueError. .. py:function:: test_interpolate_positions_three_phase_arithmetic_at_constant_velocity() At constant motor velocity, the three per-phase positions must satisfy: position_end_acquire = position_start_acquire + v * t_acquire position_end_period = position_start_acquire + v * t_period where v is the motor's velocity. Pure arithmetic check that the three np.interp calls use the correct phase timestamps. .. py:function:: test_interpolate_positions_in_scan_filter_uses_start_acquire() The in-scan filter uses position_start_acquire (not the other two phases). Construct a frame whose start_acquire is just inside p_start but whose end_acquire/end_period are further into the scan: the frame is kept (start_acquire in scan). Conversely a frame whose start_acquire is just outside p_end but whose end_acquire is past p_end is dropped (start out of scan). .. py:function:: test_interpolate_positions_widening_admits_leading_edge_frame() Widening admits a leading-edge frame the old rule rejected. Frame's [start_acquire.t, end_period.t] overlaps the motor's in-range window even though start_acquire is before p_start. .. py:function:: test_interpolate_positions_widening_admits_trailing_edge_frame() Trailing-edge analogue of the leading-edge test. A frame whose start_acquire is just inside p_end but whose end_period crosses past p_end is admitted (its exposure started inside the scan range). In contrast a frame whose entire interval is past the motor's last in-range timestamp is still rejected. .. py:function:: test_interpolate_positions_widening_rejects_all_when_motor_never_in_range() No frames are admitted if the motor never enters the range. .. py:function:: test_interpolate_positions_dedups_duplicate_image_numbers(caplog) Duplicate image_number values within the in-scan window are deduped (keep first occurrence) with a WARNING. Mirrors the motor-timestamp dedup behavior at the counter level. .. py:function:: test_pair_frames_to_positions_end_to_end() Build a fake BlueskyRun-shaped object and pair frames. .. py:function:: test_pair_frames_to_positions_trims_taxi_frames() Frames captured during taxi-in (motor below p_start) are dropped — this is the empirically-observed flyscan_3idc behaviour where the cam delivers a first frame at p_initial. .. py:function:: test_pair_frames_to_positions_index_is_timestamp() The returned DataFrame is indexed by timestamp. .. py:function:: test_pair_frames_to_positions_missing_metadata_raises() Missing start metadata -> KeyError with helpful message. .. py:function:: test_pair_frames_to_positions_missing_motor_stream_raises() Missing motor stream -> KeyError. .. py:function:: test_pair_frames_to_positions_against_observed_run_shape() Build the same data we observed in the 17:21 run (scan_id=6, UID 0f397d39) and confirm the pairing matches what an analyst would expect. The HDF counter stream was [0..36] sorted-by-time, spanning 12.27 s. Motor stream was 127 samples spanning the taxi+scan+coast range with constant scan_velocity=0.495 deg/s in the in-scan window. This isn't an exact replay (we don't have a frame-by-frame timestamp dump), but the shape is realistic and tests the integration. .. py:function:: test_array_from_ds_handles_xarray_like() Pulls a 1-D array from an xarray-like Dataset. .. py:function:: test_array_from_ds_handles_dataframe_like() Pulls a 1-D array from a pandas DataFrame column. .. py:function:: test_array_from_ds_missing_key_raises() A missing column raises KeyError naming the stream. .. py:function:: test_hdf_timestamp_diagnostic_picks_end_acquire() When hdf_t == cam_t (the end_acquire semantic), the diagnostic picks 'end_acquire' and recommends offset = -t_acquire. .. py:function:: test_hdf_timestamp_diagnostic_picks_start_acquire() When hdf_t == cam_t - t_acquire (the start_acquire semantic), the diagnostic picks 'start_acquire' and recommends offset = 0. .. py:function:: test_hdf_timestamp_diagnostic_picks_end_period() When hdf_t == cam_t + (t_period - t_acquire) (the end_period semantic), the diagnostic picks 'end_period' and recommends offset = -t_period. .. py:function:: test_hdf_timestamp_diagnostic_picks_closest_with_noise() Realistic case: HDF lags cam by a small plugin-pipeline latency (e.g. 1 ms when t_acquire is 10 ms and t_period is 100 ms). The end_acquire candidate predicts D1=0 ms; the other two predict +/-10 ms or +90 ms. 1 ms is closest to 0 so the verdict should still be 'end_acquire'. .. py:function:: test_hdf_timestamp_diagnostic_returns_expected_dict_keys() Lock the public return-dict contract. .. py:function:: test_hdf_timestamp_diagnostic_flags_noisy_data() When per-event D1 jitter is larger than t_acquire, the verdict is flagged unreliable via the noisy_data guard. Real symptom observed on the gp:m1 + adsimdet IOC at 14:22:52 on 2026-06-10 (UID 9fac2530): D1 stddev ~ 37 ms with t_acquire = 10 ms. The mean alone would falsely suggest a confident verdict; the stddev reveals the per-event jitter swamps the inter-candidate spacing (which is t_acquire wide). .. py:function:: test_hdf_timestamp_diagnostic_flags_sparse_data() When the cam/HDF monitor streams arrive at much less than the expected period (CA monitor coalescing), the verdict is flagged unreliable via the sparse_data guard. Real symptom observed on the gp:m1 + adsimdet IOC at 14:22:52 on 2026-06-10: D2 mean ~ 343 ms with t_period = 100 ms (about 3x sparser than the cam was actually producing frames). .. py:function:: test_hdf_timestamp_diagnostic_raises_when_no_in_scan_frames() If no frames fall in [p_start, p_end] (e.g. the scan window is offset from where the motor was), the diagnostic raises rather than silently producing meaningless statistics.