Testing Setup (#44)

* Add SingleRunData-level trim/split helpers

* More tests

* Cleanup

* Clean up tests

Author: alex-yang-upenn

.gitignore

@@ -17,3 +17,8 @@ perda/models/stsb-cross-encoder/
 docs/build/
 docs/build_md/
 docs/source/_autosummary/
+
+# Ignore test artifacts
+.coverage
+.pytest_cache/
+htmlcov/

README.md

@@ -13,4 +13,15 @@ Docs are autogenerated using Sphinx. To generate the docs locally, run the follo
 cd docs
 sphinx-build -b html source build  # HTML
 sphinx-build -M markdown source build_md  # Markdown
-```
+```
+
+## Testing
+
+We use `pytest` for testing. In the root directory, run the following 
+
+```bash
+pytest
+
+# Generates coverage report in htmlcov/index.html
+pytest --cov=perda --cov-report=html
+```

perda/analyzer/analyzer.py

@@ -82,7 +82,7 @@ def __init__(
             Maximum number of parsing errors before stopping. Default is 100.
         preprocessing : list[PreprocessingStep] | None, optional
             Ordered list of post-parse preprocessing steps to apply. Each step
-            is a ``SingleRunData → SingleRunData`` callable. Steps are skipped
+            is a ``SingleRunData -> SingleRunData`` callable. Steps are skipped
             with a warning if required variables are absent. Default is None.
 
         Examples
@@ -213,7 +213,7 @@ def plot(
         if self.data.concat_boundaries:
             vlines = [_to_seconds(b, unit) for b in self.data.concat_boundaries]
 
-        # Apply time range filter if specified (convert seconds → raw units for trim)
+        # Apply time range filter if specified (convert seconds -> raw units for trim)
         if ts_start is not None or ts_end is not None:
             start_raw = _from_seconds(ts_start, unit) if ts_start is not None else None
             end_raw = _from_seconds(ts_end, unit) if ts_end is not None else None

perda/core_data_structures/__init__.py

@@ -3,3 +3,4 @@
 from .joins import *
 from .resampling import *
 from .single_run_data import *
+from .split_helpers import *

perda/core_data_structures/split_helpers.py

@@ -0,0 +1,66 @@
+from .single_run_data import DataInstance, SingleRunData
+
+
+def trim_single_run_data(
+    data: SingleRunData,
+    ts_start: float,
+    ts_end: float,
+) -> SingleRunData:
+    """Return a new SingleRunData with every variable trimmed to [ts_start, ts_end].
+
+    Parameters
+    ----------
+    data : SingleRunData
+    ts_start, ts_end : float
+        Timestamps in the same unit as data.timestamp_unit.
+
+    Returns
+    -------
+    SingleRunData
+        Fresh object; the original is not mutated.
+    """
+    trimmed: dict[int, DataInstance] = {
+        var_id: di.trim(ts_start, ts_end) for var_id, di in data.id_to_instance.items()
+    }
+
+    return SingleRunData(
+        id_to_instance=trimmed,
+        cpp_name_to_id=dict(data.cpp_name_to_id),
+        id_to_cpp_name=dict(data.id_to_cpp_name),
+        id_to_descript=dict(data.id_to_descript),
+        total_data_points=sum(len(di.value_np) for di in trimmed.values()),
+        data_start_time=int(ts_start),
+        data_end_time=int(ts_end),
+        timestamp_unit=data.timestamp_unit,
+        concat_boundaries=[],
+    )
+
+
+def split_single_run_data(
+    data: SingleRunData,
+    split_timestamps: list[float],
+) -> list[SingleRunData]:
+    """Split a SingleRunData into segments defined by a list of boundary timestamps.
+
+    Each consecutive pair of timestamps in ``split_timestamps`` defines one
+    segment.  The result is keyed by 1-based segment number.
+
+    Parameters
+    ----------
+    data : SingleRunData
+    split_timestamps : list[float]
+        Ordered boundary timestamps in the same unit as data.timestamp_unit.
+        Must contain at least 2 values.
+
+    Returns
+    -------
+    List[SingleRunData]
+    """
+    if len(split_timestamps) < 2:
+        raise ValueError(f"Need at least 2 boundary timestamps")
+
+    segments = []
+    for t_start, t_end in zip(split_timestamps[:-1], split_timestamps[1:]):
+        segments.append(trim_single_run_data(data, t_start, t_end))
+
+    return segments

perda/utils/preprocessing.py

@@ -27,7 +27,7 @@
 STEERING_RAW = "ludwig.steeringWheel.raw"
 STEERING_ANGLE = "ludwig.steeringWheel.angle"
 
-# Default 3-point voltage→angle calibration for the steering pot.
+# Default 3-point voltage->angle calibration for the steering pot.
 SteeringCalibration = tuple[tuple[float, float], ...]
 DEFAULT_STEERING_CALIBRATION: SteeringCalibration = (
     (1.86, -97.0),  # max left
@@ -148,7 +148,7 @@ def convert_wheelspeeds_to_m_per_s(data: SingleRunData) -> SingleRunData:
         )
 
     print(
-        f"convert_wheelspeeds_to_m_per_s: converted {len(cols)} channels mph → m/s, backups in *_mph"
+        f"convert_wheelspeeds_to_m_per_s: converted {len(cols)} channels mph -> m/s, backups in *_mph"
     )
     return data
 
@@ -308,7 +308,7 @@ def apply_preprocessing(
     data : SingleRunData
         Parsed run data to preprocess.
     steps : list of PreprocessingStep
-        Ordered list of ``SingleRunData → SingleRunData`` callables to apply.
+        Ordered list of ``SingleRunData -> SingleRunData`` callables to apply.
 
     Returns
     -------

pyproject.toml

@@ -52,3 +52,4 @@ build-backend = "setuptools.build_meta"
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+addopts = "--import-mode=importlib"

requirements-dev.txt

@@ -8,3 +8,4 @@ black[jupyter]
 isort
 mypy
 pytest
+pytest-cov

tests/CLAUDE.md

@@ -0,0 +1,7 @@
+- Organize tests by files. Each test_* file should correspond to a single file in the codebase.
+
+- When possible, parameterize tests. Increase the number of test cases without increasing the number of functions.
+
+- Use fixtures to set up common test data or state. Use fixture.py file per test folder to keep files smaller and more maintainable. Only put widely used fixtures in the top-level conftest.py.
+
+- Do not make section header comments in test files