135 lines
3.5 KiB
Python
135 lines
3.5 KiB
Python
"""Watchable variables"""
|
||
|
||
from typing import Callable, Dict, Generic, Iterable, Optional, Set, TypeVar
|
||
|
||
|
||
Thunk = Callable[[], None]
|
||
T = TypeVar('T')
|
||
CbOldNew = Callable[[T, T], None]
|
||
CbValue = Callable[[T], None]
|
||
|
||
|
||
class Watchable(Generic[T]):
|
||
"""A non-seamless proxy to watch a variable’s value"""
|
||
|
||
def __init__(self, value: T):
|
||
self._cbs: Set[CbOldNew[T]] = set()
|
||
self._cbs_by_target: Dict[T, Set[CbOldNew[T]]] = {}
|
||
self.value = value
|
||
|
||
def get(self) -> T:
|
||
"""Get the current value of the variable"""
|
||
|
||
return self.value
|
||
|
||
def set(self, new_val: T) -> None:
|
||
"""Set the variable to a new value"""
|
||
|
||
old_val = self.value
|
||
self.value = new_val
|
||
|
||
if new_val != old_val:
|
||
for func in self._cbs:
|
||
func(old_val, new_val)
|
||
|
||
for target_func in self._cbs_by_target.get(new_val, []):
|
||
target_func(old_val, new_val)
|
||
|
||
def on_change(self, func: CbOldNew[T]) -> Thunk:
|
||
"""Add a callback to be called when the variable changes"""
|
||
|
||
self._cbs.add(func)
|
||
|
||
def del_cb() -> None:
|
||
self._cbs.remove(func)
|
||
|
||
return del_cb
|
||
|
||
def on_change_to(self, target: T) -> Callable[[CbOldNew[T]], Thunk]:
|
||
"""Add a callback to be called when the variable is set to a specific value"""
|
||
|
||
def decorator(func: CbOldNew[T]) -> Thunk:
|
||
self._cbs_by_target[target].add(func)
|
||
|
||
def del_cb() -> None:
|
||
self._cbs_by_target[target].remove(func)
|
||
|
||
return del_cb
|
||
|
||
return decorator
|
||
|
||
|
||
class WatchableSet(Set[T]):
|
||
"""A set that can be watched for changes"""
|
||
|
||
def __init__(self, iterable: Optional[Iterable[T]] = None) -> None:
|
||
if iterable is None:
|
||
super().__init__()
|
||
else:
|
||
super().__init__(iterable)
|
||
|
||
self._add_cbs: Set[CbValue[T]] = set()
|
||
self._remove_cbs: Set[CbValue[T]] = set()
|
||
self._change_cbs: Set[Thunk] = set()
|
||
|
||
def add(self, value: T) -> None:
|
||
had = value in self
|
||
super().add(value)
|
||
|
||
if not had:
|
||
for func in self._add_cbs:
|
||
func(value)
|
||
|
||
for change_func in self._change_cbs:
|
||
change_func()
|
||
|
||
def remove(self, value: T) -> None:
|
||
had = value in self
|
||
super().remove(value)
|
||
|
||
if had:
|
||
for func in self._remove_cbs:
|
||
func(value)
|
||
|
||
for change_func in self._change_cbs:
|
||
change_func()
|
||
|
||
def clear(self) -> None:
|
||
for value in super().copy():
|
||
super().remove(value)
|
||
for func in self._remove_cbs:
|
||
func(value)
|
||
|
||
for change_func in self._change_cbs:
|
||
change_func()
|
||
|
||
def on_add(self, func: CbValue[T]) -> Thunk:
|
||
"""Add a callback function to be called when an item gets added to the set"""
|
||
|
||
self._add_cbs.add(func)
|
||
|
||
def del_cb() -> None:
|
||
self._add_cbs.remove(func)
|
||
|
||
return del_cb
|
||
|
||
def on_remove(self, func: CbValue[T]) -> Thunk:
|
||
"""Add a callback function to be called when an item gets removed from the set"""
|
||
|
||
self._remove_cbs.add(func)
|
||
|
||
def del_cb() -> None:
|
||
self._remove_cbs.remove(func)
|
||
|
||
return del_cb
|
||
|
||
def on_change(self, func: Thunk) -> Thunk:
|
||
"""Add a callback function to be called when the set changes"""
|
||
|
||
self._change_cbs.add(func)
|
||
|
||
def del_cb() -> None:
|
||
self._change_cbs.remove(func)
|
||
|
||
return del_cb
|