variance adapter test unit

Author: KOVALW

src/calibrationtools/__init__.py

@@ -20,6 +20,7 @@
     AdaptIdentityVariance,
     AdaptMultivariateNormalVariance,
     AdaptNormalVariance,
+    AdaptUniformVariance,
     VarianceAdapter,
 )
 
@@ -42,4 +43,5 @@
     "AdaptNormalVariance",
     "AdaptMultivariateNormalVariance",
     "AdaptIdentityVariance",
+    "AdaptUniformVariance",
 ]

src/calibrationtools/variance_adapter.py

@@ -3,13 +3,12 @@
 
 import numpy as np
 
-from calibrationtools import PerturbationKernel
-
 from .particle_population import ParticlePopulation
 from .perturbation_kernel import (
     CompositePerturbationKernel,
     MultivariateNormalKernel,
     NormalKernel,
+    PerturbationKernel,
     UniformKernel,
 )
 

tests/test_particle_updater.py

@@ -1,49 +1,77 @@
 import pytest
+
 from calibrationtools import ParticlePopulation, _ParticleUpdater
 
+
 @pytest.fixture
 def particle_updater(seed_sequence, K, P, V):
     updater = _ParticleUpdater(K, P, V, seed_sequence)
     return updater
 
-def test_set_particle_population_normalizes_weights(particle_updater, particle_population):
+
+def test_set_particle_population_normalizes_weights(
+    particle_updater, particle_population
+):
     # Set the particle population in the updater
     particle_updater.set_particle_population(particle_population)
 
     # Check that the weights are normalized
-    assert particle_updater.particle_population.total_weight == pytest.approx(1.0)
-    
-    particle_population_unnormalized = ParticlePopulation(states=particle_population.particles, weights=[0.1, 0.1, 0.1])  # Not normalized
+    assert particle_updater.particle_population.total_weight == pytest.approx(
+        1.0
+    )
+
+    particle_population_unnormalized = ParticlePopulation(
+        states=particle_population.particles, weights=[0.1, 0.1, 0.1]
+    )  # Not normalized
     particle_updater.set_particle_population(particle_population_unnormalized)
-    
+
     # Check that the weights are normalized
-    assert particle_updater.particle_population.total_weight == pytest.approx(1.0)
-    assert particle_updater.particle_population.weights == pytest.approx([1/3, 1/3, 1/3])
+    assert particle_updater.particle_population.total_weight == pytest.approx(
+        1.0
+    )
+    assert particle_updater.particle_population.weights == pytest.approx(
+        [1 / 3, 1 / 3, 1 / 3]
+    )
+
 
 def test_sample_particle(particle_updater, particle_population):
     particle_updater.set_particle_population(particle_population)
     sampled_particle = particle_updater.sample_particle()
     assert sampled_particle in particle_population.particles
 
+
 def test_sample_perturbed_particle(particle_updater, particle_population):
     particle_updater.set_particle_population(particle_population)
     perturbed_particle = particle_updater.sample_perturbed_particle()
-    assert perturbed_particle not in particle_population.particles  # Perturbed particle should not be the same as any in the population
-    assert particle_updater.priors.probability_density(perturbed_particle) > 0  # Perturbed particle should have non-zero prior density
+    assert (
+        perturbed_particle not in particle_population.particles
+    )  # Perturbed particle should not be the same as any in the population
+    assert (
+        particle_updater.priors.probability_density(perturbed_particle) > 0
+    )  # Perturbed particle should have non-zero prior density
+
 
-def test_sample_perturbed_particle_max_attempts(particle_updater, particle_population):
+def test_sample_perturbed_particle_max_attempts(
+    particle_updater, particle_population
+):
     # Create a perturbation kernel that always produces invalid particles
     class InvalidPerturbationKernel:
         def perturb(self, current_particle, seed_sequence):
-            return {"p": -1.0, "seed": 0}  # Invalid particle outside the prior support
+            return {
+                "p": -1.0,
+                "seed": 0,
+            }  # Invalid particle outside the prior support
 
     particle_updater.perturbation_kernel = InvalidPerturbationKernel()
     particle_updater.set_particle_population(particle_population)
 
     with pytest.raises(RuntimeError):
         particle_updater.sample_perturbed_particle(max_attempts=5)
 
-def test_calculate_weight(particle_updater, particle_population, proposed_particle):
+
+def test_calculate_weight(
+    particle_updater, particle_population, proposed_particle
+):
     particle_updater.set_particle_population(particle_population)
     weight = particle_updater.calculate_weight(proposed_particle)
     assert weight >= 0  # Weights should be non-negative
@@ -52,15 +80,22 @@ def test_calculate_weight(particle_updater, particle_population, proposed_partic
     states = particle_population.particles
     weights = particle_population.weights
     transition_probs = [
-        particle_updater.perturbation_kernel.transition_probability(to_particle=proposed_particle, from_particle=p)
+        particle_updater.perturbation_kernel.transition_probability(
+            to_particle=proposed_particle, from_particle=p
+        )
         for p in states
     ]
-    weighted_probs = [w*p for w, p in zip(weights, transition_probs)]
+    weighted_probs = [w * p for w, p in zip(weights, transition_probs)]
 
-    expected_weight = particle_updater.priors.probability_density(proposed_particle) / sum(weighted_probs)
+    expected_weight = particle_updater.priors.probability_density(
+        proposed_particle
+    ) / sum(weighted_probs)
     assert weight == pytest.approx(expected_weight)
 
-def test_calculate_weight_zero_prob_perturbation(particle_updater, particle_population):
+
+def test_calculate_weight_zero_prob_perturbation(
+    particle_updater, particle_population
+):
     # Create a perturbation kernel that always produces zero transition probability
     class ZeroTransitionPerturbationKernel:
         def transition_probability(self, to_particle, from_particle):
@@ -69,6 +104,11 @@ def transition_probability(self, to_particle, from_particle):
     particle_updater.perturbation_kernel = ZeroTransitionPerturbationKernel()
     particle_updater.set_particle_population(particle_population)
 
-    proposed_particle = {"p": 0.5, "seed": 1}  # A valid particle with non-zero prior density
+    proposed_particle = {
+        "p": 0.5,
+        "seed": 1,
+    }  # A valid particle with non-zero prior density
     weight = particle_updater.calculate_weight(proposed_particle)
-    assert weight == 0.0  # Weight should be zero due to zero transition probabilities
+    assert (
+        weight == 0.0
+    )  # Weight should be zero due to zero transition probabilities

tests/test_variance_adapter.py

@@ -0,0 +1,73 @@
+import math
+import numpy as np
+import pytest
+from unittest.mock import MagicMock
+from calibrationtools import (
+    AdaptIdentityVariance,
+    AdaptMultivariateNormalVariance,
+    AdaptNormalVariance,
+    AdaptUniformVariance,
+    IndependentKernels,
+    MultivariateNormalKernel,
+    NormalKernel,
+    ParticlePopulation,
+    UniformKernel,
+)
+
+
+def test_adapt_identity_variance():
+    adapter = AdaptIdentityVariance()
+    population = MagicMock(spec=ParticlePopulation)
+    kernel = MagicMock()
+    adapter.adapt(population, kernel)
+    # No changes expected, just ensure no exceptions are raised
+
+
+def test_adapt_normal_variance():
+    adapter = AdaptNormalVariance()
+    population = ParticlePopulation(
+        states=[{'x': 1.0}, {'x': 2.0}, {'x': 3.0}]
+    )
+    kernel = NormalKernel(param='x', std_dev=1.0)
+    adapter.adapt(population, kernel)
+    expected_std_dev = math.sqrt(np.var([1.0, 2.0, 3.0]) * 2.0)
+    assert kernel.std_dev == pytest.approx(expected_std_dev)
+
+
+def test_adapt_uniform_variance():
+    adapter = AdaptUniformVariance()
+    population = ParticlePopulation(
+        states=[{'x': 1.0}, {'x': 2.0}, {'x': 3.0}]
+    )
+    kernel = UniformKernel(param='x', width=1.0)
+    adapter.adapt(population, kernel)
+    expected_width = math.sqrt(np.var([1.0, 2.0, 3.0]) * 2.0) * 2.0
+    assert kernel.width == pytest.approx(expected_width)
+
+
+def test_adapt_multivariate_normal_variance():
+    adapter = AdaptMultivariateNormalVariance()
+    population = ParticlePopulation(
+        states=[
+            {'x': 1.0, 'y': 2.0},
+            {'x': 2.0, 'y': 3.0},
+            {'x': 3.0, 'y': 4.0},
+        ]
+    )
+    kernel = MultivariateNormalKernel(params=['x','y'], cov_matrix=np.eye(2))
+    adapter.adapt(population, kernel)
+    states_matrix = np.array([[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]])
+    expected_cov_matrix = np.cov(states_matrix.T) * 2.0
+    assert np.allclose(kernel.cov_matrix, expected_cov_matrix)
+
+
+def test_adapt_composite_kernel():
+    adapter = AdaptNormalVariance()
+    population = ParticlePopulation(
+        states=[{'x': 1.0}, {'x': 2.0}, {'x': 3.0}]
+    )
+    normal_kernel = NormalKernel(param='x', std_dev=1.0)
+    composite_kernel = IndependentKernels(kernels=[normal_kernel])
+    adapter.adapt(population, composite_kernel)
+    expected_std_dev = math.sqrt(np.var([1.0, 2.0, 3.0]) * 2.0)
+    assert normal_kernel.std_dev == pytest.approx(expected_std_dev)