refactor of group 8 (tests and examples) completed: (f2843aff) · Commits · Gaetano Scandariato / PyAETNA

Changes for tests/test_gap_detection.py: 383 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		test_gap_detection.py — Unit tests for time-grid gap detection.

		Covers:
		- _build_lc_train_grid (grid_builder.py)
		- _build_visit_time_grid (grid_builder.py)
		- _build_augmented_x (gp_plotting.py)
		- GridBuilder class (grid_builder.py)
		- build_time_grid (grid_builder.py)

		Run with:
		/dati/pyenvs/pyaetnaEnv/bin/python -m pytest tests/test_gap_detection.py -v
		"""

		import os
		import sys

		import numpy as np
		import pytest

		sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

		from PyAETNA.reporting.grid_builder import (
		_build_lc_train_grid,
		_build_visit_time_grid,
		GridBuilder,
		build_time_grid,
		)
		from PyAETNA.reporting.gp_plotting import _build_augmented_x


		# ─────────────────────────────────────────────────────────────────────────────
		# Helpers
		# ─────────────────────────────────────────────────────────────────────────────

		def _two_visit_data(n_per_visit=20, gap_days=10.0):
		"""Return (t_sorted, visit_sorted) for two tight visits separated by a large gap."""
		t1 = np.linspace(0.0, 0.5, n_per_visit)
		t2 = np.linspace(gap_days, gap_days + 0.5, n_per_visit)
		t = np.concatenate([t1, t2])
		v = np.concatenate([np.ones(n_per_visit, dtype=int),
		np.full(n_per_visit, 2, dtype=int)])
		return t, v


		def _three_visit_data(n=10, gap=5.0):
		"""Return (t, visit) for three visits."""
		t1 = np.linspace(0.0, 0.2, n)
		t2 = np.linspace(gap, gap + 0.2, n)
		t3 = np.linspace(2 * gap, 2 * gap + 0.2, n)
		t = np.concatenate([t1, t2, t3])
		v = np.concatenate([np.ones(n, dtype=int),
		np.full(n, 2, dtype=int),
		np.full(n, 3, dtype=int)])
		return t, v


		# ─────────────────────────────────────────────────────────────────────────────
		# _build_lc_train_grid
		# ─────────────────────────────────────────────────────────────────────────────

		class TestBuildLcTrainGrid:
		"""Tests for _build_lc_train_grid."""

		def test_output_shapes_consistent(self):
		"""t_grid and visit_grid must have the same length."""
		t, v = _two_visit_data()
		t_grid, visit_grid, gap_edges = _build_lc_train_grid(t, v)
		assert len(t_grid) == len(visit_grid)

		def test_output_sorted_in_time(self):
		"""Output grid must be monotonically non-decreasing in time."""
		t, v = _two_visit_data()
		t_grid, _, _ = _build_lc_train_grid(t, v)
		assert np.all(np.diff(t_grid) >= 0), "Grid is not time-sorted"

		def test_gap_edges_for_two_visits(self):
		"""With two visits there should be exactly one inter-visit gap edge."""
		t, v = _two_visit_data(gap_days=10.0)
		_, _, gap_edges = _build_lc_train_grid(t, v)
		assert len(gap_edges) == 1
		t_end, t_start = gap_edges[0]
		assert t_end < t_start

		def test_gap_edges_for_three_visits(self):
		"""With three visits there should be exactly two inter-visit gap edges."""
		t, v = _three_visit_data()
		_, _, gap_edges = _build_lc_train_grid(t, v)
		assert len(gap_edges) == 2

		def test_gap_fills_inserted_between_visits(self):
		"""Grid must contain points between the two visits (in the inter-visit gap)."""
		t, v = _two_visit_data(n_per_visit=20, gap_days=10.0)
		t_grid, _, _ = _build_lc_train_grid(t, v)
		# At least one grid point should fall in the gap [0.5, 10.0]
		in_gap = (t_grid > 0.5) & (t_grid < 10.0)
		assert np.any(in_gap), "No gap-fill points inserted between visits"

		def test_training_points_present_in_output(self):
		"""All original training times must appear in (or very close to) the grid."""
		t, v = _two_visit_data(n_per_visit=10, gap_days=5.0)
		t_grid, _, _ = _build_lc_train_grid(t, v)
		for tp in t:
		dists = np.abs(t_grid - tp)
		assert np.min(dists) < 1e-10, f"Training point {tp} not found in grid"

		def test_single_visit_no_gap_edges(self):
		"""A single-visit dataset should have zero gap edges."""
		t = np.linspace(0.0, 0.5, 30)
		v = np.ones(30, dtype=int)
		_, _, gap_edges = _build_lc_train_grid(t, v)
		assert gap_edges == []

		def test_pad_time_extends_grid(self):
		"""With pad_time > 0 the grid should extend beyond the data range."""
		t, v = _two_visit_data(n_per_visit=20, gap_days=10.0)
		t_grid_no_pad, _, _ = _build_lc_train_grid(t, v, pad_time=0.0)
		t_grid_padded, _, _ = _build_lc_train_grid(t, v, pad_time=1.0)
		assert t_grid_padded.min() < t_grid_no_pad.min() - 0.5, (
		"Left padding did not extend the grid")
		assert t_grid_padded.max() > t_grid_no_pad.max() + 0.5, (
		"Right padding did not extend the grid")

		def test_gap_threshold_controls_in_visit_gap_detection(self):
		"""A very large gap_threshold should suppress in-visit gap detection."""
		# Create data with a clear in-visit gap
		t = np.concatenate([np.linspace(0.0, 0.1, 10),
		np.linspace(5.0, 5.1, 10)]) # 5-day in-visit gap
		v = np.ones(20, dtype=int)

		t_fine, _, _ = _build_lc_train_grid(t, v, gap_threshold=2.0)
		t_coarse, _, _ = _build_lc_train_grid(t, v, gap_threshold=1000.0)
		# With coarse threshold the gap is not filled densely
		in_gap = (t_coarse > 0.1) & (t_coarse < 5.0)
		# With fine threshold the in-visit gap IS filled
		in_gap_fine = (t_fine > 0.1) & (t_fine < 5.0)
		assert np.sum(in_gap_fine) >= np.sum(in_gap), (
		"Fine threshold should produce more gap-fill points")

		def test_empty_input_returns_copy(self):
		"""Empty input should return copies of the empty arrays without crashing."""
		t = np.array([], dtype=float)
		v = np.array([], dtype=int)
		t_g, v_g, ge = _build_lc_train_grid(t, v)
		assert len(t_g) == 0
		assert len(v_g) == 0
		assert ge == []

		def test_visit_labels_valid_in_output(self):
		"""Visit labels in grid must all come from the original visit set."""
		t, v = _three_visit_data()
		_, visit_grid, _ = _build_lc_train_grid(t, v)
		allowed = set(np.unique(v).tolist())
		assert set(np.unique(visit_grid).tolist()).issubset(allowed)


		# ─────────────────────────────────────────────────────────────────────────────
		# _build_visit_time_grid
		# ─────────────────────────────────────────────────────────────────────────────

		class TestBuildVisitTimeGrid:
		"""Tests for _build_visit_time_grid (used for RV data)."""

		def test_output_shapes_consistent(self):
		t, v = _two_visit_data()
		t_grid, visit_grid, _ = _build_visit_time_grid(t, v)
		assert len(t_grid) == len(visit_grid)

		def test_output_is_not_empty(self):
		t, v = _two_visit_data()
		t_grid, _, _ = _build_visit_time_grid(t, v)
		assert len(t_grid) > 0

		def test_output_sorted_in_time(self):
		t, v = _two_visit_data()
		t_grid, _, _ = _build_visit_time_grid(t, v)
		assert np.all(np.diff(t_grid) >= -1e-12), "Grid not time-sorted"

		def test_gap_edges_two_visits(self):
		t, v = _two_visit_data(gap_days=10.0)
		_, _, gap_edges = _build_visit_time_grid(t, v)
		assert len(gap_edges) == 1

		def test_gap_fills_in_inter_visit_region(self):
		"""Grid must have points inside the inter-visit gap."""
		t, v = _two_visit_data(n_per_visit=20, gap_days=10.0)
		t_grid, _, _ = _build_visit_time_grid(t, v)
		in_gap = (t_grid > 0.6) & (t_grid < 9.9)
		assert np.any(in_gap), "No gap-fill points inserted between visits"

		def test_n_grid_total_respected_approximately(self):
		"""Total grid size should be at least n_grid_total (may be slightly larger)."""
		t, v = _two_visit_data(n_per_visit=20, gap_days=8.0)
		t_grid, _, _ = _build_visit_time_grid(t, v, n_grid_total=100)
		# Grid may have more points than n_grid_total due to gap fills
		assert len(t_grid) >= 100, f"Grid size {len(t_grid)} < target 100"

		def test_single_visit(self):
		"""Single-visit datasets should produce a non-empty grid and no gap edges."""
		t = np.linspace(0.0, 1.0, 20)
		v = np.ones(20, dtype=int)
		t_grid, _, gap_edges = _build_visit_time_grid(t, v)
		assert len(t_grid) > 0
		assert gap_edges == []

		def test_pad_time_extends_grid_beyond_data(self):
		t, v = _two_visit_data(n_per_visit=10, gap_days=5.0)
		t_grid_nopad, _, _ = _build_visit_time_grid(t, v, pad_time=0.0)
		t_grid_padded, _, _ = _build_visit_time_grid(t, v, pad_time=0.5)
		assert t_grid_padded.min() < t_grid_nopad.min()
		assert t_grid_padded.max() > t_grid_nopad.max()

		def test_min_pts_per_visit_enforced(self):
		"""Each visit must have at least min_pts_per_visit grid points."""
		t, v = _two_visit_data(n_per_visit=5, gap_days=8.0)
		t_grid, visit_grid, _ = _build_visit_time_grid(
		t, v, n_grid_total=60, min_pts_per_visit=15)
		for uv in np.unique(v):
		pts = np.sum(visit_grid == uv)
		assert pts >= 15, f"Visit {uv} has only {pts} grid pts (< 15)"

		def test_visit_labels_valid(self):
		t, v = _three_visit_data()
		_, visit_grid, _ = _build_visit_time_grid(t, v)
		allowed = set(np.unique(v).tolist())
		actual = set(np.unique(visit_grid).tolist())
		assert actual.issubset(allowed)


		# ─────────────────────────────────────────────────────────────────────────────
		# _build_augmented_x
		# ─────────────────────────────────────────────────────────────────────────────

		class TestBuildAugmentedX:
		"""Tests for _build_augmented_x (gap-fill pseudo-observation builder)."""

		def test_output_length_is_train_plus_pseudo(self):
		"""x_aug length == N_train + N_pseudo where N_pseudo <= len(x_grid)."""
		x_train = np.linspace(0.0, 1.0, 10)
		x_grid = np.linspace(0.5, 2.0, 20) # some overlap, some new
		x_aug, is_real, real_order, aug_to_grid = _build_augmented_x(x_train, x_grid)
		n_pseudo = int(np.sum(~is_real))
		assert len(x_aug) == len(x_train) + n_pseudo

		def test_x_aug_is_sorted(self):
		x_train = np.linspace(0.0, 1.0, 15)
		x_grid = np.linspace(1.5, 3.0, 20)
		x_aug, _, _, _ = _build_augmented_x(x_train, x_grid)
		assert np.all(np.diff(x_aug) >= 0), "x_aug is not sorted"

		def test_is_real_marks_training_points(self):
		"""Positions matching training points must be marked as real."""
		x_train = np.array([0.0, 1.0, 2.0, 3.0])
		x_grid = np.array([0.0, 1.0, 2.0, 3.0, 4.0]) # 4.0 is new
		x_aug, is_real, _, _ = _build_augmented_x(x_train, x_grid)
		# All training points must be in x_aug as real
		for xt in x_train:
		idx = np.argmin(np.abs(x_aug - xt))
		assert is_real[idx], f"Training point {xt} not marked as real"

		def test_real_order_maps_training_to_aug_positions(self):
		"""real_order[i] must be the index of x_train[i] in x_aug."""
		x_train = np.array([0.0, 2.0, 4.0])
		x_grid = np.array([1.0, 3.0, 5.0])
		x_aug, is_real, real_order, _ = _build_augmented_x(x_train, x_grid)
		for i, xt in enumerate(x_train):
		assert abs(x_aug[real_order[i]] - xt) < 1e-9, (
		f"real_order[{i}]={real_order[i]} does not map to x_train[{i}]={xt}")

		def test_aug_to_grid_maps_pseudo_to_grid_indices(self):
		"""aug_to_grid values must be valid indices into x_grid."""
		x_train = np.linspace(0.0, 0.5, 10)
		x_grid = np.linspace(1.0, 2.0, 8) # all pseudo (no overlap)
		x_aug, is_real, _, aug_to_grid = _build_augmented_x(x_train, x_grid)
		n_pseudo = int(np.sum(~is_real))
		assert len(aug_to_grid) == n_pseudo
		assert np.all(aug_to_grid >= 0)
		assert np.all(aug_to_grid < len(x_grid))

		def test_overlapping_grid_points_not_duplicated(self):
		"""Grid points coinciding with training points should not add pseudo-obs."""
		x_train = np.linspace(0.0, 1.0, 11) # 0.0, 0.1, ..., 1.0
		x_grid = np.array([0.0, 0.1, 0.2]) # all coincide with training
		x_aug, is_real, _, aug_to_grid = _build_augmented_x(x_train, x_grid)
		# No pseudo-observations should be added
		assert len(aug_to_grid) == 0, (
		f"Expected 0 pseudo-obs for fully overlapping grid, got {len(aug_to_grid)}")

		def test_empty_grid_returns_only_training(self):
		x_train = np.linspace(0.0, 1.0, 5)
		x_grid = np.array([])
		x_aug, is_real, real_order, aug_to_grid = _build_augmented_x(x_train, x_grid)
		assert len(x_aug) == len(x_train)
		assert np.all(is_real)
		assert len(aug_to_grid) == 0

		def test_empty_training_returns_all_pseudo(self):
		x_train = np.array([])
		x_grid = np.linspace(0.0, 1.0, 5)
		x_aug, is_real, real_order, aug_to_grid = _build_augmented_x(x_train, x_grid)
		assert len(x_aug) == len(x_grid)
		assert np.all(~is_real)
		assert len(aug_to_grid) == len(x_grid)

		def test_aug_to_grid_covers_all_pseudo_unique(self):
		"""Each pseudo-observation should map to a distinct x_grid index."""
		x_train = np.linspace(0.0, 0.5, 5)
		x_grid = np.linspace(2.0, 3.0, 8) # all pseudo
		_, _, _, aug_to_grid = _build_augmented_x(x_train, x_grid)
		# All grid indices should be unique (one pseudo-obs per gap-fill point)
		assert len(aug_to_grid) == len(np.unique(aug_to_grid))


		# ─────────────────────────────────────────────────────────────────────────────
		# GridBuilder class
		# ─────────────────────────────────────────────────────────────────────────────

		class TestGridBuilderClass:
		"""Tests for the GridBuilder class in grid_builder.py."""

		def test_lc_strategy_returns_valid_grid(self):
		t, v = _two_visit_data()
		gb = GridBuilder(t, v)
		t_grid, visit_grid, gap_edges = gb.build_time_grid(strategy='lc')
		assert len(t_grid) > 0
		assert len(t_grid) == len(visit_grid)

		def test_visit_strategy_returns_valid_grid(self):
		t, v = _two_visit_data()
		gb = GridBuilder(t, v)
		t_grid, visit_grid, gap_edges = gb.build_time_grid(strategy='visit')
		assert len(t_grid) > 0
		assert len(t_grid) == len(visit_grid)

		def test_invalid_strategy_raises(self):
		t, v = _two_visit_data()
		gb = GridBuilder(t, v)
		with pytest.raises(ValueError, match="Unknown grid strategy"):
		gb.build_time_grid(strategy='invalid')

		def test_unique_visits_stored(self):
		t, v = _three_visit_data()
		gb = GridBuilder(t, v)
		assert len(gb.unique_visits) == 3

		def test_lc_strategy_has_gap_fills(self):
		t, v = _two_visit_data(gap_days=8.0)
		gb = GridBuilder(t, v)
		t_grid, _, _ = gb.build_time_grid(strategy='lc')
		in_gap = (t_grid > 0.6) & (t_grid < 7.9)
		assert np.any(in_gap)


		# ─────────────────────────────────────────────────────────────────────────────
		# build_time_grid functional API
		# ─────────────────────────────────────────────────────────────────────────────

		class TestBuildTimeGridFunctional:
		"""Tests for the build_time_grid() module-level function."""

		def test_lc_strategy(self):
		t, v = _two_visit_data()
		t_grid, visit_grid, _ = build_time_grid(t, v, strategy='lc')
		assert len(t_grid) > 0
		assert len(t_grid) == len(visit_grid)

		def test_visit_strategy(self):
		t, v = _two_visit_data()
		t_grid, visit_grid, _ = build_time_grid(t, v, strategy='visit')
		assert len(t_grid) > 0

		def test_kwargs_forwarded(self):
		t, v = _two_visit_data()
		t_small, _, _ = build_time_grid(t, v, strategy='visit',
		n_grid_total=50, min_pts_per_visit=10)
		t_large, _, _ = build_time_grid(t, v, strategy='visit',
		n_grid_total=500, min_pts_per_visit=10)
		assert len(t_large) > len(t_small)


		if __name__ == '__main__':
		import pytest as _pytest
		_pytest.main([__file__, '-v'])