From c2895dfba8eb83f7666f5017981f964bd5eaa5c6 Mon Sep 17 00:00:00 2001 From: gregorbednov Date: Sat, 20 Jun 2026 13:26:29 +0300 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?/=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- graph_n.py | 163 ++-------- graph_view.py | 831 ++++++++++++++++++++++++++++++++++++++++++++++++++ processes.py | 87 ++++++ transition.py | 53 ++++ untitled.c | 8 +- 5 files changed, 999 insertions(+), 143 deletions(-) create mode 100644 graph_view.py create mode 100644 processes.py create mode 100644 transition.py diff --git a/graph_n.py b/graph_n.py index 381f7a4..c041296 100644 --- a/graph_n.py +++ b/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 = { - ("A", "1", "A'"), - ("B", "1", "B'"), - ("AB", "A", "AB'"), - ("A'B", "A'", "A'B'"), - ("A'B", "B", "AB"), - ("A'B'", "B'", "AB'"), -} +from processes import input_thread, node, render_process, user_input_process +from tern import Tern -f = { - "000": "000", - "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", -} +def initial_values(edges) -> dict[str, Tern]: + return {x: Tern.U for triple in edges for x in triple} -class Tern(Enum): - N = -1 - U = 0 - Y = 1 +def main(): + from graph_view import active_edges, render_graph_process - @override - def __str__(self): - if self.value == -1: - return "0" - if self.value == 1: - return "1" - return "N" + edges = active_edges() + values = initial_values(edges) + commands: queue.Queue[tuple[str, Tern]] = queue.Queue() - @classmethod - def from_string(cls, s: str) -> "Tern": - if s == "N": - return Tern.U - if s == "0": - return Tern.N - if s == "1": - return Tern.Y - return cls[s] + env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False) + + threading.Thread( + target=input_thread, + args=(commands,), + daemon=True, + ).start() + + 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]): - while True: - 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) +if __name__ == "__main__": + main() diff --git a/graph_view.py b/graph_view.py new file mode 100644 index 0000000..3c5fa07 --- /dev/null +++ b/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 = ( + "Frontend is not built" + "
" + "

Frontend is not built

" + "

Run cd frontend && npm install && npm run build.

" + "
" + ).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) diff --git a/processes.py b/processes.py new file mode 100644 index 0000000..df293c2 --- /dev/null +++ b/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) diff --git a/transition.py b/transition.py new file mode 100644 index 0000000..2119a59 --- /dev/null +++ b/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", +} diff --git a/untitled.c b/untitled.c index da3b313..1a9b9cc 100644 --- a/untitled.c +++ b/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); -}