app.py

"""GraphLang interactive demo (Gradio) — DIDACTIC.

This is NOT the production GraphLang engine. It is a self-contained,
educational Python-AST reimplementation used only to illustrate the core
idea:
    source code  ->  semantic IR graph (12 kinds)  ->  hash-merge / dedup

The production engine (13 languages via tree-sitter, 238+296+242 CST node
types mapped to 12 IR kinds) is available under the MII Open License v1.1.
The benchmarks and datasets shipped alongside this demo were generated by
the REAL engine, not by this file.
"""

import ast
import hashlib
import json

import gradio as gr

from parallel_ir import detect_parallel_platform, normalize_thread_index

KINDS = {
    1: "function", 2: "if", 3: "for", 4: "while", 5: "return",
    6: "assign", 7: "call", 8: "binop", 9: "unary", 10: "var",
    11: "const", 12: "block",
}

_BINOPS = {
    ast.Add: "+", ast.Sub: "-", ast.Mult: "*", ast.Div: "/",
    ast.Eq: "==", ast.NotEq: "!=", ast.Lt: "<", ast.Gt: ">",
    ast.LtE: "<=", ast.GtE: ">=",
}


def _structural_hash(nid, nodes, memo):
    if nid in memo:
        return memo[nid]
    n = nodes[nid]
    children = tuple(_structural_hash(a, nodes, memo) for a in n["args"])
    content = json.dumps({
        "kind": n["kind"],
        "value": n.get("key", n.get("value")),
        "op": n["op"],
        "args": children,
    }, sort_keys=True)
    h = hashlib.sha256(content.encode()).hexdigest()[:12]
    memo[nid] = h
    return h


class Builder(ast.NodeVisitor):
    """Python AST -> GraphLang IR (didactic version)."""

    def __init__(self):
        self.nodes = {}
        self._n = 0
        self._names = {}

    def _canon(self, name):
        if name not in self._names:
            self._names[name] = f"v{len(self._names) + 1}"
        return self._names[name]

    def _var_node(self, name, args=None):
        return self._add("var", value=name, key=self._canon(name), args=args)

    def _add(self, kind, value=None, op="", args=None, key=None):
        self._n += 1
        nid = f"n{self._n}"
        node = {"kind": kind, "value": value, "op": op, "args": list(args or [])}
        if key is not None:
            node["key"] = key
        self.nodes[nid] = node
        return nid

    def build(self, code):
        self.nodes = {}
        self._n = 0
        self._names = {}
        tree = ast.parse(code)
        self.visit(tree)
        return self.nodes

    def visit_Module(self, node):
        return self._add("block", args=[self.visit(s) for s in node.body])

    def visit_FunctionDef(self, node):
        args = [self._var_node(a.arg) for a in node.args.args]
        body = [self.visit(s) for s in node.body]
        if len(body) == 1:
            body = body[0]
        else:
            body = self._add("block", args=body)
        return self._add("function", value=node.name, args=args + [body])

    def visit_Return(self, node):
        v = self.visit(node.value) if node.value else self._add("const", value=None)
        return self._add("return", args=[v])

    def visit_If(self, node):
        test = self.visit(node.test)
        then = self._add("block", args=[self.visit(s) for s in node.body])
        if node.orelse:
            orelse = self._add("block", args=[self.visit(s) for s in node.orelse])
            return self._add("if", args=[test, then, orelse])
        return self._add("if", args=[test, then])

    def visit_For(self, node):
        target = self._var_node(node.target.id)
        it = self.visit(node.iter)
        body = self._add("block", args=[self.visit(s) for s in node.body])
        return self._add("for", args=[target, it, body])

    def visit_While(self, node):
        test = self.visit(node.test)
        body = self._add("block", args=[self.visit(s) for s in node.body])
        return self._add("while", args=[test, body])

    def visit_Assign(self, node):
        val = self.visit(node.value)
        targets = [self._var_node(t.id) for t in node.targets
                   if isinstance(t, ast.Name)]
        return self._add("assign", args=targets + [val])

    def visit_Expr(self, node):
        return self.visit(node.value)

    def visit_Call(self, node):
        f = self.visit(node.func)
        return self._add("call", args=[f] + [self.visit(a) for a in node.args])

    def visit_Attribute(self, node):
        obj = self.visit(node.value)
        return self._var_node(node.attr, args=[obj])

    def visit_Name(self, node):
        return self._var_node(node.id)

    def visit_Constant(self, node):
        return self._add("const", value=node.value)

    def visit_BinOp(self, node):
        return self._add("binop", op=_BINOPS.get(type(node.op), "?"),
                         args=[self.visit(node.left), self.visit(node.right)])

    def visit_Compare(self, node):
        op = {ast.Eq: "==", ast.NotEq: "!=", ast.Lt: "<", ast.Gt: ">",
              ast.LtE: "<=", ast.GtE: ">="}.get(type(node.ops[0]), "?")
        return self._add("binop", op=op,
                         args=[self.visit(node.left), self.visit(node.comparators[0])])

    def visit_UnaryOp(self, node):
        op = {ast.USub: "-", ast.Not: "not", ast.UAdd: "+"}.get(type(node.op), "?")
        return self._add("unary", op=op, args=[self.visit(node.operand)])


def _graph_to_dot(nodes):
    lines = ["digraph G {", "  rankdir=TB;", '  node [shape=box, style=rounded];']
    for nid, n in nodes.items():
        label = n["kind"]
        if n.get("key"):
            label += f"\\n{n['value']} → {n['key']}"
        elif n["value"] not in (None, ""):
            label += f"\\n{n['value']}"
        if n["op"]:
            label += f" [{n['op']}]"
        lines.append(f'  {nid} [label="{label}"];')
    for nid, n in nodes.items():
        for a in n["args"]:
            lines.append(f"  {nid} -> {a};")
    lines.append("}")
    return "\n".join(lines)


def _node_hashes(nodes):
    memo = {}
    return {_structural_hash(nid, nodes, memo) for nid in nodes}


def inspect(code):
    if not code.strip():
        return "_(paste code)_", ""
    try:
        nodes = Builder().build(code)
    except SyntaxError as e:
        return f"SyntaxError: {e}", ""
    kinds = {}
    for n in nodes.values():
        kinds[n["kind"]] = kinds.get(n["kind"], 0) + 1
    summary = f"{len(nodes)} nodes — " + ", ".join(
        f"{k}×{v}" for k, v in sorted(kinds.items()))
    return summary, _graph_to_dot(nodes)


def merge(code_a, code_b):
    try:
        na = Builder().build(code_a)
        nb = Builder().build(code_b)
    except SyntaxError as e:
        return f"SyntaxError: {e}"
    ha, hb = _node_hashes(na), _node_hashes(nb)
    shared = ha & hb
    union = ha | hb
    sim = len(shared) / len(union) if union else 0.0
    total = len(na) + len(nb)
    unique = len(union)
    comp = total / unique if unique else 0.0
    return (f"Graph A: {len(na)} nodes\nGraph B: {len(nb)} nodes\n"
            f"Union (unique): {unique}\n"
            f"Structural similarity: {sim*100:.1f}%\n"
            f"Compression (A+B -> merged): {comp:.1f}x")


def parallel(code):
    plat = detect_parallel_platform(code) or "none"
    norm = normalize_thread_index(code) if plat != "none" else code
    return f"Platform: {plat}\n\nNormalized:\n{norm}"


KINDS_TABLE = "\n".join(f"| {i} | `{k}` |" for i, k in KINDS.items())

with gr.Blocks(title="GraphLang demo") as demo:
    gr.Markdown("""
# GraphLang — Universal Semantic Kernel for Code

Same intent = same graph. Paste Python code and see its canonical IR graph;
merge two snippets and measure structural deduplication.

> **Didactic demo.** This Space runs a simplified Python-AST reimplementation
> to illustrate the concept. The production engine normalizes **13 languages**
> via tree-sitter and is licensed separately (MII Open License v1.1).
> Benchmarks and datasets in the companion model repo are from the real engine.
""")

    with gr.Tabs():
        with gr.Tab("IR inspector"):
            with gr.Row():
                inp = gr.Code(language="python", lines=8,
                              value="def add(a, b):\n    return a + b",
                              label="Python code")
                with gr.Column():
                    summary = gr.Textbox(label="IR summary", interactive=False)
                    dot = gr.Code(language="dot", lines=14, label="IR graph (DOT)")
            btn = gr.Button("Build IR")
            btn.click(inspect, inputs=inp, outputs=[summary, dot])

        with gr.Tab("Merge / dedup"):
            with gr.Row():
                a = gr.Code(language="python", lines=6,
                            value="def add(a, b):\n    return a + b", label="Graph A")
                b = gr.Code(language="python", lines=6,
                            value="def add(x, y):\n    return x + y", label="Graph B")
            out = gr.Textbox(label="Result", interactive=False)
            mbtn = gr.Button("Merge")
            mbtn.click(merge, inputs=[a, b], outputs=out)

        with gr.Tab("Parallel IR"):
            pc = gr.Code(language="cpp", lines=6,
                         value="int i = threadIdx.x + blockIdx.x * blockDim.x;",
                         label="GPU code")
            pout = gr.Textbox(label="Detection + normalization", interactive=False)
            pbtn = gr.Button("Analyze")
            pbtn.click(parallel, inputs=pc, outputs=pout)

        with gr.Tab("The 12 kinds"):
            gr.Markdown("| # | Kind |\n|---|------|\n" + KINDS_TABLE)

if __name__ == "__main__":
    demo.launch()