From 289523f6f802b8b7aa4ba22f8a2f36dc07790213 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Feb 2026 20:58:57 +0000 Subject: [PATCH] Add CI/CD pipeline and comprehensive test suite Add 90 unit and integration tests covering: - detect_peaks: peak detection with various parameters (mph, mpd, threshold, valleys) - postprocess: extract_picks, calc_metrics, calc_timestamp - data_reader: normalize, normalize_long, normalize_batch, DataConfig - util: EMA, LMA, metrics, clean_queue - gamma._base: _check_shape, _check_X input validation - gamma.utils: convert_picks_csv, from_seconds, real data integrity checks Add GitHub Actions CI/CD workflow with: - Lint job (Python syntax checking) - Unit tests across Python 3.9/3.10/3.11 - Integration tests (runs after unit tests pass) - Full suite with coverage reporting - Pip caching and JUnit XML test result artifacts Also add: requirements.txt, pytest.ini, .gitignore, test fixtures (conftest.py) https://claude.ai/code/session_01P5P2JWzJQ3LcWG8Zhw44ob --- .github/workflows/ci.yml | 140 +++++++++++++++++++++++ .gitignore | 11 ++ pytest.ini | 9 ++ requirements.txt | 14 +++ tests/__init__.py | 0 tests/conftest.py | 73 ++++++++++++ tests/test_data_reader.py | 178 ++++++++++++++++++++++++++++++ tests/test_detect_peaks.py | 140 +++++++++++++++++++++++ tests/test_gamma_base.py | 75 +++++++++++++ tests/test_gamma_integration.py | 190 ++++++++++++++++++++++++++++++++ tests/test_postprocess.py | 178 ++++++++++++++++++++++++++++++ tests/test_util.py | 151 +++++++++++++++++++++++++ 12 files changed, 1159 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 pytest.ini create mode 100644 requirements.txt create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_data_reader.py create mode 100644 tests/test_detect_peaks.py create mode 100644 tests/test_gamma_base.py create mode 100644 tests/test_gamma_integration.py create mode 100644 tests/test_postprocess.py create mode 100644 tests/test_util.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..363511e --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,140 @@ +name: CI Tests + +on: + push: + branches: [master, main, "claude/**"] + pull_request: + branches: [master, main] + +jobs: + lint: + name: Lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.10" + + - name: Check Python syntax + run: | + python -m py_compile LatestPhasenetLocalTest/phasenet/detect_peaks.py + python -m py_compile LatestPhasenetLocalTest/phasenet/postprocess.py + python -m py_compile LatestPhasenetLocalTest/phasenet/util.py + python -m py_compile GaMMaTest/gamma/_base.py + python -m py_compile GaMMaTest/gamma/utils.py + + unit-tests: + name: Unit Tests + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.9", "3.10", "3.11"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Cache pip packages + uses: actions/cache@v4 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-${{ matrix.python-version }}-${{ hashFiles('requirements.txt') }} + restore-keys: | + ${{ runner.os }}-pip-${{ matrix.python-version }}- + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: Run unit tests + run: | + pytest tests/ -m unit -v --tb=short --junitxml=test-results-unit.xml + + - name: Upload unit test results + if: always() + uses: actions/upload-artifact@v4 + with: + name: unit-test-results-py${{ matrix.python-version }} + path: test-results-unit.xml + + integration-tests: + name: Integration Tests + runs-on: ubuntu-latest + needs: unit-tests + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.10" + + - name: Cache pip packages + uses: actions/cache@v4 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-3.10-${{ hashFiles('requirements.txt') }} + restore-keys: | + ${{ runner.os }}-pip-3.10- + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: Run integration tests + run: | + pytest tests/ -m integration -v --tb=short --junitxml=test-results-integration.xml + + - name: Upload integration test results + if: always() + uses: actions/upload-artifact@v4 + with: + name: integration-test-results + path: test-results-integration.xml + + all-tests-with-coverage: + name: Full Test Suite with Coverage + runs-on: ubuntu-latest + needs: unit-tests + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.10" + + - name: Cache pip packages + uses: actions/cache@v4 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-3.10-${{ hashFiles('requirements.txt') }} + restore-keys: | + ${{ runner.os }}-pip-3.10- + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: Run all tests with coverage + run: | + pytest tests/ -v --tb=short --cov=LatestPhasenetLocalTest/phasenet --cov=GaMMaTest/gamma --cov-report=term-missing --cov-report=xml:coverage.xml + + - name: Upload coverage report + if: always() + uses: actions/upload-artifact@v4 + with: + name: coverage-report + path: coverage.xml diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b5c17d5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ +*.egg-info/ +dist/ +build/ +.eggs/ +coverage.xml +test-results-*.xml +*.egg diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..33ce447 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,9 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* +addopts = -v --tb=short +markers = + unit: Unit tests for individual functions + integration: Integration tests for multi-component workflows diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..356ad77 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,14 @@ +numpy>=1.20 +pandas>=1.3 +scipy>=1.7 +scikit-learn>=0.24 +obspy>=1.3 +matplotlib>=3.4 +tqdm>=4.60 +h5py>=3.0 +pyproj>=3.0 +contexttimer>=0.3 + +# Testing +pytest>=7.0 +pytest-cov>=4.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..64796ff --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,73 @@ +"""Shared fixtures and path setup for all tests.""" + +import os +import sys + +import numpy as np +import pytest + +# Add source directories to sys.path so tests can import modules +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +PHASENET_DIR = os.path.join(PROJECT_ROOT, "LatestPhasenetLocalTest", "phasenet") +GAMMA_DIR = os.path.join(PROJECT_ROOT, "GaMMaTest") + +for path in [PHASENET_DIR, GAMMA_DIR]: + if path not in sys.path: + sys.path.insert(0, path) + + +@pytest.fixture +def project_root(): + return PROJECT_ROOT + + +@pytest.fixture +def phasenet_dir(): + return PHASENET_DIR + + +@pytest.fixture +def gamma_dir(): + return GAMMA_DIR + + +@pytest.fixture +def sample_waveform_3c(): + """A synthetic 3-component seismic waveform (nt=3000, nsta=1, nch=3).""" + np.random.seed(42) + nt, nsta, nch = 3000, 1, 3 + data = np.random.randn(nt, nsta, nch).astype(np.float32) + # Add a synthetic P-wave arrival at sample 500 + for ch in range(nch): + data[500:520, 0, ch] += 5.0 * np.sin(np.linspace(0, 4 * np.pi, 20)) + # Add a synthetic S-wave arrival at sample 1200 + for ch in range(nch): + data[1200:1240, 0, ch] += 8.0 * np.sin(np.linspace(0, 6 * np.pi, 40)) + return data + + +@pytest.fixture +def sample_predictions(): + """Synthetic model predictions with clear P and S peaks. + + Shape: (Nb=1, Nt=3000, Ns=1, Nc=3) where channels are [noise, P, S]. + """ + np.random.seed(42) + Nb, Nt, Ns, Nc = 1, 3000, 1, 3 + preds = np.zeros((Nb, Nt, Ns, Nc), dtype=np.float32) + # Noise channel is generally high + preds[:, :, :, 0] = 0.9 + + # P-wave peak at index 500 + p_signal = np.exp(-0.5 * ((np.arange(Nt) - 500) / 5) ** 2) + preds[0, :, 0, 1] = p_signal + preds[0, :, 0, 0] -= p_signal + + # S-wave peak at index 1200 + s_signal = np.exp(-0.5 * ((np.arange(Nt) - 1200) / 5) ** 2) + preds[0, :, 0, 2] = s_signal + preds[0, :, 0, 0] -= s_signal + + # Clamp noise channel + preds[:, :, :, 0] = np.clip(preds[:, :, :, 0], 0, 1) + return preds diff --git a/tests/test_data_reader.py b/tests/test_data_reader.py new file mode 100644 index 0000000..4a09229 --- /dev/null +++ b/tests/test_data_reader.py @@ -0,0 +1,178 @@ +"""Unit tests for data_reader.py normalization functions and DataConfig.""" + +import sys +import types +from unittest import mock + +import numpy as np +import pytest + +# Mock tensorflow and its submodules so data_reader can be imported without TF installed. +# The functions under test (normalize, normalize_long, normalize_batch, DataConfig) +# do not use TensorFlow at all — it's only imported at module level. +_tf_mock = types.ModuleType("tensorflow") +_tf_compat = types.ModuleType("tensorflow.compat") +_tf_compat_v1 = types.ModuleType("tensorflow.compat.v1") +_tf_compat_v1.disable_eager_execution = lambda: None +_tf_compat_v1.logging = types.SimpleNamespace(set_verbosity=lambda *a: None, ERROR=0) +_tf_mock.compat = _tf_compat +_tf_mock.compat.v1 = _tf_compat_v1 +_tf_mock.nest = types.SimpleNamespace(flatten=lambda x: x, pack_sequence_as=lambda t, v: v) +_tf_mock.data = types.SimpleNamespace(Dataset=type("Dataset", (), {"range": classmethod(lambda cls, n: None)})) +_tf_mock.numpy_function = lambda *a, **kw: None + +for mod_name in ["tensorflow", "tensorflow.compat", "tensorflow.compat.v1"]: + if mod_name not in sys.modules: + sys.modules[mod_name] = {"tensorflow": _tf_mock, "tensorflow.compat": _tf_compat, "tensorflow.compat.v1": _tf_compat_v1}[mod_name] + +# Also mock h5py and obspy if not installed +for mod_name in ["h5py", "obspy"]: + if mod_name not in sys.modules: + sys.modules[mod_name] = mock.MagicMock() + +from data_reader import normalize, normalize_long, normalize_batch, DataConfig + + +class TestDataConfig: + """Tests for the DataConfig class.""" + + @pytest.mark.unit + def test_default_values(self): + config = DataConfig() + assert config.n_channel == 3 + assert config.n_class == 3 + assert config.sampling_rate == 100 + assert config.dt == 0.01 + assert config.X_shape == [3000, 1, 3] + assert config.Y_shape == [3000, 1, 3] + + @pytest.mark.unit + def test_custom_values(self): + config = DataConfig(n_channel=1, sampling_rate=200) + assert config.n_channel == 1 + assert config.sampling_rate == 200 + + @pytest.mark.unit + def test_label_shape_default(self): + config = DataConfig() + assert config.label_shape == "gaussian" + assert config.label_width == 30 + + +class TestNormalize: + """Tests for the normalize function.""" + + @pytest.mark.unit + def test_zero_mean(self): + """After normalization, data should have approximately zero mean.""" + np.random.seed(42) + data = np.random.randn(3000, 1, 3).astype(np.float32) + 5.0 + result = normalize(data.copy()) + assert np.allclose(np.mean(result, axis=0), 0, atol=1e-5) + + @pytest.mark.unit + def test_unit_std(self): + """After normalization, data should have approximately unit std.""" + np.random.seed(42) + data = np.random.randn(3000, 1, 3).astype(np.float32) * 10.0 + result = normalize(data.copy()) + assert np.allclose(np.std(result, axis=0), 1, atol=0.05) + + @pytest.mark.unit + def test_zero_data(self): + """Zero data should not produce NaN or inf after normalization.""" + data = np.zeros((3000, 1, 3), dtype=np.float32) + result = normalize(data.copy()) + assert not np.any(np.isnan(result)) + assert not np.any(np.isinf(result)) + assert np.allclose(result, 0) + + @pytest.mark.unit + def test_preserves_shape(self): + np.random.seed(42) + data = np.random.randn(3000, 1, 3).astype(np.float32) + result = normalize(data.copy()) + assert result.shape == data.shape + + @pytest.mark.unit + def test_single_channel(self): + """Should work with single-channel data.""" + np.random.seed(42) + data = np.random.randn(3000, 1, 1).astype(np.float32) + result = normalize(data.copy()) + assert result.shape == (3000, 1, 1) + assert not np.any(np.isnan(result)) + + +class TestNormalizeLong: + """Tests for the normalize_long function (sliding window normalization).""" + + @pytest.mark.unit + def test_output_shape(self): + np.random.seed(42) + data = np.random.randn(6000, 1, 3).astype(np.float32) + result = normalize_long(data.copy()) + assert result.shape == data.shape + + @pytest.mark.unit + def test_no_nan_output(self): + np.random.seed(42) + data = np.random.randn(6000, 1, 3).astype(np.float32) + result = normalize_long(data.copy()) + assert not np.any(np.isnan(result)) + assert not np.any(np.isinf(result)) + + @pytest.mark.unit + def test_preserves_dtype(self): + data = np.random.randn(3000, 1, 3).astype(np.float32) + result = normalize_long(data.copy()) + assert result.dtype == np.float32 + + @pytest.mark.unit + def test_zero_data(self): + """Zero data should not produce NaN or inf.""" + data = np.zeros((3000, 1, 3), dtype=np.float32) + result = normalize_long(data.copy()) + assert not np.any(np.isnan(result)) + assert not np.any(np.isinf(result)) + + @pytest.mark.unit + def test_custom_window(self): + np.random.seed(42) + data = np.random.randn(3000, 1, 3).astype(np.float32) + result = normalize_long(data.copy(), window=1500) + assert result.shape == data.shape + assert not np.any(np.isnan(result)) + + +class TestNormalizeBatch: + """Tests for the normalize_batch function.""" + + @pytest.mark.unit + def test_output_shape(self): + np.random.seed(42) + data = np.random.randn(4, 3000, 1, 3).astype(np.float32) + result = normalize_batch(data.copy()) + assert result.shape == data.shape + + @pytest.mark.unit + def test_no_nan_output(self): + np.random.seed(42) + data = np.random.randn(4, 3000, 1, 3).astype(np.float32) + result = normalize_batch(data.copy()) + assert not np.any(np.isnan(result)) + assert not np.any(np.isinf(result)) + + @pytest.mark.unit + def test_zero_batch(self): + data = np.zeros((2, 3000, 1, 3), dtype=np.float32) + result = normalize_batch(data.copy()) + assert not np.any(np.isnan(result)) + assert not np.any(np.isinf(result)) + + @pytest.mark.unit + def test_single_station_batch(self): + np.random.seed(42) + data = np.random.randn(1, 3000, 1, 3).astype(np.float32) + result = normalize_batch(data.copy()) + assert result.shape == (1, 3000, 1, 3) diff --git a/tests/test_detect_peaks.py b/tests/test_detect_peaks.py new file mode 100644 index 0000000..59ed8a7 --- /dev/null +++ b/tests/test_detect_peaks.py @@ -0,0 +1,140 @@ +"""Unit tests for the detect_peaks module.""" + +import numpy as np +import pytest + +from detect_peaks import detect_peaks + + +class TestDetectPeaks: + """Tests for the detect_peaks function.""" + + @pytest.mark.unit + def test_simple_peaks(self): + """Detect peaks in a simple signal with known peak locations.""" + x = np.array([0, 1, 0, 2, 0, 3, 0, 2, 0, 1, 0], dtype=float) + ind, vals = detect_peaks(x, mpd=2, show=False) + # Peaks should be at indices 3 (value=2), 5 (value=3), 7 (value=2) + assert 5 in ind, "Peak at index 5 (value=3) should be detected" + assert len(ind) >= 1 + + @pytest.mark.unit + def test_sinusoidal_signal(self): + """Detect peaks in a sinusoidal signal.""" + t = np.linspace(0, 2 * np.pi, 200) + x = np.sin(t) + ind, vals = detect_peaks(x, mph=0.5, mpd=20, show=False) + # There should be one major peak near index 50 (pi/2) + assert len(ind) >= 1 + # Peak value should be close to 1.0 + assert np.all(vals >= 0.5) + + @pytest.mark.unit + def test_minimum_peak_height(self): + """Peaks below minimum peak height should be filtered out.""" + x = np.array([0, 0.2, 0, 0.8, 0, 0.3, 0], dtype=float) + ind, vals = detect_peaks(x, mph=0.5, show=False) + # Only the peak at index 3 (value=0.8) should pass mph=0.5 + assert len(ind) == 1 + assert ind[0] == 3 + assert np.isclose(vals[0], 0.8) + + @pytest.mark.unit + def test_minimum_peak_distance(self): + """Peaks closer than mpd should be filtered (keep tallest).""" + x = np.array([0, 3, 0, 5, 0, 2, 0, 0, 0, 4, 0], dtype=float) + ind, vals = detect_peaks(x, mpd=3, show=False) + # Peaks at 1 (3), 3 (5), 5 (2), 9 (4) + # With mpd=3: peak 3 (5) suppresses 1 (3) and 5 (2); peak 9 (4) survives + assert 3 in ind + assert 9 in ind + + @pytest.mark.unit + def test_empty_signal(self): + """An empty or very short signal should return empty array.""" + x = np.array([1.0]) + result = detect_peaks(x, show=False) + # For signals < 3 samples, returns a single empty array (not a tuple) + assert len(result) == 0 + + x = np.array([1.0, 2.0]) + result = detect_peaks(x, show=False) + assert len(result) == 0 + + @pytest.mark.unit + def test_flat_signal(self): + """A flat signal should produce no peaks (edge=None) or all edges.""" + x = np.ones(100) + ind, vals = detect_peaks(x, edge=None, show=False) + assert len(ind) == 0 + + @pytest.mark.unit + def test_valley_detection(self): + """Valley detection should find local minima.""" + x = np.array([3, 1, 3, 0, 3, 2, 3], dtype=float) + ind, vals = detect_peaks(x, valley=True, mpd=1, show=False) + # Valleys at indices 1 (1), 3 (0), 5 (2) + assert 3 in ind # deepest valley + + @pytest.mark.unit + def test_nan_handling(self): + """NaN values in the signal should be handled without crashing. + + Note: detect_peaks uses np.in1d which was removed in numpy 2.0+. + This test verifies the behavior on compatible numpy versions and + skips gracefully otherwise. + """ + np.random.seed(42) + x = np.random.randn(100) + x[40:50] = np.nan + try: + ind, vals = detect_peaks(x, show=False) + # Indices should not be in the NaN region + for i in ind: + assert i < 39 or i > 50 + except AttributeError: + # np.in1d removed in numpy >= 2.0; this is a known compatibility issue + pytest.skip("detect_peaks NaN handling requires numpy < 2.0 (uses np.in1d)") + + @pytest.mark.unit + def test_threshold_filter(self): + """Threshold parameter should filter peaks by neighbor difference.""" + x = np.array([-2, 1, -2, 2, 1, 1, 3, 0], dtype=float) + ind_low, _ = detect_peaks(x, threshold=0.5, show=False) + ind_high, _ = detect_peaks(x, threshold=2, show=False) + # Higher threshold should produce fewer or equal peaks + assert len(ind_high) <= len(ind_low) + + @pytest.mark.unit + def test_returns_correct_probabilities(self): + """The second return value should contain the actual peak amplitudes.""" + x = np.array([0, 0.3, 0, 0.7, 0, 0.9, 0], dtype=float) + ind, vals = detect_peaks(x, mph=0.2, show=False) + for i, v in zip(ind, vals): + assert np.isclose(v, x[i]) + + @pytest.mark.unit + def test_seismic_like_signal(self): + """Test with a signal mimicking seismic P and S wave probabilities.""" + np.random.seed(123) + nt = 3000 + x = np.random.uniform(0, 0.1, nt) + # P-wave peak at sample 500 + x[500] = 0.85 + x[499] = 0.3 + x[501] = 0.3 + # S-wave peak at sample 1200 + x[1200] = 0.92 + x[1199] = 0.4 + x[1201] = 0.4 + + ind, vals = detect_peaks(x, mph=0.3, mpd=50, show=False) + assert 500 in ind, "P-wave peak at 500 should be detected" + assert 1200 in ind, "S-wave peak at 1200 should be detected" + + @pytest.mark.unit + def test_both_edges(self): + """Test detection with edge='both' for flat peaks.""" + x = np.array([0, 1, 1, 0, 1, 1, 0], dtype=float) + ind, vals = detect_peaks(x, edge='both', show=False) + assert len(ind) >= 2 # both edges of both flat peaks diff --git a/tests/test_gamma_base.py b/tests/test_gamma_base.py new file mode 100644 index 0000000..d1d2a86 --- /dev/null +++ b/tests/test_gamma_base.py @@ -0,0 +1,75 @@ +"""Unit tests for the GaMMa base mixture model module.""" + +import numpy as np +import pytest + +from gamma._base import _check_shape, _check_X + + +class TestCheckShape: + """Tests for _check_shape validation function.""" + + @pytest.mark.unit + def test_valid_shape(self): + """No error for correctly shaped parameter.""" + param = np.array([1.0, 2.0, 3.0]) + _check_shape(param, (3,), "test_param") # should not raise + + @pytest.mark.unit + def test_invalid_shape(self): + """Should raise ValueError for mismatched shape.""" + param = np.array([1.0, 2.0]) + with pytest.raises(ValueError, match="should have the shape"): + _check_shape(param, (3,), "test_param") + + @pytest.mark.unit + def test_2d_shape(self): + param = np.array([[1.0, 2.0], [3.0, 4.0]]) + _check_shape(param, (2, 2), "test_param") # should not raise + + @pytest.mark.unit + def test_2d_invalid(self): + param = np.array([[1.0, 2.0], [3.0, 4.0]]) + with pytest.raises(ValueError): + _check_shape(param, (3, 2), "test_param") + + @pytest.mark.unit + def test_scalar(self): + param = np.array(5.0) + _check_shape(param, (), "test_param") # should not raise + + +class TestCheckX: + """Tests for _check_X input validation function.""" + + @pytest.mark.unit + def test_valid_input(self): + X = np.random.randn(100, 3) + result = _check_X(X, n_components=5) + assert result.shape == (100, 3) + + @pytest.mark.unit + def test_too_few_samples(self): + """Should raise when n_samples < n_components.""" + X = np.random.randn(3, 2) + with pytest.raises(ValueError, match="n_samples >= n_components"): + _check_X(X, n_components=5) + + @pytest.mark.unit + def test_wrong_features(self): + """Should raise when features don't match expected.""" + X = np.random.randn(100, 3) + with pytest.raises(ValueError, match="features"): + _check_X(X, n_features=5) + + @pytest.mark.unit + def test_converts_dtype(self): + X = np.random.randn(10, 2).astype(np.int32) + result = _check_X(X) + assert result.dtype in [np.float32, np.float64] + + @pytest.mark.unit + def test_no_constraints(self): + X = np.random.randn(5, 2) + result = _check_X(X) + assert result.shape == (5, 2) diff --git a/tests/test_gamma_integration.py b/tests/test_gamma_integration.py new file mode 100644 index 0000000..1a1cdc3 --- /dev/null +++ b/tests/test_gamma_integration.py @@ -0,0 +1,190 @@ +"""Integration tests for GaMMa utilities: convert_picks_csv and from_seconds.""" + +import os + +import numpy as np +import pandas as pd +import pytest + +from gamma.utils import convert_picks_csv, from_seconds + + +class TestFromSeconds: + """Tests for the from_seconds timestamp conversion.""" + + @pytest.mark.unit + def test_epoch_zero(self): + result = from_seconds(0) + assert result == "1970-01-01T00:00:00.000" + + @pytest.mark.unit + def test_known_timestamp(self): + # 2019-07-06T02:15:00.000 UTC + ts = 1562379300.0 + result = from_seconds(ts) + assert "2019-07-06" in result + assert "02:15:00" in result + + @pytest.mark.unit + def test_fractional_seconds(self): + ts = 1562379300.123 + result = from_seconds(ts) + assert ".123" in result + + +class TestConvertPicksCsv: + """Integration tests for convert_picks_csv.""" + + @pytest.fixture + def sample_picks_and_stations(self): + """Create minimal picks and stations DataFrames for testing.""" + picks = pd.DataFrame({ + "id": ["STA1_P", "STA1_S", "STA2_P", "STA2_S"], + "timestamp": pd.to_datetime([ + "2019-07-06T02:15:00.000", + "2019-07-06T02:15:05.000", + "2019-07-06T02:15:01.000", + "2019-07-06T02:15:07.000", + ], utc=True), + "type": ["P", "S", "P", "S"], + "prob": [0.9, 0.8, 0.85, 0.75], + "amp": [1e-5, 2e-5, 1.5e-5, 3e-5], + }) + + stations = pd.DataFrame({ + "id": ["STA1_P", "STA1_S", "STA2_P", "STA2_S"], + "x(km)": [0.0, 0.0, 10.0, 10.0], + "y(km)": [0.0, 0.0, 5.0, 5.0], + "z(km)": [0.0, 0.0, -0.5, -0.5], + }) + + config = { + "dims": ["x(km)", "y(km)", "z(km)"], + "use_amplitude": True, + } + return picks, stations, config + + @pytest.mark.integration + def test_output_shapes(self, sample_picks_and_stations): + """convert_picks_csv should return arrays of consistent sizes.""" + picks, stations, config = sample_picks_and_stations + data, locs, phase_type, phase_weight, pick_idx, pick_station_id = \ + convert_picks_csv(picks, stations, config) + + n = len(picks) + assert data.shape[0] == n + assert data.shape[1] == 2 # time + amplitude + assert locs.shape == (n, 3) # x, y, z + assert len(phase_type) == n + assert phase_weight.shape == (n, 1) + assert len(pick_idx) == n + assert len(pick_station_id) == n + + @pytest.mark.integration + def test_phase_types_lowered(self, sample_picks_and_stations): + """Phase types should be lowercased.""" + picks, stations, config = sample_picks_and_stations + _, _, phase_type, _, _, _ = convert_picks_csv(picks, stations, config) + assert all(pt in ["p", "s"] for pt in phase_type) + + @pytest.mark.integration + def test_amplitude_log_transform(self, sample_picks_and_stations): + """Amplitude column should be log10(amp * 100).""" + picks, stations, config = sample_picks_and_stations + data, _, _, _, _, _ = convert_picks_csv(picks, stations, config) + expected_amp = np.log10(picks["amp"].values * 1e2) + np.testing.assert_allclose(data[:, 1], expected_amp, rtol=1e-5) + + @pytest.mark.integration + def test_without_amplitude(self, sample_picks_and_stations): + """Without amplitude, data should only have time column.""" + picks, stations, config = sample_picks_and_stations + config["use_amplitude"] = False + data, _, _, _, _, _ = convert_picks_csv(picks, stations, config) + assert data.shape[1] == 1 # time only + + @pytest.mark.integration + def test_nan_stations_filtered(self): + """Picks with missing station coordinates should be filtered out.""" + picks = pd.DataFrame({ + "id": ["STA1_P", "MISSING_P"], + "timestamp": pd.to_datetime([ + "2019-07-06T02:15:00.000", + "2019-07-06T02:15:01.000", + ], utc=True), + "type": ["P", "P"], + "prob": [0.9, 0.8], + "amp": [1e-5, 2e-5], + }) + stations = pd.DataFrame({ + "id": ["STA1_P"], + "x(km)": [0.0], + "y(km)": [0.0], + "z(km)": [0.0], + }) + config = {"dims": ["x(km)", "y(km)", "z(km)"], "use_amplitude": True} + data, locs, phase_type, phase_weight, pick_idx, pick_station_id = \ + convert_picks_csv(picks, stations, config) + # MISSING_P has no station match, so should be filtered + assert data.shape[0] == 1 + + +class TestGammaPicksDataIntegrity: + """Integration test that loads real picks.csv and station data to verify format compatibility.""" + + @pytest.fixture + def real_data_paths(self, gamma_dir): + picks_csv = os.path.join(gamma_dir, "picks.csv") + station_csv = os.path.join(gamma_dir, "tests", "SCSN_station_response.csv") + if not os.path.exists(picks_csv) or not os.path.exists(station_csv): + pytest.skip("Real test data not available") + return picks_csv, station_csv + + @pytest.mark.integration + def test_load_real_picks(self, real_data_paths): + """Real picks.csv should load without errors and have expected columns.""" + picks_csv, _ = real_data_paths + picks = pd.read_csv(picks_csv) + required_cols = ["station_id", "phase_time", "phase_score", "phase_type"] + for col in required_cols: + assert col in picks.columns, f"Missing column: {col}" + assert len(picks) > 0 + + @pytest.mark.integration + def test_load_real_stations(self, real_data_paths): + """Real station CSV should load and have expected columns.""" + _, station_csv = real_data_paths + stations = pd.read_csv(station_csv) + required_cols = ["id", "latitude", "longitude", "elevation(m)"] + for col in required_cols: + assert col in stations.columns, f"Missing column: {col}" + assert len(stations) > 0 + + @pytest.mark.integration + def test_picks_stations_id_overlap(self, real_data_paths): + """Picks should reference station IDs that exist in the stations file.""" + picks_csv, station_csv = real_data_paths + picks = pd.read_csv(picks_csv) + stations = pd.read_csv(station_csv) + pick_ids = set(picks["station_id"].unique()) + station_ids = set(stations["id"].unique()) + overlap = pick_ids & station_ids + assert len(overlap) > 0, "No overlap between pick station IDs and station IDs" + + @pytest.mark.integration + def test_phase_types_valid(self, real_data_paths): + """All phase types should be P or S.""" + picks_csv, _ = real_data_paths + picks = pd.read_csv(picks_csv) + valid_types = {"P", "S"} + actual_types = set(picks["phase_type"].unique()) + assert actual_types.issubset(valid_types), \ + f"Unexpected phase types: {actual_types - valid_types}" + + @pytest.mark.integration + def test_phase_scores_in_range(self, real_data_paths): + """All phase scores should be between 0 and 1.""" + picks_csv, _ = real_data_paths + picks = pd.read_csv(picks_csv) + assert picks["phase_score"].min() >= 0 + assert picks["phase_score"].max() <= 1.0 diff --git a/tests/test_postprocess.py b/tests/test_postprocess.py new file mode 100644 index 0000000..e945830 --- /dev/null +++ b/tests/test_postprocess.py @@ -0,0 +1,178 @@ +"""Unit tests for the postprocess module (extract_picks, calc_metrics, calc_timestamp).""" + +import os +import tempfile + +import numpy as np +import pytest + +from postprocess import extract_picks, calc_metrics, calc_timestamp + + +class TestExtractPicks: + """Tests for the extract_picks function.""" + + @pytest.mark.unit + def test_basic_pick_extraction(self, sample_predictions): + """extract_picks should find P and S picks from clear prediction peaks.""" + picks = extract_picks( + sample_predictions, + file_names=["test_file.mseed"], + begin_times=["2019-07-06T02:15:00.000+00:00"], + station_ids=[["STA01"]], + dt=0.01, + ) + assert len(picks) >= 2, "Should detect at least P and S picks" + + phase_types = [p["phase_type"] for p in picks] + assert "P" in phase_types, "Should detect a P-phase pick" + assert "S" in phase_types, "Should detect an S-phase pick" + + @pytest.mark.unit + def test_pick_fields(self, sample_predictions): + """Each pick should have the required fields.""" + picks = extract_picks(sample_predictions, dt=0.01) + required_fields = [ + "file_name", "station_id", "begin_time", + "phase_index", "phase_time", "phase_score", "phase_type", "dt", + ] + for pick in picks: + for field in required_fields: + assert field in pick, f"Pick missing field: {field}" + + @pytest.mark.unit + def test_default_file_names(self, sample_predictions): + """When file_names is None, defaults should be generated.""" + picks = extract_picks(sample_predictions, dt=0.01) + assert all(p["file_name"] == "0000" for p in picks) + + @pytest.mark.unit + def test_default_station_ids(self, sample_predictions): + """When station_ids is None, defaults should be generated.""" + picks = extract_picks(sample_predictions, dt=0.01) + assert all(p["station_id"] == "0000" for p in picks) + + @pytest.mark.unit + def test_default_begin_times(self, sample_predictions): + """When begin_times is None, epoch time should be used.""" + picks = extract_picks(sample_predictions, dt=0.01) + for pick in picks: + assert "1970-01-01" in pick["begin_time"] + + @pytest.mark.unit + def test_pick_time_ordering(self, sample_predictions): + """P-wave pick should come before S-wave pick (P arrives first).""" + picks = extract_picks(sample_predictions, dt=0.01) + p_picks = [p for p in picks if p["phase_type"] == "P"] + s_picks = [p for p in picks if p["phase_type"] == "S"] + if p_picks and s_picks: + assert p_picks[0]["phase_index"] < s_picks[0]["phase_index"] + + @pytest.mark.unit + def test_phase_score_range(self, sample_predictions): + """Phase scores should be between 0 and 1.""" + picks = extract_picks(sample_predictions, dt=0.01) + for pick in picks: + assert 0 <= pick["phase_score"] <= 1.0 + + @pytest.mark.unit + def test_phase_index_near_expected(self, sample_predictions): + """Pick indices should be near the injected peaks (500 for P, 1200 for S).""" + picks = extract_picks(sample_predictions, dt=0.01) + p_indices = [p["phase_index"] for p in picks if p["phase_type"] == "P"] + s_indices = [p["phase_index"] for p in picks if p["phase_type"] == "S"] + assert any(abs(idx - 500) <= 5 for idx in p_indices), \ + f"P pick index should be near 500, got {p_indices}" + assert any(abs(idx - 1200) <= 5 for idx in s_indices), \ + f"S pick index should be near 1200, got {s_indices}" + + @pytest.mark.unit + def test_no_picks_for_noise(self): + """A pure noise prediction (low values) should yield no picks.""" + preds = np.random.uniform(0, 0.1, (1, 3000, 1, 3)).astype(np.float32) + preds[:, :, :, 0] = 0.9 + picks = extract_picks(preds, dt=0.01) + assert len(picks) == 0 + + @pytest.mark.unit + def test_bytes_file_names(self, sample_predictions): + """Byte-encoded file names should be decoded properly.""" + picks = extract_picks( + sample_predictions, + file_names=[b"test_file.mseed"], + dt=0.01, + ) + assert all(p["file_name"] == "test_file.mseed" for p in picks) + + @pytest.mark.unit + def test_multiple_batches(self): + """Multiple batches should each produce independent picks.""" + Nb, Nt, Ns, Nc = 3, 3000, 1, 3 + preds = np.zeros((Nb, Nt, Ns, Nc), dtype=np.float32) + preds[:, :, :, 0] = 0.9 + for b in range(Nb): + peak_loc = 300 + b * 500 + p_signal = np.exp(-0.5 * ((np.arange(Nt) - peak_loc) / 5) ** 2) + preds[b, :, 0, 1] = p_signal + picks = extract_picks(preds, dt=0.01) + file_names = set(p["file_name"] for p in picks) + assert len(file_names) == Nb + + +class TestCalcTimestamp: + """Tests for the calc_timestamp function.""" + + @pytest.mark.unit + def test_basic_timestamp(self): + result = calc_timestamp("2019-07-06T02:15:00.000", 5.12) + assert result == "2019-07-06T02:15:05.120" + + @pytest.mark.unit + def test_zero_offset(self): + result = calc_timestamp("2019-07-06T02:15:00.000", 0.0) + assert result == "2019-07-06T02:15:00.000" + + @pytest.mark.unit + def test_fractional_seconds(self): + result = calc_timestamp("2019-07-06T02:15:00.000", 0.01) + assert result == "2019-07-06T02:15:00.010" + + @pytest.mark.unit + def test_large_offset_crosses_minute(self): + result = calc_timestamp("2019-07-06T02:15:50.000", 15.0) + assert result == "2019-07-06T02:16:05.000" + + +class TestCalcMetrics: + """Tests for the calc_metrics function.""" + + @pytest.mark.unit + def test_perfect_detection(self): + """Perfect detection: all true picks detected, no false positives.""" + precision, recall, f1 = calc_metrics(nTP=10, nP=10, nT=10) + assert precision == 1.0 + assert recall == 1.0 + assert f1 == 1.0 + + @pytest.mark.unit + def test_half_precision(self): + """Half the positive picks are true positives.""" + precision, recall, f1 = calc_metrics(nTP=5, nP=10, nT=5) + assert precision == 0.5 + assert recall == 1.0 + + @pytest.mark.unit + def test_half_recall(self): + """Only half the true picks are detected.""" + precision, recall, f1 = calc_metrics(nTP=5, nP=5, nT=10) + assert precision == 1.0 + assert recall == 0.5 + + @pytest.mark.unit + def test_f1_calculation(self): + """F1 should be the harmonic mean of precision and recall.""" + precision, recall, f1 = calc_metrics(nTP=6, nP=10, nT=8) + expected_p = 6 / 10 + expected_r = 6 / 8 + expected_f1 = 2 * expected_p * expected_r / (expected_p + expected_r) + assert np.isclose(f1, expected_f1) diff --git a/tests/test_util.py b/tests/test_util.py new file mode 100644 index 0000000..4d235f1 --- /dev/null +++ b/tests/test_util.py @@ -0,0 +1,151 @@ +"""Unit tests for util.py (EMA, LMA, metrics, clean_queue).""" + +import numpy as np +import pytest + +from util import EMA, LMA, metrics, clean_queue, clean_queue_thread + + +class TestEMA: + """Tests for the Exponential Moving Average class.""" + + @pytest.mark.unit + def test_first_value(self): + """First call should return the input value itself.""" + ema = EMA(alpha=0.9) + result = ema(5.0) + assert result == 5.0 + + @pytest.mark.unit + def test_smoothing(self): + """EMA should smooth values towards the running average.""" + ema = EMA(alpha=0.9) + ema(10.0) + result = ema(20.0) + # result = 0.9 * 10 + 0.1 * 20 = 11.0 + assert np.isclose(result, 11.0) + + @pytest.mark.unit + def test_high_alpha_slow_change(self): + """High alpha means slow response to new values.""" + ema = EMA(alpha=0.99) + ema(0.0) + for _ in range(100): + val = ema(1.0) + # After 100 iterations, should be close to but not quite 1.0 + assert val < 1.0 + assert val > 0.5 + + @pytest.mark.unit + def test_zero_alpha_instant_change(self): + """Alpha=0 means the EMA always equals the latest value.""" + ema = EMA(alpha=0.0) + ema(10.0) + result = ema(20.0) + assert result == 20.0 + + @pytest.mark.unit + def test_value_property(self): + ema = EMA(alpha=0.5) + ema(10.0) + assert ema.value == 10.0 + + +class TestLMA: + """Tests for the Linear Moving Average class.""" + + @pytest.mark.unit + def test_first_value(self): + lma = LMA() + result = lma(5.0) + assert result == 5.0 + + @pytest.mark.unit + def test_running_mean(self): + """LMA should compute the running mean.""" + lma = LMA() + lma(2.0) + result = lma(4.0) + # Mean of [2, 4] = 3 + assert np.isclose(result, 3.0) + + @pytest.mark.unit + def test_multiple_values(self): + lma = LMA() + values = [1.0, 2.0, 3.0, 4.0, 5.0] + for v in values: + result = lma(v) + assert np.isclose(result, 3.0) # mean of 1..5 + + @pytest.mark.unit + def test_value_property(self): + lma = LMA() + lma(10.0) + assert lma.value == 10.0 + + +class TestMetrics: + """Tests for the metrics function.""" + + @pytest.mark.unit + def test_perfect_score(self): + precision, recall, f1 = metrics(10, 10, 10) + assert precision == 1.0 + assert recall == 1.0 + assert f1 == 1.0 + + @pytest.mark.unit + def test_low_precision(self): + precision, recall, f1 = metrics(2, 10, 2) + assert precision == 0.2 + assert recall == 1.0 + + @pytest.mark.unit + def test_low_recall(self): + precision, recall, f1 = metrics(2, 2, 10) + assert precision == 1.0 + assert recall == 0.2 + + @pytest.mark.unit + def test_f1_harmonic_mean(self): + p, r, f1 = metrics(3, 5, 8) + expected_f1 = 2 * (3 / 5) * (3 / 8) / ((3 / 5) + (3 / 8)) + assert np.isclose(f1, expected_f1) + + +class TestCleanQueue: + """Tests for the clean_queue and clean_queue_thread functions.""" + + @pytest.mark.unit + def test_removes_zeros(self): + picks = [[0, 1, 0, 2, 0], [3, 0, 4]] + result = clean_queue(picks) + assert result == [[1, 2], [3, 4]] + + @pytest.mark.unit + def test_all_zeros(self): + picks = [[0, 0, 0]] + result = clean_queue(picks) + assert result == [[]] + + @pytest.mark.unit + def test_no_zeros(self): + picks = [[1, 2, 3]] + result = clean_queue(picks) + assert result == [[1, 2, 3]] + + @pytest.mark.unit + def test_empty_list(self): + picks = [[]] + result = clean_queue(picks) + assert result == [[]] + + @pytest.mark.unit + def test_clean_queue_thread(self): + result = clean_queue_thread([0, 5, 0, 10, 0]) + assert result == [5, 10] + + @pytest.mark.unit + def test_clean_queue_thread_empty(self): + result = clean_queue_thread([0, 0, 0]) + assert result == []