helper.py

import sys
import re
import os
import datetime
import uuid
import logging
import warnings
import string
import time
import enum
import ipaddress
from typing import NamedTuple, Any, List, Optional, Dict, Union, TYPE_CHECKING, Tuple
from ast import literal_eval
from decimal import Decimal, localcontext, InvalidOperation as InvalidDecimalOperation
from itertools import repeat
from orderly_set import StableSetEq as SetOrderedBase  # median: 1.0867 s for cache test, 5.63s for all tests
from threading import Timer

if TYPE_CHECKING:
    from pytz.tzinfo import BaseTzInfo


class np_type:
    pass


class pydantic_base_model_type:
    pass


class SetOrdered(SetOrderedBase):
    def __repr__(self):
        return str(list(self))


try:
    import numpy as np
except ImportError:  # pragma: no cover. The case without Numpy is tested locally only.
    np = None  # pragma: no cover.
    np_array_factory = 'numpy not available'  # pragma: no cover.
    np_ndarray = np_type  # pragma: no cover.
    np_bool_ = np_type  # pragma: no cover.
    np_int8 = np_type  # pragma: no cover.
    np_int16 = np_type  # pragma: no cover.
    np_int32 = np_type  # pragma: no cover.
    np_int64 = np_type  # pragma: no cover.
    np_uint8 = np_type  # pragma: no cover.
    np_uint16 = np_type  # pragma: no cover.
    np_uint32 = np_type  # pragma: no cover.
    np_uint64 = np_type  # pragma: no cover.
    np_intp = np_type  # pragma: no cover.
    np_uintp = np_type  # pragma: no cover.
    np_float32 = np_type  # pragma: no cover.
    np_float64 = np_type  # pragma: no cover.
    np_double = np_type  # pragma: no cover.
    np_floating = np_type  # pragma: no cover.
    np_complex64 = np_type  # pragma: no cover.
    np_complex128 = np_type  # pragma: no cover.
    np_cdouble = np_type  # pragma: no cover.
    np_complexfloating = np_type  # pragma: no cover.
else:
    np_array_factory = np.array
    np_ndarray = np.ndarray
    np_bool_ = np.bool_
    np_int8 = np.int8
    np_int16 = np.int16
    np_int32 = np.int32
    np_int64 = np.int64
    np_uint8 = np.uint8
    np_uint16 = np.uint16
    np_uint32 = np.uint32
    np_uint64 = np.uint64
    np_intp = np.intp
    np_uintp = np.uintp
    np_float32 = np.float32
    np_float64 = np.float64
    np_double = np.double  # np.float_ is an alias for np.double and is being removed by NumPy 2.0
    np_floating = np.floating
    np_complex64 = np.complex64
    np_complex128 = np.complex128
    np_cdouble = np.cdouble  # np.complex_ is an alias for np.cdouble and is being removed by NumPy 2.0
    np_complexfloating = np.complexfloating

numpy_numbers = (
    np_int8, np_int16, np_int32, np_int64, np_uint8,
    np_uint16, np_uint32, np_uint64, np_intp, np_uintp,
    np_float32, np_float64, np_double, np_floating, np_complex64,
    np_complex128, np_cdouble,)

numpy_complex_numbers = (
    np_complexfloating, np_complex64, np_complex128, np_cdouble,
)

numpy_dtypes = set(numpy_numbers)
numpy_dtypes.add(np_bool_)  # type: ignore

numpy_dtype_str_to_type = {
    item.__name__: item for item in numpy_dtypes
}

try:
    from pydantic.main import BaseModel as PydanticBaseModel  # type: ignore
except ImportError:
    PydanticBaseModel = pydantic_base_model_type


logger = logging.getLogger(__name__)

py_major_version = sys.version_info.major
py_minor_version = sys.version_info.minor

py_current_version = Decimal("{}.{}".format(py_major_version, py_minor_version))

py2 = py_major_version == 2
py3 = py_major_version == 3
py4 = py_major_version == 4


NUMERICS = frozenset(string.digits)


class EnumBase(str, enum.Enum):
    def __repr__(self):
        """
        We need to add a single quotes so we can easily copy the value when we do ipdb.
        """
        return f"'{self.name}'"

    def __str__(self):
        return self.name


def _int_or_zero(value):
    """
    Tries to extract some number from a string.

    12c becomes 12
    """
    try:
        return int(value)
    except Exception:
        result = []
        for char in value:
            if char in NUMERICS:
                result.append(char)
        if result:
            return int(''.join(result))
        return 0


def get_semvar_as_integer(version):
    """
    Converts:

    '1.23.5' to 1023005
    """
    version = version.split('.')
    if len(version) > 3:
        version = version[:3]
    elif len(version) < 3:
        version.extend(['0'] * (3 - len(version)))

    return sum([10**(i * 3) * _int_or_zero(v) for i, v in enumerate(reversed(version))])


# we used to use OrderedDictPlus when dictionaries in Python were not ordered.
dict_ = dict

if py4:
    logger.warning('Python 4 is not supported yet. Switching logic to Python 3.')  # pragma: no cover
    py3 = True  # pragma: no cover

if py2:  # pragma: no cover
    sys.exit('Python 2 is not supported anymore. The last version of DeepDiff that supported Py2 was 3.3.0')

pypy3 = py3 and hasattr(sys, "pypy_translation_info")


if np and get_semvar_as_integer(np.__version__) < 1019000:
    sys.exit('The minimum required Numpy version is 1.19.0. Please upgrade your Numpy package.')

strings = (str, bytes)  # which are both basestring
unicode_type = str
bytes_type = bytes
only_complex_number = (complex,) + numpy_complex_numbers
only_numbers = (int, float, complex, Decimal) + numpy_numbers
datetimes = (datetime.datetime, datetime.date, datetime.timedelta, datetime.time)
ipranges = (ipaddress.IPv4Interface, ipaddress.IPv6Interface, ipaddress.IPv4Network, ipaddress.IPv6Network)
uuids = (uuid.UUID, )
times = (datetime.datetime, datetime.time)
numbers: Tuple = only_numbers + datetimes
booleans = (bool, np_bool_)

basic_types = strings + numbers + uuids + booleans + (type(None), )

class IndexedHash(NamedTuple):
    indexes: List
    item: Any

current_dir = os.path.dirname(os.path.abspath(__file__))

ID_PREFIX = '!>*id'

KEY_TO_VAL_STR = "{}:{}"

TREE_VIEW = 'tree'
TEXT_VIEW = 'text'
DELTA_VIEW = '_delta'

ENUM_INCLUDE_KEYS = ['__objclass__', 'name', 'value']


def short_repr(item, max_length=15):
    """Short representation of item if it is too long"""
    item = repr(item)
    if len(item) > max_length:
        item = '{}...{}'.format(item[:max_length - 3], item[-1])
    return item


class ListItemRemovedOrAdded:  # pragma: no cover
    """Class of conditions to be checked"""
    pass


class OtherTypes:
    def __repr__(self):
        return "Error: {}".format(self.__class__.__name__)  # pragma: no cover

    __str__ = __repr__


class Skipped(OtherTypes):
    pass


class Unprocessed(OtherTypes):
    pass


class NotHashed(OtherTypes):
    pass


class NotPresent:  # pragma: no cover
    """
    In a change tree, this indicated that a previously existing object has been removed -- or will only be added
    in the future.
    We previously used None for this but this caused problem when users actually added and removed None. Srsly guys? :D
    """

    def __repr__(self):
        return 'not present'  # pragma: no cover

    __str__ = __repr__


class CannotCompare(Exception):
    """
    Exception when two items cannot be compared in the compare function.
    """
    pass


unprocessed = Unprocessed()
skipped = Skipped()
not_hashed = NotHashed()
notpresent = NotPresent()

# Disabling remapping from old to new keys since the mapping is deprecated.
RemapDict = dict_


# class RemapDict(dict_):
#     """
#     DISABLED
#     Remap Dictionary.

#     For keys that have a new, longer name, remap the old key to the new key.
#     Other keys that don't have a new name are handled as before.
#     """

#     def __getitem__(self, old_key):
#         new_key = EXPANDED_KEY_MAP.get(old_key, old_key)
#         if new_key != old_key:
#             logger.warning(
#                 "DeepDiff Deprecation: %s is renamed to %s. Please start using "
#                 "the new unified naming convention.", old_key, new_key)
#         if new_key in self:
#             return self.get(new_key)
#         else:  # pragma: no cover
#             raise KeyError(new_key)


class indexed_set(set):
    """
    A set class that lets you get an item by index

    >>> a = indexed_set()
    >>> a.add(10)
    >>> a.add(20)
    >>> a[0]
    10
    """


def add_to_frozen_set(parents_ids, item_id):
    return parents_ids | {item_id}


def convert_item_or_items_into_set_else_none(items):
    if items:
        if isinstance(items, strings):
            items = {items}
        else:
            items = set(items)
    else:
        items = None
    return items


def add_root_to_paths(paths):
    """
    Sometimes the users want to just pass
    [key] instead of root[key] for example.
    Here we automatically add all sorts of variations that might match
    the path they were supposed to pass. 
    """
    if paths is None:
        return
    result = SetOrdered()
    for path in paths:
        if path.startswith('root'):
            result.add(path)
        else:
            if path.isdigit():
                result.add(f"root['{path}']")
                result.add(f"root[{path}]")
            elif path[0].isdigit():
                result.add(f"root['{path}']")
            else:
                result.add(f"root.{path}")
                result.add(f"root['{path}']")
    return result


RE_COMPILED_TYPE = type(re.compile(''))


def convert_item_or_items_into_compiled_regexes_else_none(items):
    if items:
        if isinstance(items, (strings, RE_COMPILED_TYPE)):
            items = [items]
        items = [i if isinstance(i, RE_COMPILED_TYPE) else re.compile(i) for i in items]
    else:
        items = None
    return items


def get_id(obj):
    """
    Adding some characters to id so they are not just integers to reduce the risk of collision.
    """
    return "{}{}".format(ID_PREFIX, id(obj))


def get_type(obj):
    """
    Get the type of object or if it is a class, return the class itself.
    """
    if isinstance(obj, np_ndarray):
        return obj.dtype.type  # type: ignore
    return obj if type(obj) is type else type(obj)


def numpy_dtype_string_to_type(dtype_str):
    return numpy_dtype_str_to_type[dtype_str]


def type_in_type_group(item, type_group):
    return get_type(item) in type_group


def type_is_subclass_of_type_group(item, type_group):
    return isinstance(item, type_group) \
        or (isinstance(item, type) and issubclass(item, type_group)) \
        or type_in_type_group(item, type_group)


def get_doc(doc_filename):
    try:
        with open(os.path.join(current_dir, '../docs/', doc_filename), 'r') as doc_file:
            doc = doc_file.read()
    except Exception:  # pragma: no cover
        doc = 'Failed to load the docstrings. Please visit: https://zepworks.com/deepdiff/current/'  # pragma: no cover
    return doc


number_formatting = {
    "f": r'{:.%sf}',
    "e": r'{:.%se}',
}


def number_to_string(number, significant_digits, number_format_notation="f"):
    """
    Convert numbers to string considering significant digits.
    """
    try:
        using = number_formatting[number_format_notation]
    except KeyError:
        raise ValueError("number_format_notation got invalid value of {}. The valid values are 'f' and 'e'".format(number_format_notation)) from None

    if not isinstance(number, numbers):  # type: ignore
        return number
    elif isinstance(number, Decimal):
        with localcontext() as ctx:
            # Precision = number of integer digits + significant_digits
            # Using number//1 to get the integer part of the number
            ctx.prec = len(str(abs(number // 1))) + significant_digits
            try:
                number = number.quantize(Decimal('0.' + '0' * significant_digits))
            except InvalidDecimalOperation:
                # Sometimes rounding up causes a higher precision to be needed for the quantize operation
                # For example '999.99999999' will become '1000.000000' after quantize
                ctx.prec += 1
                number = number.quantize(Decimal('0.' + '0' * significant_digits))
    elif isinstance(number, only_complex_number):  # type: ignore
        # Case for complex numbers.
        number = number.__class__(
            "{real}+{imag}j".format(  # type: ignore
                real=number_to_string(
                    number=number.real,  # type: ignore
                    significant_digits=significant_digits,
                    number_format_notation=number_format_notation
                ),
                imag=number_to_string(
                    number=number.imag,  # type: ignore
                    significant_digits=significant_digits,
                    number_format_notation=number_format_notation
                )
            )  # type: ignore
        )
    else:
        number = round(number=number, ndigits=significant_digits)  # type: ignore

        if significant_digits == 0:
            number = int(number)

    if number == 0.0:
        # Special case for 0: "-0.xx" should compare equal to "0.xx"
        number = abs(number)  # type: ignore

    # Cast number to string
    result = (using % significant_digits).format(number)
    # https://bugs.python.org/issue36622
    if number_format_notation == 'e':
        # Removing leading 0 for exponential part.
        result = re.sub(
            pattern=r'(?<=e(\+|\-))0(?=\d)+',
            repl=r'',
            string=result
        )
    return result


class DeepDiffDeprecationWarning(DeprecationWarning):
    """
    Use this warning instead of DeprecationWarning
    """
    pass


def cartesian_product(a, b):
    """
    Get the Cartesian product of two iterables

    **parameters**

    a: list of lists
    b: iterable to do the Cartesian product
    """

    for i in a:
        for j in b:
            yield i + (j,)


def cartesian_product_of_shape(dimentions, result=None):
    """
    Cartesian product of a dimentions iterable.
    This is mainly used to traverse Numpy ndarrays.

    Each array has dimentions that are defines in ndarray.shape
    """
    if result is None:
        result = ((),)  # a tuple with an empty tuple
    for dimension in dimentions:
        result = cartesian_product(result, range(dimension))
    return result


def get_numpy_ndarray_rows(obj, shape=None):
    """
    Convert a multi dimensional numpy array to list of rows
    """
    if shape is None:
        shape = obj.shape

    dimentions = shape[:-1]
    for path_tuple in cartesian_product_of_shape(dimentions):
        result = obj
        for index in path_tuple:
            result = result[index]
        yield path_tuple, result


class _NotFound:

    def __eq__(self, other):
        return False

    __req__ = __eq__

    def __repr__(self):
        return 'not found'

    __str__ = __repr__


not_found = _NotFound()

warnings.simplefilter('once', DeepDiffDeprecationWarning)


class RepeatedTimer:
    """
    Threaded Repeated Timer by MestreLion
    https://stackoverflow.com/a/38317060/1497443
    """

    def __init__(self, interval, function, *args, **kwargs):
        self._timer = None
        self.interval = interval
        self.function = function
        self.args = args
        self.start_time = time.time()
        self.kwargs = kwargs
        self.is_running = False
        self.start()

    def _get_duration_sec(self):
        return int(time.time() - self.start_time)

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        self.kwargs.update(duration=self._get_duration_sec())
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        duration = self._get_duration_sec()
        if self._timer is not None:
            self._timer.cancel()
        self.is_running = False
        return duration


def _eval_decimal(params):
    return Decimal(params)


def _eval_datetime(params):
    params = f'({params})'
    params = literal_eval(params)
    return datetime.datetime(*params)


def _eval_date(params):
    params = f'({params})'
    params = literal_eval(params)
    return datetime.date(*params)


LITERAL_EVAL_PRE_PROCESS = [
    ('Decimal(', ')', _eval_decimal),
    ('datetime.datetime(', ')', _eval_datetime),
    ('datetime.date(', ')', _eval_date),
]


def literal_eval_extended(item):
    """
    An extended version of literal_eval
    """
    try:
        return literal_eval(item)
    except (SyntaxError, ValueError):
        for begin, end, func in LITERAL_EVAL_PRE_PROCESS:
            if item.startswith(begin) and item.endswith(end):
                # Extracting and removing extra quotes so for example "Decimal('10.1')" becomes "'10.1'" and then '10.1'
                params = item[len(begin): -len(end)].strip('\'\"')
                return func(params)
        raise


def time_to_seconds(t:datetime.time) -> int:
    return (t.hour * 60 + t.minute) * 60 + t.second


def datetime_normalize(
    truncate_datetime:Union[str, None],
    obj:Union[datetime.datetime, datetime.time],
    default_timezone: Union[
        datetime.timezone, "BaseTzInfo"
    ] = datetime.timezone.utc,
) -> Any:
    if truncate_datetime:
        if truncate_datetime == 'second':
            obj = obj.replace(microsecond=0)
        elif truncate_datetime == 'minute':
            obj = obj.replace(second=0, microsecond=0)
        elif truncate_datetime == 'hour':
            obj = obj.replace(minute=0, second=0, microsecond=0)
        elif truncate_datetime == 'day':
            obj = obj.replace(hour=0, minute=0, second=0, microsecond=0)
    if isinstance(obj, datetime.datetime):
        if has_timezone(obj):
            obj = obj.astimezone(default_timezone)
        else:
            obj = obj.replace(tzinfo=default_timezone)
    elif isinstance(obj, datetime.time):
        return time_to_seconds(obj)
    return obj


def has_timezone(dt):
    """
    Function to check if a datetime object has a timezone

    Checking dt.tzinfo.utcoffset(dt) ensures that the datetime object is truly timezone-aware
    because some datetime objects may have a tzinfo attribute that is not None but still
    doesn't provide a valid offset.

    Certain tzinfo objects, such as pytz.timezone(None), can exist but do not provide meaningful UTC offset information.
    If tzinfo is present but calling .utcoffset(dt) returns None, the datetime is not truly timezone-aware.
    """
    return dt.tzinfo is not None and dt.tzinfo.utcoffset(dt) is not None


def get_truncate_datetime(truncate_datetime) -> Union[str, None]:
    """
    Validates truncate_datetime value
    """
    if truncate_datetime not in {None, 'second', 'minute', 'hour', 'day'}:
        raise ValueError("truncate_datetime must be second, minute, hour or day")
    return truncate_datetime


def cartesian_product_numpy(*arrays):
    """
    Cartesian product of Numpy arrays by Paul Panzer
    https://stackoverflow.com/a/49445693/1497443
    """
    la = len(arrays)
    dtype = np.result_type(*arrays)  # type: ignore
    arr = np.empty((la, *map(len, arrays)), dtype=dtype)  # type: ignore
    idx = slice(None), *repeat(None, la)
    for i, a in enumerate(arrays):
        arr[i, ...] = a[idx[:la - i]]
    return arr.reshape(la, -1).T


def diff_numpy_array(A, B):
    """
    Numpy Array A - B
    return items in A that are not in B
    By Divakar
    https://stackoverflow.com/a/52417967/1497443
    """
    return A[~np.isin(A, B)]  # type: ignore


PYTHON_TYPE_TO_NUMPY_TYPE = {
    int: np_int64,
    float: np_float64,
    Decimal: np_float64
}


def get_homogeneous_numpy_compatible_type_of_seq(seq):
    """
    Return with the numpy dtype if the array can be converted to a non-object numpy array.
    Originally written by mgilson https://stackoverflow.com/a/13252348/1497443
    This is the modified version.
    """
    iseq = iter(seq)
    first_type = type(next(iseq))
    if first_type in {int, float, Decimal}:
        type_ = first_type if all((type(x) is first_type) for x in iseq) else False
        return PYTHON_TYPE_TO_NUMPY_TYPE.get(type_, False)
    else:
        return False


def detailed__dict__(obj, ignore_private_variables=True, ignore_keys=frozenset(), include_keys=None):
    """
    Get the detailed dictionary of an object.

    This is used so we retrieve object properties too.
    """
    if include_keys:
        result = {}
        for key in include_keys:
            try:
                value = getattr(obj, key)
            except Exception:
                pass
            else:
                if not callable(value) or key == '__objclass__':  # We don't want to compare functions, however for backward compatibility, __objclass__ needs to be reported.
                    result[key] = value
    else:
        result = obj.__dict__.copy()  # A shallow copy
        private_var_prefix = f"_{obj.__class__.__name__}__"  # The semi private variables in Python get this prefix
        for key in ignore_keys:
            if key in result or (
                ignore_private_variables and key.startswith('__') and not key.startswith(private_var_prefix)
            ):
                del result[key]
        for key in dir(obj):
            if key not in result and key not in ignore_keys and (
                    not ignore_private_variables or (
                        ignore_private_variables and not key.startswith('__') and not key.startswith(private_var_prefix)
                    )
            ):
                value = getattr(obj, key)
                if not callable(value):
                    result[key] = value
    return result


def named_tuple_repr(self):
    fields = []
    for field, value in self._asdict().items():
        # Only include fields that do not have their default value
        if field in self._field_defaults:
            if value != self._field_defaults[field]:
                fields.append(f"{field}={value!r}")
        else:
            fields.append(f"{field}={value!r}")

    return f"{self.__class__.__name__}({', '.join(fields)})"


class OpcodeTag(EnumBase):
    insert = 'insert'
    delete = 'delete'
    equal = 'equal'
    replace = 'replace'  # type: ignore
    # swapped = 'swapped'  # in the future we should support reporting of items swapped with each other


class Opcode(NamedTuple):
    tag: str
    t1_from_index: int
    t1_to_index: int
    t2_from_index: int
    t2_to_index: int
    old_values: Optional[List[Any]] = None
    new_values: Optional[List[Any]] = None

    __repr__ = __str__ = named_tuple_repr


class FlatDataAction(EnumBase):
    values_changed = 'values_changed'
    type_changes = 'type_changes'
    set_item_added = 'set_item_added'
    set_item_removed = 'set_item_removed'
    dictionary_item_added = 'dictionary_item_added'
    dictionary_item_removed = 'dictionary_item_removed'
    iterable_item_added = 'iterable_item_added'
    iterable_item_removed = 'iterable_item_removed'
    iterable_item_moved = 'iterable_item_moved'
    iterable_items_inserted = 'iterable_items_inserted'  # opcode
    iterable_items_deleted = 'iterable_items_deleted'  # opcode
    iterable_items_replaced = 'iterable_items_replaced'  # opcode
    iterable_items_equal = 'iterable_items_equal'  # opcode
    attribute_removed = 'attribute_removed'
    attribute_added = 'attribute_added'
    unordered_iterable_item_added = 'unordered_iterable_item_added'
    unordered_iterable_item_removed = 'unordered_iterable_item_removed'
    initiated = "initiated"


OPCODE_TAG_TO_FLAT_DATA_ACTION = {
    OpcodeTag.insert: FlatDataAction.iterable_items_inserted,
    OpcodeTag.delete: FlatDataAction.iterable_items_deleted,
    OpcodeTag.replace: FlatDataAction.iterable_items_replaced,
    OpcodeTag.equal: FlatDataAction.iterable_items_equal,
}

FLAT_DATA_ACTION_TO_OPCODE_TAG = {v: i for i, v in OPCODE_TAG_TO_FLAT_DATA_ACTION.items()}


UnkownValueCode: str = 'unknown___'


class FlatDeltaRow(NamedTuple):
    path: List
    action: FlatDataAction
    value: Optional[Any] = UnkownValueCode
    old_value: Optional[Any] = UnkownValueCode
    type: Optional[Any] = UnkownValueCode
    old_type: Optional[Any] = UnkownValueCode
    new_path: Optional[List] = None
    t1_from_index: Optional[int] = None
    t1_to_index: Optional[int] = None
    t2_from_index: Optional[int] = None
    t2_to_index: Optional[int] = None

    __repr__ = __str__ = named_tuple_repr


JSON = Union[Dict[str, str], List[str], List[int], Dict[str, "JSON"], List["JSON"], str, int, float, bool, None]


class SummaryNodeType(EnumBase):
    dict = 'dict'
    list = 'list'
    leaf = 'leaf'