id3c.tests.test_flyscan_3idc_analysis#

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#

`test_interpolate_positions_simple_linear`()	At constant velocity, each frame should land at the
`test_interpolate_positions_trims_outside_scan_range`()	Frames whose interpolated positions fall outside
`test_interpolate_positions_handles_out_of_order_motor_stream`()	Record-order-shuffled motor stream still produces correct
`test_interpolate_positions_handles_duplicate_timestamps`()	Motor samples with repeated timestamps are deduped (keep
`test_interpolate_positions_drops_extrapolation`(caplog)	HDF frames with timestamps outside the motor stream's time
`test_interpolate_positions_empty_hdf_returns_empty_frame`()	No HDF frames -> empty result, no crash.
`test_interpolate_positions_requires_two_motor_samples`()	Single-sample motor stream cannot be interpolated.
`test_interpolate_positions_shape_mismatch_raises`()	Shape mismatches between paired arrays raise ValueError.
`test_interpolate_positions_three_phase_arithmetic_at_constant_velocity`()	At constant motor velocity, the three per-phase positions
`test_interpolate_positions_in_scan_filter_uses_start_acquire`()	The in-scan filter uses position_start_acquire (not the
`test_interpolate_positions_widening_admits_leading_edge_frame`()	Widening admits a leading-edge frame the old rule rejected.
`test_interpolate_positions_widening_admits_trailing_edge_frame`()	Trailing-edge analogue of the leading-edge test.
`test_interpolate_positions_widening_rejects_all_when_motor_never_in_range`()	No frames are admitted if the motor never enters the range.
`test_interpolate_positions_dedups_duplicate_image_numbers`(caplog)	Duplicate image_number values within the in-scan window are
`test_pair_frames_to_positions_end_to_end`()	Build a fake BlueskyRun-shaped object and pair frames.
`test_pair_frames_to_positions_trims_taxi_frames`()	Frames captured during taxi-in (motor below p_start) are
`test_pair_frames_to_positions_index_is_timestamp`()	The returned DataFrame is indexed by timestamp.
`test_pair_frames_to_positions_missing_metadata_raises`()	Missing start metadata -> KeyError with helpful message.
`test_pair_frames_to_positions_missing_motor_stream_raises`()	Missing motor stream -> KeyError.
`test_pair_frames_to_positions_against_observed_run_shape`()	Build the same data we observed in the 17:21 run (scan_id=6,
`test_array_from_ds_handles_xarray_like`()	Pulls a 1-D array from an xarray-like Dataset.
`test_array_from_ds_handles_dataframe_like`()	Pulls a 1-D array from a pandas DataFrame column.
`test_array_from_ds_missing_key_raises`()	A missing column raises KeyError naming the stream.
`test_hdf_timestamp_diagnostic_picks_end_acquire`()	When hdf_t == cam_t (the end_acquire semantic), the diagnostic
`test_hdf_timestamp_diagnostic_picks_start_acquire`()	When hdf_t == cam_t - t_acquire (the start_acquire semantic),
`test_hdf_timestamp_diagnostic_picks_end_period`()	When hdf_t == cam_t + (t_period - t_acquire) (the end_period
`test_hdf_timestamp_diagnostic_picks_closest_with_noise`()	Realistic case: HDF lags cam by a small plugin-pipeline
`test_hdf_timestamp_diagnostic_returns_expected_dict_keys`()	Lock the public return-dict contract.
`test_hdf_timestamp_diagnostic_flags_noisy_data`()	When per-event D1 jitter is larger than t_acquire, the verdict
`test_hdf_timestamp_diagnostic_flags_sparse_data`()	When the cam/HDF monitor streams arrive at much less than the
`test_hdf_timestamp_diagnostic_raises_when_no_in_scan_frames`()	If no frames fall in [p_start, p_end] (e.g. the scan window

Module Contents#

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_simple_linear()[source]#

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.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_trims_outside_scan_range()[source]#: Frames whose interpolated positions fall outside [p_start, p_end] are dropped.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_handles_out_of_order_motor_stream()[source]#: 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).

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_handles_duplicate_timestamps()[source]#: 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.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_drops_extrapolation(caplog)[source]#: HDF frames with timestamps outside the motor stream’s time range are dropped (extrapolation rejected, logged at WARNING).

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_empty_hdf_returns_empty_frame()[source]#: No HDF frames -> empty result, no crash.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_requires_two_motor_samples()[source]#: Single-sample motor stream cannot be interpolated.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_shape_mismatch_raises()[source]#: Shape mismatches between paired arrays raise ValueError.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_three_phase_arithmetic_at_constant_velocity()[source]#

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.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_in_scan_filter_uses_start_acquire()[source]#: 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).

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_widening_admits_leading_edge_frame()[source]#

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.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_widening_admits_trailing_edge_frame()[source]#

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.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_widening_rejects_all_when_motor_never_in_range()[source]#: No frames are admitted if the motor never enters the range.

id3c.tests.test_flyscan_3idc_analysis.test_interpolate_positions_dedups_duplicate_image_numbers(caplog)[source]#: 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.

id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_end_to_end()[source]#: Build a fake BlueskyRun-shaped object and pair frames.

id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_trims_taxi_frames()[source]#: 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.

id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_index_is_timestamp()[source]#: The returned DataFrame is indexed by timestamp.

id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_missing_metadata_raises()[source]#: Missing start metadata -> KeyError with helpful message.

id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_missing_motor_stream_raises()[source]#: Missing motor stream -> KeyError.

id3c.tests.test_flyscan_3idc_analysis.test_pair_frames_to_positions_against_observed_run_shape()[source]#

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.

id3c.tests.test_flyscan_3idc_analysis.test_array_from_ds_handles_xarray_like()[source]#: Pulls a 1-D array from an xarray-like Dataset.

id3c.tests.test_flyscan_3idc_analysis.test_array_from_ds_handles_dataframe_like()[source]#: Pulls a 1-D array from a pandas DataFrame column.

id3c.tests.test_flyscan_3idc_analysis.test_array_from_ds_missing_key_raises()[source]#: A missing column raises KeyError naming the stream.

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_end_acquire()[source]#: When hdf_t == cam_t (the end_acquire semantic), the diagnostic picks ‘end_acquire’ and recommends offset = -t_acquire.

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_start_acquire()[source]#: When hdf_t == cam_t - t_acquire (the start_acquire semantic), the diagnostic picks ‘start_acquire’ and recommends offset = 0.

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_end_period()[source]#: When hdf_t == cam_t + (t_period - t_acquire) (the end_period semantic), the diagnostic picks ‘end_period’ and recommends offset = -t_period.

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_picks_closest_with_noise()[source]#: 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’.

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_returns_expected_dict_keys()[source]#: Lock the public return-dict contract.

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_flags_noisy_data()[source]#

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).

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_flags_sparse_data()[source]#

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).

id3c.tests.test_flyscan_3idc_analysis.test_hdf_timestamp_diagnostic_raises_when_no_in_scan_frames()[source]#: 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.