Compare commits
2 Commits
c5eaba88b7
...
488d1e8bff
Author | SHA1 | Date | |
---|---|---|---|
![]() |
488d1e8bff | ||
![]() |
6d1a6907d2 |
@ -1 +1,2 @@
|
||||
from .gear import *
|
||||
from .gear import Gear
|
||||
from .effect import Effect
|
||||
|
@ -3,11 +3,8 @@ from typing import Any, 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(
|
||||
def connect[S: Hashable, T: Hashable](
|
||||
source: Gear[S],
|
||||
to_target_val: Callable[[S], T],
|
||||
to_source: Callable[[S, T], S],
|
||||
|
@ -1,31 +1,40 @@
|
||||
from typing import Any, Callable, cast
|
||||
|
||||
from gears import Gear
|
||||
from .wrap_aware_tuple import WrapAwareTuple
|
||||
from typing import Any, Generic, Hashable, TypeVar, Callable, TypeVarTuple
|
||||
|
||||
class Effect[*Ts]:
|
||||
def __init__(self, fn: Callable[[*Ts], None], gears: WrapAwareTuple[Gear, *Ts]):
|
||||
for gear in gears:
|
||||
gear.effects.append(self)
|
||||
|
||||
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()
|
||||
self.on_change(gears[0]())
|
||||
|
||||
def on_change(self):
|
||||
self._fn(*self._gears.unwrap(lambda x: x()))
|
||||
def on_change(self, _value: Any):
|
||||
self._fn(*(gear() for gear in self._gears))
|
||||
|
||||
def effect_of[*Ts, *G_Ts](
|
||||
*gears: Gear[Any]
|
||||
) -> Callable[[Callable[[*Ts], None]], Effect]:
|
||||
# No deduction, no tears, only cast now
|
||||
ts_gears = cast(WrapAwareTuple[Gear, *Ts], WrapAwareTuple.prewrapped(gears))
|
||||
return effect_of_typechecked(ts_gears)
|
||||
|
||||
def effect_of_typechecked[*Ts](
|
||||
gears: WrapAwareTuple[Gear, *Ts]
|
||||
) -> Callable[[Callable[[*Ts], None]], Effect]:
|
||||
def decorator(fn: Callable[[*Ts], None]) -> Effect:
|
||||
return Effect(fn, gears)
|
||||
def effect_of[T0: Hashable](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[T0: Hashable, T1: Hashable](
|
||||
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[T0: Hashable, T1: Hashable, T2: Hashable](
|
||||
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
|
||||
|
@ -1,14 +1,10 @@
|
||||
from typing import Any, Hashable, TypeVar, TYPE_CHECKING
|
||||
from typing import Any, Generic, Hashable, TypeVar, Callable
|
||||
|
||||
T = TypeVar("T", bound=Hashable)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from gears.effect import Effect
|
||||
|
||||
class Gear[T]():
|
||||
class Gear[T: Hashable]:
|
||||
def __init__(self, value: T):
|
||||
self._value = value
|
||||
self.effects: list['Effect'] = []
|
||||
self.effects = []
|
||||
|
||||
def get(self) -> T:
|
||||
return self._value
|
||||
@ -21,4 +17,4 @@ class Gear[T]():
|
||||
return
|
||||
self._value = value
|
||||
for effect in self.effects:
|
||||
effect.on_change()
|
||||
effect.on_change(value)
|
||||
|
@ -1,90 +0,0 @@
|
||||
|
||||
from typing import cast, Callable, reveal_type, TYPE_CHECKING
|
||||
|
||||
# no-op "Wrap" type
|
||||
type Identity[X] = X
|
||||
|
||||
class WrapAwareTuple[Wrap, *Ts](tuple[Wrap, ...]):
|
||||
'''
|
||||
Wrap is a wrapper type around each of the types in Ts, in order.
|
||||
|
||||
For example:
|
||||
|
||||
a: Wrapped[Wrap, int, str, float] = Wrapped(Wrap[int], Wrap[str], Wrap[float])
|
||||
|
||||
This lets us keep track of type transformations, starting from the Identity mapping.
|
||||
'''
|
||||
|
||||
@staticmethod
|
||||
def id[*NewTs](*args: *NewTs) -> 'WrapAwareTuple[Identity, *NewTs]':
|
||||
'''
|
||||
Create a new WrapAwareTuple, using the Identity psuedo-wrapper.
|
||||
'''
|
||||
return WrapAwareTuple[Identity, *NewTs](args)
|
||||
|
||||
@staticmethod
|
||||
def prewrapped[NewWrap, *NewTs](args: tuple[NewWrap, ...]) -> 'WrapAwareTuple[NewWrap, *NewTs]':
|
||||
'''
|
||||
Create a new WrapAwareTuple, using the given wrapper type.
|
||||
'''
|
||||
return WrapAwareTuple[NewWrap, *NewTs](args)
|
||||
|
||||
def wrap[NewWrap](self, mapper: Callable[[Wrap], NewWrap]) -> 'WrapAwareTuple[NewWrap, *Ts]':
|
||||
'''
|
||||
Wrap all existing elements with a new wrapping function.
|
||||
|
||||
This function must unwrap the current wrapping type, if non-Identity, and apply a new one.
|
||||
|
||||
To pop the wrapped value from a tuple wrapper:
|
||||
>>> Wrapped[tuple, int](3).wrap[Identity](lambda x: x[0])
|
||||
|
||||
To wrap an unwrapped value into a tuple:
|
||||
>>> Wrapped[Identity, int](3).wrap(tuple)
|
||||
'''
|
||||
# pyright supports Union[*Ts], but we still can't do applicative functor things with it
|
||||
return WrapAwareTuple[NewWrap, *Ts](mapper(v) for v in self)
|
||||
|
||||
def unwrap(self, mapper: Callable[[Wrap], Identity]) -> tuple[*Ts]:
|
||||
'''
|
||||
Unwrap all elements in the tuple, returning a normal tuple.
|
||||
|
||||
This is the same as calling wrap with the appropriate wrapper, except we simplify the signature and
|
||||
return type for this case.
|
||||
'''
|
||||
return cast(tuple[*Ts], self.wrap(mapper))
|
||||
|
||||
def list_fmap[*Ts](t: tuple[*Ts]):
|
||||
match t:
|
||||
case (head, *tail):
|
||||
return ([head], *list_fmap(tuple(tail)))
|
||||
case _:
|
||||
return ()
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
def _fmap_scratchpad():
|
||||
reveal_type(list_fmap((1,"s",3.0)))
|
||||
# Type of "list_fmap((1, "s", 3))" is "tuple[list[*tuple[int, str, float]], *tuple[list[*tuple[Unknown, ...]] | Unknown, ...]] | tuple[()]"
|
||||
# ...wanted to get "tuple[list[int], list[str], list[float]]"; RIP
|
||||
|
||||
def _wrapaware_scratchpad():
|
||||
|
||||
atom = WrapAwareTuple.id(3)
|
||||
reveal_type(atom)
|
||||
# Type of "atom" is "Wrapped[Unknown, int]"
|
||||
# (variable) atom: Wrapped[Identity[Unknown], int]
|
||||
|
||||
compound_3 = WrapAwareTuple.id(3, "str", 9.0)
|
||||
reveal_type(compound_3)
|
||||
# Type of "compound_3" is "Wrapped[Unknown, int, str, float]"
|
||||
# (variable) compound_3: Wrapped[Identity[Unknown], int, str, float]
|
||||
|
||||
tupled_compound_3 = compound_3.wrap(tuple)
|
||||
reveal_type(tupled_compound_3)
|
||||
# Type of "tupled_compound_3" is "Wrapped[tuple[Unknown, ...], int, str, float]"
|
||||
# (variable) tupled_compound_3: Wrapped[tuple[Identity[Unknown], ...], int, str, float]
|
||||
|
||||
restored_compound_3 = tupled_compound_3.wrap(lambda x: x[0])
|
||||
reveal_type(restored_compound_3)
|
||||
# Type of "restored_compound_3" is "Wrapped[Unknown, int, str, float]"
|
||||
# (variable) restored_compound_3: Wrapped[Unknown, int, str, float]
|
@ -3,7 +3,7 @@ from typing import NamedTuple
|
||||
from unittest.mock import MagicMock
|
||||
from gears import Gear
|
||||
from gears.connections import connect
|
||||
from gears.effect import effect_of
|
||||
from gears.effect import effect_of, effect_of_2
|
||||
|
||||
|
||||
def test_get_set():
|
||||
@ -35,7 +35,7 @@ def test_effect_of_2():
|
||||
|
||||
last_arg: tuple[str, int] | None = None
|
||||
|
||||
@effect_of(g1, g2)
|
||||
@effect_of_2(g1, g2)
|
||||
def value_changed(v1: str, v2: int):
|
||||
nonlocal last_arg
|
||||
last_arg = (v1, v2)
|
||||
|
Loading…
x
Reference in New Issue
Block a user