path: root/.venv/lib/python3.12/site-packages/deepdiff/distance.py

import math
import datetime
from typing import TYPE_CHECKING, Callable, Protocol, Any
from deepdiff.deephash import DeepHash
from deepdiff.helper import (
    DELTA_VIEW, numbers, strings, add_to_frozen_set, not_found, only_numbers, np, np_float64, time_to_seconds,
    cartesian_product_numpy, np_ndarray, np_array_factory, get_homogeneous_numpy_compatible_type_of_seq, dict_,
    CannotCompare)
from collections.abc import Mapping, Iterable

if TYPE_CHECKING:
    from deepdiff.diff import DeepDiffProtocol

    class DistanceProtocol(DeepDiffProtocol, Protocol):
        hashes: dict
        deephash_parameters: dict
        iterable_compare_func: Callable | None
        math_epsilon: float
        cutoff_distance_for_pairs: float

        def __get_item_rough_length(self, item, parent:str="root") -> float:
            ...

        def _to_delta_dict(
            self,
            directed: bool = True,
            report_repetition_required: bool = True,
            always_include_values: bool = False,
        ) -> dict:
            ...

        def __calculate_item_deephash(self, item: Any) -> None:
            ...



DISTANCE_CALCS_NEEDS_CACHE = "Distance calculation can not happen once the cache is purged. Try with _cache='keep'"


class DistanceMixin:

    def _get_rough_distance(self: "DistanceProtocol"):
        """
        Gives a numeric value for the distance of t1 and t2 based on how many operations are needed to convert
        one to the other.

        This is a similar concept to the Levenshtein Edit Distance but for the structured data and is it is designed
        to be between 0 and 1.

        A distance of zero means the objects are equal and a distance of 1 is very far.

        Note: The distance calculation formula is subject to change in future. Use the distance results only as a
        way of comparing the distances of pairs of items with other pairs rather than an absolute distance
        such as the one provided by Levenshtein edit distance.

        Info: The current algorithm is based on the number of operations that are needed to convert t1 to t2 divided
        by the number of items that make up t1 and t2.
        """

        _distance = get_numeric_types_distance(
            self.t1, self.t2, max_=self.cutoff_distance_for_pairs, use_log_scale=self.use_log_scale, log_scale_similarity_threshold=self.log_scale_similarity_threshold)

        if _distance is not not_found:
            return _distance

        item = self if self.view == DELTA_VIEW else self._to_delta_dict(report_repetition_required=False)
        diff_length = _get_item_length(item)

        if diff_length == 0:
            return 0

        t1_len = self.__get_item_rough_length(self.t1)
        t2_len = self.__get_item_rough_length(self.t2)

        return diff_length / (t1_len + t2_len)

    def __get_item_rough_length(self: "DistanceProtocol", item, parent='root'):
        """
        Get the rough length of an item.
        It is used as a part of calculating the rough distance between objects.

        **parameters**

        item: The item to calculate the rough length for
        parent: It is only used for DeepHash reporting purposes. Not really useful here.
        """
        if not hasattr(self, 'hashes'):
            raise RuntimeError(DISTANCE_CALCS_NEEDS_CACHE)
        length = DeepHash.get_key(self.hashes, key=item, default=None, extract_index=1)
        if length is None:
            self.__calculate_item_deephash(item)
            length = DeepHash.get_key(self.hashes, key=item, default=None, extract_index=1)
        return length

    def __calculate_item_deephash(self: "DistanceProtocol", item: Any) -> None:
        DeepHash(
            item,
            hashes=self.hashes,
            parent='root',
            apply_hash=True,
            **self.deephash_parameters,
        )

    def _precalculate_distance_by_custom_compare_func(
            self: "DistanceProtocol", hashes_added, hashes_removed, t1_hashtable, t2_hashtable, _original_type):
        pre_calced_distances = dict_()
        for added_hash in hashes_added:
            for removed_hash in hashes_removed:
                try:
                    is_close_distance = self.iterable_compare_func(t2_hashtable[added_hash].item, t1_hashtable[removed_hash].item)
                except CannotCompare:
                    pass
                else:
                    if is_close_distance:
                        # an arbitrary small distance if math_epsilon is not defined
                        distance = self.math_epsilon or 0.000001
                    else:
                        distance = 1
                    pre_calced_distances["{}--{}".format(added_hash, removed_hash)] = distance

        return pre_calced_distances

    def _precalculate_numpy_arrays_distance(
            self: "DistanceProtocol", hashes_added, hashes_removed, t1_hashtable, t2_hashtable, _original_type):

        # We only want to deal with 1D arrays.
        if isinstance(t2_hashtable[next(iter(hashes_added))].item, (np_ndarray, list)):
            return

        pre_calced_distances = dict_()
        added = [t2_hashtable[k].item for k in hashes_added]
        removed = [t1_hashtable[k].item for k in hashes_removed]

        if _original_type is None:
            added_numpy_compatible_type = get_homogeneous_numpy_compatible_type_of_seq(added)
            removed_numpy_compatible_type = get_homogeneous_numpy_compatible_type_of_seq(removed)
            if added_numpy_compatible_type and added_numpy_compatible_type == removed_numpy_compatible_type:
                _original_type = added_numpy_compatible_type
        if _original_type is None:
            return

        added = np_array_factory(added, dtype=_original_type)
        removed = np_array_factory(removed, dtype=_original_type)

        pairs = cartesian_product_numpy(added, removed)

        pairs_transposed = pairs.T

        distances = _get_numpy_array_distance(
            pairs_transposed[0], pairs_transposed[1],
            max_=self.cutoff_distance_for_pairs,
            use_log_scale=self.use_log_scale,
            log_scale_similarity_threshold=self.log_scale_similarity_threshold,
        )

        i = 0
        for added_hash in hashes_added:
            for removed_hash in hashes_removed:
                pre_calced_distances["{}--{}".format(added_hash, removed_hash)] = distances[i]
                i += 1
        return pre_calced_distances


def _get_item_length(item, parents_ids=frozenset([])):
    """
    Get the number of operations in a diff object.
    It is designed mainly for the delta view output
    but can be used with other dictionary types of view outputs too.
    """
    length = 0
    if isinstance(item, Mapping):
        for key, subitem in item.items():
            # dedupe the repetition report so the number of times items have shown up does not affect the distance.
            if key in {'iterable_items_added_at_indexes', 'iterable_items_removed_at_indexes'}:
                new_subitem = dict_()
                for path_, indexes_to_items in subitem.items():
                    used_value_ids = set()
                    new_indexes_to_items = dict_()
                    for k, v in indexes_to_items.items():
                        v_id = id(v)
                        if v_id not in used_value_ids:
                            used_value_ids.add(v_id)
                            new_indexes_to_items[k] = v
                    new_subitem[path_] = new_indexes_to_items
                subitem = new_subitem

            # internal keys such as _numpy_paths should not count towards the distance
            if isinstance(key, strings) and (key.startswith('_') or key == 'deep_distance' or key == 'new_path'):
                continue

            item_id = id(subitem)
            if parents_ids and item_id in parents_ids:
                continue
            parents_ids_added = add_to_frozen_set(parents_ids, item_id)
            length += _get_item_length(subitem, parents_ids_added)
    elif isinstance(item, numbers):
        length = 1
    elif isinstance(item, strings):
        length = 1
    elif isinstance(item, Iterable):
        for subitem in item:
            item_id = id(subitem)
            if parents_ids and item_id in parents_ids:
                continue
            parents_ids_added = add_to_frozen_set(parents_ids, item_id)
            length += _get_item_length(subitem, parents_ids_added)
    elif isinstance(item, type):  # it is a class
        length = 1
    else:
        if hasattr(item, '__dict__'):
            for subitem in item.__dict__:
                item_id = id(subitem)
                parents_ids_added = add_to_frozen_set(parents_ids, item_id)
                length += _get_item_length(subitem, parents_ids_added)
    return length


def _get_numbers_distance(num1, num2, max_=1, use_log_scale=False, log_scale_similarity_threshold=0.1):
    """
    Get the distance of 2 numbers. The output is a number between 0 to the max.
    The reason is the
    When max is returned means the 2 numbers are really far, and 0 means they are equal.
    """
    if num1 == num2:
        return 0
    if use_log_scale:
        distance = logarithmic_distance(num1, num2)
        if distance < 0:
            return 0
        return distance
    if not isinstance(num1, float):
        num1 = float(num1)
    if not isinstance(num2, float):
        num2 = float(num2)
    # Since we have a default cutoff of 0.3 distance when
    # getting the pairs of items during the ingore_order=True
    # calculations, we need to make the divisor of comparison very big
    # so that any 2 numbers can be chosen as pairs.
    divisor = (num1 + num2) / max_
    if divisor == 0:
        return max_
    try:
        return min(max_, abs((num1 - num2) / divisor))
    except Exception:  # pragma: no cover. I don't think this line will ever run but doesn't hurt to leave it.
        return max_  # pragma: no cover


def _numpy_div(a, b, replace_inf_with=1):
    max_array = np.full(shape=a.shape, fill_value=replace_inf_with, dtype=np_float64)
    result = np.divide(a, b, out=max_array, where=b != 0, dtype=np_float64)
    # wherever 2 numbers are the same, make sure the distance is zero. This is mainly for 0 divided by zero.
    result[a == b] = 0
    return result

# To deal with numbers close to zero
MATH_LOG_OFFSET = 1e-10

def numpy_apply_log_keep_sign(array, offset=MATH_LOG_OFFSET):
    # Calculate the absolute value and add the offset
    abs_plus_offset = np.abs(array) + offset
    
    # Calculate the logarithm
    log_values = np.log(abs_plus_offset)
    
    # Apply the original signs to the log values
    signed_log_values = np.copysign(log_values, array)
    
    return signed_log_values


def logarithmic_similarity(a: numbers, b: numbers, threshold: float=0.1) -> float:
    """
    A threshold of 0.1 translates to about 10.5% difference.
    A threshold of 0.5 translates to about 65% difference.
    A threshold of 0.05 translates to about 5.1% difference.
    """
    return logarithmic_distance(a, b) < threshold


def logarithmic_distance(a: numbers, b: numbers) -> float:
    # Apply logarithm to the absolute values and consider the sign
    a = float(a)
    b = float(b)
    log_a = math.copysign(math.log(abs(a) + MATH_LOG_OFFSET), a)
    log_b = math.copysign(math.log(abs(b) + MATH_LOG_OFFSET), b)

    return abs(log_a - log_b)


def _get_numpy_array_distance(num1, num2, max_=1, use_log_scale=False, log_scale_similarity_threshold=0.1):
    """
    Get the distance of 2 numbers. The output is a number between 0 to the max.
    The reason is the
    When max is returned means the 2 numbers are really far, and 0 means they are equal.
    """
    # Since we have a default cutoff of 0.3 distance when
    # getting the pairs of items during the ingore_order=True
    # calculations, we need to make the divisor of comparison very big
    # so that any 2 numbers can be chosen as pairs.
    if use_log_scale:
        num1 = numpy_apply_log_keep_sign(num1)
        num2 = numpy_apply_log_keep_sign(num2)

    divisor = (num1 + num2) / max_
    result = _numpy_div((num1 - num2), divisor, replace_inf_with=max_)

    distance_array = np.clip(np.absolute(result), 0, max_)
    if use_log_scale:
        distance_array[distance_array < log_scale_similarity_threshold] = 0
    return distance_array


def _get_datetime_distance(date1, date2, max_, use_log_scale, log_scale_similarity_threshold):
    return _get_numbers_distance(date1.timestamp(), date2.timestamp(), max_)


def _get_date_distance(date1, date2, max_, use_log_scale, log_scale_similarity_threshold):
    return _get_numbers_distance(date1.toordinal(), date2.toordinal(), max_)


def _get_timedelta_distance(timedelta1, timedelta2, max_, use_log_scale, log_scale_similarity_threshold):
    return _get_numbers_distance(timedelta1.total_seconds(), timedelta2.total_seconds(), max_)


def _get_time_distance(time1, time2, max_, use_log_scale, log_scale_similarity_threshold):
    return _get_numbers_distance(time_to_seconds(time1), time_to_seconds(time2), max_)


TYPES_TO_DIST_FUNC = [
    (only_numbers, _get_numbers_distance),
    (datetime.datetime, _get_datetime_distance),
    (datetime.date, _get_date_distance),
    (datetime.timedelta, _get_timedelta_distance),
    (datetime.time, _get_time_distance),
]


def get_numeric_types_distance(num1, num2, max_, use_log_scale=False, log_scale_similarity_threshold=0.1):
    for type_, func in TYPES_TO_DIST_FUNC:
        if isinstance(num1, type_) and isinstance(num2, type_):
            return func(num1, num2, max_, use_log_scale, log_scale_similarity_threshold)
    return not_found