diff --git a/experanto/experiment.py b/experanto/experiment.py index abfda7a..fa739c9 100644 --- a/experanto/experiment.py +++ b/experanto/experiment.py @@ -118,11 +118,33 @@ def _load_devices(self) -> None: **interp_conf, # type: ignore[arg-type] ) - self.devices[d.name] = dev - self.start_time = dev.start_time - self.end_time = dev.end_time + if ( + dev.start_time is None + or dev.end_time is None + or not np.isfinite(dev.start_time) + or not np.isfinite(dev.end_time) + ): + logger.warning( + "Device %s has undefined start_time or end_time and will be " + "excluded from the experiment-wide time range.", + d.name, + ) + else: + self.start_time = min(self.start_time, dev.start_time) + self.end_time = max(self.end_time, dev.end_time) + self.devices[d.name] = dev logger.info("Parsing finished") + if not self.devices: + raise ValueError( + "Experiment time range could not be determined: no devices with valid start_time and end_time were found." + ) + elif self.start_time > self.end_time: + raise ValueError( + "Experiment time range could not be determined: at least one device " + "must define finite start_time and end_time." + ) + @property def device_names(self): return tuple(self.devices.keys()) diff --git a/tests/create_experiment.py b/tests/create_experiment.py new file mode 100644 index 0000000..2a97c07 --- /dev/null +++ b/tests/create_experiment.py @@ -0,0 +1,69 @@ +import shutil +from contextlib import contextmanager + +import numpy as np +import yaml + + +@contextmanager +def make_sequence_device( + root, name, start, end, sampling_rate=10.0, n_signals=5, override_meta=None +): + """Create a single sequence device folder under root.""" + device_root = root / name + try: + (device_root / "meta").mkdir(parents=True, exist_ok=True) + + n_samples = ( + int((end - start) * sampling_rate) + 1 + ) # +1 to include both start and end as sample points + timestamps = np.linspace(start, end, n_samples) + data = np.random.rand(n_samples, n_signals) + + np.save(device_root / "timestamps.npy", timestamps) + np.save(device_root / "data.npy", data) + + meta = { + "start_time": start, + "end_time": end, + "modality": "sequence", + "sampling_rate": sampling_rate, + "phase_shift_per_signal": False, + "is_mem_mapped": False, + "n_signals": n_signals, + "n_timestamps": n_samples, + "dtype": "float64", + } + if override_meta: + meta.update(override_meta) + with open(device_root / "meta.yml", "w") as f: + yaml.safe_dump(meta, f) + + yield device_root + + finally: + shutil.rmtree(device_root) + + +def make_modality_config(*device_names, sampling_rates=None, offsets=None): + if sampling_rates is None: + sampling_rates = [10.0] * len(device_names) + elif isinstance(sampling_rates, (int, float)): + sampling_rates = [sampling_rates] * len(device_names) + + if offsets is None: + offsets = [0.0] * len(device_names) + elif isinstance(offsets, (int, float)): + offsets = [offsets] * len(device_names) + + assert len(device_names) == len( + sampling_rates + ), f"sampling_rates length {len(sampling_rates)} does not match device_names length {len(device_names)}" + assert len(device_names) == len( + offsets + ), f"offsets length {len(offsets)} does not match device_names length {len(device_names)}" + + return { + name: {"interpolation": {"sampling_rate": sr, "offset": off}} + for name, sr, off in zip(device_names, sampling_rates, offsets, strict=True) + } diff --git a/tests/test_experiment.py b/tests/test_experiment.py new file mode 100644 index 0000000..72d5f2a --- /dev/null +++ b/tests/test_experiment.py @@ -0,0 +1,206 @@ +import logging +from contextlib import ExitStack + +import numpy as np +import pytest + +from experanto.experiment import Experiment + +from .create_experiment import make_modality_config, make_sequence_device + +DEVICE_TIME_RANGE_CASES = [ + # Single device: start and end should match that device's range + ([(2.0, 9.0)], 2.0, 9.0), + # Two devices with different ranges: start should be min, end should be max + ([(1.0, 8.0), (0.0, 10.0)], 0.0, 10.0), + # Three devices with different ranges: start should be min, end should be max + ([(0.0, 10.0), (1.0, 8.0), (2.0, 9.0)], 0.0, 10.0), + # Devices with non-overlapping ranges: start should be min, end should be max + ([(0.0, 3.0), (7.0, 8.0)], 0.0, 8.0), + # Devices with identical ranges: start and end should match that range + ([(1.0, 5.0), (1.0, 5.0)], 1.0, 5.0), + # Large time stamps: start should be min, end should be max + ([(1e9, 1e9 + 100), (1e9 - 50, 1e9 + 50)], 1e9 - 50, 1e9 + 100), +] + +DEVICE_TIME_RANGE_IDS = [ + "single_device", + "two_devices_different_ranges", + "three_devices_different_ranges", + "non_overlapping_ranges", + "identical_ranges", + "large_time_stamps", +] + +# Inverted range is intentionally separate from INVALID_META_CASES — +# None/NaN/inf are caught per-device before being added to self.devices, +# whereas start > end is only caught after all devices are loaded. +INVALID_META_CASES = [ + {"start_time": None, "end_time": None}, # Both missing + {"start_time": None, "end_time": 10.0}, # Missing start_time + {"start_time": 0.0, "end_time": None}, # Missing end_time + {"start_time": float("inf"), "end_time": 10.0}, # Infinite start_time + {"start_time": 0.0, "end_time": float("inf")}, # Infinite end_time + {"start_time": float("-inf"), "end_time": 10.0}, # Negative Infinite start_time + {"start_time": 0.0, "end_time": float("-inf")}, # Negative Infinite end_time + {"start_time": float("nan"), "end_time": 10.0}, # NaN start_time + {"start_time": 0.0, "end_time": float("nan")}, # NaN end_time +] + +INVALID_META_IDS = [ + "both_missing", + "missing_start_time", + "missing_end_time", + "infinite_start_time", + "infinite_end_time", + "negative_infinite_start_time", + "negative_infinite_end_time", + "nan_start_time", + "nan_end_time", +] + + +# Test for union of device time ranges +@pytest.mark.parametrize("n_signals", [5, 20]) +@pytest.mark.parametrize( + "device_ranges, expected_start, expected_end", + DEVICE_TIME_RANGE_CASES, + ids=DEVICE_TIME_RANGE_IDS, +) +def test_experiment_start_end_time_reflects_union( + tmp_path, device_ranges, expected_start, expected_end, n_signals +): + """ + Experiment.start_time and end_time should reflect the union of all + device time ranges — earliest start and latest end across all devices. + """ + device_names = [f"device_{i}" for i in range(len(device_ranges))] + + with ExitStack() as stack: + for name, (start, end) in zip(device_names, device_ranges, strict=True): + stack.enter_context( + make_sequence_device( + tmp_path, + name, + start=start, + end=end, + n_signals=n_signals, + sampling_rate=float(np.random.randint(5, 30)), + ) + ) + + experiment = Experiment( + root_folder=tmp_path, + modality_config=make_modality_config( + *device_names, offsets=[float(np.random.rand()) for _ in device_names] + ), + ) + + assert experiment.start_time == ( + expected_start + ), f"Expected start_time={expected_start}, got {experiment.start_time}" + assert experiment.end_time == ( + expected_end + ), f"Expected end_time={expected_end}, got {experiment.end_time}" + + +# Safety check +@pytest.mark.parametrize("override_meta", INVALID_META_CASES, ids=INVALID_META_IDS) +def test_experiment_invalid_metadata(tmp_path, override_meta): + """ + Experiment should raise an error when initialized with invalid metadata. + Covers cases where start_time or end_time is None, NaN, or infinite. + """ + with make_sequence_device( + tmp_path, + "device_0", + start=0.0, + end=10.0, + override_meta=override_meta, + ): + with pytest.raises( + ValueError, match="Experiment time range could not be determined" + ): + Experiment( + root_folder=tmp_path, + modality_config=make_modality_config("device_0"), + ) + + +def test_experiment_inverted_time_range_raises(tmp_path): + """ + Experiment should raise ValueError when start_time > end_time. + This is a separate guard from invalid metadata (None/NaN/inf) because it + only becomes apparent after all devices are loaded and the overall time range is computed. + """ + with make_sequence_device( + tmp_path, + "device_0", + start=0.0, + end=10.0, + override_meta={"start_time": 5.0, "end_time": 2.0}, + ): + with pytest.raises( + ValueError, match="Experiment time range could not be determined" + ): + Experiment( + root_folder=tmp_path, + modality_config=make_modality_config("device_0"), + ) + + +@pytest.mark.parametrize("override_meta", INVALID_META_CASES, ids=INVALID_META_IDS) +def test_experiment_skips_invalid_devices(tmp_path, override_meta, caplog): + """ + Experiment should skip devices with invalid start_time or end_time and + log a warning, but still initialize successfully if at least one valid + device is present. The experiment time range should reflect only the + valid device. + """ + start_val = np.random.lognormal(mean=0.0, sigma=1.0) # Strictly positive float + duration_val = np.random.lognormal(mean=0.0, sigma=1.0) + end_val = start_val + duration_val + + start_nonval = np.random.lognormal(mean=0.0, sigma=1.0) + duration_nonval = np.random.lognormal(mean=0.0, sigma=1.0) + end_nonval = start_nonval + duration_nonval + + with ExitStack() as stack: + # Valid device with proper metadata + stack.enter_context( + make_sequence_device( + tmp_path, + "valid_device", + start=start_val, + end=end_val, + ) + ) + # Invalid device with missing start_time and end_time + stack.enter_context( + make_sequence_device( + tmp_path, + "invalid_device", + start=start_nonval, + end=end_nonval, + override_meta=override_meta, + ) + ) + + with caplog.at_level(logging.WARNING, logger="experanto.experiment"): + experiment = Experiment( + root_folder=tmp_path, + modality_config=make_modality_config("valid_device", "invalid_device"), + ) + + assert "valid_device" in experiment.devices + assert "invalid_device" not in experiment.devices + + assert experiment.start_time == ( + start_val + ), f"Expected start_time={start_val}, got {experiment.start_time}" + assert experiment.end_time == ( + end_val + ), f"Expected end_time={end_val}, got {experiment.end_time}" + assert any( + "invalid_device" in message for message in caplog.messages + ), "Expected warning about invalid_device was skipped"