test_field_readout_consistency.py

"""Read-only fields must implement the same neighborhood means on every graph."""

from __future__ import annotations

import math
from fractions import Fraction

import networkx as nx
import numpy as np
import pytest

import tnfr.physics.extended as extended
from tnfr.alias import set_attr
from tnfr.constants.aliases import ALIAS_DNFR, ALIAS_THETA
from tnfr.physics.canonical import (
    compute_phase_curvature,
    compute_phase_gradient,
    compute_structural_potential,
    estimate_coherence_length,
    estimate_coherence_length_with_provenance,
)
from tnfr.physics.extended import compute_dnfr_flux, compute_phase_current
from tnfr.physics.fields import measure_phase_symmetry
from tnfr.physics.telemetry import compute_structural_telemetry
from tnfr.physics.vectorized_ops import compute_dnfr_flux_vectorized
from tnfr.utils.cache import reset_global_cache


@pytest.mark.parametrize(
    "graph_type", [nx.Graph, nx.DiGraph, nx.MultiGraph, nx.MultiDiGraph]
)
def test_fields_match_unique_neighbor_means(graph_type):
    """Successors, loops and parallel edges use the documented G.neighbors set."""
    graph = graph_type()
    graph.add_nodes_from(range(4))
    graph.add_edges_from([(0, 0), (0, 1), (0, 1), (1, 2)])
    phases = {0: 0.1, 1: 2 * math.pi - 0.2, 2: 0.8, 3: 1.5}
    pressure = {0: 0.2, 1: 0.5, 2: 1.0, 3: 0.3}
    for node in graph:
        set_attr(graph.nodes[node], ALIAS_THETA, phases[node])
        set_attr(graph.nodes[node], ALIAS_DNFR, pressure[node])

    expected = {key: {} for key in ("grad_phi", "curv_phi", "j_phi", "j_dnfr")}
    for node in graph:
        neighbors = list(graph.neighbors(node))
        if not neighbors:
            for field in expected.values():
                field[node] = 0.0
            continue
        differences = np.array([phases[other] - phases[node] for other in neighbors])
        wrapped = (differences + math.pi) % (2 * math.pi) - math.pi
        mean_phase = np.angle(
            np.mean(np.exp(1j * np.array([phases[n] for n in neighbors])))
        )
        expected["grad_phi"][node] = float(np.mean(np.abs(wrapped)))
        expected["curv_phi"][node] = (phases[node] - mean_phase + math.pi) % (
            2 * math.pi
        ) - math.pi
        expected["j_phi"][node] = float(np.mean(np.sin(wrapped)))
        expected["j_dnfr"][node] = float(
            np.mean([pressure[n] - pressure[node] for n in neighbors])
        )

    standalone = {
        "grad_phi": compute_phase_gradient(graph),
        "curv_phi": compute_phase_curvature(graph),
        "j_phi": compute_phase_current(graph),
        "j_dnfr": compute_dnfr_flux(graph),
    }
    telemetry = compute_structural_telemetry(graph)
    for key in expected:
        assert standalone[key] == pytest.approx(expected[key], abs=1e-12), key
        assert telemetry[key] == pytest.approx(expected[key], abs=1e-12), key


def test_uniform_pressure_has_zero_directed_flux():
    """An equilibrium pressure field cannot acquire a sink from normalization."""
    graph = nx.DiGraph([(0, 1), (1, 2)])
    for node in graph:
        set_attr(graph.nodes[node], ALIAS_DNFR, 0.5)
    assert compute_dnfr_flux(graph) == pytest.approx({node: 0.0 for node in graph})


def test_disconnected_potential_has_no_cross_component_source():
    """Unreachable sources contribute exactly zero, without a finite sentinel."""
    graph = nx.empty_graph(2)
    for node in graph:
        set_attr(graph.nodes[node], ALIAS_DNFR, 0.5)
    assert compute_structural_potential(graph) == {0: 0.0, 1: 0.0}


def test_coherence_length_provenance_matches_scalar_readout():
    graph = nx.path_graph(8)
    for node in graph:
        set_attr(graph.nodes[node], ALIAS_DNFR, 0.1 * (node + 1))
    estimate = estimate_coherence_length_with_provenance(graph)
    assert estimate.method in {"autocorrelation_fit", "spectral_gap", "unavailable"}
    assert estimate.distance_weighting != "unspecified"
    assert estimate.sample_selection != "unspecified"
    assert estimate.fit_quality != "unspecified"
    assert estimate.positive_mode_selection != "unspecified"
    assert estimate.graph_regime == "undirected; connected"
    if estimate.method != "unavailable":
        assert estimate.value == pytest.approx(estimate_coherence_length(graph))


def test_coherence_length_gap_fallback_does_not_cache_mode_shapes():
    graph = nx.path_graph(3)
    estimate = estimate_coherence_length_with_provenance(graph)
    assert estimate.method == "spectral_gap"
    assert "vecs" not in graph.graph["_tnfr_spectrum_cache"]


def test_coherence_length_directed_flat_field_refuses_symmetric_fallback():
    graph = nx.DiGraph([(0, 1), (1, 2)])
    estimate = estimate_coherence_length_with_provenance(graph)
    assert estimate.method == "unavailable"
    assert estimate.graph_regime == "directed; connected"


def test_landmark_potential_is_zero_at_equilibrium():
    """The landmark correction excludes self-interaction even for 0/0."""
    graph = nx.path_graph(12)
    assert compute_structural_potential(graph, landmark_ratio=0.5) == {
        node: 0.0 for node in graph
    }


@pytest.mark.parametrize("phase", [1e-16, -1e-16, math.pi, -math.pi, 1e308])
@pytest.mark.parametrize("vectorized", [True, False])
def test_phase_current_preserves_represented_sine_and_antisymmetry(
    phase, vectorized, monkeypatch
):
    graph = nx.Graph([(0, 1)])
    set_attr(graph.nodes[0], ALIAS_THETA, 0.0)
    set_attr(graph.nodes[1], ALIAS_THETA, phase)
    if not vectorized:

        def unavailable(*args, **kwargs):
            raise RuntimeError("exercise loop fallback")

        monkeypatch.setattr(extended, "compute_phase_current_vectorized", unavailable)
    reset_global_cache()
    current = compute_phase_current(graph)
    assert current[0] == pytest.approx(math.sin(phase), rel=2e-15, abs=0.0)
    assert current[1] == -current[0]


@pytest.mark.parametrize("vectorized", [True, False])
def test_phase_current_rejects_overflow_instead_of_caching_nan(vectorized, monkeypatch):
    graph = nx.Graph([(0, 1)])
    for node, phase in enumerate((1e308, -1e308)):
        set_attr(graph.nodes[node], ALIAS_THETA, phase)
    if not vectorized:

        def unavailable(*args, **kwargs):
            raise RuntimeError("exercise loop fallback")

        monkeypatch.setattr(extended, "compute_phase_current_vectorized", unavailable)
    reset_global_cache()
    with pytest.raises(ValueError, match="differences must be finite"):
        compute_phase_current(graph)


@pytest.mark.parametrize(
    "pressures",
    [
        (1e308, 1e308, 1e308),
        (0.0, 1e308, 1.0, -1e308),
        (-1e308, 1e308, -1e308),
    ],
    ids=("uniform-large-offset", "signed-cancellation", "overflowing-pair-difference"),
)
@pytest.mark.parametrize("vectorized", [True, False], ids=("vector", "fallback"))
def test_pressure_flux_retains_representable_neighbor_contrast(
    pressures, vectorized, monkeypatch
):
    graph = nx.MultiDiGraph()
    graph.add_nodes_from(range(len(pressures)))
    graph.add_edges_from(
        (0, neighbor, {"weight": 0.0, "length": 1e150})
        for neighbor in range(1, len(pressures))
    )
    # Parallel conductance and an incoming-only arc cannot reweight successors.
    graph.add_edge(0, 1, weight=7.0, length=1e150)
    graph.add_node("incoming", **{ALIAS_DNFR[0]: pressures[0]})
    graph.add_edge("incoming", 0, weight=5.0, length=1e150)
    for node, pressure in enumerate(pressures):
        set_attr(graph.nodes[node], ALIAS_DNFR, pressure)
    if not vectorized:

        def unavailable(*args, **kwargs):
            raise RuntimeError("exercise loop fallback")

        monkeypatch.setattr(extended, "compute_dnfr_flux_vectorized", unavailable)
    reset_global_cache()
    center = Fraction.from_float(pressures[0])
    expected = float(
        sum(
            (Fraction.from_float(value) - center for value in pressures[1:]), Fraction()
        )
        / (len(pressures) - 1)
    )
    flux = compute_dnfr_flux(graph)
    assert flux[0] == expected
    assert all(flux[node] == 0.0 for node in graph if node != 0)
    # The telemetry path imports the same array owner independently of the
    # standalone fallback switch, so this checks both actual consumers.
    assert compute_structural_telemetry(graph)["j_dnfr"] == flux


@pytest.mark.parametrize("vectorized", [True, False], ids=("vector", "fallback"))
def test_unrepresentable_final_pressure_flux_is_not_cached(vectorized, monkeypatch):
    graph = nx.DiGraph([(0, 1)])
    for node, pressure in enumerate((-1e308, 1e308)):
        set_attr(graph.nodes[node], ALIAS_DNFR, pressure)
    if not vectorized:

        def unavailable(*args, **kwargs):
            raise RuntimeError("exercise loop fallback")

        monkeypatch.setattr(extended, "compute_dnfr_flux_vectorized", unavailable)
    reset_global_cache()
    with pytest.raises(ValueError, match="finite floating-point range"):
        compute_dnfr_flux(graph)
    set_attr(graph.nodes[1], ALIAS_DNFR, -1e308)
    assert compute_dnfr_flux(graph) == {0: 0.0, 1: 0.0}


def test_pressure_flux_array_cannot_silently_ignore_wrong_neighbor_counts():
    with pytest.raises(ValueError, match="neighbor counts"):
        compute_dnfr_flux_vectorized(
            np.array([0.0, 2.0]), np.array([1]), np.array([0]), np.array([2, 0])
        )


def test_pressure_flux_rejects_output_dtype_overflow():
    arguments = (np.array([0.0, 1e100]), np.array([1]), np.array([0]), np.array([1, 0]))
    assert compute_dnfr_flux_vectorized(*arguments).tolist() == [1e100, 0.0]
    with pytest.raises(ValueError, match="output dtype range"):
        compute_dnfr_flux_vectorized(*arguments, dtype=np.float32)


@pytest.mark.parametrize("bad", [False, "0.0", math.nan, Fraction(1, 2**2000)])
def test_phase_symmetry_rejects_invalid_authoritative_data(bad):
    graph = nx.empty_graph(1)
    set_attr(graph.nodes[0], ALIAS_THETA, 0.0)
    graph.nodes[0][ALIAS_THETA[0]] = bad
    with pytest.raises((TypeError, ValueError)):
        measure_phase_symmetry(graph)


def test_phase_symmetry_distinguishes_empty_sample_and_undefined_mean():
    graph = nx.empty_graph(2)
    assert measure_phase_symmetry(graph) == 0.0  # Explicit legacy convention.
    for node, phase in enumerate((0.0, math.pi)):
        set_attr(graph.nodes[node], ALIAS_THETA, phase)
    with pytest.raises(ValueError, match="vanishing resultant"):
        measure_phase_symmetry(graph)


def test_phase_symmetry_is_axial_and_uses_primitive_phasors():
    graph = nx.empty_graph(3)
    for node, phase in enumerate((0.0, 0.0, math.pi)):
        set_attr(graph.nodes[node], ALIAS_THETA, phase)
    assert measure_phase_symmetry(graph) == pytest.approx(1.0)
    for node in graph:
        set_attr(graph.nodes[node], ALIAS_THETA, 1e308)
    assert measure_phase_symmetry(graph) == pytest.approx(1.0)