import inspect
import sys
import traceback
from collections.abc import Callable
from contextlib import contextmanager, suppress
from typing import Any, Generic, TypeVar, cast
from weakref import WeakKeyDictionary, ref
from typing_extensions import Self, overload, override
P = TypeVar("P")
NoArgListener = Callable[[], None]
InstanceListener = Callable[[Any], None]
InstanceValueListener = Callable[[Any, Any], None]
InstanceNewOldListener = Callable[[Any, Any, Any], None]
AnyListener = NoArgListener | InstanceListener | InstanceValueListener | InstanceNewOldListener
class _Obs(Generic[P]):
"""
Internal holder for Property value and change listeners
"""
__slots__ = ("value", "_listeners")
def __init__(self, value: P):
self.value = value
# This will keep any added listener even if it is not referenced anymore
# and would be garbage collected
self._listeners: dict[AnyListener, InstanceNewOldListener] = dict()
def add(
self,
callback: AnyListener,
):
"""Add a callback to the list of listeners"""
self._listeners[callback] = _Obs._normalize_callback(callback)
def remove(self, callback):
"""Remove a callback from the list of listeners"""
if callback in self._listeners:
del self._listeners[callback]
@property
def listeners(self) -> list[InstanceNewOldListener]:
return list(self._listeners.values())
@staticmethod
def _normalize_callback(callback) -> InstanceNewOldListener:
"""Normalizes the callback so every callback can be invoked with the same signature."""
signature = inspect.signature(callback)
with suppress(TypeError):
signature.bind(1, 1)
return lambda instance, new, old: callback(instance, new)
with suppress(TypeError):
signature.bind(1, 1, 1)
return lambda instance, new, old: callback(instance, new, old)
with suppress(TypeError):
signature.bind(1)
return lambda instance, new, old: callback(instance)
with suppress(TypeError):
signature.bind()
return lambda instance, new, old: callback()
raise TypeError("Callback is not callable")
[docs]
class Property(Generic[P]):
"""An observable property which triggers observers when changed.
.. code-block:: python
def log_change(instance, value):
print("Something changed")
class MyObject:
name = Property()
my_obj = MyObject()
bind(my_obj, "name", log_change)
unbind(my_obj, "name", log_change)
my_obj.name = "Hans"
# > Something changed
Properties provide a less verbose way to implement the observer pattern in comparison to
using the `property` decorator.
Args:
default: Default value which is returned, if no value set before
default_factory: A callable which returns the default value.
Will be called with the property and the instance
"""
__slots__ = ("name", "default_factory", "obs")
name: str
"""Attribute name of the property"""
default_factory: Callable[[Any, Any], P]
"""Default factory to create the initial value"""
obs: WeakKeyDictionary[Any, _Obs]
"""Weak dictionary to hold the value and listeners"""
def __init__(
self,
default: P | None = None,
default_factory: Callable[[Any, Any], P] | None = None,
):
if default_factory is None:
default_factory = lambda prop, instance: cast(P, default)
self.default_factory = default_factory
self.obs: WeakKeyDictionary[Any, _Obs] = WeakKeyDictionary()
def _get_obs(self, instance) -> _Obs:
obs = self.obs.get(instance)
if obs is None:
obs = _Obs(self.default_factory(self, instance))
self.obs[instance] = obs
return obs
[docs]
def get(self, instance) -> P:
"""Get value for owner instance"""
obs = self._get_obs(instance)
return obs.value
[docs]
def set(self, instance, value):
"""Set value for owner instance"""
obs = self._get_obs(instance)
if obs.value != value:
old = obs.value
obs.value = value
self.dispatch(instance, value, old)
[docs]
def dispatch(self, instance, value, old_value):
"""Notifies every listener, which subscribed to the change event.
Args:
instance: Property instance
value: new value set
old_value: previous value
"""
obs = self._get_obs(instance)
for listener in obs.listeners:
try:
listener(instance, value, old_value)
except Exception:
print(
f"Change listener for {instance}.{self.name} = {value} raised an exception!",
file=sys.stderr,
)
traceback.print_exc()
[docs]
def bind(self, instance, callback):
"""Binds a function to the change event of the property.
A reference to the function will be kept.
Args:
instance: The instance to bind the callback to.
callback: The callback to bind.
"""
obs = self._get_obs(instance)
# Instance methods are bound methods, which can not be referenced by normal `ref()`
# if listeners would be a WeakSet, we would have to add listeners as WeakMethod
# ourselves into `WeakSet.data`.
obs.add(callback)
[docs]
def unbind(self, instance, callback):
"""Unbinds a function from the change event of the property.
Args:
instance: The target instance.
callback: The callback to unbind.
"""
obs = self._get_obs(instance)
obs.remove(callback)
def __set_name__(self, owner, name):
self.name = name
@overload
def __get__(self, instance: None, instance_type) -> Self: ...
@overload
def __get__(self, instance: Any, instance_type) -> P: ...
def __get__(self, instance: Any | None, instance_type) -> Self | P:
if instance is None:
return self
return self.get(instance)
def __set__(self, instance, value: P):
self.set(instance, value)
[docs]
def bind(instance, property: str, callback):
"""Bind a function to the change event of the property.
A reference to the function will be kept, so that it will be still
invoked even if it would normally have been garbage collected:
.. code-block:: python
def log_change(instance, value):
print(f"Value of {instance} changed to {value}")
class MyObject:
name = Property()
my_obj = MyObject()
bind(my_obj, "name", log_change)
my_obj.name = "Hans"
# > Value of <__main__.MyObject ...> changed to Hans
Args:
instance: Instance owning the property
property: Name of the property
callback: Function to call
Returns:
None
"""
t = type(instance)
prop = getattr(t, property)
if not isinstance(prop, Property):
raise ValueError(f"{t.__name__}.{property} is not an arcade.gui.Property")
prop.bind(instance, callback)
[docs]
def unbind(instance, property: str, callback):
"""Unbinds a function from the change event of the property.
.. code-block:: python
def log_change(instance, value):
print("Something changed")
class MyObject:
name = Property()
my_obj = MyObject()
bind(my_obj, "name", log_change)
unbind(my_obj, "name", log_change)
my_obj.name = "Hans"
# > Something changed
Args:
instance: Instance owning the property
property: Name of the property
callback: Function to unbind
Returns:
None
"""
t = type(instance)
prop = getattr(t, property)
if isinstance(prop, Property):
prop.unbind(instance, callback)
class _ObservableDict(dict):
"""Internal class to observe changes inside a native python dict."""
__slots__ = ("prop", "obj")
def __init__(self, prop: Property, instance, *args):
self.prop: Property = prop
self.obj = ref(instance)
super().__init__(*args)
@contextmanager
def _dispatch(self):
"""This is a context manager which will dispatch the change event
when the context is exited.
"""
old_value = self.copy()
yield
self.prop.dispatch(self.obj(), self, old_value)
@override
def __setitem__(self, key, value):
with self._dispatch():
dict.__setitem__(self, key, value)
@override
def __delitem__(self, key):
with self._dispatch():
dict.__delitem__(self, key)
@override
def clear(self):
with self._dispatch():
dict.clear(self)
@override
def pop(self, *args):
with self._dispatch():
return dict.pop(self, *args)
@override
def popitem(self):
with self._dispatch():
return dict.popitem(self)
@override
def setdefault(self, *args):
with self._dispatch():
return dict.setdefault(self, *args)
@override
def update(self, *args):
with self._dispatch():
dict.update(self, *args)
K = TypeVar("K")
V = TypeVar("V")
[docs]
class DictProperty(Property[dict[K, V]], Generic[K, V]):
"""Property that represents a dict.
Only dict are allowed. Any other classes are forbidden.
"""
def __init__(self):
super().__init__(default_factory=_ObservableDict)
[docs]
@override
def set(self, instance, value: dict):
"""Set value for owner instance, wraps the dict into an observable dict."""
value = _ObservableDict(self, instance, value)
super().set(instance, value)
class _ObservableList(list):
"""Internal class to observe changes inside a native python list.
Args:
prop: Property instance
instance: Instance owning the property
*args: List of arguments to pass to the list
"""
__slots__ = ("prop", "obj")
def __init__(self, prop: Property, instance, *args):
self.prop: Property = prop
self.obj = ref(instance)
super().__init__(*args)
@contextmanager
def _dispatch(self):
"""Dispatches the change event.
This is a context manager which will dispatch the change event
when the context is exited.
"""
old_value = self.copy()
yield
self.prop.dispatch(self.obj(), self, old_value)
@override
def __setitem__(self, key, value):
with self._dispatch():
list.__setitem__(self, key, value)
@override
def __delitem__(self, key):
with self._dispatch():
list.__delitem__(self, key)
@override
def __iadd__(self, *args):
with self._dispatch():
list.__iadd__(self, *args)
return self
@override
def __imul__(self, *args):
with self._dispatch():
list.__imul__(self, *args)
return self
@override
def append(self, *args):
"""Proxy for list.append() which dispatches the change event."""
with self._dispatch():
list.append(self, *args)
@override
def clear(self):
"""Proxy for list.clear() which dispatches the change event."""
with self._dispatch():
list.clear(self)
@override
def remove(self, *args):
"""Proxy for list.remove() which dispatches the change event."""
with self._dispatch():
list.remove(self, *args)
@override
def insert(self, *args):
"""Proxy for list.insert() which dispatches the change event."""
with self._dispatch():
list.insert(self, *args)
@override
def pop(self, *args):
"""Proxy for list.pop() which dispatches the change"""
with self._dispatch():
result = list.pop(self, *args)
return result
@override
def extend(self, *args):
"""Proxy for list.extend() which dispatches the change event."""
with self._dispatch():
list.extend(self, *args)
@override
def sort(self, **kwargs):
"""Proxy for list.sort() which dispatches the change event."""
with self._dispatch():
list.sort(self, **kwargs)
@override
def reverse(self):
"""Proxy for list.reverse() which dispatches the change event."""
with self._dispatch():
list.reverse(self)
[docs]
class ListProperty(Property[list[P]], Generic[P]):
"""Property that represents a list.
Only list are allowed. Any other classes are forbidden.
"""
def __init__(self):
super().__init__(default_factory=_ObservableList)
[docs]
@override
def set(self, instance, value: list):
"""Set value for owner instance, wraps the list into an observable list."""
value = _ObservableList(self, instance, value)
super().set(instance, value)