Test set x deprecation

sbi/inference/posteriors/ensemble_posterior.py

@@ -479,14 +479,20 @@ def allow_iid_x(self) -> bool:
         )
 
     def set_x(self, x_o: Optional[Tensor]):
-        """Check the shape of the observed data and, if valid, set it."""
-        if x_o is not None:
-            x_o = process_x(x_o).to(  # type: ignore
-                self.device
-            )
-        self._x_o = x_o
-        for comp_potential in self.potential_fns:
-            comp_potential.set_x(x_o)
+        """Check the shape of the observed data and, if valid, set it.
+
+        DEPRECATED: Use bind() instead. This method delegates to bind() internally.
+        """
+        import warnings
+
+        warnings.warn(
+            "set_x() is deprecated, use bind() instead",
+            FutureWarning,
+            stacklevel=2,
+        )
+        bound = self.bind(x_o)
+        self._x_o = bound._x_o
+        self.potential_fns = bound.potential_fns
 
     def bind(
         self,

sbi/inference/posteriors/vi_posterior.py

@@ -880,6 +880,10 @@ def train(
                 break
         # Training finished:
         self._trained_on = x
+
+        # Bind potential_fn to the trained x so it can be used for sampling/evaluation
+        self.potential_fn = self.potential_fn.bind(x)
+
         if self._mode == "amortized":
             warnings.warn(
                 "Switching from amortized to single-x mode. "

sbi/inference/potentials/base_potential.py

@@ -30,7 +30,10 @@ def __init__(
         """
         self.device = device
         self.prior = prior
-        self.set_x(x_o)
+        if x_o is not None:
+            x_o = process_x(x_o).to(self.device)
+        self._x_o = x_o
+        self._x_is_iid = True
 
     @abstractmethod
     def __call__(self, theta: Tensor, track_gradients: bool = True) -> Tensor:
@@ -53,11 +56,20 @@ def x_is_iid(self) -> bool:
             )
 
     def set_x(self, x_o: Optional[Tensor], x_is_iid: Optional[bool] = True):
-        """Check the shape of the observed data and, if valid, set it."""
-        if x_o is not None:
-            x_o = process_x(x_o).to(self.device)
-        self._x_o = x_o
-        self._x_is_iid = x_is_iid
+        """Check the shape of the observed data and, if valid, set it.
+
+        DEPRECATED: Use bind() instead. This method delegates to bind() internally.
+        """
+        import warnings
+
+        warnings.warn(
+            "set_x() is deprecated, use bind() instead",
+            FutureWarning,
+            stacklevel=2,
+        )
+        bound = self.bind(x_o, x_is_iid=x_is_iid)
+        self._x_o = bound._x_o
+        self._x_is_iid = bound._x_is_iid
 
     @property
     def x_o(self) -> Tensor:

sbi/inference/potentials/likelihood_based_potential.py

@@ -96,13 +96,17 @@ def to(self, device: Union[str, torch.device]) -> "LikelihoodBasedPotential":
 
     def bind(self, x_o: Tensor, x_is_iid: bool = True) -> "LikelihoodBasedPotential":
         """Create new potential with x bound, without mutable state."""
+        from sbi.utils.user_input_checks import process_x
+
         bound = LikelihoodBasedPotential(
             likelihood_estimator=self.likelihood_estimator,
             prior=self.prior,
             x_o=None,
             device=self.device,
         )
-        bound.set_x(x_o, x_is_iid=x_is_iid)
+        x_o = process_x(x_o).to(self.device)
+        bound._x_o = x_o
+        bound._x_is_iid = x_is_iid
         return bound
 
     def __call__(self, theta: Tensor, track_gradients: bool = True) -> Tensor:

sbi/inference/potentials/posterior_based_potential.py

@@ -102,20 +102,35 @@ def to(self, device: Union[str, torch.device]) -> "PosteriorBasedPotential":
 
     def bind(self, x_o: Tensor, x_is_iid: bool = False) -> "PosteriorBasedPotential":
         """Create new potential with x bound, without mutable state."""
+        from sbi.utils.user_input_checks import process_x
+
         bound = PosteriorBasedPotential(
             posterior_estimator=self.posterior_estimator,
             prior=self.prior,
             x_o=None,
             device=self.device,
         )
-        bound.set_x(x_o, x_is_iid=x_is_iid)
+        x_o = process_x(x_o).to(self.device)
+        bound._x_o = x_o
+        bound._x_is_iid = x_is_iid
         return bound
 
     def set_x(self, x_o: Optional[Tensor], x_is_iid: Optional[bool] = False):
         """
         Check the shape of the observed data and, if valid, set it.
+
+        DEPRECATED: Use bind() instead. This method delegates to bind() internally.
         """
-        super().set_x(x_o, x_is_iid=x_is_iid)
+        import warnings
+
+        warnings.warn(
+            "set_x() is deprecated, use bind() instead",
+            FutureWarning,
+            stacklevel=2,
+        )
+        bound = self.bind(x_o, x_is_iid=x_is_iid)
+        self._x_o = bound._x_o
+        self._x_is_iid = bound._x_is_iid
 
     def __call__(self, theta: Tensor, track_gradients: bool = True) -> Tensor:
         r"""Returns the potential for posterior-based methods.

sbi/inference/potentials/ratio_based_potential.py

@@ -84,13 +84,17 @@ def to(self, device: Union[str, torch.device]) -> "RatioBasedPotential":
 
     def bind(self, x_o: Tensor, x_is_iid: bool = True) -> "RatioBasedPotential":
         """Create new potential with x bound, without mutable state."""
+        from sbi.utils.user_input_checks import process_x
+
         bound = RatioBasedPotential(
             ratio_estimator=self.ratio_estimator,
             prior=self.prior,
             x_o=None,
             device=self.device,
         )
-        bound.set_x(x_o, x_is_iid=x_is_iid)
+        x_o = process_x(x_o).to(self.device)
+        bound._x_o = x_o
+        bound._x_is_iid = x_is_iid
         return bound
 
     def __call__(self, theta: Tensor, track_gradients: bool = True) -> Tensor:

sbi/inference/potentials/vector_field_potential.py

@@ -61,6 +61,8 @@ def __init__(
         self.vector_field_estimator.eval()
         self.iid_method = iid_method
         self.iid_params = iid_params
+        self.guidance_method: Optional[str] = None
+        self.guidance_params: Optional[Dict[str, Any]] = None
         self.neural_ode_backend = neural_ode_backend
 
         neural_ode_kwargs = neural_ode_kwargs or {}
@@ -110,6 +112,8 @@ def set_x(
 
         Rebuilds the continuous normalizing flow if the observed data is set.
 
+        DEPRECATED: Use bind() instead. This method delegates to bind() internally.
+
         Args:
             x_o: The observed data.
             x_is_iid: Whether the observed data is IID (if batch_dim>1).
@@ -119,15 +123,32 @@ def set_x(
                 `IIDScoreFunction`.
             ode_kwargs: Additional keyword arguments for the neural ODE.
         """
-        super().set_x(x_o, x_is_iid)
-        self.iid_method = iid_method or self.iid_method
-        self.iid_params = iid_params
-        self.guidance_method = guidance_method
-        self.guidance_params = guidance_params
-        if not x_is_iid and (self._x_o is not None):
-            self.flow = self.rebuild_flow(**ode_kwargs)
-        elif self._x_o is not None:
-            self.flows = self.rebuild_flows_for_batch(**ode_kwargs)
+        import warnings
+
+        warnings.warn(
+            "set_x() is deprecated, use bind() instead",
+            FutureWarning,
+            stacklevel=2,
+        )
+        bound = self.bind(
+            x_o,
+            x_is_iid=x_is_iid,
+            iid_method=iid_method,
+            iid_params=iid_params,
+            guidance_method=guidance_method,
+            guidance_params=guidance_params,
+            **ode_kwargs,
+        )
+        self._x_o = bound._x_o
+        self._x_is_iid = bound._x_is_iid
+        self.iid_method = bound.iid_method
+        self.iid_params = bound.iid_params
+        self.guidance_method = bound.guidance_method
+        self.guidance_params = bound.guidance_params
+        if not x_is_iid and (bound._x_o is not None):
+            self.flow = bound.flow
+        elif bound._x_o is not None:
+            self.flows = bound.flows
 
     def bind(
         self,
@@ -140,27 +161,31 @@ def bind(
         **ode_kwargs,
     ) -> "VectorFieldBasedPotential":
         """Create new potential with x bound, without mutable state."""
+        from sbi.utils.user_input_checks import process_x
+
         bound = VectorFieldBasedPotential(
             vector_field_estimator=self.vector_field_estimator,
             prior=self.prior,
             x_o=None,
             device=self.device,
-            iid_method=self.iid_method,
-            iid_params=self.iid_params,
+            iid_method=iid_method or self.iid_method,
+            iid_params=iid_params if iid_params is not None else self.iid_params,
             neural_ode_backend=self.neural_ode_backend,
         )
         bound.neural_ode.params.update(self.neural_ode.params)
-        # Transfer learned neural ODE parameters (e.g., from training) to the bound
-        # potential. This preserves the trained flow model when conditioning on new x.
-        bound.set_x(
-            x_o,
-            x_is_iid=x_is_iid,
-            iid_method=iid_method,
-            iid_params=iid_params,
-            guidance_method=guidance_method,
-            guidance_params=guidance_params,
-            **ode_kwargs,
+        x_o = process_x(x_o).to(self.device)
+        bound._x_o = x_o
+        bound._x_is_iid = x_is_iid
+        bound.guidance_method = (
+            guidance_method if guidance_method is not None else self.guidance_method
+        )
+        bound.guidance_params = (
+            guidance_params if guidance_params is not None else self.guidance_params
         )
+        if not x_is_iid and (bound._x_o is not None):
+            bound.flow = bound.rebuild_flow(**ode_kwargs)
+        elif bound._x_o is not None:
+            bound.flows = bound.rebuild_flows_for_batch(**ode_kwargs)
         return bound
 
     def __call__(

sbi/samplers/vi/vi_divergence_optimizers.py

@@ -479,7 +479,7 @@ def generate_elbo_particles(
             samples = self.q.rsample(torch.Size((num_samples,)))
             log_q = self.q.log_prob(samples)
 
-        self.potential_fn.set_x(x_o)
+        self.potential_fn = self.potential_fn.bind(x_o)
         log_potential = self.potential_fn(samples)
         elbo = log_potential - log_q
         return elbo
@@ -638,7 +638,7 @@ def _loss_q_proposal(self, x_o: Tensor) -> Tuple[Tensor, Tensor]:
         if hasattr(self.q, "clear_cache"):
             self.q.clear_cache()
         logq = self.q.log_prob(samples)
-        self.potential_fn.set_x(x_o)
+        self.potential_fn = self.potential_fn.bind(x_o)
         logp = self.potential_fn(samples)
         with torch.no_grad():
             logweights = logp - logq

sbi/utils/conditional_density_utils.py

@@ -15,7 +15,6 @@
     MixtureDensityEstimator,
 )
 from sbi.utils.torchutils import ensure_theta_batched
-from sbi.utils.user_input_checks import process_x
 
 
 def compute_corrcoeff(probs: Tensor, limits: Tensor):
@@ -418,11 +417,20 @@ def x_is_iid(self) -> bool:
             )
 
     def set_x(self, x_o: Optional[Tensor], x_is_iid: Optional[bool] = True):
-        """Check the shape of the observed data and, if valid, set it."""
-        if x_o is not None:
-            x_o = process_x(x_o).to(self.device)
-        self._x_is_iid = x_is_iid
-        self.potential_fn = self.potential_fn.bind(x_o, x_is_iid=x_is_iid)
+        """Check the shape of the observed data and, if valid, set it.
+
+        DEPRECATED: Use bind() instead. This method delegates to bind() internally.
+        """
+        import warnings
+
+        warnings.warn(
+            "set_x() is deprecated, use bind() instead",
+            FutureWarning,
+            stacklevel=2,
+        )
+        bound = self.bind(x_o, x_is_iid=x_is_iid)
+        self._x_is_iid = bound._x_is_iid
+        self.potential_fn = bound.potential_fn
 
     def bind(self, x_o: Tensor, x_is_iid: bool = True) -> "ConditionedPotential":
         """Create new potential with x bound, without mutable state."""

sbi/utils/torchutils.py

@@ -677,13 +677,14 @@ def _base_recursor(
         _active: Object identities on the active recursion path. Used internally
             to avoid infinite recursion in cyclic object graphs.
     """
-    if _active is None:
-        _active = set()
+    _active_holder: Optional[set[int]] = _active
+    if _active_holder is None:
+        _active_holder = set()
 
     obj_id = id(obj)
-    if obj_id in _active:
+    if obj_id in _active_holder:
         return
-    _active.add(obj_id)
+    _active_holder.add(obj_id)
 
     if isinstance(obj, Module) and check(obj):
         action(obj)
@@ -698,7 +699,7 @@ def _base_recursor(
                     key=k,
                     check=check,
                     action=action,
-                    _active=_active,
+                    _active=_active_holder,
                 )
     elif isinstance(obj, type):
         # Skip class/type objects to avoid modifying immutable C extension types
@@ -715,20 +716,20 @@ def _base_recursor(
                     key=k,
                     check=check,
                     action=action,
-                    _active=_active,
+                    _active=_active_holder,
                 )
     elif isinstance(obj, (List, Tuple, Generator)):
         new_obj = []
         for o in obj:
             if check(o):
                 new_obj.append(action(o))
             else:
-                _base_recursor(o, check=check, action=action, _active=_active)
+                _base_recursor(o, check=check, action=action, _active=_active_holder)
                 new_obj.append(o)
         if parent is not None and key is not None:
             setattr(parent, key, type(obj)(new_obj))  # type: ignore
 
-    _active.remove(obj_id)
+    _active_holder.remove(obj_id)
 
 
 def move_all_tensor_to_device(obj: object, device: Union[str, torch.device]) -> None: