summaryrefslogtreecommitdiffstats
path: root/polymatrix/statemonad/mixins.py
blob: 1a13440c3fa7ea1d88a63dff23e8029d0bef0724 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import abc
import dataclasses
from typing import Callable, Tuple, TypeVar, Generic


class StateCacheMixin(abc.ABC):
    @property
    @abc.abstractmethod
    def cache(self) -> dict:
        ...


State = TypeVar('State', bound=StateCacheMixin)
U = TypeVar('U')
V = TypeVar('V')


class StateMonadMixin(
    Generic[State, U],
    abc.ABC,    
):
    @property
    @abc.abstractmethod
    def apply_func(self) -> Callable[[State], tuple[State, U]]:
        ...

    def map(self, fn: Callable[[U], V]) -> 'StateMonadMixin[State, V]':

        def internal_map(state: State) -> Tuple[State, U]:
            n_state, val = self.apply(state)
            return n_state, fn(val)

        return dataclasses.replace(self, apply_func=internal_map)

    def flat_map(self, fn: Callable[[U], 'StateMonadMixin']) -> 'StateMonadMixin[State, V]':

        def internal_map(state: State) -> Tuple[State, V]:
            n_state, val = self.apply(state)
            return fn(val).apply(n_state)

        return dataclasses.replace(self, apply_func=internal_map)

    def zip(self, other: 'StateMonadMixin') -> 'StateMonadMixin':
        def internal_map(state: State) -> Tuple[State, V]:
            state, val1 = self.apply(state)
            state, val2 = other.apply(state)
            return state, (val1, val2)

        return dataclasses.replace(self, apply_func=internal_map)

    def cache(self) -> 'StateMonadMixin':
        def internal_map(state: State) -> Tuple[State, V]:
            if self in state.cache:
                return state, state.cache[self]

            state, val = self.apply(state)

            state = dataclasses.replace(
                state, 
                cache=state.cache | {self: val},
            )

            return state, val

        return dataclasses.replace(self, apply_func=internal_map)

    def apply(self, state: State) -> Tuple[State, U]:
        return self.apply_func(state)