🔒 signals

[flint moat]

_hook: Callable[[Observable[Any]], None] | None = None

class Observable[T]:
    _value: T

    def __init__(self) -> None:
        self._subscribers: dict[int, Callable[[T], None]] = {}

    def subscribe(self, callback: Callable[[T], None]) -> Callable[[], None]:
        h = hash(callback)
        self._subscribers[h] = callback

        def unsubscribe():
            del self._subscribers[h]

        return unsubscribe

    @property
    def value(self) -> T:
        if _hook is not None:
            _hook(self)
        return self._value

    @value.setter
    def value(self, new_value: T) -> None:
        if self._value != new_value:
            self._value = new_value
            for subscriber in self._subscribers.values():
                subscriber(new_value)

class Signal[T](Observable[T]):
    def __init__(self, initial: T) -> None:
        super().__init__()
        self._value: T = initial

class Computed[T](Observable[T]):
    def __init__(self, compute: Callable[[], T]) -> None:
        super().__init__()
        self._compute: Callable[[], T] = compute
        global _hook
        old_hook = _hook
        self._subscriptions: list[Callable[[], None]] = []
        def hook(signal: Observable[Any]) -> None:
            def subscription(_value: Any):
                self._recompute()
            self._subscriptions.append(signal.subscribe(subscription))
        _hook = hook
        self._value: T = compute()
        _hook = old_hook

    def __del__(self) -> None:
        for subscription in self._subscriptions:
            subscription()

    def _recompute(self) -> None:
        self.value: T = self._compute()

def effect(callback: Callable[[], None]) -> Callable[[], None]:
    def subscription(_value: Any) -> None:
        callback()
    global _hook
    old_hook = _hook
    def hook(signal: Observable[Any]) -> None:
        signal.subscribe(subscription)
    _hook = hook
    callback()
    _hook = old_hook
    return callback

[still light]

@flint moat

Python help channel opened

Remember to:

• Ask your Python question, not if you can ask or if there's an expert who can help.
• Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
• Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.

[still light]

@flint moat

Python help channel closed for inactivity

This help channel has been closed. Feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.