diff --git a/tests/test_gsolve_algorithms.py b/tests/test_gsolve_algorithms.py new file mode 100644 index 0000000..0316787 --- /dev/null +++ b/tests/test_gsolve_algorithms.py @@ -0,0 +1,300 @@ +# GSolve - gravity processing software. +# Copyright (c) 2026 Earth Sciences New Zealand. +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. + +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# SPDX-License-Identifier: GPLv3 + +"""Tests for solver core logic in gsolve.gsolve_algorithms.""" + +from __future__ import annotations + +import warnings + +import numpy as np +import pandas as pd +import pytest + +from gsolve.gsolve_algorithms import call_gsolve_lstsq, g_solver_lstsq +from gsolve.gsolve_outputs import GSolveResults + + +def _obs_df(include_meter_reading: bool = False) -> pd.DataFrame: + """Create a small, stable observation set for solver tests.""" + data: dict[str, object] = { + "site_id": ["A", "B", "A", "B", "A", "B"], + "gravity": [1000.0, 1002.0, 1001.0, 1003.0, 1001.5, 1003.5], + "timedelta": [0.0, 0.5, 1.0, 0.0, 0.5, 1.0], + "loop": ["L1", "L1", "L1", "L2", "L2", "L2"], + } + if include_meter_reading: + data["meter_reading_mgal"] = [ + 1000.0, + 1002.1, + 1000.9, + 1003.1, + 1001.6, + 1003.4, + ] + return pd.DataFrame(data) + + +def _ref_df(index: list[str] | None = None) -> pd.DataFrame: + """Create a reference-site table keyed by site_id index.""" + if index is None: + index = ["A", "B"] + grav = [1000.2 + i for i in range(len(index))] + return pd.DataFrame( + {"reference_gravity": grav}, + index=pd.Index(index, name="site_id"), + ) + + +def _solver_inputs(n_obs: int = 6, use_two_loops: bool = True) -> dict: + """Build raw ndarray inputs for direct g_solver_lstsq tests.""" + site_ids = np.array(["A", "B", "A", "B", "A", "B"][:n_obs]) + obs_loop = np.array(["L1", "L1", "L1", "L2", "L2", "L2"][:n_obs]) + if not use_two_loops: + obs_loop = np.array(["L1"] * n_obs) + + return { + "obs_g": np.array([1000.0, 1002.0, 1001.0, 1003.0, 1001.5, 1003.5][:n_obs]), + "obs_site_id": site_ids, + "obs_timedelta": np.array([0.0, 0.5, 1.0, 0.0, 0.5, 1.0][:n_obs]), + "ties_site_id": np.array(["A", "B"]), + "ties_g": np.array([1000.2, 1001.2]), + "obs_loop": obs_loop, + } + + +def _solver_inputs_robust() -> dict: + """Build a better-conditioned synthetic dataset for clipping tests.""" + n_obs = 18 + site_ids = np.array(["A", "B", "C"] * 6) + obs_loop = np.array((["L1"] * 6) + (["L2"] * 6) + (["L3"] * 6)) + obs_timedelta = np.array([0.0, 0.5, 1.0, 1.5, 2.0, 2.5] * 3) + obs_g = np.array( + [ + 1000.0, + 1001.8, + 1003.4, + 1000.6, + 1002.2, + 1003.8, + 1000.2, + 1002.0, + 1003.6, + 1000.8, + 1002.4, + 1004.0, + 1000.3, + 1002.1, + 1003.7, + 1000.9, + 1002.5, + 1060.0, + ] + ) + + return { + "obs_g": obs_g, + "obs_site_id": site_ids, + "obs_timedelta": obs_timedelta, + "ties_site_id": np.array(["A", "B", "C"]), + "ties_g": np.array([1000.2, 1001.2, 1002.2]), + "obs_loop": obs_loop, + } + + +class TestCallGSolveLstsq: + def test_no_tie_sites_raises(self) -> None: + obs = _obs_df() + ref = _ref_df(index=["X", "Y"]) + + with pytest.raises(ValueError, match="no tie sites"): + call_gsolve_lstsq(obs=obs, ref_sites=ref, method=1) + + def test_returns_gsolve_results_object(self) -> None: + obs = _obs_df() + ref = _ref_df() + + result = call_gsolve_lstsq( + obs=obs, + ref_sites=ref, + method=2, + use_loops=True, + percentile_clipping=100.0, + ) + + assert isinstance(result, GSolveResults) + assert result.params.method == 2 + assert result.obs_solution.shape[0] == obs.shape[0] + assert set(result.site_solution.index) == {"A", "B"} + + def test_calibration_enabled_returns_float(self) -> None: + obs = _obs_df(include_meter_reading=True) + ref = _ref_df() + + result = call_gsolve_lstsq( + obs=obs, + ref_sites=ref, + method=1, + calculate_calibration_factor=True, + ) + + assert isinstance(result.calibration_factor, float) + assert np.isfinite(result.calibration_factor) + + def test_calibration_enabled_without_meter_reading_raises(self) -> None: + obs = _obs_df(include_meter_reading=False) + ref = _ref_df() + + with pytest.raises(KeyError, match="meter_reading_mgal"): + call_gsolve_lstsq( + obs=obs, + ref_sites=ref, + method=1, + calculate_calibration_factor=True, + ) + + +class TestGSolverLstsqValidation: + def test_invalid_method_raises(self) -> None: + kwargs = _solver_inputs() + + with pytest.raises(ValueError, match="invalid method"): + g_solver_lstsq( + **kwargs, + method=99, + use_loops=True, + calculate_calibration_factor=False, + ) + + @pytest.mark.parametrize("percentile", [-0.1, 100.1]) + def test_invalid_percentile_raises(self, percentile: float) -> None: + kwargs = _solver_inputs() + + with pytest.raises(ValueError, match="must be between 0 and 100 inclusive"): + g_solver_lstsq( + **kwargs, + method=1, + use_loops=True, + calculate_calibration_factor=False, + percentile_clipping=percentile, + ) + + def test_missing_non_detided_values_raises_for_calibration(self) -> None: + kwargs = _solver_inputs() + + with pytest.raises(ValueError, match="obs_g_not_detided must be provided"): + g_solver_lstsq( + **kwargs, + method=1, + use_loops=True, + calculate_calibration_factor=True, + obs_g_not_detided=None, + ) + + +class TestGSolverLstsqBranches: + def test_use_loops_false_returns_single_loop_terms(self) -> None: + kwargs = _solver_inputs(use_two_loops=False) + gravity, residuals, gravity_var, drift, baseline, cal, mask = g_solver_lstsq( + **kwargs, + method=1, + use_loops=False, + calculate_calibration_factor=False, + percentile_clipping=100.0, + ) + + assert gravity.shape == (2, 1) + assert residuals.shape[0] == kwargs["obs_g"].shape[0] + assert gravity_var.shape[0] == 2 + assert drift.shape == (1,) + assert baseline.shape == (1,) + assert cal is None + assert mask.dtype == bool + assert np.all(mask) + + def test_method_3_executes_and_returns_expected_shapes(self) -> None: + kwargs = _solver_inputs(use_two_loops=True) + gravity, residuals, gravity_var, drift, baseline, cal, mask = g_solver_lstsq( + **kwargs, + method=3, + use_loops=True, + calculate_calibration_factor=False, + percentile_clipping=100.0, + ) + + assert gravity.shape[0] == 2 + assert residuals.shape[0] == kwargs["obs_g"].shape[0] + assert gravity_var.shape[0] == 2 + assert drift.shape == (2,) + assert baseline.shape == (2,) + assert cal is None + assert np.all(np.isfinite(gravity)) + assert np.all(np.isfinite(gravity_var)) + assert mask.shape[0] == kwargs["obs_g"].shape[0] + + def test_percentile_100_keeps_all_observations(self) -> None: + kwargs = _solver_inputs() + _, _, _, _, _, _, mask = g_solver_lstsq( + **kwargs, + method=2, + use_loops=True, + calculate_calibration_factor=False, + percentile_clipping=100.0, + ) + assert np.all(mask) + + def test_percentile_clipping_excludes_outlier(self) -> None: + kwargs = _solver_inputs(n_obs=6) + kwargs["obs_g"] = np.array([1000.0, 1002.0, 1001.0, 1003.0, 1001.5, 1200.0]) + + _, _, _, _, _, _, mask = g_solver_lstsq( + **kwargs, + method=1, + use_loops=True, + calculate_calibration_factor=False, + percentile_clipping=80.0, + ) + + assert mask.shape[0] == kwargs["obs_g"].shape[0] + assert np.sum(~mask) >= 1 + + def test_percentile_clipping_robust_geometry_has_finite_outputs(self) -> None: + kwargs = _solver_inputs_robust() + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + gravity, residuals, gravity_var, drift, baseline, cal, mask = ( + g_solver_lstsq( + **kwargs, + method=1, + use_loops=True, + calculate_calibration_factor=False, + percentile_clipping=95.0, + ) + ) + + runtime_warnings = [w for w in caught if issubclass(w.category, RuntimeWarning)] + assert not runtime_warnings + assert gravity.shape[0] == 3 + assert residuals.shape[0] == kwargs["obs_g"].shape[0] + assert gravity_var.shape[0] == 3 + assert drift.shape == (3,) + assert baseline.shape == (3,) + assert cal is None + assert np.sum(mask) >= 7 + assert np.sum(~mask) >= 1 + assert np.all(np.isfinite(gravity)) + assert np.all(np.isfinite(gravity_var)) diff --git a/tests/test_reports.py b/tests/test_reports.py new file mode 100644 index 0000000..dcb80d8 --- /dev/null +++ b/tests/test_reports.py @@ -0,0 +1,306 @@ +# GSolve - gravity processing software. +# Copyright (c) 2026 Earth Sciences New Zealand. +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. + +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# SPDX-License-Identifier: GPLv3 + +"""Tests for report assembly and consistency checks in gsolve.reports.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd +import pytest + +import gsolve.reports as reports_module +from gsolve import GravityObservations, GravitySites +from gsolve.gsolve_outputs import GSolveResults +from gsolve.reports import GSolveReport + + +def _make_observations() -> GravityObservations: + return GravityObservations( + site_id=["A", "B", "A", "B", "A", "B"], + meter_reading_mgal=[1000.0, 1002.0, 1001.0, 1003.0, 1001.5, 1003.5], + meter_id=["G106"] * 6, + datetime=[ + "2021-01-01T00:00:00", + "2021-01-01T00:10:00", + "2021-01-01T00:20:00", + "2021-01-01T01:00:00", + "2021-01-01T01:10:00", + "2021-01-01T01:20:00", + ], + loop=["L1", "L1", "L1", "L2", "L2", "L2"], + comment=["o1", "o2", "o3", "o4", "o5", "o6"], + ) + + +def _make_sites() -> GravitySites: + return GravitySites( + site_id=["A", "B"], + latitude=[-41.0, -41.1], + longitude=[174.0, 174.1], + height_ellipsoidal=[10.0, 20.0], + ) + + +def _make_results(obs: GravityObservations) -> GSolveResults: + obs_input = obs.data.loc[:, ["site_id", "loop", "loop_tdelta"]].copy() + obs_input = obs_input.rename(columns={"loop_tdelta": "timedelta"}) + ref_input = pd.DataFrame( + {"reference_gravity": [1000.2, 1001.2]}, + index=pd.Index(["A", "B"], name="site_id"), + ) + + results = GSolveResults( + method=1, + use_loops=True, + calculate_calibration_factor=False, + percentile_clipping=95.0, + ) + results.set_inputs(obs_input, ref_input) + + site_grav = np.array([1000.3, 1001.3]) + residuals = np.array([0.01, -0.02, 0.01, -0.01, 0.0, 0.01]) + site_var = np.array([0.01, 0.02]) + drift = np.array([0.01, -0.01]) + baseline = np.array([0.0, 0.0]) + calibration_factor = None + mask = np.array([True, True, True, True, True, False]) + + results.set_solutions( + (site_grav, residuals, site_var, drift, baseline, calibration_factor, mask) + ) + return results + + +@dataclass +class _DummyAnomalies: + data: pd.DataFrame + params: dict + tcorr_params: dict | None = None + + +@dataclass +class _DummyTerrainCorrections: + data: pd.DataFrame + params: dict + + +class TestGSolveReportAssembly: + def test_init_requires_required_inputs(self) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + + with pytest.raises(ValueError, match="must be provided"): + GSolveReport(observations=None, sites=sites, results=results) # type: ignore[arg-type] + + def test_assembly_builds_site_obs_loop_tables(self) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + + report = GSolveReport(observations=obs, sites=sites, results=results) + + assert "n_obs_input" in report.site_data.columns + assert "n_obs_used" in report.site_data.columns + assert "in_loops" in report.site_data.columns + assert "included_in_solution" in report.obs_data.columns + assert "has_site_solution" in report.obs_data.columns + assert "n_obs_input" in report.loop_data.columns + assert "n_obs_used" in report.loop_data.columns + assert not any(c.endswith("_duplicate") for c in report.site_data.columns) + assert not any(c.endswith("_duplicate") for c in report.obs_data.columns) + + def test_params_sections_are_populated(self) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + + report = GSolveReport(observations=obs, sites=sites, results=results) + + assert "observations" in report.params + assert "solution" in report.params + + +class TestGSolveReportTerrainCorrectionConsistency: + def test_inconsistent_terrain_corrections_raise_with_anomaly_tcorr(self) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + + anomalies = _DummyAnomalies( + data=pd.DataFrame({"anomaly": [1.0, 2.0]}, index=pd.Index(["A", "B"])), + params={"method": "dummy"}, + tcorr_params={"tcorr_zone_1": {"min_dist": 0.0, "max_dist": 1000.0}}, + ) + terrain_bad = _DummyTerrainCorrections( + data=pd.DataFrame({"tcorr_zone_1": [0.1, 0.2]}, index=pd.Index(["A", "B"])), + params={"tcorr_zone_1": {"min_dist": 0.0, "max_dist": 900.0}}, + ) + + with pytest.raises(ValueError, match="terrain_corrections are inconsistent"): + GSolveReport( + observations=obs, + sites=sites, + results=results, + anomalies=anomalies, # type: ignore[arg-type] + terrain_corrections=terrain_bad, # type: ignore[arg-type] + ) + + def test_consistent_terrain_corrections_are_accepted(self) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + + tc_params = {"tcorr_zone_1": {"min_dist": 0.0, "max_dist": 1000.0}} + anomalies = _DummyAnomalies( + data=pd.DataFrame({"anomaly": [1.0, 2.0]}, index=pd.Index(["A", "B"])), + params={"method": "dummy"}, + tcorr_params=tc_params, + ) + terrain_ok = _DummyTerrainCorrections( + data=pd.DataFrame({"tcorr_zone_1": [0.1, 0.2]}, index=pd.Index(["A", "B"])), + params=tc_params, + ) + + report = GSolveReport( + observations=obs, + sites=sites, + results=results, + anomalies=anomalies, # type: ignore[arg-type] + terrain_corrections=terrain_ok, # type: ignore[arg-type] + ) + + assert report.terrain_correction_data is not None + assert "tcorr_zone_1" in report.terrain_correction_data.columns + + +class TestGSolveReportToExcel: + def test_to_excel_raises_if_workbook_exists_and_error(self, tmp_path: Path) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + report = GSolveReport(observations=obs, sites=sites, results=results) + + out_file = tmp_path / "report.xlsx" + out_file.touch() + + with pytest.raises(ValueError, match="already exists"): + report.to_excel(out_file, if_workbook_exists="error") + + def test_to_excel_append_raises_if_existing_worksheets( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + report = GSolveReport(observations=obs, sites=sites, results=results) + + out_file = tmp_path / "report.xlsx" + out_file.touch() + + monkeypatch.setattr( + reports_module, + "get_excel_worksheets", + lambda _fname: ["observations", "sites"], + ) + + with pytest.raises(ValueError, match="already exist"): + report.to_excel( + out_file, + if_workbook_exists="append", + if_sheet_exists="error", + ) + + def test_to_excel_writes_expected_sheets_without_terrain( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + report = GSolveReport(observations=obs, sites=sites, results=results) + + calls: list[dict[str, Any]] = [] + + def _fake_write_excel_worksheet(df, filename, sheet_name, **kwargs): + calls.append( + {"sheet_name": sheet_name, "kwargs": kwargs, "shape": df.shape} + ) + + monkeypatch.setattr( + reports_module, "write_excel_worksheet", _fake_write_excel_worksheet + ) + + out_file = tmp_path / "report.xlsx" + report.to_excel(out_file, if_workbook_exists="replace", if_sheet_exists="new") + + assert [c["sheet_name"] for c in calls] == [ + "observations", + "sites", + "loops", + "metadata", + ] + assert calls[0]["kwargs"]["if_workbook_exists"] == "replace" + assert calls[1]["kwargs"]["if_workbook_exists"] == "append" + assert calls[2]["kwargs"]["if_workbook_exists"] == "append" + assert calls[3]["kwargs"]["if_workbook_exists"] == "append" + + def test_to_excel_writes_terrain_sheet_when_present( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + obs = _make_observations() + sites = _make_sites() + results = _make_results(obs) + + terrain_ok = _DummyTerrainCorrections( + data=pd.DataFrame({"tcorr_zone_1": [0.1, 0.2]}, index=pd.Index(["A", "B"])), + params={"tcorr_zone_1": {"min_dist": 0.0, "max_dist": 1000.0}}, + ) + report = GSolveReport( + observations=obs, + sites=sites, + results=results, + terrain_corrections=terrain_ok, # type: ignore[arg-type] + ) + # Keep this test focused on worksheet-writing flow; in this test setup the + # dummy terrain parameters are plain dict values (not metadata-serializable). + report.params = { + k: v for k, v in report.params.items() if hasattr(v, "to_series") + } + + sheet_names: list[str] = [] + + def _fake_write_excel_worksheet(_df, _filename, sheet_name, **_kwargs): + sheet_names.append(sheet_name) + + monkeypatch.setattr( + reports_module, "write_excel_worksheet", _fake_write_excel_worksheet + ) + + out_file = tmp_path / "report.xlsx" + report.to_excel(out_file, if_workbook_exists="replace", if_sheet_exists="new") + + assert sheet_names == [ + "observations", + "sites", + "loops", + "terrain_corrections", + "metadata", + ]