siamese_equations.py

import math

import networkx as nx
import numpy as np
import pandas as pd
from gensim.models import KeyedVectors
from scipy.stats import describe


def weighted_hunter_busybody(nm: dict) -> float:
    """
    Calculate a weighted hunter-busybody metric for a network based on its properties.

    This metric combines the mean edge weights, average clustering coefficient, and
    characteristic path length of the network. Higher values indicate networks with
    stronger local connectivity (clustering) and edge weights, but shorter path lengths.

    Args:
        nm: Dictionary containing network metrics including 'weights_mean',
            'avg_clustering_coefficient', and 'Characteristic_path_length'

    Returns:
        float: The weighted hunter-busybody score calculated as:
              weights_mean + avg_clustering_coefficient - Characteristic_path_length
    """
    return (
        nm["weights_mean"]
        + nm["avg_clustering_coefficient"]
        - nm["Characteristic_path_length"]
    )


def pairwise_distances(words: list[str], W2V: KeyedVectors) -> list[float]:
    """
    Calculate pairwise semantic distances between consecutive words using word embeddings.

    Args:
        words: List of words to calculate distances between
        W2V: Gensim KeyedVectors word embedding model

    Returns:
        List of semantic distances between consecutive word pairs, where distance is
        1 minus the cosine similarity. Only includes pairs where both words are in
        the embedding vocabulary.
    """
    distances = []

    for i in range(len(words) - 1):
        word1, word2 = words[i], words[i + 1]
        if word1 in W2V and word2 in W2V:  # Check if words are in W2V vocabulary
            distances.append(1 - W2V.similarity(word1, word2))

    return distances


def forward_flow(words: list[str], W2V: KeyedVectors) -> list[float]:
    """
    Calculate the forward flow measure for a sequence of words using word embeddings.

    Forward flow measures the average semantic distance between consecutive words
    in increasingly longer prefixes of the word sequence.

    Note that there are several methods to calculate forward flow, this is one implementation using W2V.

    Args:
        words: List of words to calculate forward flow for
        W2V: Gensim KeyedVectors word embedding model

    Returns:
        List of forward flow values for each prefix length starting from 2 words
    """
    distances = []
    for i in range(2, len(words) + 1):
        distances.append(sum(pairwise_distances(words[:i], W2V)) / len(words[:i]))
    return distances


def herons_formula(a, b, c):
    """
    Calculate the area of a triangle using Heron's formula if the sides form a valid triangle.

    Parameters:
        a (float): Length of the first side.
        b (float): Length of the second side.
        c (float): Length of the third side.

    Returns:
        float: Area of the triangle, or 0 if the sides do not form a valid triangle.
    """
    # Check the triangle inequality
    if a + b > c and a + c > b and b + c > a:
        s = (a + b + c) / 2  # semi-perimeter
        return math.sqrt(s * (s - a) * (s - b) * (s - c))
    else:
        return 0  # Invalid triangle


def calculate_triangle_areas(graph):
    """
    Calculate the areas of all triangles in a weighted NetworkX graph, validating triangle inequality constraints.

    Parameters:
        graph (nx.Graph): A NetworkX graph with edge weights representing distances.

    Returns:
        list: A list of tuples (triangle, area, valid) where `triangle` is a tuple of node indices,
              `area` is the area of the triangle, and `valid` is a boolean indicating if the triangle is valid.
    """
    triangles = [triangle for triangle in nx.cycle_basis(graph) if len(triangle) == 3]
    triangle_areas = []

    for triangle in triangles:
        u, v, w = triangle

        # Get the edge weights (interpreted as distances)
        try:
            a = graph[u][v]["weight"]
            b = graph[v][w]["weight"]
            c = graph[w][u]["weight"]

            # Calculate the area and validate the triangle
            area = herons_formula(a, b, c)
            valid_triangle = area > 0

            # Append the result as a tuple (triangle, area, valid)
            triangle_areas.append((triangle, area, valid_triangle))
        except KeyError:
            # Skip if any edge weight is missing
            triangle_areas.append((triangle, None, False))

    return triangle_areas


def compute_relevant_weighted_graph_metrics(graph):
    """
    Compute a suite of relevant metrics for a fully connected, undirected, weighted NetworkX graph.

    Args:
        graph (nx.Graph): A fully connected, undirected, weighted NetworkX graph.

    Returns:
        dict: A dictionary containing relevant graph metrics for weighted, fully connected graphs.
    """
    if not nx.is_connected(graph):
        raise ValueError("The graph must be fully connected.")
    if not nx.is_weighted(graph):
        raise ValueError("The graph must have weighted edges.")

    # Number of nodes
    num_nodes = graph.number_of_nodes()

    # Average weighted degree (mean of node strengths)
    weighted_degrees = dict(graph.degree(weight="weight"))
    avg_weighted_degree = sum(weighted_degrees.values()) / num_nodes

    # Average clustering coefficient (weighted)
    avg_clustering_coefficient = nx.average_clustering(graph, weight="weight")

    triangle_areas = calculate_triangle_areas(graph)

    # Compute the average triangle area, ignoring invalid triangles
    valid_areas = [area for _, area, valid in triangle_areas if valid]
    average_triangle_area = np.mean(valid_areas) if valid_areas else 0

    # Compute the ratio of valid triangles
    ratio_valid_triangles = (
        sum(valid for _, _, valid in triangle_areas) / len(triangle_areas)
        if triangle_areas
        else 0
    )

    return {
        "avg_weighted_degree": avg_weighted_degree,
        "avg_clustering_coefficient": avg_clustering_coefficient,
        "average_triangle_area": average_triangle_area,
        "ratio_valid_triangles": ratio_valid_triangles,
    }


def calculate_node_gravity_1(word_list: list[str], W2V: object) -> dict[str, float]:
    """
    Calculate the gravity of each node in a semantic graph based on Stephen's gravity formula.

    Gravity G(i) for node i is calculated as: sum of edge weights connected to node i divided by node i's weight.
    Edge weights are computed as cosine distances between word embeddings.

    Args:
        word_list (list[str]): List of words to analyze
        W2V (object): Word2Vec model object for calculating word similarities

    Returns:
        dict[str, float]: Dictionary mapping each word to its calculated gravity value.
            Returns all zeros if any node has zero cumulative edge weight.
    """
    all_edge_weights = []
    cummulative_weight = 0
    for i, word1 in enumerate(word_list):
        surrounding_edge_weights = []
        for j, word2 in enumerate(word_list):
            if i == j or word1 == word2:
                continue
            # Calculate cosine distance between word embeddings
            distance = 1 - W2V.similarity(word1, word2)
            surrounding_edge_weights.append(distance)
            cummulative_weight = sum(surrounding_edge_weights)
        if cummulative_weight == 0:
            logger.warning(f"Zero weight in {word_list}")
            return {word: 0 for word in word_list}
        all_edge_weights.append(cummulative_weight)
    node_weights = calculate_node_weights(word_list)
    return {
        word: all_edge_weights[i] / node_weights[word] if node_weights[word] != 0 else 0
        for i, word in enumerate(word_list)
    }


def describe_node_gravity_1(word_list: list[str], W2V: object) -> dict[str, float]:
    """
    Calculate descriptive statistics for node gravity values in a semantic graph.

    Takes the node gravity values calculated by calculate_node_gravity_1() and computes
    various statistical measures including min, max, mean (both overall and excluding zeros),
    variance, skewness, and kurtosis.

    Parameters:
    - word_list (list[str]): List of words to analyze
    - W2V (object): Word2Vec model object for calculating word similarities

    Returns:
    - dict[str, float]: Dictionary containing the following statistical measures:
        - node_gravity_1_min: Minimum gravity value
        - node_gravity_1_max: Maximum gravity value
        - node_gravity_1_mean_non_0: Mean of non-zero gravity values
        - node_gravity_1_mean: Mean of all gravity values
        - node_gravity_1_variance: Variance of gravity values
        - node_gravity_1_skewness: Skewness of gravity values
        - node_gravity_1_kurtosis: Kurtosis of gravity values
    """
    # sourcery skip: inline-immediately-returned-variable
    node_gravity_1 = list(calculate_node_gravity_1(word_list, W2V).values())
    non_0_gravity_1 = [x for x in node_gravity_1 if x != 0]
    gravity_descriptions = describe(node_gravity_1)
    node_gravity_1_description = {
        "node_gravity_1_min": gravity_descriptions.minmax[0],
        "node_gravity_1_max": gravity_descriptions.minmax[1],
        "node_gravity_1_mean_non_0": sum(non_0_gravity_1)
        / len(
            non_0_gravity_1
        ),  # if a node weight is 0, the word is excluded from mean calculation
        "node_gravity_1_mean": gravity_descriptions.mean,
        "node_gravity_1_variance": gravity_descriptions.variance,
        "node_gravity_1_skewness": gravity_descriptions.skewness,
        "node_gravity_1_kurtosis": gravity_descriptions.kurtosis,
    }
    return node_gravity_1_description


def calculate_node_weights(word_list: list[str]) -> dict[str, float]:
    """
    Calculate the weights of the nodes in the graph based on word frequencies.

    Loads word frequencies from a CSV file and normalizes them using log2 scaling.
    For words not found in the frequency data, assigns a weight of 0.

    Parameters:
    - word_list (list[str]): A list of words to calculate weights for

    Returns:
    - dict[str, float]: A dictionary mapping words to their normalized frequency weights.
        Words not found in frequency data will have weight 0.
    """
    # onethirdm dictionary
    word_dict = pd.read_csv(f"sources/unigram_freq.csv")
    # work with a word list, but take threshold from referencing snafu data
    # TODO: why dont we normalize by the max count
    word_dict["normalized_count"] = np.log2(word_dict["count"]) / np.log2(
        word_dict["count"].max()
    )

    return {
        word: (
            word_dict[word_dict["word"] == word]["normalized_count"].values[0]
            if word in word_dict["word"].unique()
            else 0
        )
        for word in word_list
    }


def calculate_node_gravity_2(word_list: list[str], W2V: object) -> dict[str, float]:
    """
    Calculate gravity values for nodes in a semantic graph using a modified gravity model.

    The gravity model is typically used with k-shell decomposition, but since we have fully
    connected graphs, k-shell decomposition is not appropriate. The model uses semantic
    distances between nodes and normalized word frequencies as node weights.

    The gravity formula is: G(i) = sum(w(i) * w(j) / d**2) for all j != i
    where w(i) and w(j) are node weights and d is the semantic distance between nodes.

    Note: The semantic distance is not the same as shortest path length between nodes.

    Parameters:
    - word_list (list[str]): List of words to analyze
    - W2V (object): Word2Vec model object for calculating word similarities

    Returns:
    - dict[str, float]: Dictionary mapping words to their calculated gravity values
    """
    node_weights = calculate_node_weights(word_list)
    gravity_2 = []
    for i, word1 in enumerate(word_list):
        node_gravity_2 = []
        for j, word2 in enumerate(word_list):
            if i == j or word1 == word2:
                continue
            sq_distance = (1 - W2V.similarity(word1, word2)) ** 2
            weight_products = (node_weights[word1] * node_weights[word2]) / sq_distance
            node_gravity_2.append(weight_products)
        gravity_2.append(sum(node_gravity_2))

    return {word: gravity_2[i] for i, word in enumerate(word_list)}


def describe_node_gravity_2(word_list: list[str], W2V: object) -> dict[str, float]:
    """
    Calculate descriptive statistics for node gravity values.

    Parameters:
    - word_list (list[str]): List of words to analyze
    - W2V (object): Word2Vec model object for calculating word similarities

    Returns:
    - dict[str, float]: Dictionary containing descriptive statistics:
        - node_gravity_2_min: Minimum gravity value
        - node_gravity_2_max: Maximum gravity value
        - node_gravity_2_mean_non_0: Mean of non-zero gravity values
        - node_gravity_2_mean: Mean of all gravity values
        - node_gravity_2_variance: Variance of gravity values
        - node_gravity_2_skewness: Skewness of gravity values
        - node_gravity_2_kurtosis: Kurtosis of gravity values
    """
    # sourcery skip: inline-immediately-returned-variable
    node_gravity_2 = list(calculate_node_gravity_2(word_list, W2V).values())
    non_0_gravity_2 = [x for x in node_gravity_2 if x != 0]
    gravity_descriptions = describe(node_gravity_2)
    node_gravity_2_description = {
        "node_gravity_2_min": gravity_descriptions.minmax[0],
        "node_gravity_2_max": gravity_descriptions.minmax[1],
        "node_gravity_2_mean_non_0": sum(non_0_gravity_2)
        / len(
            non_0_gravity_2
        ),  # if a node weight is 0, the word is excluded from mean calculation
        "node_gravity_2_mean": gravity_descriptions.mean,
        "node_gravity_2_variance": gravity_descriptions.variance,
        "node_gravity_2_skewness": gravity_descriptions.skewness,
        "node_gravity_2_kurtosis": gravity_descriptions.kurtosis,
    }
    return node_gravity_2_description


def describe_edge_weights(G):
    """
    Calculate the average edge weight of the graph.

    Parameters:
    - G (networkx.Graph): The input graph.

    Returns:
    - float: The average edge weight.
    """
    # Extract the edge weights from the graph
    edge_weights = nx.get_edge_attributes(G, "weight")
    weights = list(edge_weights.values())

    return describe(weights)


def describe_node_weights(word_list) -> dict:
    """
    Calculate the node weights and average weight of the graph.
    Parameters:
    - word_list: input CFA list

    Returns:
    - dict: A dictionary with describing node weights.
    """
    # dictionary with word as key and weights as value
    node_weights = list(calculate_node_weights(word_list).values())
    non_0_weights = [x for x in node_weights if x != 0]
    node_descriptions = describe(node_weights)
    node_weight_description = {
        "node_weights_min": node_descriptions.minmax[0],
        "node_weights_max": node_descriptions.minmax[1],
        "node_weights_mean_non_0": sum(non_0_weights)
        / len(
            non_0_weights
        ),  # if a node weight is 0, the word is excluded from mean calculation
        "node_weights_mean": node_descriptions.mean,
        "node_weights_variance": node_descriptions.variance,
        "node_weights_skewness": node_descriptions.skewness,
        "node_weights_kurtosis": node_descriptions.kurtosis,
    }
    return node_weight_description