File: coffee_expanded.py

package info (click to toggle)
automat 25.4.16-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 492 kB
  • sloc: python: 3,657; makefile: 15
file content (174 lines) | stat: -rw-r--r-- 4,879 bytes parent folder | download
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Protocol

from automat import TypeMachineBuilder


@dataclass
class Beans:
    description: str


@dataclass
class Water:
    "It's Water"


@dataclass
class Carafe:
    "It's a carafe"
    full: bool = False


@dataclass
class Ready:
    beans: Beans
    water: Water
    carafe: Carafe

    def brew(self) -> Mixture:
        print(f"brewing {self.beans} with {self.water} in {self.carafe}")
        return Mixture(self.beans, self.water)


@dataclass
class Mixture:
    beans: Beans
    water: Water


class Brewer(Protocol):
    def brew_button(self) -> None:
        "The user pressed the 'brew' button."

    def wait_a_while(self) -> Mixture:
        "Allow some time to pass."

    def put_in_beans(self, beans: Beans) -> None:
        "The user put in some beans."

    def put_in_water(self, water: Water) -> None:
        "The user put in some water."

    def put_in_carafe(self, carafe: Carafe) -> None:
        "The user put the mug"


class _BrewerInternals(Brewer, Protocol):
    def _ready(self, beans: Beans, water: Water, carafe: Carafe) -> None:
        "We are ready with all of our inputs."


@dataclass
class Light:
    on: bool = False


@dataclass
class BrewCore:
    "state for the brew process"
    ready_light: Light
    brew_light: Light
    beans: Beans | None = None
    water: Water | None = None
    carafe: Carafe | None = None
    brewing: Mixture | None = None


def _coffee_machine() -> TypeMachineBuilder[_BrewerInternals, BrewCore]:
    """
    Best practice: these functions are all fed in to the builder, they don't
    need to call each other, so they don't need to be defined globally.  Use a
    function scope to avoid littering a module with states and such.
    """
    builder = TypeMachineBuilder(_BrewerInternals, BrewCore)
    # reveal_type(builder)

    not_ready = builder.state("not_ready")

    def ready_factory(
        brewer: _BrewerInternals,
        core: BrewCore,
        beans: Beans,
        water: Water,
        carafe: Carafe,
    ) -> Ready:
        return Ready(beans, water, carafe)

    def mixture_factory(brewer: _BrewerInternals, core: BrewCore) -> Mixture:
        # We already do have a 'ready' but it's State-Specific Data which makes
        # it really annoying to relay on to the *next* state without passing it
        # through the state core.  requiring the factory to take SSD inherently
        # means that it could only work with transitions away from a single
        # state, which would not be helpful, although that *is* what we want
        # here.

        assert core.beans is not None
        assert core.water is not None
        assert core.carafe is not None

        return Mixture(core.beans, core.water)

    ready = builder.state("ready", ready_factory)
    brewing = builder.state("brewing", mixture_factory)

    def ready_check(brewer: _BrewerInternals, core: BrewCore) -> None:
        if (
            core.beans is not None
            and core.water is not None
            and core.carafe is not None
            and core.carafe.full is not None
        ):
            brewer._ready(core.beans, core.water, core.carafe)

    @not_ready.upon(Brewer.put_in_beans).loop()
    def put_beans(brewer: _BrewerInternals, core: BrewCore, beans: Beans) -> None:
        core.beans = beans
        ready_check(brewer, core)

    @not_ready.upon(Brewer.put_in_water).loop()
    def put_water(brewer: _BrewerInternals, core: BrewCore, water: Water) -> None:
        core.water = water
        ready_check(brewer, core)

    @not_ready.upon(Brewer.put_in_carafe).loop()
    def put_carafe(brewer: _BrewerInternals, core: BrewCore, carafe: Carafe) -> None:
        core.carafe = carafe
        ready_check(brewer, core)

    @not_ready.upon(_BrewerInternals._ready).to(ready)
    def get_ready(
        brewer: _BrewerInternals,
        core: BrewCore,
        beans: Beans,
        water: Water,
        carafe: Carafe,
    ) -> None:
        print("ready output")

    @ready.upon(Brewer.brew_button).to(brewing)
    def brew(brewer: _BrewerInternals, core: BrewCore, ready: Ready) -> None:
        core.brew_light.on = True
        print("BREW CALLED")
        core.brewing = ready.brew()

    @brewing.upon(_BrewerInternals.wait_a_while).to(not_ready)
    def brewed(brewer: _BrewerInternals, core: BrewCore, mixture: Mixture) -> Mixture:
        core.brew_light.on = False
        return mixture

    return builder


CoffeeMachine: Callable[[BrewCore], Brewer] = _coffee_machine().build()

if __name__ == "__main__":
    machine = CoffeeMachine(core := BrewCore(Light(), Light()))
    machine.put_in_beans(Beans("light roast"))
    machine.put_in_water(Water())
    machine.put_in_carafe(Carafe())
    machine.brew_button()
    brewed = machine.wait_a_while()
    print(brewed)