deepcad-canonical-equivalence

#!/usr/bin/env python3
"""DeepCAD canonical-equivalence solver.

Reads a pairs JSON, decides for each pair whether program_a and program_b
describe the same canonical solid, and writes predictions.json.
"""

import json
import re
import sys
from collections import Counter


# --------------------------------------------------------------------------
# Command classification
# --------------------------------------------------------------------------
SUPPORTED = {"param", "sketch", "rect", "circle", "slot", "extrude"}
PROFILE_KINDS = {"rect", "circle", "slot"}
ANNOTATION = {"constraint", "dimension", "note", "metadata", "view", "comment"}


def is_construction(cmd):
    return bool(cmd.get("construction", False))


# --------------------------------------------------------------------------
# Numeric helpers
# --------------------------------------------------------------------------
def norm_num(x):
    """Quantise to the 1e-6 tolerance and kill negative zero."""
    v = round(float(x), 6) + 0.0
    if v == 0:
        v = 0.0
    return v


# --------------------------------------------------------------------------
# Expression evaluator: numeric constants, params, + - * /, parentheses
# --------------------------------------------------------------------------
_TOKEN_NUM = re.compile(r"[0-9]*\.?[0-9]+([eE][+-]?[0-9]+)?")
_TOKEN_ID = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")


def tokenize(s):
    tokens = []
    i = 0
    n = len(s)
    while i < n:
        c = s[i]
        if c.isspace():
            i += 1
            continue
        if c in "+-*/()":
            tokens.append(c)
            i += 1
            continue
        if c.isdigit() or c == ".":
            m = _TOKEN_NUM.match(s, i)
            if not m:
                raise ValueError("bad number in %r" % s)
            tokens.append(("num", float(m.group())))
            i = m.end()
            continue
        m = _TOKEN_ID.match(s, i)
        if m:
            tokens.append(("id", m.group()))
            i = m.end()
            continue
        raise ValueError("bad char %r in %r" % (c, s))
    return tokens


class Evaluator:
    def __init__(self, params):
        # params: name -> raw expr (string or number)
        self.params = params
        self.memo = {}
        self.stack = set()

    def value(self, name):
        if name in self.memo:
            return self.memo[name]
        if name not in self.params:
            raise ValueError("undefined param %r" % name)
        if name in self.stack:
            raise ValueError("cyclic param %r" % name)
        self.stack.add(name)
        v = self.eval(self.params[name])
        self.stack.discard(name)
        self.memo[name] = v
        return v

    def eval(self, expr):
        if isinstance(expr, bool):
            raise ValueError("bool not numeric")
        if isinstance(expr, (int, float)):
            return float(expr)
        toks = tokenize(str(expr))
        self._toks = toks
        self._pos = 0
        v = self._expr()
        if self._pos != len(toks):
            raise ValueError("trailing tokens in %r" % expr)
        return v

    def _peek(self):
        if self._pos < len(self._toks):
            return self._toks[self._pos]
        return None

    def _next(self):
        t = self._toks[self._pos]
        self._pos += 1
        return t

    def _expr(self):
        v = self._term()
        while True:
            t = self._peek()
            if t == "+":
                self._next()
                v += self._term()
            elif t == "-":
                self._next()
                v -= self._term()
            else:
                break
        return v

    def _term(self):
        v = self._factor()
        while True:
            t = self._peek()
            if t == "*":
                self._next()
                v *= self._factor()
            elif t == "/":
                self._next()
                v /= self._factor()
            else:
                break
        return v

    def _factor(self):
        t = self._peek()
        if t == "+":
            self._next()
            return self._factor()
        if t == "-":
            self._next()
            return -self._factor()
        if t == "(":
            self._next()
            v = self._expr()
            if self._peek() != ")":
                raise ValueError("missing )")
            self._next()
            return v
        if isinstance(t, tuple) and t[0] == "num":
            self._next()
            return t[1]
        if isinstance(t, tuple) and t[0] == "id":
            self._next()
            return self.value(t[1])
        raise ValueError("unexpected token %r" % (t,))


# --------------------------------------------------------------------------
# Profile canonicalisation
# --------------------------------------------------------------------------
def canon_field(name, v, ev):
    if isinstance(v, list):
        return [canon_field(name, x, ev) for x in v]
    if isinstance(v, bool):
        return v
    if isinstance(v, (int, float)):
        num = float(v)
    elif isinstance(v, str):
        try:
            num = ev.eval(v)
        except Exception:
            return v  # opaque token inside a profile field
    else:
        return v
    if name == "angle":
        num = num % 180.0
    return norm_num(num)


def canon_profile(cmd, ev, planes):
    plane = planes.get(cmd.get("sketch"))
    fields = {}
    for k, v in cmd.items():
        if k in ("cmd", "id", "sketch", "construction"):
            continue
        fields[k] = canon_field(k, v, ev)
    return json.dumps(
        {"kind": cmd["cmd"], "plane": plane, "fields": fields},
        sort_keys=True,
    )


# --------------------------------------------------------------------------
# Extrude canonicalisation
# --------------------------------------------------------------------------
def canon_direction(v, ev):
    if isinstance(v, str):
        return v.lower()
    if isinstance(v, list):
        return [canon_field("dir", x, ev) for x in v]
    if isinstance(v, (int, float)) and not isinstance(v, bool):
        return norm_num(v)
    return v


def canon_extrude_content(cmd, ev, profiles):
    op = str(cmd.get("operation", "new")).lower()
    extent = str(cmd.get("extent", "one_side")).lower()
    direction = canon_direction(cmd.get("direction", "normal"), ev)
    depth = norm_num(ev.eval(cmd.get("depth", 0)))

    refs = []
    if "profiles" in cmd:
        refs.extend(cmd["profiles"])
    if "profile" in cmd:
        refs.append(cmd["profile"])

    prof_sigs = []
    for r in refs:
        if r in profiles:
            prof_sigs.append(profiles[r])
    prof_sigs.sort()

    content = {
        "op": op,
        "extent": extent,
        "direction": direction,
        "depth": depth,
        "profiles": prof_sigs,
    }
    return json.dumps(content, sort_keys=True)


def partition_key(cmd):
    if "body" in cmd:
        return ("L", str(cmd["body"]))
    if "channel" in cmd:
        return ("L", str(cmd["channel"]))
    return ("DEFAULT",)


# --------------------------------------------------------------------------
# Program signature
# --------------------------------------------------------------------------
def program_signature(program):
    # Build param table
    param_table = {}
    for cmd in program:
        if cmd.get("cmd") == "param":
            param_table[cmd.get("name")] = cmd.get("expr")
    ev = Evaluator(param_table)

    # Build sketch plane table
    planes = {}
    for cmd in program:
        if cmd.get("cmd") == "sketch":
            plane = cmd.get("plane")
            if isinstance(plane, str):
                plane = plane.upper()
            planes[cmd.get("id")] = plane

    # Build profile table (non-construction profiles only)
    profiles = {}
    for cmd in program:
        if cmd.get("cmd") in PROFILE_KINDS and not is_construction(cmd):
            profiles[cmd.get("id")] = canon_profile(cmd, ev, planes)

    # Collect extrude features and opaque payloads
    features = []  # (partition_key, content_json)
    opaque = Counter()
    for cmd in program:
        if is_construction(cmd):
            continue
        kind = cmd.get("cmd")
        if kind in ANNOTATION:
            continue
        if kind == "extrude":
            content = canon_extrude_content(cmd, ev, profiles)
            features.append((partition_key(cmd), content))
        elif kind in SUPPORTED:
            continue  # param/sketch/profile: consumed via tables
        else:
            payload = {k: v for k, v in cmd.items() if k != "id"}
            opaque[json.dumps(payload, sort_keys=True)] += 1

    # Group features by partition, canonicalise as multiset of group-multisets
    groups = {}
    for key, content in features:
        groups.setdefault(key, []).append(content)
    group_sigs = Counter()
    for key, contents in groups.items():
        contents.sort()
        group_sigs[json.dumps(contents, sort_keys=True)] += 1

    return group_sigs, opaque


def equivalent(program_a, program_b):
    ga, oa = program_signature(program_a)
    gb, ob = program_signature(program_b)
    return ga == gb and oa == ob


def sig_string(program):
    g, o = program_signature(program)
    return json.dumps(
        {"groups": sorted(g.elements()), "opaque": sorted(o.elements())},
        sort_keys=True,
    )


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------
def main():
    if len(sys.argv) != 3:
        sys.stderr.write("usage: solve.py <input_json> <output_json>\n")
        sys.exit(2)

    with open(sys.argv[1]) as f:
        data = json.load(f)

    predictions = []
    for pair in data.get("pairs", []):
        pa = pair.get("program_a", [])
        pb = pair.get("program_b", [])
        try:
            eq = equivalent(pa, pb)
        except Exception:
            eq = False
        rec = {"pair_id": pair.get("pair_id"), "equivalent": bool(eq)}
        try:
            rec["signature_a"] = sig_string(pa)
            rec["signature_b"] = sig_string(pb)
        except Exception:
            pass
        predictions.append(rec)

    with open(sys.argv[2], "w") as f:
        json.dump({"predictions": predictions}, f, indent=2)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""DeepCAD canonical-equivalence solver.

Reads a pairs JSON, decides for each pair whether program_a and program_b
describe the same canonical solid, and writes predictions.json.
"""

import json
import re
import sys
from collections import Counter


# --------------------------------------------------------------------------
# Command classification
# --------------------------------------------------------------------------
SUPPORTED = {"param", "sketch", "rect", "circle", "slot", "extrude"}
PROFILE_KINDS = {"rect", "circle", "slot"}
ANNOTATION = {"constraint", "dimension", "note", "metadata", "view", "comment"}


def is_construction(cmd):
    return bool(cmd.get("construction", False))


# --------------------------------------------------------------------------
# Numeric helpers
# --------------------------------------------------------------------------
def norm_num(x):
    """Quantise to the 1e-6 tolerance and kill negative zero."""
    v = round(float(x), 6) + 0.0
    if v == 0:
        v = 0.0
    return v


# --------------------------------------------------------------------------
# Expression evaluator: numeric constants, params, + - * /, parentheses
# --------------------------------------------------------------------------
_TOKEN_NUM = re.compile(r"[0-9]*\.?[0-9]+([eE][+-]?[0-9]+)?")
_TOKEN_ID = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")


def tokenize(s):
    tokens = []
    i = 0
    n = len(s)
    while i < n:
        c = s[i]
        if c.isspace():
            i += 1
            continue
        if c in "+-*/()":
            tokens.append(c)
            i += 1
            continue
        if c.isdigit() or c == ".":
            m = _TOKEN_NUM.match(s, i)
            if not m:
                raise ValueError("bad number in %r" % s)
            tokens.append(("num", float(m.group())))
            i = m.end()
            continue
        m = _TOKEN_ID.match(s, i)
        if m:
            tokens.append(("id", m.group()))
            i = m.end()
            continue
        raise ValueError("bad char %r in %r" % (c, s))
    return tokens


class Evaluator:
    def __init__(self, params):
        # params: name -> raw expr (string or number)
        self.params = params
        self.memo = {}
        self.stack = set()

    def value(self, name):
        if name in self.memo:
            return self.memo[name]
        if name not in self.params:
            raise ValueError("undefined param %r" % name)
        if name in self.stack:
            raise ValueError("cyclic param %r" % name)
        self.stack.add(name)
        v = self.eval(self.params[name])
        self.stack.discard(name)
        self.memo[name] = v
        return v

    def eval(self, expr):
        if isinstance(expr, bool):
            raise ValueError("bool not numeric")
        if isinstance(expr, (int, float)):
            return float(expr)
        toks = tokenize(str(expr))
        self._toks = toks
        self._pos = 0
        v = self._expr()
        if self._pos != len(toks):
            raise ValueError("trailing tokens in %r" % expr)
        return v

    def _peek(self):
        if self._pos < len(self._toks):
            return self._toks[self._pos]
        return None

    def _next(self):
        t = self._toks[self._pos]
        self._pos += 1
        return t

    def _expr(self):
        v = self._term()
        while True:
            t = self._peek()
            if t == "+":
                self._next()
                v += self._term()
            elif t == "-":
                self._next()
                v -= self._term()
            else:
                break
        return v

    def _term(self):
        v = self._factor()
        while True:
            t = self._peek()
            if t == "*":
                self._next()
                v *= self._factor()
            elif t == "/":
                self._next()
                v /= self._factor()
            else:
                break
        return v

    def _factor(self):
        t = self._peek()
        if t == "+":
            self._next()
            return self._factor()
        if t == "-":
            self._next()
            return -self._factor()
        if t == "(":
            self._next()
            v = self._expr()
            if self._peek() != ")":
                raise ValueError("missing )")
            self._next()
            return v
        if isinstance(t, tuple) and t[0] == "num":
            self._next()
            return t[1]
        if isinstance(t, tuple) and t[0] == "id":
            self._next()
            return self.value(t[1])
        raise ValueError("unexpected token %r" % (t,))


# --------------------------------------------------------------------------
# Profile canonicalisation
# --------------------------------------------------------------------------
def canon_field(name, v, ev):
    if isinstance(v, list):
        return [canon_field(name, x, ev) for x in v]
    if isinstance(v, bool):
        return v
    if isinstance(v, (int, float)):
        num = float(v)
    elif isinstance(v, str):
        try:
            num = ev.eval(v)
        except Exception:
            return v  # opaque token inside a profile field
    else:
        return v
    if name == "angle":
        num = num % 180.0
    return norm_num(num)


def canon_profile(cmd, ev, planes):
    plane = planes.get(cmd.get("sketch"))
    fields = {}
    for k, v in cmd.items():
        if k in ("cmd", "id", "sketch", "construction"):
            continue
        fields[k] = canon_field(k, v, ev)
    return json.dumps(
        {"kind": cmd["cmd"], "plane": plane, "fields": fields},
        sort_keys=True,
    )


# --------------------------------------------------------------------------
# Extrude canonicalisation
# --------------------------------------------------------------------------
def canon_direction(v, ev):
    if isinstance(v, str):
        return v.lower()
    if isinstance(v, list):
        return [canon_field("dir", x, ev) for x in v]
    if isinstance(v, (int, float)) and not isinstance(v, bool):
        return norm_num(v)
    return v


def canon_extrude_content(cmd, ev, profiles):
    op = str(cmd.get("operation", "new")).lower()
    extent = str(cmd.get("extent", "one_side")).lower()
    direction = canon_direction(cmd.get("direction", "normal"), ev)
    depth = norm_num(ev.eval(cmd.get("depth", 0)))

    refs = []
    if "profiles" in cmd:
        refs.extend(cmd["profiles"])
    if "profile" in cmd:
        refs.append(cmd["profile"])

    prof_sigs = []
    for r in refs:
        if r in profiles:
            prof_sigs.append(profiles[r])
    prof_sigs.sort()

    content = {
        "op": op,
        "extent": extent,
        "direction": direction,
        "depth": depth,
        "profiles": prof_sigs,
    }
    return json.dumps(content, sort_keys=True)


def partition_key(cmd):
    if "body" in cmd:
        return ("L", str(cmd["body"]))
    if "channel" in cmd:
        return ("L", str(cmd["channel"]))
    return ("DEFAULT",)


# --------------------------------------------------------------------------
# Program signature
# --------------------------------------------------------------------------
def program_signature(program):
    # Build param table
    param_table = {}
    for cmd in program:
        if cmd.get("cmd") == "param":
            param_table[cmd.get("name")] = cmd.get("expr")
    ev = Evaluator(param_table)

    # Build sketch plane table
    planes = {}
    for cmd in program:
        if cmd.get("cmd") == "sketch":
            plane = cmd.get("plane")
            if isinstance(plane, str):
                plane = plane.upper()
            planes[cmd.get("id")] = plane

    # Build profile table (non-construction profiles only)
    profiles = {}
    for cmd in program:
        if cmd.get("cmd") in PROFILE_KINDS and not is_construction(cmd):
            profiles[cmd.get("id")] = canon_profile(cmd, ev, planes)

    # Collect extrude features and opaque payloads
    features = []  # (partition_key, content_json)
    opaque = Counter()
    for cmd in program:
        if is_construction(cmd):
            continue
        kind = cmd.get("cmd")
        if kind in ANNOTATION:
            continue
        if kind == "extrude":
            content = canon_extrude_content(cmd, ev, profiles)
            features.append((partition_key(cmd), content))
        elif kind in SUPPORTED:
            continue  # param/sketch/profile consumed via lookup tables
        else:
            payload = {k: v for k, v in cmd.items() if k != "id"}
            opaque[json.dumps(payload, sort_keys=True)] += 1

    # Group features by partition -> multiset of group-multisets
    groups = {}
    for key, content in features:
        groups.setdefault(key, []).append(content)
    group_sigs = Counter()
    for key, contents in groups.items():
        contents.sort()
        group_sigs[json.dumps(contents, sort_keys=True)] += 1

    return group_sigs, opaque


def equivalent(program_a, program_b):
    ga, oa = program_signature(program_a)
    gb, ob = program_signature(program_b)
    return ga == gb and oa == ob


def sig_string(program):
    g, o = program_signature(program)
    return json.dumps(
        {"groups": sorted(g.elements()), "opaque": sorted(o.elements())},
        sort_keys=True,
    )


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------
def main():
    if len(sys.argv) != 3:
        sys.stderr.write("usage: solve.py <input_json> <output_json>\n")
        return 2

    with open(sys.argv[1]) as f:
        data = json.load(f)

    predictions = []
    for pair in data.get("pairs", []):
        pa = pair.get("program_a", [])
        pb = pair.get("program_b", [])
        try:
            eq = equivalent(pa, pb)
        except Exception:
            eq = False
        rec = {"pair_id": pair.get("pair_id"), "equivalent": bool(eq)}
        try:
            rec["signature_a"] = sig_string(pa)
            rec["signature_b"] = sig_string(pb)
        except Exception:
            pass
        predictions.append(rec)

    with open(sys.argv[2], "w") as f:
        json.dump({"predictions": predictions}, f, indent=2)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())