deepcad-canonical-equivalence

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

Reads a JSON file of program pairs and decides, for each pair, whether the two
compact CAD command programs describe the same canonical solid.  The decision is
purely symbolic/structural: we normalize the aliases described in the task and
compare canonicalized field values.
"""

import sys
import json
import ast
import operator
import hashlib

# ---------------------------------------------------------------------------
# Command classification
# ---------------------------------------------------------------------------

# Annotation / metadata command kinds that never affect the solid.
METADATA_KINDS = {"constraint", "dimension", "note", "metadata", "view", "comment"}

# Command kinds we interpret with defined semantics.  Anything else that is not
# a construction command and not a metadata kind is an opaque solid-affecting
# payload.
PROFILE_KINDS = {"rect", "circle", "slot"}

DEFAULT_PARTITION = "\x00__default_partition__"

# ---------------------------------------------------------------------------
# Arithmetic expression evaluation
# ---------------------------------------------------------------------------

_BINOPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
}


def _eval_node(node, resolve):
    if isinstance(node, ast.Expression):
        return _eval_node(node.body, resolve)
    if isinstance(node, ast.Constant):
        if isinstance(node.value, bool) or not isinstance(node.value, (int, float)):
            raise ValueError("non-numeric constant")
        return float(node.value)
    if isinstance(node, ast.Num):  # pragma: no cover - legacy python
        return float(node.n)
    if isinstance(node, ast.BinOp) and type(node.op) in _BINOPS:
        return _BINOPS[type(node.op)](
            _eval_node(node.left, resolve), _eval_node(node.right, resolve)
        )
    if isinstance(node, ast.UnaryOp):
        val = _eval_node(node.operand, resolve)
        if isinstance(node.op, ast.USub):
            return -val
        if isinstance(node.op, ast.UAdd):
            return +val
        raise ValueError("bad unary op")
    if isinstance(node, ast.Name):
        return resolve(node.id)
    raise ValueError("unsupported expression node: %r" % (node,))


def _eval_value(expr, resolve):
    if isinstance(expr, bool):
        raise ValueError("bool not numeric")
    if isinstance(expr, (int, float)):
        return float(expr)
    if isinstance(expr, str):
        tree = ast.parse(expr, mode="eval")
        return _eval_node(tree, resolve)
    raise ValueError("cannot evaluate %r" % (expr,))


class ParamResolver:
    """Lazily evaluates parameters, allowing params to reference other params."""

    def __init__(self, defs):
        self.defs = defs
        self.cache = {}
        self.active = set()

    def resolve(self, name):
        if name in self.cache:
            return self.cache[name]
        if name in self.active:
            raise ValueError("cyclic parameter reference: %s" % name)
        if name not in self.defs:
            raise ValueError("unknown parameter: %s" % name)
        self.active.add(name)
        try:
            val = _eval_value(self.defs[name], self.resolve)
        finally:
            self.active.discard(name)
        self.cache[name] = val
        return val


# ---------------------------------------------------------------------------
# Numeric canonicalization
# ---------------------------------------------------------------------------

def _fnum(x):
    """Round a number to the comparison tolerance and format deterministically."""
    x = round(float(x) + 0.0, 6)
    if x == 0:
        x = 0.0
    return format(x, ".6f")


def _leaf(value, resolve):
    """Canonicalize a scalar geometry value (number or arithmetic expression)."""
    try:
        return _fnum(_eval_value(value, resolve))
    except Exception:
        return "raw:" + str(value)


def _canon_val(value, resolve):
    if isinstance(value, list):
        return [_canon_val(v, resolve) for v in value]
    return _leaf(value, resolve)


# ---------------------------------------------------------------------------
# Profile & extrude canonicalization
# ---------------------------------------------------------------------------

def _profile_sig(cmd, plane, resolve):
    kind = cmd.get("cmd")
    canon = {}
    for key, val in cmd.items():
        if key in ("cmd", "id", "sketch", "construction"):
            continue
        if kind == "slot" and key == "angle":
            try:
                ang = _eval_value(val, resolve) % 180.0
                canon[key] = _fnum(ang)
            except Exception:
                canon[key] = "raw:" + str(val)
        else:
            canon[key] = _canon_val(val, resolve)
    items = [[k, canon[k]] for k in sorted(canon.keys())]
    plane_str = plane if plane else "\x00__noplane__"
    return ["profile", kind, plane_str, items]


def _feature_sig(cmd, profile_table, resolve):
    op = str(cmd.get("operation", "new")).lower()
    extent = str(cmd.get("extent", "one_side")).lower()
    direction = str(cmd.get("direction", "normal")).lower()
    if "depth" in cmd:
        depth = _leaf(cmd["depth"], resolve)
    else:
        depth = _fnum(0)

    pids = []
    if "profile" in cmd:
        pids.append(cmd["profile"])
    if "profiles" in cmd:
        extend = cmd["profiles"]
        if isinstance(extend, list):
            pids.extend(extend)
        else:
            pids.append(extend)

    prof_strings = []
    for pid in pids:
        if pid in profile_table:
            prof_strings.append(json.dumps(profile_table[pid], sort_keys=True))
    prof_strings.sort()

    sig = ["extrude", op, extent, direction, depth, prof_strings]

    if "body" in cmd:
        label = cmd["body"]
    elif "channel" in cmd:
        label = cmd["channel"]
    else:
        label = None
    return sig, label


# ---------------------------------------------------------------------------
# Program signature
# ---------------------------------------------------------------------------

def program_signature(program):
    params = {}
    sketches = {}
    profile_cmds = []
    extrude_cmds = []
    opaque = []

    if not isinstance(program, list):
        program = []

    for cmd in program:
        if not isinstance(cmd, dict):
            opaque.append(json.dumps(cmd, sort_keys=True))
            continue
        if cmd.get("construction") is True:
            continue
        kind = cmd.get("cmd")
        if kind in METADATA_KINDS:
            continue
        if kind == "param":
            params[cmd.get("name")] = cmd.get("expr")
        elif kind == "sketch":
            plane = cmd.get("plane")
            sketches[cmd.get("id")] = str(plane).lower() if plane is not None else ""
        elif kind in PROFILE_KINDS:
            profile_cmds.append(cmd)
        elif kind == "extrude":
            extrude_cmds.append(cmd)
        else:
            payload = {k: v for k, v in cmd.items() if k != "id"}
            opaque.append(json.dumps(payload, sort_keys=True))

    resolver = ParamResolver(params).resolve

    profile_table = {}
    for cmd in profile_cmds:
        plane = sketches.get(cmd.get("sketch"))
        profile_table[cmd.get("id")] = _profile_sig(cmd, plane, resolver)

    features = []
    for cmd in extrude_cmds:
        sig, label = _feature_sig(cmd, profile_table, resolver)
        features.append((sig, label))

    # Partition topology: group features by body/channel label; unlabeled
    # features share a single default partition.  Labels are renamable, so the
    # canonical form is the multiset of groups (each group a multiset of feature
    # signatures) with the label strings dropped.
    groups = {}
    for sig, label in features:
        key = label if label is not None else DEFAULT_PARTITION
        groups.setdefault(key, []).append(json.dumps(sig, sort_keys=True))

    partition = sorted(json.dumps(sorted(g)) for g in groups.values())
    opaque_sig = sorted(opaque)

    return json.dumps({"partition": partition, "opaque": opaque_sig}, sort_keys=True)


def _short(sig):
    return hashlib.md5(sig.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

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

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

    pairs = data.get("pairs", []) if isinstance(data, dict) else []
    predictions = []
    for pair in pairs:
        pid = pair.get("pair_id")
        sig_a = program_signature(pair.get("program_a", []))
        sig_b = program_signature(pair.get("program_b", []))
        predictions.append({
            "pair_id": pid,
            "equivalent": sig_a == sig_b,
            "signature_a": _short(sig_a),
            "signature_b": _short(sig_b),
        })

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


if __name__ == "__main__":
    sys.exit(main(sys.argv))

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

Reads a JSON file of program pairs and decides, for each pair, whether the two
compact CAD command programs describe the same canonical solid.  The decision is
purely symbolic/structural: we normalize the aliases described in the task and
compare canonicalized field values.
"""

import sys
import json
import ast
import operator
import hashlib

# ---------------------------------------------------------------------------
# Command classification
# ---------------------------------------------------------------------------

# Annotation / metadata command kinds that never affect the solid.
METADATA_KINDS = {"constraint", "dimension", "note", "metadata", "view", "comment"}

# Command kinds we interpret with defined semantics.  Anything else that is not
# a construction command and not a metadata kind is an opaque solid-affecting
# payload.
PROFILE_KINDS = {"rect", "circle", "slot"}

DEFAULT_PARTITION = "\x00__default_partition__"

# ---------------------------------------------------------------------------
# Arithmetic expression evaluation
# ---------------------------------------------------------------------------

_BINOPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
}


def _eval_node(node, resolve):
    if isinstance(node, ast.Expression):
        return _eval_node(node.body, resolve)
    if isinstance(node, ast.Constant):
        if isinstance(node.value, bool) or not isinstance(node.value, (int, float)):
            raise ValueError("non-numeric constant")
        return float(node.value)
    if isinstance(node, ast.Num):  # pragma: no cover - legacy python
        return float(node.n)
    if isinstance(node, ast.BinOp) and type(node.op) in _BINOPS:
        return _BINOPS[type(node.op)](
            _eval_node(node.left, resolve), _eval_node(node.right, resolve)
        )
    if isinstance(node, ast.UnaryOp):
        val = _eval_node(node.operand, resolve)
        if isinstance(node.op, ast.USub):
            return -val
        if isinstance(node.op, ast.UAdd):
            return +val
        raise ValueError("bad unary op")
    if isinstance(node, ast.Name):
        return resolve(node.id)
    raise ValueError("unsupported expression node: %r" % (node,))


def _eval_value(expr, resolve):
    if isinstance(expr, bool):
        raise ValueError("bool not numeric")
    if isinstance(expr, (int, float)):
        return float(expr)
    if isinstance(expr, str):
        tree = ast.parse(expr, mode="eval")
        return _eval_node(tree, resolve)
    raise ValueError("cannot evaluate %r" % (expr,))


class ParamResolver:
    """Lazily evaluates parameters, allowing params to reference other params."""

    def __init__(self, defs):
        self.defs = defs
        self.cache = {}
        self.active = set()

    def resolve(self, name):
        if name in self.cache:
            return self.cache[name]
        if name in self.active:
            raise ValueError("cyclic parameter reference: %s" % name)
        if name not in self.defs:
            raise ValueError("unknown parameter: %s" % name)
        self.active.add(name)
        try:
            val = _eval_value(self.defs[name], self.resolve)
        finally:
            self.active.discard(name)
        self.cache[name] = val
        return val


# ---------------------------------------------------------------------------
# Numeric canonicalization
# ---------------------------------------------------------------------------

def _fnum(x):
    """Round a number to the comparison tolerance and format deterministically."""
    x = round(float(x) + 0.0, 6)
    if x == 0:
        x = 0.0
    return format(x, ".6f")


def _leaf(value, resolve):
    """Canonicalize a scalar geometry value (number or arithmetic expression)."""
    try:
        return _fnum(_eval_value(value, resolve))
    except Exception:
        return "raw:" + str(value)


def _canon_val(value, resolve):
    if isinstance(value, list):
        return [_canon_val(v, resolve) for v in value]
    return _leaf(value, resolve)


# ---------------------------------------------------------------------------
# Profile & extrude canonicalization
# ---------------------------------------------------------------------------

def _profile_sig(cmd, plane, resolve):
    kind = cmd.get("cmd")
    canon = {}
    for key, val in cmd.items():
        if key in ("cmd", "id", "sketch", "construction"):
            continue
        if kind == "slot" and key == "angle":
            try:
                ang = _eval_value(val, resolve) % 180.0
                canon[key] = _fnum(ang)
            except Exception:
                canon[key] = "raw:" + str(val)
        else:
            canon[key] = _canon_val(val, resolve)
    items = [[k, canon[k]] for k in sorted(canon.keys())]
    plane_str = plane if plane else "\x00__noplane__"
    return ["profile", kind, plane_str, items]


def _feature_sig(cmd, profile_table, resolve):
    op = str(cmd.get("operation", "new")).lower()
    extent = str(cmd.get("extent", "one_side")).lower()
    direction = str(cmd.get("direction", "normal")).lower()
    if "depth" in cmd:
        depth = _leaf(cmd["depth"], resolve)
    else:
        depth = _fnum(0)

    pids = []
    if "profile" in cmd:
        pids.append(cmd["profile"])
    if "profiles" in cmd:
        extra = cmd["profiles"]
        if isinstance(extra, list):
            pids.extend(extra)
        else:
            pids.append(extra)

    prof_strings = []
    for pid in pids:
        if pid in profile_table:
            prof_strings.append(json.dumps(profile_table[pid], sort_keys=True))
    prof_strings.sort()

    sig = ["extrude", op, extent, direction, depth, prof_strings]

    if "body" in cmd:
        label = cmd["body"]
    elif "channel" in cmd:
        label = cmd["channel"]
    else:
        label = None
    return sig, label


# ---------------------------------------------------------------------------
# Program signature
# ---------------------------------------------------------------------------

def program_signature(program):
    params = {}
    sketches = {}
    profile_cmds = []
    extrude_cmds = []
    opaque = []

    if not isinstance(program, list):
        program = []

    for cmd in program:
        if not isinstance(cmd, dict):
            opaque.append(json.dumps(cmd, sort_keys=True))
            continue
        if cmd.get("construction") is True:
            continue
        kind = cmd.get("cmd")
        if kind in METADATA_KINDS:
            continue
        if kind == "param":
            params[cmd.get("name")] = cmd.get("expr")
        elif kind == "sketch":
            plane = cmd.get("plane")
            sketches[cmd.get("id")] = str(plane).lower() if plane is not None else ""
        elif kind in PROFILE_KINDS:
            profile_cmds.append(cmd)
        elif kind == "extrude":
            extrude_cmds.append(cmd)
        else:
            payload = {k: v for k, v in cmd.items() if k != "id"}
            opaque.append(json.dumps(payload, sort_keys=True))

    resolver = ParamResolver(params).resolve

    profile_table = {}
    for cmd in profile_cmds:
        plane = sketches.get(cmd.get("sketch"))
        profile_table[cmd.get("id")] = _profile_sig(cmd, plane, resolver)

    features = []
    for cmd in extrude_cmds:
        sig, label = _feature_sig(cmd, profile_table, resolver)
        features.append((sig, label))

    # Partition topology: group features by body/channel label; unlabeled
    # features share a single default partition.  Labels are renamable, so the
    # canonical form is the multiset of groups (each group a multiset of feature
    # signatures) with the label strings dropped.
    groups = {}
    for sig, label in features:
        key = label if label is not None else DEFAULT_PARTITION
        groups.setdefault(key, []).append(json.dumps(sig, sort_keys=True))

    partition = sorted(json.dumps(sorted(g)) for g in groups.values())
    opaque_sig = sorted(opaque)

    return json.dumps({"partition": partition, "opaque": opaque_sig}, sort_keys=True)


def _short(sig):
    return hashlib.md5(sig.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

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

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

    pairs = data.get("pairs", []) if isinstance(data, dict) else []
    predictions = []
    for pair in pairs:
        pid = pair.get("pair_id")
        sig_a = program_signature(pair.get("program_a", []))
        sig_b = program_signature(pair.get("program_b", []))
        predictions.append({
            "pair_id": pid,
            "equivalent": sig_a == sig_b,
            "signature_a": _short(sig_a),
            "signature_b": _short(sig_b),
        })

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


if __name__ == "__main__":
    sys.exit(main(sys.argv))