Merge pull request #1585 from FlorianPfaff/copilot/add-filters-and-tests

Add tests for AbstractDummyFilter and HypersphericalDummyFilter

Author: FlorianPfaff

pyrecest/filters/__init__.py

@@ -1,18 +1,22 @@
+from .abstract_dummy_filter import AbstractDummyFilter
 from .abstract_axial_filter import AbstractAxialFilter
 from .abstract_filter import AbstractFilter
 from .abstract_particle_filter import AbstractParticleFilter
 from .euclidean_particle_filter import EuclideanParticleFilter
+from .hyperspherical_dummy_filter import HypersphericalDummyFilter
 from .hypertoroidal_particle_filter import HypertoroidalParticleFilter
 from .kalman_filter import KalmanFilter
 from .manifold_mixins import EuclideanFilterMixin, HypertoroidalFilterMixin
 
 __all__ = [
+    "AbstractDummyFilter",
     "AbstractAxialFilter",
     "AbstractFilter",
     "EuclideanFilterMixin",
     "HypertoroidalFilterMixin",
     "AbstractParticleFilter",
     "HypertoroidalParticleFilter",
+    "HypersphericalDummyFilter",
     "KalmanFilter",
     "EuclideanParticleFilter",
 ]

pyrecest/filters/abstract_dummy_filter.py

@@ -0,0 +1,49 @@
+from .abstract_filter import AbstractFilter
+
+
+class AbstractDummyFilter(AbstractFilter):
+    """Abstract dummy filter that does nothing on predictions and updates.
+
+    Subclasses should call super().__init__ with the initial distribution.
+    """
+
+    def __init__(self, initial_filter_state):
+        AbstractFilter.__init__(self, initial_filter_state)
+
+    @property
+    def dist(self):
+        return self._filter_state
+
+    @property
+    def filter_state(self):
+        return self._filter_state
+
+    @filter_state.setter
+    def filter_state(self, new_state):
+        # Do nothing - the dummy filter state is fixed at initialization
+        pass
+
+    def set_state(self, dist):
+        assert dist.dim == self.dim
+        # Do nothing
+
+    def predict_identity(self, noise_distribution):
+        pass
+
+    def predict_nonlinear(self, f, *args, **kwargs):
+        pass
+
+    def predict_nonlinear_via_transition_density(self, transition_density, *args):
+        pass
+
+    def update_identity(self, noise_distribution, measurement):
+        pass
+
+    def update_nonlinear(self, likelihood, measurement=None):
+        pass
+
+    def get_estimate(self):
+        return self.dist
+
+    def get_point_estimate(self):
+        return self.dist.sample(1)[0]

pyrecest/filters/hyperspherical_dummy_filter.py

@@ -0,0 +1,27 @@
+from pyrecest.distributions.hypersphere_subset.hyperspherical_uniform_distribution import (
+    HypersphericalUniformDistribution,
+)
+
+from .abstract_dummy_filter import AbstractDummyFilter
+from .manifold_mixins import HypersphericalFilterMixin
+
+
+class HypersphericalDummyFilter(AbstractDummyFilter, HypersphericalFilterMixin):
+    """Hyperspherical dummy filter initialized with a uniform distribution.
+
+    This filter does nothing on predictions and updates, always returning
+    samples from the initial uniform distribution as point estimates.
+    """
+
+    def __init__(self, dim):
+        """Initialize HypersphericalDummyFilter.
+
+        Parameters:
+            dim (int >= 2): Manifold dimension of the hypersphere (e.g. 2 for S^2).
+        """
+        assert dim >= 2, "dim must be at least 2"
+        HypersphericalFilterMixin.__init__(self)
+        AbstractDummyFilter.__init__(self, HypersphericalUniformDistribution(dim))
+
+    def get_point_estimate(self):
+        return AbstractDummyFilter.get_point_estimate(self)

pyrecest/tests/filters/test_hyperspherical_dummy_filter.py

@@ -0,0 +1,85 @@
+import unittest
+
+import numpy.testing as npt
+
+# pylint: disable=no-name-in-module,no-member
+from pyrecest.backend import array, linalg
+from pyrecest.distributions.hypersphere_subset.hyperspherical_uniform_distribution import (
+    HypersphericalUniformDistribution,
+)
+from pyrecest.distributions import VonMisesFisherDistribution
+from pyrecest.filters.hyperspherical_dummy_filter import HypersphericalDummyFilter
+
+
+class HypersphericalDummyFilterTest(unittest.TestCase):
+    def setUp(self):
+        self.filter_s2 = HypersphericalDummyFilter(2)
+        self.filter_s3 = HypersphericalDummyFilter(3)
+
+    def test_dim_s2(self):
+        self.assertEqual(self.filter_s2.dim, 2)
+
+    def test_dim_s3(self):
+        self.assertEqual(self.filter_s3.dim, 3)
+
+    def test_assert_dim_too_small(self):
+        with self.assertRaises(AssertionError):
+            HypersphericalDummyFilter(1)
+
+    def test_filter_state_is_uniform(self):
+        self.assertIsInstance(
+            self.filter_s2.filter_state, HypersphericalUniformDistribution
+        )
+
+    def test_get_point_estimate_unit_norm_s2(self):
+        est = self.filter_s2.get_point_estimate()
+        self.assertEqual(est.shape, (3,))
+        npt.assert_allclose(linalg.norm(est), 1.0, atol=1e-10)
+
+    def test_get_point_estimate_unit_norm_s3(self):
+        est = self.filter_s3.get_point_estimate()
+        self.assertEqual(est.shape, (4,))
+        npt.assert_allclose(linalg.norm(est), 1.0, atol=1e-10)
+
+    def test_predict_identity_is_noop(self):
+        noise = VonMisesFisherDistribution(array([0.0, 0.0, 1.0]), 1.0)
+        state_before = self.filter_s2.filter_state
+        self.filter_s2.predict_identity(noise)
+        self.assertIs(self.filter_s2.filter_state, state_before)
+
+    def test_predict_nonlinear_is_noop(self):
+        state_before = self.filter_s2.filter_state
+        self.filter_s2.predict_nonlinear(lambda x: x)
+        self.assertIs(self.filter_s2.filter_state, state_before)
+
+    def test_update_identity_is_noop(self):
+        noise = VonMisesFisherDistribution(array([0.0, 0.0, 1.0]), 1.0)
+        measurement = array([0.0, 0.0, 1.0])
+        state_before = self.filter_s2.filter_state
+        self.filter_s2.update_identity(noise, measurement)
+        self.assertIs(self.filter_s2.filter_state, state_before)
+
+    def test_update_nonlinear_is_noop(self):
+        state_before = self.filter_s2.filter_state
+        self.filter_s2.update_nonlinear(lambda z, x: x)
+        self.assertIs(self.filter_s2.filter_state, state_before)
+
+    def test_filter_state_setter_is_noop(self):
+        original_state = self.filter_s2.filter_state
+        new_dist = HypersphericalUniformDistribution(2)
+        self.filter_s2.filter_state = new_dist
+        self.assertIs(self.filter_s2.filter_state, original_state)
+
+    def test_set_state_is_noop(self):
+        original_state = self.filter_s2.filter_state
+        new_dist = HypersphericalUniformDistribution(2)
+        self.filter_s2.set_state(new_dist)
+        self.assertIs(self.filter_s2.filter_state, original_state)
+
+    def test_get_estimate_returns_distribution(self):
+        est = self.filter_s2.get_estimate()
+        self.assertIsInstance(est, HypersphericalUniformDistribution)
+
+
+if __name__ == "__main__":
+    unittest.main()