Fix threshold subscription to prevent duplicate callbacks

Changes the ContinuousObservable threshold system from storing callbacks
directly to using a signal-based subscription pattern:

- Change `_thresholds` from dict to set (stores threshold values only)
- Update `add_threshold()` to check for existing subscriptions before
  subscribing to prevent duplicate callback invocations
- Separate concerns: thresholds define WHICH values to watch, observers
  define WHO to notify when crossed
- Add ValueError when attempting to add threshold to non-ContinuousObservable

This fixes an issue where registering multiple thresholds with the same
callback would cause it to be called multiple times per threshold crossing.

Author: EwoutH

mesa/experimental/mesa_signals/mesa_signal.py

@@ -476,7 +476,21 @@ def add_threshold(self, observable_name: str, threshold: float, callback: Callab
         """Convenience method for adding thresholds."""
         obs = getattr(type(self), observable_name)
         if isinstance(obs, ContinuousObservable):
-            obs._thresholds[threshold] = callback
+            obs._thresholds.add(threshold)
+
+            # Check if callback is already subscribed
+            existing_subscribers = self.subscribers.get(observable_name, {}).get(
+                "threshold_crossed", []
+            )
+            already_subscribed = any(
+                ref() == callback for ref in existing_subscribers if ref() is not None
+            )
+
+            # Only subscribe if not already subscribed
+            if not already_subscribed:
+                self.observe(observable_name, "threshold_crossed", callback)
+        else:
+            raise ValueError(f"{observable_name} is not a ContinuousObservable")
 
 
 class ContinuousObservable(Observable):
@@ -487,7 +501,7 @@ def __init__(self, initial_value: float, rate_func: Callable):
         super().__init__(fallback_value=initial_value)
         self.signal_types.add("threshold_crossed")
         self._rate_func = rate_func
-        self._thresholds = {}  # threshold_value -> callback
+        self._thresholds = set()
 
     def __set__(self, instance: HasObservables, value):
         """Set the value, ensuring we store a ContinuousState."""
@@ -624,13 +638,13 @@ def check_thresholds(self, old_value: float, new_value: float) -> list:
             List of (threshold_value, direction) tuples for crossed thresholds
         """
         crossed = []
-        for threshold_value in self.thresholds:
+        for threshold in self.thresholds:
             # Crossed upward
-            if old_value < threshold_value <= new_value:
-                crossed.append((threshold_value, "up"))
+            if old_value < threshold <= new_value:
+                crossed.append((threshold, "up"))
             # Crossed downward
-            elif new_value <= threshold_value < old_value:
-                crossed.append((threshold_value, "down"))
+            elif new_value <= threshold < old_value:
+                crossed.append((threshold, "down"))
         return crossed