initial code-up

gears/__init__.py

@@ -0,0 +1 @@
+from .gear import *

gears/connections.py

@@ -0,0 +1,44 @@
+from typing import Any, Generic, Hashable, TypeVar, Callable
+
+from gears.effect import effect_of
+from gears.gear import Gear
+
+S = TypeVar("S", bound=Hashable)
+T = TypeVar("T", bound=Hashable)
+
+
+def connect(
+    source: Gear[S],
+    to_target_val: Callable[[S], T],
+    to_source: Callable[[S, T], S] | None = None,
+) -> Gear[T]:
+    target = Gear(to_target_val(source()))
+
+    # TODO: check for determinism maybe
+    # TODO: magic setter
+
+    entry_guard = False
+
+    @effect_of(source)
+    def set_target_val(source_val: S):
+        nonlocal entry_guard
+        if entry_guard:
+            return
+        entry_guard = True
+        try:
+            target.set(to_target_val(source_val))
+        finally:
+            entry_guard = False
+
+    @effect_of(target)
+    def set_source_val(target_val: T):
+        nonlocal entry_guard
+        if entry_guard:
+            return
+        entry_guard = True
+        try:
+            source.set(to_source(source(), target_val))
+        finally:
+            entry_guard = False
+
+    return target

gears/effect.py

@@ -0,0 +1,47 @@
+from gears import Gear
+from typing import Any, Generic, Hashable, TypeVar, Callable, TypeVarTuple
+
+
+class Effect:
+    def __init__(self, fn: Callable[..., None], *gears: Gear):
+        self._fn = fn
+        self._gears = gears
+        for gear in gears:
+            gear.effects.append(self)
+
+        self.on_change(gears[0]())
+
+    def on_change(self, _value: Any):
+        self._fn(*(gear() for gear in self._gears))
+
+
+T0 = TypeVar("T0", bound=Hashable)
+T1 = TypeVar("T1", bound=Hashable)
+T2 = TypeVar("T2", bound=Hashable)
+
+# Ts = TypeVarTuple("Ts", bound=Hashable)
+
+
+def effect_of(g0: Gear[T0]) -> Callable[[Callable[[T0], None]], Effect]:
+    def decorator(fn: Callable[[T0], None]) -> Effect:
+        return Effect(fn, g0)
+
+    return decorator
+
+
+def effect_of_2(
+    g0: Gear[T0], g1: Gear[T1]
+) -> Callable[[Callable[[T0, T1], None]], Effect]:
+    def decorator(fn: Callable[[T0, T1], None]) -> Effect:
+        return Effect(fn, g0, g1)
+
+    return decorator
+
+
+def effect_of_3(
+    g0: Gear[T0], g1: Gear[T1], g2: Gear[T2]
+) -> Callable[[Callable[[T0, T1, T2], None]], Effect]:
+    def decorator(fn: Callable[[T0, T1, T2], None]) -> Effect:
+        return Effect(fn, g0, g1, g2)
+
+    return decorator

gears/gear.py

@@ -0,0 +1,22 @@
+from typing import Any, Generic, Hashable, TypeVar, Callable
+
+T = TypeVar("T", bound=Hashable)
+
+
+class Gear(Generic[T]):
+    def __init__(self, value: T):
+        self._value = value
+        self.effects = []
+
+    def get(self) -> T:
+        return self._value
+
+    def __call__(self) -> T:
+        return self.get()
+
+    def set(self, value: T):
+        if value == self._value:
+            return
+        self._value = value
+        for effect in self.effects:
+            effect.on_change(value)

pyproject.toml

@@ -0,0 +1,4 @@
+[project]
+name = "gears" # REQUIRED, is the only field that cannot be marked as dynamic.
+
+version = "0.0.0"

tests/test_signals.py

@@ -0,0 +1,120 @@
+from dataclasses import dataclass
+from typing import NamedTuple
+from unittest.mock import MagicMock
+from gears import Gear
+from gears.connections import connect
+from gears.effect import effect_of, effect_of_2
+
+
+def test_get_set():
+    g = Gear(1)
+    assert g.get() == 1
+    assert g() == 1
+    g.set(3)
+
+    assert g.get() == 3
+    assert g() == 3
+
+
+def test_basic_effect():
+    g = Gear("E")
+
+    last_arg: str | None = None
+
+    @effect_of(g)
+    def value_changed(v: str):
+        nonlocal last_arg
+        last_arg = v
+
+    assert last_arg == "E"
+
+
+def test_effect_of_2():
+    g1 = Gear("E")
+    g2 = Gear(3)
+
+    last_arg: tuple[str, int] | None = None
+
+    @effect_of_2(g1, g2)
+    def value_changed(v1: str, v2: int):
+        nonlocal last_arg
+        last_arg = (v1, v2)
+
+    assert last_arg == ("E", 3)
+    g2.set(4)
+    assert last_arg == ("E", 4)
+
+
+def test_connect():
+    my_str = Gear("123")
+    my_int = connect(my_str, lambda s: int(s), lambda _, i: str(i))
+
+    assert my_int() == 123
+    my_int.set(321)
+    assert my_str() == "321"
+
+
+def test_connect_multiplication():
+    mul1 = Gear(10.0)
+    mul2 = connect(mul1, lambda x: x * 2.0, lambda _, y: y / 2.0)
+    mul3 = connect(mul1, lambda x: x * 3.0, lambda _, y: y / 3.0)
+    assert mul2() == 20
+    assert mul3() == 30
+    mul2.set(30)
+    assert mul1() == 15
+    assert mul3() == 45
+
+
+def test_connect_property():
+    class Foo(NamedTuple):
+        bar: int
+        baz: int
+
+    foo = Gear(Foo(bar=1, baz=2))
+    bar = connect(
+        foo,
+        lambda foo: foo.bar,
+        lambda foo, bar: Foo(bar=bar, baz=foo.baz),
+    )
+
+    assert bar() == 1
+    bar.set(5)
+    assert bar() == 5
+    assert foo() == Foo(5, 2)
+
+    class IntEntryOld:
+        on_value_changed: Signal[int]
+        _val: int
+
+        def get_val(self, val: int): ...
+        def set_val(self, val: int): ...
+
+    class IntEntry:
+        def __init__(self, value: Gear[int]): ...
+
+    int_entry = IntEntry()
+
+    effect_of(bar)(int_entry.set_val)
+    int_entry.on_value_changed.connect(bar.set)
+
+    @effect_of(bar)
+    def update_ui(bar: int):
+        int_entry.set_val(bar)
+
+    def update_model(bar_val: int):
+        bar.set(bar_val)
+
+    int_entry.on_value_changed.connect(update_model)
+
+
+def hello(my_callback):
+    my_callback(5)
+
+
+def test_demonstrate_magicmock():
+    m = MagicMock()
+    # print(f"{m.foo(4).bar[0]=}")
+    # m.
+    hello(m)
+    # assert m.call_count == 1
+    m.assert_called_once_with(5)