Implement ObservableCollection.map, map_to_float, map_to_int, other helpful methods, supporting classes and tests

Author: FancyNeuron

src/spellbind/float_collections.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import operator
+from abc import ABC, abstractmethod
+from functools import cached_property
+from typing import Iterable, Callable, Any, TypeVar
+
+from typing_extensions import TypeIs, override
+
+from spellbind.float_values import FloatValue, FloatConstant
+from spellbind.observable_collections import ObservableCollection, ReducedValue, CombinedValue, ValueCollection, \
+    MappedObservableBag
+from spellbind.observable_sequences import ObservableList, TypedValueList, ValueSequence, UnboxedValueSequence, \
+    ObservableSequence
+from spellbind.values import Value
+
+
+_S = TypeVar("_S")
+
+
+class ObservableFloatCollection(ObservableCollection[float], ABC):
+    @property
+    def summed(self) -> FloatValue:
+        return self.reduce_to_float(add_reducer=operator.add, remove_reducer=operator.sub, initial=0.0)
+
+    @property
+    def multiplied(self) -> FloatValue:
+        return self.reduce_to_float(add_reducer=operator.mul, remove_reducer=operator.truediv, initial=1.0)
+
+
+class MappedToFloatBag(MappedObservableBag[float], ObservableFloatCollection):
+    pass
+
+
+class ObservableFloatSequence(ObservableSequence[float], ObservableFloatCollection, ABC):
+    pass
+
+
+class ObservableFloatList(ObservableList[float], ObservableFloatSequence):
+    pass
+
+
+class FloatValueCollection(ValueCollection[float], ABC):
+    @property
+    def summed(self) -> FloatValue:
+        return self.unboxed.reduce_to_float(add_reducer=operator.add, remove_reducer=operator.sub, initial=0.0)
+
+    @property
+    @abstractmethod
+    def unboxed(self) -> ObservableFloatCollection: ...
+
+
+class CombinedFloatValue(CombinedValue[float], FloatValue):
+    def __init__(self, collection: ObservableCollection[_S], combiner: Callable[[Iterable[_S]], float]) -> None:
+        super().__init__(collection=collection, combiner=combiner)
+
+
+class ReducedFloatValue(ReducedValue[float], FloatValue):
+    def __init__(self,
+                 collection: ObservableCollection[_S],
+                 add_reducer: Callable[[float, _S], float],
+                 remove_reducer: Callable[[float, _S], float],
+                 initial: float):
+        super().__init__(collection=collection,
+                         add_reducer=add_reducer,
+                         remove_reducer=remove_reducer,
+                         initial=initial)
+
+
+class UnboxedFloatValueSequence(UnboxedValueSequence[float], ObservableFloatSequence):
+    def __init__(self, sequence: FloatValueSequence) -> None:
+        super().__init__(sequence)
+
+
+class FloatValueSequence(ValueSequence[float], FloatValueCollection, ABC):
+    @cached_property
+    @override
+    def unboxed(self) -> ObservableFloatSequence:
+        return UnboxedFloatValueSequence(self)
+
+
+class FloatValueList(TypedValueList[float], FloatValueSequence):
+    def __init__(self, values: Iterable[float | Value[float]] | None = None):
+        def is_float(value: Any) -> TypeIs[float]:
+            return isinstance(value, float)
+        super().__init__(values, checker=is_float, constant_factory=FloatConstant.of)

src/spellbind/int_collections.py

@@ -7,8 +7,10 @@
 
 from typing_extensions import TypeIs, override
 
+from spellbind.float_values import FloatValue
 from spellbind.int_values import IntValue, IntConstant
-from spellbind.observable_collections import ObservableCollection, ReducedValue, CombinedValue, ValueCollection
+from spellbind.observable_collections import ObservableCollection, ReducedValue, CombinedValue, ValueCollection, \
+    MappedObservableBag
 from spellbind.observable_sequences import ObservableList, TypedValueList, ValueSequence, UnboxedValueSequence, \
     ObservableSequence
 from spellbind.values import Value
@@ -27,6 +29,10 @@ def multiplied(self) -> IntValue:
         return self.reduce_to_int(add_reducer=operator.mul, remove_reducer=operator.floordiv, initial=1)
 
 
+class MappedToIntBag(MappedObservableBag[int], ObservableIntCollection):
+    pass
+
+
 class ObservableIntSequence(ObservableSequence[int], ObservableIntCollection, ABC):
     pass
 
@@ -50,6 +56,11 @@ def __init__(self, collection: ObservableCollection[_S], combiner: Callable[[Ite
         super().__init__(collection=collection, combiner=combiner)
 
 
+class CombinedFloatValue(CombinedValue[float], FloatValue):
+    def __init__(self, collection: ObservableCollection[_S], combiner: Callable[[Iterable[_S]], float]) -> None:
+        super().__init__(collection=collection, combiner=combiner)
+
+
 class ReducedIntValue(ReducedValue[int], IntValue):
     def __init__(self,
                  collection: ObservableCollection[_S],

src/spellbind/observable_collections.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import functools
+import logging
 from abc import ABC, abstractmethod
-from typing import TypeVar, Generic, Collection, Callable, Iterable, Iterator, Any
+from typing import TypeVar, Generic, Collection, Callable, Iterable, Iterator, Any, TYPE_CHECKING
 
 from typing_extensions import override
 
@@ -11,16 +12,24 @@
 from spellbind.bool_values import BoolValue
 from spellbind.deriveds import Derived
 from spellbind.event import BiEvent, ValueEvent
+from spellbind.float_values import FloatValue
 from spellbind.int_values import IntValue, IntVariable
 from spellbind.observables import ValuesObservable, ValueObservable, Observer, ValueObserver, BiObserver, \
     Subscription
 from spellbind.str_values import StrValue
 from spellbind.values import Value, EMPTY_FROZEN_SET
 
+if TYPE_CHECKING:
+    from spellbind.float_collections import ObservableFloatCollection
+    from spellbind.int_collections import ObservableIntCollection
+
+
 _S = TypeVar("_S")
 _S_co = TypeVar("_S_co", covariant=True)
 _T = TypeVar("_T")
 
+_logger = logging.getLogger(__name__)
+
 
 class ObservableCollection(Collection[_S_co], Generic[_S_co], ABC):
     @property
@@ -56,6 +65,11 @@ def combine_to_int(self, combiner: Callable[[Iterable[_S_co]], int]) -> IntValue
 
         return CombinedIntValue(self, combiner=combiner)
 
+    def combine_to_float(self, combiner: Callable[[Iterable[_S_co]], float]) -> 'FloatValue':
+        from spellbind.float_collections import CombinedFloatValue
+
+        return CombinedFloatValue(self, combiner=combiner)
+
     def reduce(self,
                add_reducer: Callable[[_T, _S_co], _T],
                remove_reducer: Callable[[_T, _S_co], _T],
@@ -79,17 +93,41 @@ def reduce_to_str(self,
     def reduce_to_int(self,
                       add_reducer: Callable[[int, _S_co], int],
                       remove_reducer: Callable[[int, _S_co], int],
-                      initial: int) -> IntValue:
+                      initial: int = 0) -> IntValue:
         from spellbind.int_collections import ReducedIntValue
 
         return ReducedIntValue(self,
                                add_reducer=add_reducer,
                                remove_reducer=remove_reducer,
                                initial=initial)
 
-    def filter_to_bag(self, predicate: Callable[[_S_co], bool]) -> FilteredObservableBag[_S_co]:
+    def reduce_to_float(self,
+                        add_reducer: Callable[[float, _S_co], float],
+                        remove_reducer: Callable[[float, _S_co], float],
+                        initial: float = 0.) -> FloatValue:
+        from spellbind.float_collections import ReducedFloatValue
+
+        return ReducedFloatValue(self,
+                                 add_reducer=add_reducer,
+                                 remove_reducer=remove_reducer,
+                                 initial=initial)
+
+    def filter_to_bag(self, predicate: Callable[[_S_co], bool]) -> ObservableCollection[_S_co]:
         return FilteredObservableBag(self, predicate)
 
+    def map(self, transform: Callable[[_S_co], _T]) -> ObservableCollection[_T]:
+        return MappedObservableBag(self, transform)
+
+    def map_to_float(self, transform: Callable[[_S_co], float]) -> ObservableFloatCollection:
+        from spellbind.float_collections import MappedToFloatBag
+
+        return MappedToFloatBag(self, transform)
+
+    def map_to_int(self, transform: Callable[[_S_co], int]) -> ObservableIntCollection:
+        from spellbind.int_collections import MappedToIntBag
+
+        return MappedToIntBag(self, transform)
+
 
 class ReducedValue(Value[_S], Generic[_S]):
     def __init__(self,
@@ -156,18 +194,6 @@ def is_observed(self, by: Callable[..., Any] | None = None) -> bool:
 
 
 class ValueCollection(ObservableCollection[Value[_S]], Generic[_S], ABC):
-    @override
-    def reduce_to_int(self,
-                      add_reducer: Callable[[int, Value[_S]], int],
-                      remove_reducer: Callable[[int, Value[_S]], int],
-                      initial: int) -> IntValue:
-        from spellbind.int_collections import ReducedIntValue
-
-        return ReducedIntValue(self,
-                               add_reducer=add_reducer,
-                               remove_reducer=remove_reducer,
-                               initial=initial)
-
     def value_iterable(self) -> Iterable[_S]:
         return (value.value for value in self)
 
@@ -281,6 +307,52 @@ def _clear(self) -> None:
                 self._action_event(clear_action())
 
 
+class MappedObservableBag(_ObservableBagBase[_S], Generic[_S]):
+    def __init__(self, source: ObservableCollection[_T], transform: Callable[[_T], _S]) -> None:
+        super().__init__(tuple(transform(item) for item in source))
+        self._source = source
+        self._transform = transform
+
+        self._source.on_change.observe(self._on_source_action)
+
+    def _on_source_action(self, action: CollectionAction[Any]) -> None:
+        if isinstance(action, ClearAction):
+            self._clear()
+        elif isinstance(action, ReverseAction):
+            pass
+        elif isinstance(action, DeltasAction):
+            mapped_action = action.map(self._transform)
+            total_count = self._len_value.value
+            for delta in mapped_action.delta_actions:
+                if delta.is_add:
+                    self._item_counts[delta.value] = self._item_counts.get(delta.value, 0) + 1
+                    total_count += 1
+                else:
+                    count = self._item_counts.get(delta.value, 0)
+                    if count > 0:
+                        if count == 1:
+                            del self._item_counts[delta.value]
+                        else:
+                            self._item_counts[delta.value] = count - 1
+                        total_count -= 1
+                    else:
+                        _logger.warning(
+                            f"Attempted to remove {delta.value!r} from {self.__class__.__name__}, "
+                            f"but item not present. Source collection may be inconsistent with the mapped collection."
+                        )
+
+            if self._is_observed():
+                with self._len_value.set_delay_notify(total_count):
+                    self._action_event(mapped_action)
+                    self._deltas_event(mapped_action)
+            else:
+                self._len_value.value = total_count
+
+    @override
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({list(self)!r})"
+
+
 class FilteredObservableBag(_ObservableBagBase[_S], Generic[_S]):
     def __init__(self, source: ObservableCollection[_S], predicate: Callable[[_S], bool]) -> None:
         super().__init__(tuple(item for item in source if predicate(item)))
@@ -310,6 +382,11 @@ def _on_source_action(self, action: CollectionAction[_S]) -> None:
                             else:
                                 self._item_counts[delta.value] = count - 1
                             total_count -= 1
+                        else:
+                            _logger.warning(
+                                f"Attempted to remove {delta.value!r} from {self.__class__.__name__}, "
+                                f"but item not present. Source collection may be inconsistent with the filtered collection."
+                            )
 
                 if self._is_observed():
                     with self._len_value.set_delay_notify(total_count):

src/spellbind/observable_sequences.py

@@ -35,6 +35,7 @@ class ObservableSequence(Sequence[_S_co], ObservableCollection[_S_co], Generic[_
     def on_change(self) -> ValueObservable[AtIndicesDeltasAction[_S_co] | ClearAction[_S_co] | ReverseAction[_S_co] | ElementsChangedAction[_S_co]]: ...
 
     @abstractmethod
+    @override
     def map(self, transformer: Callable[[_S_co], _T]) -> ObservableSequence[_T]: ...
 
     @override
@@ -848,25 +849,25 @@ def __eq__(self, other: object) -> bool:
 
 
 class MappedIndexObservableSequence(IndexObservableSequenceBase[_S], Generic[_S]):
-    def __init__(self, mapped_from: IndexObservableSequence[_T], map_func: Callable[[_T], _S]) -> None:
-        super().__init__(map_func(item) for item in mapped_from)
-        self._mapped_from = mapped_from
-        self._map_func = map_func
+    def __init__(self, source: IndexObservableSequence[_T], transform: Callable[[_T], _S]) -> None:
+        super().__init__(transform(item) for item in source)
+        self._source = source
+        self._transform = transform
 
         def on_action(other_action: AtIndicesDeltasAction[_T] | ClearAction[_T] | ReverseAction[_T]) -> None:
             if isinstance(other_action, AtIndicesDeltasAction):
                 if isinstance(other_action, ExtendAction):
-                    self._extend((self._map_func(item) for item in other_action.items))
+                    self._extend((self._transform(item) for item in other_action.items))
                 else:
                     for delta in other_action.delta_actions:
                         if delta.is_add:
-                            value: _S = self._map_func(delta.value)
+                            value: _S = self._transform(delta.value)
                             self._values.insert(delta.index, value)
                         else:
                             del self._values[delta.index]
                     if self._is_observed():
                         with self._len_value.set_delay_notify(len(self._values)):
-                            action = other_action.map(self._map_func)
+                            action = other_action.map(self._transform)
                             self._action_event(action)
                             self._deltas_event(action)
                     else:
@@ -876,7 +877,7 @@ def on_action(other_action: AtIndicesDeltasAction[_T] | ClearAction[_T] | Revers
             elif isinstance(other_action, ReverseAction):
                 self._reverse()
 
-        mapped_from.on_change.observe(on_action)
+        source.on_change.observe(on_action)
 
     def _is_observed(self) -> bool:
         return self._action_event.is_observed() or self._deltas_event.is_observed()