87 lines
2.1 KiB
Python
87 lines
2.1 KiB
Python
|
import queue
|
||
|
|
|
||
|
|
import simpy
|
||
|
|
from prompt_toolkit import PromptSession
|
||
|
|
from prompt_toolkit.patch_stdout import patch_stdout
|
||
|
|
|
||
|
|
from tern import Tern
|
||
|
|
from transition import TRANSITIONS
|
||
|
|
|
||
|
|
|
||
|
|
CommandQueue = queue.Queue[tuple[str, Tern]]
|
||
|
|
|
||
|
|
|
||
|
|
def node(env: simpy.Environment, node: tuple[str, str, str], values: dict[str, Tern]):
|
||
|
|
while True:
|
||
|
|
nn, uu, yy = node
|
||
|
|
old = "".join(str(values[x]) for x in node)
|
||
|
|
new = TRANSITIONS[old]
|
||
|
|
|
||
|
|
n, u, y = new
|
||
|
|
|
||
|
|
values[nn], values[uu], values[yy] = (
|
||
|
|
Tern.from_string(n),
|
||
|
|
Tern.from_string(u),
|
||
|
|
Tern.from_string(y),
|
||
|
|
)
|
||
|
|
yield env.timeout(1)
|
||
|
|
|
||
|
|
|
||
|
|
def frame(values: dict[str, Tern]) -> str:
|
||
|
|
return " ".join(f"{k}={values[k]}" for k in sorted(values))
|
||
|
|
|
||
|
|
|
||
|
|
def render_process(
|
||
|
|
env: simpy.Environment,
|
||
|
|
values: dict[str, Tern],
|
||
|
|
):
|
||
|
|
last_frame: str | None = None
|
||
|
|
|
||
|
|
while True:
|
||
|
|
current_frame = frame(values)
|
||
|
|
|
||
|
|
if current_frame != last_frame:
|
||
|
|
print(f"[t={env.now:5.1f}] {current_frame}")
|
||
|
|
last_frame = current_frame
|
||
|
|
|
||
|
|
yield env.timeout(0.1)
|
||
|
|
|
||
|
|
|
||
|
|
def input_thread(commands: CommandQueue):
|
||
|
|
session: PromptSession[str] = PromptSession("> ")
|
||
|
|
|
||
|
|
with patch_stdout():
|
||
|
|
while True:
|
||
|
|
line = session.prompt().strip()
|
||
|
|
|
||
|
|
if line in {"q", "quit", "exit"}:
|
||
|
|
commands.put(("__quit__", Tern.U))
|
||
|
|
break
|
||
|
|
|
||
|
|
try:
|
||
|
|
name, value = line.split()
|
||
|
|
commands.put((name, Tern.from_string(value)))
|
||
|
|
except Exception:
|
||
|
|
print("format: A 1 | A 0 | A N | quit")
|
||
|
|
|
||
|
|
|
||
|
|
def user_input_process(
|
||
|
|
env: simpy.Environment,
|
||
|
|
values: dict[str, Tern],
|
||
|
|
commands: CommandQueue,
|
||
|
|
):
|
||
|
|
while True:
|
||
|
|
while not commands.empty():
|
||
|
|
name, value = commands.get_nowait()
|
||
|
|
|
||
|
|
if name == "__quit__":
|
||
|
|
return
|
||
|
|
|
||
|
|
if name not in values:
|
||
|
|
print(f"unknown node: {name}")
|
||
|
|
continue
|
||
|
|
|
||
|
|
values[name] = value
|
||
|
|
print(f"[t={env.now}] user set {name} = {value}")
|
||
|
|
|
||
|
|
yield env.timeout(0.1)
|