Resolve test helper merge conflicts

Author: BitForge95

experanto/experiment.py

@@ -118,11 +118,33 @@ def _load_devices(self) -> None:
                     **{str(k): v for k, v in dict(interp_conf).items()},
                 )
 
-            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())

tests/create_experiment.py

@@ -2,6 +2,9 @@
 import shutil
 from contextlib import contextmanager
 
+import numpy as np
+import yaml
+
 from .create_sequence_data import _generate_sequence_data
 
 DEFAULT_CONFIG = {
@@ -45,11 +48,73 @@ def create_experiment(
 
         for device_id, device_kwargs in enumerate(devices_kwargs):
             device_path = tmp_path / f"device_{device_id}"
-            _generate_sequence_data(
-                str(device_path), **device_kwargs
-            )  # pyright: ignore
+            _generate_sequence_data(str(device_path), **device_kwargs)  # pyright: ignore
 
         yield tmp_path
     finally:
         if tmp_path.exists():
             shutil.rmtree(tmp_path)
+
+
+@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)
+    }

tests/test_experiment.py

@@ -1,3 +1,5 @@
+import logging
+from contextlib import ExitStack
 from unittest.mock import MagicMock
 
 import numpy as np
@@ -6,7 +8,12 @@
 from experanto.experiment import Experiment
 from experanto.interpolators import Interpolator
 
-from .create_experiment import create_experiment, get_default_config
+from .create_experiment import (
+    create_experiment,
+    get_default_config,
+    make_modality_config,
+    make_sequence_device,
+)
 
 
 class DummyInterpolator(Interpolator):
@@ -180,3 +187,201 @@ def test_experiment_multi_device_interpolation(tmp_path, return_valid, device):
             else:
                 assert isinstance(data, np.ndarray)
                 assert data.shape == (2, 10)
+
+
+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"
+    )