|
| 1 | +"""A fused depth carries the side view's own status: which values were reconstructed.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import numpy as np |
| 6 | + |
| 7 | +from track2data.exporters.schema import depth_columns |
| 8 | +from track2data.fusion.fuse import fuse |
| 9 | +from track2data.metrics.binning import slice_psess |
| 10 | + |
| 11 | +from .builders import make_pair, settings |
| 12 | + |
| 13 | + |
| 14 | +def _fused(**side_updates): |
| 15 | + top, side, pair = make_pair(fusion=side_updates.pop("fusion", None)) |
| 16 | + n = side.xy.shape[0] |
| 17 | + # the tracker's own side positions; NaN where the preprocessing later filled a gap |
| 18 | + side.raw_xy_rows = side.xy.copy() |
| 19 | + for rows, animal in side_updates.pop("interpolated", []): |
| 20 | + side.raw_xy_rows[rows, animal] = np.nan |
| 21 | + jumped = side_updates.pop("jumped", []) |
| 22 | + if jumped: |
| 23 | + side.jump_replaced = np.zeros((n, 3), dtype=bool) |
| 24 | + for rows, animal in jumped: |
| 25 | + side.jump_replaced[rows, animal] = True |
| 26 | + return top, side, pair, fuse(top, side, pair, same_video=False) |
| 27 | + |
| 28 | + |
| 29 | +def test_depth_interpolated_marks_the_sides_filled_frames_for_that_fish(): |
| 30 | + _, _, _, fused = _fused(interpolated=[(slice(10, 20), 0)]) |
| 31 | + flag = fused.psess.depth_interpolated |
| 32 | + assert flag.shape == fused.psess.depth.shape and flag.dtype == bool |
| 33 | + expected = np.zeros_like(flag) |
| 34 | + expected[10:20, 0] = True |
| 35 | + np.testing.assert_array_equal(flag, expected) |
| 36 | + assert np.isfinite(fused.psess.depth[10:20, 0]).all() # still a number, now marked |
| 37 | + |
| 38 | + |
| 39 | +def test_depth_jump_replaced_marks_the_sides_replaced_frames(): |
| 40 | + _, _, _, fused = _fused(jumped=[(slice(30, 35), 1)]) |
| 41 | + flag = fused.psess.depth_jump_replaced |
| 42 | + expected = np.zeros_like(flag) |
| 43 | + expected[30:35, 1] = True |
| 44 | + np.testing.assert_array_equal(flag, expected) |
| 45 | + |
| 46 | + |
| 47 | +def test_a_side_without_a_jump_mask_gives_an_all_false_flag_not_none(): |
| 48 | + _, _, _, fused = _fused() |
| 49 | + assert fused.psess.depth_jump_replaced is not None |
| 50 | + assert not fused.psess.depth_jump_replaced.any() |
| 51 | + assert not fused.psess.depth_interpolated.any() |
| 52 | + |
| 53 | + |
| 54 | +def test_the_flags_follow_the_frame_offset(): |
| 55 | + top, side, pair = make_pair(fusion=settings(frame_offset=5)) |
| 56 | + side.raw_xy_rows = side.xy.copy() |
| 57 | + side.raw_xy_rows[15, 0] = np.nan # side row 15 is top frame 10 |
| 58 | + fused = fuse(top, side, pair, same_video=False) |
| 59 | + assert fused.psess.depth_interpolated[:, 0].nonzero()[0].tolist() == [10] |
| 60 | + |
| 61 | + |
| 62 | +def test_a_flag_is_false_where_the_depth_is_missing(): |
| 63 | + top, side, pair = make_pair() |
| 64 | + side.raw_xy_rows = side.xy.copy() |
| 65 | + side.raw_xy_rows[3, 0] = np.nan |
| 66 | + side.xy[3, 0, 1] = 50.0 # filled, but outside the water column, so no depth |
| 67 | + fused = fuse(top, side, pair, same_video=False) |
| 68 | + assert np.isnan(fused.psess.depth[3, 0]) |
| 69 | + assert not fused.psess.depth_interpolated[3, 0] |
| 70 | + |
| 71 | + |
| 72 | +def test_separator_rows_are_false(): |
| 73 | + top, side, pair = make_pair() |
| 74 | + tracked = np.ones(100, dtype=bool) |
| 75 | + tracked[30:70] = False |
| 76 | + side.tracked_mask = tracked |
| 77 | + side.raw_xy_rows = side.xy.copy() |
| 78 | + side.raw_xy_rows[:, :] = np.nan # every tracked side frame counts as filled |
| 79 | + fused = fuse(top, side, pair, same_video=False) |
| 80 | + sep = fused.psess.separator_mask |
| 81 | + assert sep is not None and sep.any() |
| 82 | + assert not fused.psess.depth_interpolated[sep].any() |
| 83 | + assert fused.psess.depth_interpolated[~sep].all() |
| 84 | + |
| 85 | + |
| 86 | +def test_slice_psess_slices_the_flags(): |
| 87 | + _, _, _, fused = _fused(interpolated=[(slice(10, 20), 0)], jumped=[(slice(30, 35), 1)]) |
| 88 | + window = slice_psess(fused.psess, 8, 22) |
| 89 | + assert window.depth_interpolated.shape == (14, 3) |
| 90 | + assert window.depth_interpolated[2:12, 0].all() and not window.depth_interpolated[:2].any() |
| 91 | + assert window.depth_jump_replaced.shape == (14, 3) and not window.depth_jump_replaced.any() |
| 92 | + |
| 93 | + |
| 94 | +def test_the_export_columns_carry_the_flags_beside_depth(): |
| 95 | + _, _, _, fused = _fused(interpolated=[(slice(10, 20), 0)], jumped=[(slice(30, 35), 1)]) |
| 96 | + cols = depth_columns(fused.psess) |
| 97 | + assert list(cols) == [ |
| 98 | + "depth_fraction", "depth_cm", "depth_interpolated", "depth_jump_replaced" |
| 99 | + ] |
| 100 | + assert cols["depth_interpolated"].dtype == bool |
| 101 | + flat = fused.psess.depth_interpolated.reshape(-1) |
| 102 | + np.testing.assert_array_equal(cols["depth_interpolated"], flat) |
| 103 | + assert cols["depth_jump_replaced"].sum() == 5 |
| 104 | + |
| 105 | + |
| 106 | +def test_a_session_without_a_depth_exports_no_flag_columns(): |
| 107 | + top, _, _ = make_pair() |
| 108 | + assert depth_columns(top) == {} |
0 commit comments