cognee-hackathon-project-main/falsify/utils.py

"""
FALSIFY presentation helpers — console state + interactive graph visualization.

These functions turn the belief graph into the two things a judge actually sees:

* :func:`print_graph_state` — the BEFORE/AFTER console beat, each node tagged with a
  colored ``ALIVE`` / ``REFUTED`` / ``INVALIDATED`` / ``SUPERSEDED`` marker.
* :func:`visualize_belief_graph` — a single self-contained HTML file (vis-network via
  CDN, no build step) that colors nodes by truth-state so the cascade is legible at a
  glance: green = alive, red (dashed) = refuted, grey = invalidated, amber = superseded.
* :func:`get_belief_summary` — counts by type/state, used in output and tests.
* :func:`verify_propagation` — a small assertion helper for tests.

Everything degrades gracefully on an empty graph and never hard-depends on ``rich``.
"""

from __future__ import annotations

import html
import json
import logging
import os
from typing import Dict, List, Optional

from falsify import graph_ops
from falsify.models import TruthState

logger = logging.getLogger("falsify.utils")

# truth-state -> (console tag, hex color for the graph)
_STATE_STYLE = {
    TruthState.ALIVE.value: ("ALIVE", "#22c55e"),
    TruthState.REFUTED.value: ("REFUTED", "#ef4444"),
    TruthState.INVALIDATED.value: ("INVALIDATED", "#9ca3af"),
    TruthState.SUPERSEDED.value: ("SUPERSEDED", "#f59e0b"),
    TruthState.FORGOTTEN.value: ("FORGOTTEN", "#4b5563"),
}

# ANSI colors for console tags (fall back to plain text if not a TTY).
_ANSI = {
    "ALIVE": "\033[92m",
    "REFUTED": "\033[91m",
    "INVALIDATED": "\033[90m",
    "SUPERSEDED": "\033[93m",
    "FORGOTTEN": "\033[90m",
}
_RESET = "\033[0m"


def _colored(tag: str) -> str:
    """Return an ANSI-colored tag if stdout is a TTY, else the plain tag."""
    if os.environ.get("NO_COLOR") or not _stdout_is_tty():
        return tag
    return f"{_ANSI.get(tag, '')}{tag}{_RESET}"


def _stdout_is_tty() -> bool:
    try:
        import sys

        return bool(sys.stdout.isatty())
    except Exception:
        return False


async def _state_of(node_ids: List[str]) -> Dict[str, str]:
    """Return ``{id: single-state-string}`` (first alignment entry, default alive)."""
    truth = await graph_ops.get_truth(node_ids)
    return {nid: (align[0] if align else TruthState.ALIVE.value) for nid, align in truth.items()}


async def get_belief_summary(dataset: Optional[str] = None) -> Dict[str, Dict[str, int]]:
    """Count nodes grouped by node type and truth-state.

    Returns e.g. ``{"Hypothesis": {"alive": 2, "superseded": 1}, "Conclusion": {...}}``.
    Node type is inferred from the node's ``type`` property, falling back to the
    dominant embeddable field present.
    """
    nodes, _edges = await graph_ops.load_graph()
    if not nodes:
        return {}
    states = await _state_of([str(nid) for nid, _p in nodes])
    summary: Dict[str, Dict[str, int]] = {}
    for nid, props in nodes:
        ntype = _infer_type(props)
        state = states.get(str(nid), TruthState.ALIVE.value)
        summary.setdefault(ntype, {})
        summary[ntype][state] = summary[ntype].get(state, 0) + 1
    return summary


def _infer_type(props: dict) -> str:
    """Best-effort node-type label from properties."""
    if props.get("type"):
        return str(props["type"])
    for field, label in (
        ("question", "InvestigationQuestion"),
        ("claim", "Evidence"),
        ("statement", "Hypothesis/Conclusion"),
        ("text", "Assertion"),
    ):
        if props.get(field):
            return label
    return "Node"


async def print_graph_state(title: str, dataset: Optional[str] = None) -> None:
    """Print every node with a colored truth-state tag (the BEFORE/AFTER beat)."""
    nodes, _edges = await graph_ops.load_graph()
    print(f"\n{'=' * 64}\n  {title}\n{'=' * 64}")
    if not nodes:
        print("  (empty graph)")
        return
    states = await _state_of([str(nid) for nid, _p in nodes])
    # Stable, readable ordering: questions, hypotheses, evidence, conclusions.
    order = {"InvestigationQuestion": 0, "Hypothesis/Conclusion": 1, "Evidence": 2, "Assertion": 3}
    rows = []
    for nid, props in nodes:
        ntype = _infer_type(props)
        state = states.get(str(nid), TruthState.ALIVE.value)
        tag = _STATE_STYLE.get(state, ("ALIVE", ""))[0]
        label = graph_ops.node_label(props)
        rows.append((order.get(ntype, 9), ntype, tag, label))
    for _o, ntype, tag, label in sorted(rows, key=lambda r: r[0]):
        clipped = label if len(label) <= 66 else label[:63] + "..."
        print(f"  [{_colored(tag):<22}] {ntype:<22} {clipped}")
    print()


async def visualize_belief_graph(
    out_path: str = "output/graph.html",
    dataset: Optional[str] = None,
    title: str = "FALSIFY belief graph",
) -> Optional[str]:
    """Write a self-contained interactive HTML visualization of the belief graph.

    Nodes are colored by truth-state (green/red/grey/amber). Refuted nodes are drawn
    with a dashed red border so the cascade result reads at a glance. Returns the
    written path, or ``None`` if the graph is empty.
    """
    nodes, edges = await graph_ops.load_graph()
    if not nodes:
        logger.info("visualize_belief_graph: empty graph, nothing to draw")
        return None

    states = await _state_of([str(nid) for nid, _p in nodes])

    vis_nodes = []
    for nid, props in nodes:
        nid = str(nid)
        state = states.get(nid, TruthState.ALIVE.value)
        tag, color = _STATE_STYLE.get(state, ("ALIVE", "#22c55e"))
        label = graph_ops.node_label(props)
        short = label if len(label) <= 40 else label[:37] + "..."
        vis_nodes.append(
            {
                "id": nid,
                "label": short,
                "title": f"{_infer_type(props)} — {tag}\n{html.escape(label)}",
                "color": {"background": color, "border": "#111827"},
                "shapeProperties": {"borderDashes": state == TruthState.REFUTED.value},
                "borderWidth": 3 if state == TruthState.REFUTED.value else 1,
                "font": {"color": "#0b1020"},
            }
        )

    vis_edges = []
    for src, dst, rel, props in edges:
        vis_edges.append(
            {
                "from": str(src),
                "to": str(dst),
                "label": rel,
                "arrows": "to",
                "font": {"align": "middle", "size": 10},
                "color": {"color": "#94a3b8"},
            }
        )

    os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
    doc = _HTML_TEMPLATE.replace("__TITLE__", html.escape(title)) \
        .replace("__NODES__", json.dumps(vis_nodes)) \
        .replace("__EDGES__", json.dumps(vis_edges))
    with open(out_path, "w", encoding="utf-8") as fh:
        fh.write(doc)
    logger.info("wrote visualization to %s (%d nodes, %d edges)", out_path, len(vis_nodes), len(vis_edges))
    return out_path


async def verify_propagation(refuted_id: str, expected_affected: List[str]) -> bool:
    """Test helper: assert every ``expected_affected`` id is now non-alive.

    Returns True iff the refuted node is refuted and each expected dependent is in a
    dead state (refuted/invalidated/forgotten or absent from the graph).
    """
    ids = [str(refuted_id)] + [str(x) for x in expected_affected]
    truth = await graph_ops.get_truth(ids)
    dead = {TruthState.REFUTED.value, TruthState.INVALIDATED.value, TruthState.FORGOTTEN.value}
    nodes, _edges = await graph_ops.load_graph()
    present = {str(nid) for nid, _p in nodes}

    r_align = truth.get(str(refuted_id), [])
    if TruthState.REFUTED.value not in r_align:
        return False
    for dep in expected_affected:
        dep = str(dep)
        if dep not in present:  # forgotten (deleted) counts as affected
            continue
        if not any(s in dead for s in truth.get(dep, [TruthState.ALIVE.value])):
            return False
    return True


_HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8" />
  <title>__TITLE__</title>
  <script src="https://unpkg.com/vis-network/standalone/umd/vis-network.min.js"></script>
  <style>
    body { margin:0; font-family: ui-sans-serif, system-ui, sans-serif; background:#0b1020; color:#e5e7eb; }
    #hdr { padding:14px 18px; font-size:18px; font-weight:600; border-bottom:1px solid #1f2937; }
    #legend { padding:8px 18px; font-size:13px; color:#9ca3af; }
    .chip { display:inline-block; width:11px; height:11px; border-radius:3px; margin:0 5px 0 14px; vertical-align:middle; }
    #net { width:100%; height:calc(100vh - 92px); }
  </style>
</head>
<body>
  <div id="hdr">__TITLE__</div>
  <div id="legend">
    <span class="chip" style="background:#22c55e"></span>alive
    <span class="chip" style="background:#ef4444"></span>refuted
    <span class="chip" style="background:#9ca3af"></span>invalidated
    <span class="chip" style="background:#f59e0b"></span>superseded
  </div>
  <div id="net"></div>
  <script>
    const nodes = new vis.DataSet(__NODES__);
    const edges = new vis.DataSet(__EDGES__);
    const container = document.getElementById('net');
    const options = {
      physics: { stabilization: true, barnesHut: { gravitationalConstant: -8000, springLength: 150 } },
      nodes: { shape: 'box', margin: 10, widthConstraint: { maximum: 200 } },
      edges: { smooth: { type: 'cubicBezier' } }
    };
    new vis.Network(container, { nodes, edges }, options);
  </script>
</body>
</html>
"""