deepcad-canonical-equivalence

#!/usr/bin/env python3
"""DeepCAD Canonical Equivalence solver.

Reads an input JSON of program pairs, decides for each pair whether both
programs describe the same canonical solid, and writes predictions.json.

Usage:
    python3 solve.py <input_json> <output_json>
"""

import ast
import json
import sys


ANNOTATION = {"constraint", "dimension", "note", "metadata", "view", "comment"}
PROFILE_KINDS = {"rect", "circle", "slot"}


def round6(x):
    """Quantize to the 1e-6 tolerance grid, normalizing negative zero."""
    r = round(float(x), 6)
    if r == 0:
        r = 0.0
    return r


def canon(obj):
    """Canonical, order-stable JSON string.

    sort_keys canonicalizes object key order (JSON objects are unordered)
    while preserving list order, which is required for opaque payloads.
    """
    return json.dumps(obj, sort_keys=True, separators=(",", ":"))


class Evaluator:
    """Evaluates scalar arithmetic expressions over a parameter environment."""

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

    def get(self, name):
        if name in self.cache:
            return self.cache[name]
        if name not in self.params:
            raise KeyError(name)
        if name in self.stack:
            raise ValueError("cyclic parameter reference")
        self.stack.add(name)
        try:
            val = self.eval_expr(self.params[name])
        finally:
            self.stack.discard(name)
        self.cache[name] = val
        return val

    def eval_expr(self, expr):
        node = ast.parse(str(expr), mode="eval").body
        return self._ev(node)

    def _ev(self, node):
        if isinstance(node, ast.BinOp):
            l = self._ev(node.left)
            r = self._ev(node.right)
            if isinstance(node.op, ast.Add):
                return l + r
            if isinstance(node.op, ast.Sub):
                return l - r
            if isinstance(node.op, ast.Mult):
                return l * r
            if isinstance(node.op, ast.Div):
                return l / r
            raise ValueError("unsupported operator")
        if isinstance(node, ast.UnaryOp):
            v = self._ev(node.operand)
            if isinstance(node.op, ast.USub):
                return -v
            if isinstance(node.op, ast.UAdd):
                return v
            raise ValueError("unsupported unary operator")
        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.Name):
            return self.get(node.id)
        raise ValueError("unsupported expression node")


def eval_value(v, ev):
    """Evaluate a scalar field to a rounded float, or leave non-numeric as-is."""
    if isinstance(v, bool):
        return v
    if isinstance(v, (int, float)):
        return round6(v)
    if isinstance(v, str):
        try:
            return round6(ev.eval_expr(v))
        except Exception:
            return v
    return v


def deep_eval(v, ev):
    if isinstance(v, list):
        return [deep_eval(x, ev) for x in v]
    return eval_value(v, ev)


def parse_program(prog):
    params, sketches, profiles, extrudes, opaque = {}, {}, {}, [], []
    for cmd in prog:
        if not isinstance(cmd, dict):
            # Non-dict command: treat as opaque literal payload.
            opaque.append(cmd)
            continue
        if cmd.get("construction") is True:
            continue
        kind = cmd.get("cmd")
        if kind in ANNOTATION:
            continue
        if kind == "param":
            params[cmd.get("name")] = cmd.get("expr")
        elif kind == "sketch":
            plane = cmd.get("plane")
            sketches[cmd.get("id")] = plane.upper() if isinstance(plane, str) else plane
        elif kind in PROFILE_KINDS:
            profiles[cmd.get("id")] = cmd
        elif kind == "extrude":
            extrudes.append(cmd)
        else:
            opaque.append(cmd)
    return params, sketches, profiles, extrudes, opaque


def profile_sig(pcmd, sketches, ev):
    plane = sketches.get(pcmd.get("sketch"))
    kind = pcmd.get("cmd")
    fields = {}
    for k, v in pcmd.items():
        if k in ("cmd", "id", "sketch", "construction"):
            continue
        fields[k] = deep_eval(v, ev)
    if kind == "slot" and isinstance(fields.get("angle"), (int, float)) and not isinstance(fields.get("angle"), bool):
        fields["angle"] = round6(fields["angle"] % 180)
    return {"kind": kind, "plane": plane, "fields": fields}


def feature_sig(ecmd, sketches, profiles, ev):
    op = ecmd.get("operation", "new")
    op = op.lower() if isinstance(op, str) else op
    extent = ecmd.get("extent", "one_side")
    extent = extent.lower() if isinstance(extent, str) else extent
    direction = ecmd.get("direction", "normal")
    if isinstance(direction, str):
        direction = direction.lower()
    else:
        direction = deep_eval(direction, ev)
    depth = ecmd.get("depth")
    depth = deep_eval(depth, ev) if depth is not None else None

    if "profiles" in ecmd:
        prof_ids = ecmd["profiles"]
        if not isinstance(prof_ids, list):
            prof_ids = [prof_ids]
    elif "profile" in ecmd:
        prof_ids = [ecmd["profile"]]
    else:
        prof_ids = []
    psigs = sorted(
        canon(profile_sig(profiles[pid], sketches, ev))
        for pid in prof_ids
        if pid in profiles
    )
    return {
        "op": op,
        "extent": extent,
        "direction": direction,
        "depth": depth,
        "profiles": psigs,
    }


def partition_key(ecmd):
    """Partition label used for co-membership (value only, not namespace)."""
    if "body" in ecmd:
        return ("L", canon(ecmd["body"]))
    if "channel" in ecmd:
        return ("L", canon(ecmd["channel"]))
    return ("DEFAULT",)


def program_signature(prog):
    params, sketches, profiles, extrudes, opaque = parse_program(prog)
    ev = Evaluator(params)

    # Group extrude features by partition; discard label spelling, keep topology.
    groups = {}
    for ecmd in extrudes:
        fsig = canon(feature_sig(ecmd, sketches, profiles, ev))
        pk = partition_key(ecmd)
        groups.setdefault(pk, []).append(fsig)
    group_list = sorted(sorted(v) for v in groups.values())
    partition_sig = canon(group_list)

    # Opaque commands: unordered multiset, payload literal minus top-level id.
    opaque_sigs = []
    for c in opaque:
        if isinstance(c, dict):
            payload = {k: v for k, v in c.items() if k != "id"}
        else:
            payload = c
        opaque_sigs.append(canon(payload))
    opaque_sigs.sort()

    return canon([partition_sig, opaque_sigs])


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], "r") as f:
        data = json.load(f)

    predictions = []
    for pair in data.get("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": sig_a,
            "signature_b": sig_b,
        })

    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 an input JSON of program pairs, decides for each pair whether both
programs describe the same canonical solid, and writes predictions.json.

Usage:
    python3 solve.py <input_json> <output_json>
"""

import ast
import json
import sys


ANNOTATION = {"constraint", "dimension", "note", "metadata", "view", "comment"}
PROFILE_KINDS = {"rect", "circle", "slot"}


def round6(x):
    """Quantize to the 1e-6 tolerance grid, normalizing negative zero."""
    r = round(float(x), 6)
    if r == 0:
        r = 0.0
    return r


def canon(obj):
    """Canonical, order-stable JSON string.

    sort_keys canonicalizes object key order (JSON objects are unordered)
    while preserving list order, which is required for opaque payloads.
    """
    return json.dumps(obj, sort_keys=True, separators=(",", ":"))


class Evaluator:
    """Evaluates scalar arithmetic expressions over a parameter environment."""

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

    def get(self, name):
        if name in self.cache:
            return self.cache[name]
        if name not in self.params:
            raise KeyError(name)
        if name in self.stack:
            raise ValueError("cyclic parameter reference")
        self.stack.add(name)
        try:
            val = self.eval_expr(self.params[name])
        finally:
            self.stack.discard(name)
        self.cache[name] = val
        return val

    def eval_expr(self, expr):
        node = ast.parse(str(expr), mode="eval").body
        return self._ev(node)

    def _ev(self, node):
        if isinstance(node, ast.BinOp):
            l = self._ev(node.left)
            r = self._ev(node.right)
            if isinstance(node.op, ast.Add):
                return l + r
            if isinstance(node.op, ast.Sub):
                return l - r
            if isinstance(node.op, ast.Mult):
                return l * r
            if isinstance(node.op, ast.Div):
                return l / r
            raise ValueError("unsupported operator")
        if isinstance(node, ast.UnaryOp):
            v = self._ev(node.operand)
            if isinstance(node.op, ast.USub):
                return -v
            if isinstance(node.op, ast.UAdd):
                return v
            raise ValueError("unsupported unary operator")
        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.Name):
            return self.get(node.id)
        raise ValueError("unsupported expression node")


def eval_value(v, ev):
    """Evaluate a scalar field to a rounded float, or leave non-numeric as-is."""
    if isinstance(v, bool):
        return v
    if isinstance(v, (int, float)):
        return round6(v)
    if isinstance(v, str):
        try:
            return round6(ev.eval_expr(v))
        except Exception:
            return v
    return v


def deep_eval(v, ev):
    if isinstance(v, list):
        return [deep_eval(x, ev) for x in v]
    return eval_value(v, ev)


def parse_program(prog):
    params, sketches, profiles, extrudes, opaque = {}, {}, {}, [], []
    for cmd in prog:
        if not isinstance(cmd, dict):
            # Non-dict command: treat as opaque literal payload.
            opaque.append(cmd)
            continue
        if cmd.get("construction") is True:
            continue
        kind = cmd.get("cmd")
        if kind in ANNOTATION:
            continue
        if kind == "param":
            params[cmd.get("name")] = cmd.get("expr")
        elif kind == "sketch":
            plane = cmd.get("plane")
            sketches[cmd.get("id")] = plane.upper() if isinstance(plane, str) else plane
        elif kind in PROFILE_KINDS:
            profiles[cmd.get("id")] = cmd
        elif kind == "extrude":
            extrudes.append(cmd)
        else:
            opaque.append(cmd)
    return params, sketches, profiles, extrudes, opaque


def profile_sig(pcmd, sketches, ev):
    plane = sketches.get(pcmd.get("sketch"))
    kind = pcmd.get("cmd")
    fields = {}
    for k, v in pcmd.items():
        if k in ("cmd", "id", "sketch", "construction"):
            continue
        fields[k] = deep_eval(v, ev)
    angle = fields.get("angle")
    if kind == "slot" and isinstance(angle, (int, float)) and not isinstance(angle, bool):
        fields["angle"] = round6(angle % 180)
    return {"kind": kind, "plane": plane, "fields": fields}


def feature_sig(ecmd, sketches, profiles, ev):
    op = ecmd.get("operation", "new")
    op = op.lower() if isinstance(op, str) else op
    extent = ecmd.get("extent", "one_side")
    extent = extent.lower() if isinstance(extent, str) else extent
    direction = ecmd.get("direction", "normal")
    if isinstance(direction, str):
        direction = direction.lower()
    else:
        direction = deep_eval(direction, ev)
    depth = ecmd.get("depth")
    depth = deep_eval(depth, ev) if depth is not None else None

    if "profiles" in ecmd:
        prof_ids = ecmd["profiles"]
        if not isinstance(prof_ids, list):
            prof_ids = [prof_ids]
    elif "profile" in ecmd:
        prof_ids = [ecmd["profile"]]
    else:
        prof_ids = []
    psigs = sorted(
        canon(profile_sig(profiles[pid], sketches, ev))
        for pid in prof_ids
        if pid in profiles
    )
    return {
        "op": op,
        "extent": extent,
        "direction": direction,
        "depth": depth,
        "profiles": psigs,
    }


def partition_key(ecmd):
    """Partition label used for co-membership (value only, not namespace)."""
    if "body" in ecmd:
        return ("L", canon(ecmd["body"]))
    if "channel" in ecmd:
        return ("L", canon(ecmd["channel"]))
    return ("DEFAULT",)


def program_signature(prog):
    params, sketches, profiles, extrudes, opaque = parse_program(prog)
    ev = Evaluator(params)

    # Group extrude features by partition; discard label spelling, keep topology.
    groups = {}
    for ecmd in extrudes:
        fsig = canon(feature_sig(ecmd, sketches, profiles, ev))
        pk = partition_key(ecmd)
        groups.setdefault(pk, []).append(fsig)
    group_list = sorted(sorted(v) for v in groups.values())
    partition_sig = canon(group_list)

    # Opaque commands: unordered multiset, payload literal minus top-level id.
    opaque_sigs = []
    for c in opaque:
        if isinstance(c, dict):
            payload = {k: v for k, v in c.items() if k != "id"}
        else:
            payload = c
        opaque_sigs.append(canon(payload))
    opaque_sigs.sort()

    return canon([partition_sig, opaque_sigs])


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], "r", encoding="utf-8") as f:
        data = json.load(f)

    predictions = []
    for pair in data.get("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": bool(sig_a == sig_b),
            "signature_a": sig_a,
            "signature_b": sig_b,
        })

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

    return 0


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