Загрузить файлы в «/»

graph_n.py

@@ -1,149 +1,40 @@
 import queue
 import threading
-from enum import Enum
-from typing import override
 
-import simpy
 import simpy.rt
-from prompt_toolkit import PromptSession
-from prompt_toolkit.patch_stdout import patch_stdout
 
-nodes_n = {
+from processes import input_thread, node, render_process, user_input_process
-    ("A", "1", "A'"),
+from tern import Tern
-    ("B", "1", "B'"),
-    ("AB", "A", "AB'"),
-    ("A'B", "A'", "A'B'"),
-    ("A'B", "B", "AB"),
-    ("A'B'", "B'", "AB'"),
-}
 
 
-f = {
+def initial_values(edges) -> dict[str, Tern]:
-    "000": "000",
+    return {x: Tern.U for triple in edges for x in triple}
-    "001": "0NN",
-    "00N": "000",
-    "010": "NNN",
-    "011": "011",
-    "01N": "011",
-    "0N0": "000",
-    "0N1": "011",
-    "0NN": "0NN",
-    "100": "NN0",
-    "101": "NNN",
-    "10N": "NNN",
-    "110": "110",
-    "111": "111",
-    "11N": "11N",
-    "1N0": "110",
-    "1N1": "111",
-    "1NN": "11N",
-    "N00": "000",
-    "N01": "NNN",
-    "N0N": "000",
-    "N10": "110",
-    "N11": "N11",
-    "N1N": "N1N",
-    "NN0": "NN0",
-    "NN1": "NN1",
-    "NNN": "NNN",
-}
 
 
-class Tern(Enum):
+def main():
-    N = -1
+    from graph_view import active_edges, render_graph_process
-    U = 0
-    Y = 1
 
-    @override
+    edges = active_edges()
-    def __str__(self):
+    values = initial_values(edges)
-        if self.value == -1:
+    commands: queue.Queue[tuple[str, Tern]] = queue.Queue()
-            return "0"
-        if self.value == 1:
-            return "1"
-        return "N"
 
-    @classmethod
+    env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)
-    def from_string(cls, s: str) -> "Tern":
+
-        if s == "N":
+    threading.Thread(
-            return Tern.U
+        target=input_thread,
-        if s == "0":
+        args=(commands,),
-            return Tern.N
+        daemon=True,
-        if s == "1":
+    ).start()
-            return Tern.Y
+
-        return cls[s]
+    for n in edges:
+        _ = env.process(node(env, n, values))
+
+    _ = env.process(user_input_process(env, values, commands))
+    _ = env.process(render_process(env, values))
+    _ = env.process(render_graph_process(env, values))
+
+    _ = env.run(until=float("inf"))
 
 
-def node(env: simpy.Environment, node: tuple[str, str, str], values: dict[str, Tern]):
+if __name__ == "__main__":
-    while True:
+    main()
-        nn, uu, yy = node
-        old = "".join(str(values[x]) for x in node)
-        new = f[old]
-
-        n, u, y = new
-
-        values[nn], values[uu], values[yy] = (
-            Tern.from_string(n),
-            Tern.from_string(u),
-            Tern.from_string(y),
-        )
-        print(values)
-        yield env.timeout(1)
-
-
-def input_thread(commands: queue.Queue[tuple[str, Tern]]):
-    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: queue.Queue[tuple[str, Tern]],
-):
-    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)
-
-
-ls = {x: Tern.U for triple in nodes_n for x in triple}
-commands: queue.Queue[tuple[str, Tern]] = queue.Queue()
-
-env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)
-
-threading.Thread(
-    target=input_thread,
-    args=(commands,),
-    daemon=True,
-).start()
-
-for n in nodes_n:
-    _ = env.process(node(env, n, ls))
-
-_ = env.process(user_input_process(env, ls, commands))
-
-_ = env.run(until=100)

graph_view.py

@@ -0,0 +1,831 @@
+from __future__ import annotations
+
+import json
+import mimetypes
+import os
+import queue
+import re
+import threading
+import time
+import urllib.parse
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any, Callable
+
+import simpy
+
+from tern import Tern
+from transition import NODES_N, TRANSITIONS
+
+
+def active_edges() -> set[tuple[str, str, str]]:
+    """Graph the live engine drives."""
+    return NODES_N
+
+
+PIN_ID = "0"
+DEFAULT_HOST = "127.0.0.1"
+DEFAULT_PORT = 8765
+FRONTEND_DIR = Path(__file__).with_name("frontend").joinpath("dist")
+SESSION_ID_RE = re.compile(r"[^a-zA-Z0-9_-]+")
+
+
+def normalize_pin_value(value: object) -> str:
+    text = str(value if value is not None else "N").strip()
+
+    if text in {"1", "+"} or text.lower() == "true":
+        return "+"
+
+    if text in {"-1", "0", "-"} or text.lower() == "false":
+        return "0"
+
+    if text in {"", "N", "?", "unknown"}:
+        return "N"
+
+    return text
+
+
+def tern_event_value(value: Tern) -> str:
+    match value:
+        case Tern.Y:
+            return "+"
+        case Tern.N:
+            return "0"
+        case Tern.U:
+            return "N"
+
+
+def event_value_to_tern(value: object) -> Tern:
+    match normalize_pin_value(value):
+        case "+":
+            return Tern.Y
+        case "0":
+            return Tern.N
+        case _:
+            return Tern.U
+
+
+def snapshot_values(values: dict[str, Tern]) -> dict[str, str]:
+    return {name: tern_event_value(value) for name, value in sorted(values.items())}
+
+
+def graph_topology(values: dict[str, Tern]) -> dict[str, Any]:
+    current_values = snapshot_values(values)
+    edges: list[dict[str, str]] = []
+
+    for index, (nn, uu, yy) in enumerate(sorted(active_edges())):
+        edges.append(
+            {
+                "id": f"{uu}->{nn}:0:{index}",
+                "source": uu,
+                "target": nn,
+                "label": "0",
+            }
+        )
+        edges.append(
+            {
+                "id": f"{uu}->{yy}:1:{index}",
+                "source": uu,
+                "target": yy,
+                "label": "+",
+            }
+        )
+
+    return {
+        "nodes": [
+            {
+                "id": name,
+                "label": name,
+                "pins": [PIN_ID],
+                "value": value,
+            }
+            for name, value in current_values.items()
+        ],
+        "edges": edges,
+        "values": current_values,
+    }
+
+
+def transition_char(value: object) -> str:
+    """Event value ("+"/"0"/"N") -> TRANSITIONS char ("1"/"0"/"N")."""
+    return str(event_value_to_tern(value))
+
+
+def char_event_value(char: str) -> str:
+    """TRANSITIONS char ("1"/"0"/"N") -> event value ("+"/"0"/"N")."""
+    return tern_event_value(Tern.from_string(char))
+
+
+def derive_triples(topology: dict[str, Any]) -> list[tuple[str, str, str]]:
+    """Invert graph_topology(): a node with a "0" out-edge (-> nn) and a "+"
+    out-edge (-> yy) is a transition node, yielding the triple (nn, source, yy).
+    """
+    zero_target: dict[str, str] = {}
+    plus_target: dict[str, str] = {}
+
+    for edge in topology.get("edges", []):
+        source = edge.get("source")
+        target = edge.get("target")
+        if not source or not target:
+            continue
+
+        label = normalize_pin_value(edge.get("label", ""))
+        if label == "0":
+            zero_target.setdefault(source, target)
+        elif label == "+":
+            plus_target.setdefault(source, target)
+
+    triples = {
+        (zero_target[source], source, plus_target[source])
+        for source in zero_target
+        if source in plus_target
+    }
+
+    return sorted(triples, key=lambda triple: (triple[1], triple[0], triple[2]))
+
+
+class EventHub:
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._subscribers: set[queue.Queue[dict[str, Any]]] = set()
+        self._latest: dict[str, Any] | None = None
+
+    def subscribe(self) -> queue.Queue[dict[str, Any]]:
+        events: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=100)
+
+        with self._lock:
+            self._subscribers.add(events)
+            if self._latest is not None:
+                events.put_nowait(self._latest)
+
+        return events
+
+    def unsubscribe(self, events: queue.Queue[dict[str, Any]]) -> None:
+        with self._lock:
+            self._subscribers.discard(events)
+
+    def publish(self, event: dict[str, Any]) -> None:
+        with self._lock:
+            self._latest = event
+            subscribers = list(self._subscribers)
+
+        for events in subscribers:
+            try:
+                events.put_nowait(event)
+            except queue.Full:
+                try:
+                    _ = events.get_nowait()
+                except queue.Empty:
+                    pass
+                try:
+                    events.put_nowait(event)
+                except queue.Full:
+                    pass
+
+
+class GraphSession:
+    def __init__(
+        self,
+        session_id: str,
+        name: str,
+        topology: dict[str, Any],
+        *,
+        editable: bool = True,
+        kind: str = "user",
+        on_value_update: Callable[[str, str, str], None] | None = None,
+    ):
+        self.id = session_id
+        self.name = name
+        self.kind = kind
+        self.editable = editable
+        self.created_at = time.time()
+        self.event_hub = EventHub()
+        self._lock = threading.Lock()
+        self._topology = self._normalize_topology(topology)
+        self._values = self._values_from_topology(self._topology, topology.get("values", {}))
+        self._frame = 0
+        self._on_value_update = on_value_update
+        self._triples = derive_triples(self._topology)
+        self.steppable = kind != "simpy" and bool(self._triples)
+
+        self._publish_initial()
+
+    def summary(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "id": self.id,
+                "name": self.name,
+                "kind": self.kind,
+                "editable": self.editable,
+                "steppable": self.steppable,
+                "createdAt": self.created_at,
+                "frame": self._frame,
+                "nodeCount": len(self._topology["nodes"]),
+                "edgeCount": len(self._topology["edges"]),
+            }
+
+    def graph(self) -> dict[str, Any]:
+        with self._lock:
+            topology = json.loads(json.dumps(self._topology))
+            values = self._public_values()
+
+            for node in topology["nodes"]:
+                node_values = self._values.get(node["id"], {})
+                node["value"] = node_values.get(PIN_ID, "N")
+
+            topology["values"] = values
+            return topology
+
+    def set_value(self, node_id: str, pin_id: str, value: object, *, time_value: float | None = None) -> dict[str, Any]:
+        normalized_value = normalize_pin_value(value)
+
+        with self._lock:
+            if not self.editable:
+                raise ValueError("session is not editable")
+
+            node = self._ensure_node(node_id, pin_id)
+            old_value = node.get(pin_id, "N")
+
+            if old_value == normalized_value:
+                return {
+                    "frame": self._frame,
+                    "time": time_value,
+                    "changed": [],
+                }
+
+            node[pin_id] = normalized_value
+            self._frame += 1
+            event = {
+                "frame": self._frame,
+                "changed": [
+                    {
+                        "node": node_id,
+                        "pin": pin_id,
+                        "from": old_value,
+                        "to": normalized_value,
+                    }
+                ],
+            }
+
+            if time_value is not None:
+                event["time"] = time_value
+
+        if self._on_value_update is not None:
+            self._on_value_update(node_id, pin_id, normalized_value)
+
+        self.event_hub.publish(event)
+        return event
+
+    def step(self, *, time_value: float | None = None) -> dict[str, Any]:
+        """Advance the simulation by one tick across all derived triples.
+
+        Triples are applied in deterministic order, each reading the current
+        (possibly already-mutated) values — mirroring the live engine's
+        sequential, shared-state update within a tick.
+        """
+        with self._lock:
+            if not self.steppable:
+                raise ValueError("session is not steppable")
+
+            changed: list[dict[str, str]] = []
+
+            for nn, uu, yy in self._triples:
+                old = "".join(transition_char(self._ensure_node(name, PIN_ID)[PIN_ID]) for name in (nn, uu, yy))
+                new = TRANSITIONS.get(old)
+                if new is None:
+                    continue
+
+                for name, char in zip((nn, uu, yy), new):
+                    node = self._ensure_node(name, PIN_ID)
+                    next_value = char_event_value(char)
+                    old_value = node[PIN_ID]
+                    if old_value == next_value:
+                        continue
+
+                    node[PIN_ID] = next_value
+                    changed.append({"node": name, "pin": PIN_ID, "from": old_value, "to": next_value})
+
+            self._frame += 1
+            event: dict[str, Any] = {"frame": self._frame, "changed": changed}
+            if time_value is not None:
+                event["time"] = time_value
+
+        self.event_hub.publish(event)
+        return event
+
+    def sync_external_values(
+        self,
+        values: dict[str, str],
+        *,
+        frame: int,
+        time_value: float,
+        changed: list[dict[str, str]],
+    ) -> None:
+        with self._lock:
+            self._frame = frame
+            for node, value in values.items():
+                self._ensure_node(node, PIN_ID)[PIN_ID] = normalize_pin_value(value)
+
+        if changed:
+            self.event_hub.publish(
+                {
+                    "frame": frame,
+                    "time": time_value,
+                    "changed": changed,
+                }
+            )
+
+    def _publish_initial(self) -> None:
+        changed = []
+
+        with self._lock:
+            for node, pins in sorted(self._values.items()):
+                for pin, value in sorted(pins.items()):
+                    changed.append(
+                        {
+                            "node": node,
+                            "pin": pin,
+                            "from": value,
+                            "to": value,
+                        }
+                    )
+
+        self.event_hub.publish(
+            {
+                "frame": 0,
+                "changed": changed,
+            }
+        )
+
+    def _ensure_node(self, node_id: str, pin_id: str) -> dict[str, str]:
+        if node_id not in self._values:
+            self._values[node_id] = {}
+            self._topology["nodes"].append(
+                {
+                    "id": node_id,
+                    "label": node_id,
+                    "pins": [pin_id],
+                    "value": "N",
+                }
+            )
+
+        if pin_id not in self._values[node_id]:
+            self._values[node_id][pin_id] = "N"
+
+        node_definition = next((node for node in self._topology["nodes"] if node["id"] == node_id), None)
+        if node_definition is not None:
+            pins = node_definition.setdefault("pins", [])
+            if pin_id not in pins:
+                pins.append(pin_id)
+
+        return self._values[node_id]
+
+    def _public_values(self) -> dict[str, str | dict[str, str]]:
+        result: dict[str, str | dict[str, str]] = {}
+
+        for node, pins in self._values.items():
+            if set(pins) == {PIN_ID}:
+                result[node] = pins[PIN_ID]
+            else:
+                result[node] = dict(sorted(pins.items()))
+
+        return result
+
+    def _normalize_topology(self, topology: dict[str, Any]) -> dict[str, Any]:
+        nodes_by_id: dict[str, dict[str, Any]] = {}
+
+        for raw_node in topology.get("nodes", []):
+            node_id = str(raw_node.get("id", "")).strip()
+            if not node_id:
+                continue
+
+            pins = [str(pin) for pin in raw_node.get("pins", [PIN_ID]) if str(pin)]
+            nodes_by_id[node_id] = {
+                "id": node_id,
+                "label": str(raw_node.get("label") or node_id),
+                "pins": pins or [PIN_ID],
+                "value": normalize_pin_value(raw_node.get("value", "N")),
+            }
+
+        edges: list[dict[str, str]] = []
+        for index, raw_edge in enumerate(topology.get("edges", [])):
+            source = str(raw_edge.get("source", "")).strip()
+            target = str(raw_edge.get("target", "")).strip()
+
+            if not source or not target:
+                continue
+
+            nodes_by_id.setdefault(source, {"id": source, "label": source, "pins": [PIN_ID], "value": "N"})
+            nodes_by_id.setdefault(target, {"id": target, "label": target, "pins": [PIN_ID], "value": "N"})
+
+            edge_id = str(raw_edge.get("id") or f"{source}->{target}:{index}")
+            edges.append(
+                {
+                    "id": edge_id,
+                    "source": source,
+                    "target": target,
+                    "label": str(raw_edge.get("label") or ""),
+                }
+            )
+
+        return {
+            "nodes": [nodes_by_id[node_id] for node_id in sorted(nodes_by_id)],
+            "edges": edges,
+        }
+
+    def _values_from_topology(self, topology: dict[str, Any], raw_values: object) -> dict[str, dict[str, str]]:
+        values: dict[str, dict[str, str]] = {}
+
+        for node in topology["nodes"]:
+            node_id = node["id"]
+            values[node_id] = {}
+            raw_value = raw_values.get(node_id, node.get("value", "N")) if isinstance(raw_values, dict) else node.get("value", "N")
+
+            if isinstance(raw_value, dict):
+                for pin, value in raw_value.items():
+                    values[node_id][str(pin)] = normalize_pin_value(value)
+
+            for pin in node.get("pins", [PIN_ID]):
+                values[node_id].setdefault(str(pin), normalize_pin_value(raw_value if not isinstance(raw_value, dict) else "N"))
+
+        return values
+
+
+class SessionRegistry:
+    def __init__(self, simpy_values: dict[str, Tern]):
+        self._lock = threading.Lock()
+        self._sessions: dict[str, GraphSession] = {}
+        self.simpy_values = simpy_values
+        self.simpy_session = GraphSession(
+            "simpy",
+            "SimPy live",
+            graph_topology(simpy_values),
+            kind="simpy",
+            on_value_update=self._update_simpy_value,
+        )
+        self._sessions[self.simpy_session.id] = self.simpy_session
+
+    def list_sessions(self) -> dict[str, Any]:
+        with self._lock:
+            sessions = [session.summary() for session in self._sessions.values()]
+
+        return {"sessions": sorted(sessions, key=lambda session: (session["kind"] != "simpy", session["createdAt"]))}
+
+    def get(self, session_id: str) -> GraphSession:
+        with self._lock:
+            session = self._sessions.get(session_id)
+
+        if session is None:
+            raise KeyError(session_id)
+
+        return session
+
+    def create(self, name: str, topology: dict[str, Any]) -> GraphSession:
+        base_id = SESSION_ID_RE.sub("-", name.strip().lower()).strip("-") or "graph"
+        session_id = base_id
+
+        with self._lock:
+            suffix = 2
+            while session_id in self._sessions:
+                session_id = f"{base_id}-{suffix}"
+                suffix += 1
+
+            session = GraphSession(session_id, name.strip() or session_id, topology)
+            self._sessions[session_id] = session
+
+        return session
+
+    def sync_simpy(self, values: dict[str, str], *, frame: int, time_value: float, changed: list[dict[str, str]]) -> None:
+        self.simpy_session.sync_external_values(values, frame=frame, time_value=time_value, changed=changed)
+
+    def _update_simpy_value(self, node_id: str, pin_id: str, value: str) -> None:
+        if pin_id != PIN_ID:
+            return
+
+        if node_id in self.simpy_values:
+            self.simpy_values[node_id] = event_value_to_tern(value)
+
+
+class GraphHTTPServer(ThreadingHTTPServer):
+    allow_reuse_address = True
+    daemon_threads = True
+
+    def __init__(self, address: tuple[str, int], values: dict[str, Tern], registry: SessionRegistry):
+        super().__init__(address, GraphRequestHandler)
+        self.values = values
+        self.registry = registry
+        self.frontend_dir = FRONTEND_DIR.resolve()
+
+
+class GraphRequestHandler(BaseHTTPRequestHandler):
+    server: GraphHTTPServer
+
+    def do_OPTIONS(self):
+        self.send_response(HTTPStatus.NO_CONTENT)
+        self._send_cors_headers()
+        self.end_headers()
+
+    def do_GET(self):
+        path = urllib.parse.urlparse(self.path).path
+
+        if path == "/api/graph":
+            self._send_json(self.server.registry.get("simpy").graph())
+            return
+
+        if path == "/api/events":
+            self._send_events(self.server.registry.get("simpy"))
+            return
+
+        if path == "/api/sessions":
+            self._send_json(self.server.registry.list_sessions())
+            return
+
+        session_id, action = self._session_route(path)
+        if session_id and action == "graph":
+            try:
+                self._send_json(self.server.registry.get(session_id).graph())
+            except KeyError:
+                self._send_error_json(HTTPStatus.NOT_FOUND, "session not found")
+            return
+
+        if session_id and action == "events":
+            try:
+                self._send_events(self.server.registry.get(session_id))
+            except KeyError:
+                self._send_error_json(HTTPStatus.NOT_FOUND, "session not found")
+            return
+
+        self._send_static(path)
+
+    def do_POST(self):
+        path = urllib.parse.urlparse(self.path).path
+
+        if path == "/api/sessions":
+            try:
+                payload = self._read_json()
+            except ValueError as error:
+                self._send_error_json(HTTPStatus.BAD_REQUEST, str(error))
+                return
+
+            topology = payload.get("graph")
+
+            if not isinstance(topology, dict):
+                self._send_error_json(HTTPStatus.BAD_REQUEST, "expected graph topology")
+                return
+
+            session = self.server.registry.create(str(payload.get("name") or "Graph"), topology)
+            self._send_json({"session": session.summary(), "graph": session.graph()}, status=HTTPStatus.CREATED)
+            return
+
+        session_id, action = self._session_route(path)
+        if session_id and action == "step":
+            try:
+                event = self.server.registry.get(session_id).step()
+            except KeyError:
+                self._send_error_json(HTTPStatus.NOT_FOUND, "session not found")
+                return
+            except ValueError as error:
+                self._send_error_json(HTTPStatus.BAD_REQUEST, str(error))
+                return
+
+            self._send_json({"event": event})
+            return
+
+        self._send_error_json(HTTPStatus.NOT_FOUND, "unknown endpoint")
+
+    def do_PATCH(self):
+        path = urllib.parse.urlparse(self.path).path
+        session_id, action = self._session_route(path)
+
+        if session_id and action == "values":
+            try:
+                payload = self._read_json()
+            except ValueError as error:
+                self._send_error_json(HTTPStatus.BAD_REQUEST, str(error))
+                return
+
+            try:
+                event = self.server.registry.get(session_id).set_value(
+                    str(payload.get("node") or ""),
+                    str(payload.get("pin") or PIN_ID),
+                    payload.get("value", "N"),
+                )
+            except KeyError:
+                self._send_error_json(HTTPStatus.NOT_FOUND, "session not found")
+                return
+            except ValueError as error:
+                self._send_error_json(HTTPStatus.BAD_REQUEST, str(error))
+                return
+
+            self._send_json({"event": event})
+            return
+
+        self._send_error_json(HTTPStatus.NOT_FOUND, "unknown endpoint")
+
+    def log_message(self, format: str, *args: object) -> None:
+        return
+
+    def _session_route(self, path: str) -> tuple[str | None, str | None]:
+        parts = [urllib.parse.unquote(part) for part in path.strip("/").split("/")]
+
+        if len(parts) == 4 and parts[:2] == ["api", "sessions"]:
+            return parts[2], parts[3]
+
+        return None, None
+
+    def _read_json(self) -> dict[str, Any]:
+        length = int(self.headers.get("Content-Length", "0"))
+        body = self.rfile.read(length).decode("utf-8") if length else "{}"
+
+        try:
+            payload = json.loads(body)
+        except json.JSONDecodeError as error:
+            raise ValueError(f"invalid JSON: {error}") from error
+
+        if not isinstance(payload, dict):
+            raise ValueError("expected JSON object")
+
+        return payload
+
+    def _send_json(self, payload: dict[str, Any], *, status: HTTPStatus = HTTPStatus.OK) -> None:
+        body = json.dumps(payload, ensure_ascii=True).encode("utf-8")
+        self.send_response(status)
+        self._send_cors_headers()
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def _send_error_json(self, status: HTTPStatus, message: str) -> None:
+        self._send_json({"error": message}, status=status)
+
+    def _send_events(self, session: GraphSession) -> None:
+        self.send_response(HTTPStatus.OK)
+        self._send_cors_headers()
+        self.send_header("Content-Type", "text/event-stream; charset=utf-8")
+        self.send_header("Cache-Control", "no-cache")
+        self.send_header("Connection", "keep-alive")
+        self.send_header("X-Accel-Buffering", "no")
+        self.end_headers()
+
+        events = session.event_hub.subscribe()
+
+        try:
+            while True:
+                try:
+                    event = events.get(timeout=15)
+                    self._write_sse("frame", event)
+                except queue.Empty:
+                    self.wfile.write(b": ping\n\n")
+                    self.wfile.flush()
+        except (BrokenPipeError, ConnectionResetError, OSError):
+            pass
+        finally:
+            session.event_hub.unsubscribe(events)
+
+    def _write_sse(self, event_name: str, payload: dict[str, Any]) -> None:
+        data = json.dumps(payload, ensure_ascii=True)
+        self.wfile.write(f"event: {event_name}\n".encode("utf-8"))
+        self.wfile.write(f"data: {data}\n\n".encode("utf-8"))
+        self.wfile.flush()
+
+    def _send_static(self, path: str) -> None:
+        if not self.server.frontend_dir.exists():
+            self._send_missing_frontend()
+            return
+
+        requested = "/index.html" if path == "/" else path
+        target = (self.server.frontend_dir / requested.lstrip("/")).resolve()
+
+        try:
+            target.relative_to(self.server.frontend_dir)
+        except ValueError:
+            self.send_error(HTTPStatus.FORBIDDEN)
+            return
+
+        if target.is_dir():
+            target = target / "index.html"
+
+        if not target.exists():
+            target = self.server.frontend_dir / "index.html"
+
+        if not target.exists():
+            self._send_missing_frontend()
+            return
+
+        content = target.read_bytes()
+        content_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
+
+        if target.suffix == ".js":
+            content_type = "text/javascript"
+
+        self.send_response(HTTPStatus.OK)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(content)))
+        self.end_headers()
+        self.wfile.write(content)
+
+    def _send_missing_frontend(self) -> None:
+        body = (
+            "<!doctype html><title>Frontend is not built</title>"
+            "<main style='font-family: system-ui; padding: 32px'>"
+            "<h1>Frontend is not built</h1>"
+            "<p>Run <code>cd frontend && npm install && npm run build</code>.</p>"
+            "</main>"
+        ).encode("utf-8")
+
+        self.send_response(HTTPStatus.SERVICE_UNAVAILABLE)
+        self.send_header("Content-Type", "text/html; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def _send_cors_headers(self) -> None:
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Access-Control-Allow-Methods", "GET, POST, PATCH, OPTIONS")
+        self.send_header("Access-Control-Allow-Headers", "Content-Type")
+
+
+_server_lock = threading.Lock()
+_server: GraphHTTPServer | None = None
+_server_url: str | None = None
+_registry: SessionRegistry | None = None
+
+
+def ensure_graph_server(values: dict[str, Tern]) -> str:
+    global _registry
+    global _server
+    global _server_url
+
+    with _server_lock:
+        if _server is not None and _server_url is not None:
+            return _server_url
+
+        host = os.environ.get("SIMPY_TERNS_GRAPH_HOST", DEFAULT_HOST)
+        port = int(os.environ.get("SIMPY_TERNS_GRAPH_PORT", DEFAULT_PORT))
+        _registry = SessionRegistry(values)
+
+        for candidate_port in range(port, port + 50):
+            try:
+                server = GraphHTTPServer((host, candidate_port), values, _registry)
+                break
+            except OSError:
+                continue
+        else:
+            raise RuntimeError(f"could not bind graph server starting at {host}:{port}")
+
+        _server = server
+        _server_url = f"http://{host}:{server.server_port}"
+
+        threading.Thread(
+            target=server.serve_forever,
+            name="graph-http-server",
+            daemon=True,
+        ).start()
+
+        return _server_url
+
+
+def render_graph_process(
+    env: simpy.Environment,
+    values: dict[str, Tern],
+):
+    url = ensure_graph_server(values)
+    print(f"[graph] {url}")
+
+    last_values: dict[str, str] | None = None
+    frame_number = 0
+
+    while True:
+        current_values = snapshot_values(values)
+
+        if last_values is None:
+            changed = [
+                {"node": node, "pin": PIN_ID, "from": value, "to": value}
+                for node, value in current_values.items()
+            ]
+        else:
+            changed = [
+                {
+                    "node": node,
+                    "pin": PIN_ID,
+                    "from": last_values.get(node, "N"),
+                    "to": value,
+                }
+                for node, value in current_values.items()
+                if last_values.get(node) != value
+            ]
+
+        if changed:
+            frame_number += 1
+            if _registry is not None:
+                _registry.sync_simpy(
+                    current_values,
+                    frame=frame_number,
+                    time_value=env.now,
+                    changed=changed,
+                )
+
+        last_values = current_values
+        yield env.timeout(0.1)

processes.py

@@ -0,0 +1,87 @@
+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)

transition.py

@@ -0,0 +1,53 @@
+NODES_N = {
+    ("3", "A", "1"),
+    ("2", "1", "B"),
+    ("3", "2", "C"),
+}
+
+NODES_N0 = {
+    ("A", "1", "A'"),
+    ("B", "1", "B'"),
+    ("A B", "A", "AB'"),
+    ("A'B", "A'", "A'B'"),
+    ("A'B", "B", "AB"),
+    ("A'B'", "B'", "AB'"),
+}
+
+TRANSITIONS: dict[str, str] = {
+    "000": "000",
+    "001": "011",  # CHANGED (was "0NN") — NOT(0) fires: child wins
+    "00N": "000",
+    "010": "NNN",
+    "011": "011",
+    "01N": "011",
+    "0N0": "000",
+    "0N1": "011",
+    "0NN": "0NN",
+    "100": "110",  # CHANGED (was "NN0") — NOT(0) fires: child wins
+    "101": "101",  # CHANGED (was "NNN") — NOT(1) holds: result is P="0"
+    "10N": "10N",  # CHANGED (was "NNN") — NOT waiting state: hold
+    "110": "110",
+    "111": "111",
+    "11N": "11N",
+    "1N0": "110",
+    "1N1": "111",
+    "1NN": "11N",
+    "N00": "000",
+    "N01": "NNN",
+    "N0N": "000",  # UNCHANGED — required, per design constraint
+    "N10": "110",
+    "N11": "N11",
+    "N1N": "N1N",
+    "NN0": "NN0",
+    "NN1": "N11",
+    "NNN": "NNN",
+}
+
+# Original table kept for reference / A-B testing.
+TRANSITIONS_ORIGINAL: dict[str, str] = {
+    **TRANSITIONS,
+    "100": "NN0",
+    "101": "NNN",
+    "10N": "NNN",
+    "001": "0NN",
+}

untitled.c

@@ -18,7 +18,7 @@ void rec (char *str) {
     }
 }
 
-int main2() {
+int main(void) {
     char str[] = "0000\0";
     printf("digraph {\n");
     rec(str);
@@ -34,9 +34,3 @@ int main2() {
     }
     printf("\n}\n");
 }
-
-int main(void) {
-    unsigned short int x = 65535; // 1111 1111 1111 1111
-    short int y = x;
-    printf("%d", y);
-}