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

import copy
import logging
from typing import List, Dict, IO, Callable, Set, Union, Optional
from functools import partial, cmp_to_key
from collections.abc import Mapping
from copy import deepcopy
from deepdiff import DeepDiff
from deepdiff.serialization import pickle_load, pickle_dump
from deepdiff.helper import (
    strings, numbers,
    np_ndarray, np_array_factory, numpy_dtypes, get_doc,
    not_found, numpy_dtype_string_to_type, dict_,
    Opcode, FlatDeltaRow, UnkownValueCode, FlatDataAction,
    OPCODE_TAG_TO_FLAT_DATA_ACTION,
    FLAT_DATA_ACTION_TO_OPCODE_TAG,
    SetOrdered,
)
from deepdiff.path import (
    _path_to_elements, _get_nested_obj, _get_nested_obj_and_force,
    GET, GETATTR, parse_path, stringify_path,
)
from deepdiff.anyset import AnySet
from deepdiff.summarize import summarize

logger = logging.getLogger(__name__)


VERIFICATION_MSG = 'Expected the old value for {} to be {} but it is {}. Error found on: {}. You may want to set force=True, especially if this delta is created by passing flat_rows_list or flat_dict_list'
ELEM_NOT_FOUND_TO_ADD_MSG = 'Key or index of {} is not found for {} for setting operation.'
TYPE_CHANGE_FAIL_MSG = 'Unable to do the type change for {} from to type {} due to {}'
VERIFY_BIDIRECTIONAL_MSG = ('You have applied the delta to an object that has '
                            'different values than the original object the delta was made from.')
FAIL_TO_REMOVE_ITEM_IGNORE_ORDER_MSG = 'Failed to remove index[{}] on {}. It was expected to be {} but got {}'
DELTA_NUMPY_OPERATOR_OVERRIDE_MSG = (
    'A numpy ndarray is most likely being added to a delta. '
    'Due to Numpy override the + operator, you can only do: delta + ndarray '
    'and NOT ndarray + delta')
BINIARY_MODE_NEEDED_MSG = "Please open the file in the binary mode and pass to Delta by passing 'b' in open(..., 'b'): {}"
DELTA_AT_LEAST_ONE_ARG_NEEDED = 'At least one of the diff, delta_path or delta_file arguments need to be passed.'
INVALID_ACTION_WHEN_CALLING_GET_ELEM = 'invalid action of {} when calling _get_elem_and_compare_to_old_value'
INVALID_ACTION_WHEN_CALLING_SIMPLE_SET_ELEM = 'invalid action of {} when calling _simple_set_elem_value'
INVALID_ACTION_WHEN_CALLING_SIMPLE_DELETE_ELEM = 'invalid action of {} when calling _simple_set_elem_value'
UNABLE_TO_GET_ITEM_MSG = 'Unable to get the item at {}: {}'
UNABLE_TO_GET_PATH_MSG = 'Unable to get the item at {}'
INDEXES_NOT_FOUND_WHEN_IGNORE_ORDER = 'Delta added to an incompatible object. Unable to add the following items at the specific indexes. {}'
NUMPY_TO_LIST = 'NUMPY_TO_LIST'
NOT_VALID_NUMPY_TYPE = "{} is not a valid numpy type."

doc = get_doc('delta.rst')


class DeltaError(ValueError):
    """
    Delta specific errors
    """
    pass


class DeltaNumpyOperatorOverrideError(ValueError):
    """
    Delta Numpy Operator Override Error
    """
    pass


class Delta:

    __doc__ = doc

    def __init__(
        self,
        diff: Union[DeepDiff, Mapping, str, bytes, None]=None,
        delta_path: Optional[str]=None,
        delta_file: Optional[IO]=None,
        delta_diff: Optional[dict]=None,
        flat_dict_list: Optional[List[Dict]]=None,
        flat_rows_list: Optional[List[FlatDeltaRow]]=None,
        deserializer: Callable=pickle_load,
        log_errors: bool=True,
        mutate: bool=False,
        raise_errors: bool=False,
        safe_to_import: Optional[Set[str]]=None,
        serializer: Callable=pickle_dump,
        verify_symmetry: Optional[bool]=None,
        bidirectional: bool=False,
        always_include_values: bool=False,
        iterable_compare_func_was_used: Optional[bool]=None,
        force: bool=False,
    ):
        # for pickle deserializer:
        if hasattr(deserializer, '__code__') and 'safe_to_import' in set(deserializer.__code__.co_varnames):
            _deserializer = deserializer
        else:
            def _deserializer(obj, safe_to_import=None):
                result = deserializer(obj)
                if result.get('_iterable_opcodes'):
                    _iterable_opcodes = {}
                    for path, op_codes in result['_iterable_opcodes'].items():
                        _iterable_opcodes[path] = []
                        for op_code in op_codes:
                            _iterable_opcodes[path].append(
                                Opcode(
                                    **op_code
                                )
                            )
                    result['_iterable_opcodes'] = _iterable_opcodes
                return result


        self._reversed_diff = None

        if verify_symmetry is not None:
            logger.warning(
                "DeepDiff Deprecation: use bidirectional instead of verify_symmetry parameter."
            )
            bidirectional = verify_symmetry

        self.bidirectional = bidirectional
        if bidirectional:
            self.always_include_values = True  # We need to include the values in bidirectional deltas
        else:
            self.always_include_values = always_include_values

        if diff is not None:
            if isinstance(diff, DeepDiff):
                self.diff = diff._to_delta_dict(directed=not bidirectional, always_include_values=self.always_include_values)
            elif isinstance(diff, Mapping):
                self.diff = diff
            elif isinstance(diff, strings):
                self.diff = _deserializer(diff, safe_to_import=safe_to_import)
        elif delta_path:
            with open(delta_path, 'rb') as the_file:
                content = the_file.read()
            self.diff = _deserializer(content, safe_to_import=safe_to_import)
        elif delta_diff:
            self.diff = delta_diff
        elif delta_file:
            try:
                content = delta_file.read()
            except UnicodeDecodeError as e:
                raise ValueError(BINIARY_MODE_NEEDED_MSG.format(e)) from None
            self.diff = _deserializer(content, safe_to_import=safe_to_import)
        elif flat_dict_list:
            # Use copy to preserve original value of flat_dict_list in calling module
            self.diff = self._from_flat_dicts(copy.deepcopy(flat_dict_list))
        elif flat_rows_list:
            self.diff = self._from_flat_rows(copy.deepcopy(flat_rows_list))
        else:
            raise ValueError(DELTA_AT_LEAST_ONE_ARG_NEEDED)

        self.mutate = mutate
        self.raise_errors = raise_errors
        self.log_errors = log_errors
        self._numpy_paths = self.diff.get('_numpy_paths', False)
        # When we create the delta from a list of flat dictionaries, details such as iterable_compare_func_was_used get lost.
        # That's why we allow iterable_compare_func_was_used to be explicitly set.
        self._iterable_compare_func_was_used = self.diff.get('_iterable_compare_func_was_used', iterable_compare_func_was_used)
        self.serializer = serializer
        self.deserializer = deserializer
        self.force = force
        if force:
            self.get_nested_obj = _get_nested_obj_and_force
        else:
            self.get_nested_obj = _get_nested_obj
        self.reset()

    def __repr__(self):
        return "<Delta: {}>".format(summarize(self.diff, max_length=100))

    def reset(self):
        self.post_process_paths_to_convert = dict_()

    def __add__(self, other):
        if isinstance(other, numbers) and self._numpy_paths:  # type: ignore
            raise DeltaNumpyOperatorOverrideError(DELTA_NUMPY_OPERATOR_OVERRIDE_MSG)
        if self.mutate:
            self.root = other
        else:
            self.root = deepcopy(other)
        self._do_pre_process()
        self._do_values_changed()
        self._do_set_item_added()
        self._do_set_item_removed()
        self._do_type_changes()
        # NOTE: the remove iterable action needs to happen BEFORE
        # all the other iterables to match the reverse of order of operations in DeepDiff
        self._do_iterable_opcodes()
        self._do_iterable_item_removed()
        self._do_iterable_item_added()
        self._do_ignore_order()
        self._do_dictionary_item_added()
        self._do_dictionary_item_removed()
        self._do_attribute_added()
        self._do_attribute_removed()
        self._do_post_process()

        other = self.root
        # removing the reference to other
        del self.root
        self.reset()
        return other

    __radd__ = __add__

    def __rsub__(self, other):
        if self._reversed_diff is None:
            self._reversed_diff = self._get_reverse_diff()
        self.diff, self._reversed_diff = self._reversed_diff, self.diff
        result = self.__add__(other)
        self.diff, self._reversed_diff = self._reversed_diff, self.diff
        return result

    def _raise_or_log(self, msg, level='error'):
        if self.log_errors:
            getattr(logger, level)(msg)
        if self.raise_errors:
            raise DeltaError(msg)

    def _do_verify_changes(self, path, expected_old_value, current_old_value):
        if self.bidirectional and expected_old_value != current_old_value:
            if isinstance(path, str):
                path_str = path
            else:
                path_str = stringify_path(path, root_element=('', GETATTR))
            self._raise_or_log(VERIFICATION_MSG.format(
                path_str, expected_old_value, current_old_value, VERIFY_BIDIRECTIONAL_MSG))

    def _get_elem_and_compare_to_old_value(
        self,
        obj,
        path_for_err_reporting,
        expected_old_value,
        elem=None,
        action=None,
        forced_old_value=None,
        next_element=None,
    ):
        # if forced_old_value is not None:
        try:
            if action == GET:
                current_old_value = obj[elem]
            elif action == GETATTR:
                current_old_value = getattr(obj, elem)  # type: ignore
            else:
                raise DeltaError(INVALID_ACTION_WHEN_CALLING_GET_ELEM.format(action))
        except (KeyError, IndexError, AttributeError, TypeError) as e:
            if self.force:
                if forced_old_value is None:
                    if next_element is None or isinstance(next_element, str):
                        _forced_old_value = {}
                    else:
                        _forced_old_value = []    
                else:
                    _forced_old_value = forced_old_value
                if action == GET:
                    if isinstance(obj, list):
                        if isinstance(elem, int) and elem < len(obj):
                            obj[elem] = _forced_old_value
                        else:
                            obj.append(_forced_old_value)
                    else:
                        obj[elem] = _forced_old_value
                elif action == GETATTR:
                    setattr(obj, elem, _forced_old_value)  # type: ignore
                return _forced_old_value
            current_old_value = not_found
            if isinstance(path_for_err_reporting, (list, tuple)):
                path_for_err_reporting = '.'.join([i[0] for i in path_for_err_reporting])
            if self.bidirectional:
                self._raise_or_log(VERIFICATION_MSG.format(
                    path_for_err_reporting,
                    expected_old_value, current_old_value, e))
            else:
                self._raise_or_log(UNABLE_TO_GET_PATH_MSG.format(
                    path_for_err_reporting))
        return current_old_value

    def _simple_set_elem_value(self, obj, path_for_err_reporting, elem=None, value=None, action=None):
        """
        Set the element value directly on an object
        """
        try:
            if action == GET:
                try:
                    obj[elem] = value
                except IndexError:
                    if elem == len(obj):
                        obj.append(value)
                    else:
                        self._raise_or_log(ELEM_NOT_FOUND_TO_ADD_MSG.format(elem, path_for_err_reporting))
            elif action == GETATTR:
                setattr(obj, elem, value)  # type: ignore
            else:
                raise DeltaError(INVALID_ACTION_WHEN_CALLING_SIMPLE_SET_ELEM.format(action))
        except (KeyError, IndexError, AttributeError, TypeError) as e:
            self._raise_or_log('Failed to set {} due to {}'.format(path_for_err_reporting, e))

    def _coerce_obj(self, parent, obj, path, parent_to_obj_elem,
                    parent_to_obj_action, elements, to_type, from_type):
        """
        Coerce obj and mark it in post_process_paths_to_convert for later to be converted back.
        Also reassign it to its parent to replace the old object.
        """
        self.post_process_paths_to_convert[elements[:-1]] = {'old_type': to_type, 'new_type': from_type}
        # If this function is going to ever be used to convert numpy arrays, uncomment these lines:
        # if from_type is np_ndarray:
        #     obj = obj.tolist()
        # else:
        obj = to_type(obj)

        if parent:
            # Making sure that the object is re-instated inside the parent especially if it was immutable
            # and we had to turn it into a mutable one. In such cases the object has a new id.
            self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
                                        value=obj, action=parent_to_obj_action)
        return obj

    def _set_new_value(self, parent, parent_to_obj_elem, parent_to_obj_action,
                       obj, elements, path, elem, action, new_value):
        """
        Set the element value on an object and if necessary convert the object to the proper mutable type
        """
        if isinstance(obj, tuple):
            # convert this object back to a tuple later
            obj = self._coerce_obj(
                parent, obj, path, parent_to_obj_elem,
                parent_to_obj_action, elements,
                to_type=list, from_type=tuple)
        if elem != 0 and self.force and isinstance(obj, list) and len(obj) == 0:
            # it must have been a dictionary    
            obj = {}
            self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
                                        value=obj, action=parent_to_obj_action)
        self._simple_set_elem_value(obj=obj, path_for_err_reporting=path, elem=elem,
                                    value=new_value, action=action)

    def _simple_delete_elem(self, obj, path_for_err_reporting, elem=None, action=None):
        """
        Delete the element directly on an object
        """
        try:
            if action == GET:
                del obj[elem]
            elif action == GETATTR:
                del obj.__dict__[elem]
            else:
                raise DeltaError(INVALID_ACTION_WHEN_CALLING_SIMPLE_DELETE_ELEM.format(action))
        except (KeyError, IndexError, AttributeError) as e:
            self._raise_or_log('Failed to set {} due to {}'.format(path_for_err_reporting, e))

    def _del_elem(self, parent, parent_to_obj_elem, parent_to_obj_action,
                  obj, elements, path, elem, action):
        """
        Delete the element value on an object and if necessary convert the object to the proper mutable type
        """
        obj_is_new = False
        if isinstance(obj, tuple):
            # convert this object back to a tuple later
            self.post_process_paths_to_convert[elements[:-1]] = {'old_type': list, 'new_type': tuple}
            obj = list(obj)
            obj_is_new = True
        self._simple_delete_elem(obj=obj, path_for_err_reporting=path, elem=elem, action=action)
        if obj_is_new and parent:
            # Making sure that the object is re-instated inside the parent especially if it was immutable
            # and we had to turn it into a mutable one. In such cases the object has a new id.
            self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
                                        value=obj, action=parent_to_obj_action)

    def _do_iterable_item_added(self):
        iterable_item_added = self.diff.get('iterable_item_added', {})
        iterable_item_moved = self.diff.get('iterable_item_moved')

        # First we need to create a placeholder for moved items.
        # This will then get replaced below after we go through added items.
        # Without this items can get double added because moved store the new_value and does not need item_added replayed
        if iterable_item_moved:
            added_dict = {v["new_path"]: None for k, v in iterable_item_moved.items()}
            iterable_item_added.update(added_dict)

        if iterable_item_added:
            self._do_item_added(iterable_item_added, insert=True)

        if iterable_item_moved:
            added_dict = {v["new_path"]: v["value"] for k, v in iterable_item_moved.items()}
            self._do_item_added(added_dict, insert=False)

    def _do_dictionary_item_added(self):
        dictionary_item_added = self.diff.get('dictionary_item_added')
        if dictionary_item_added:
            self._do_item_added(dictionary_item_added, sort=False)

    def _do_attribute_added(self):
        attribute_added = self.diff.get('attribute_added')
        if attribute_added:
            self._do_item_added(attribute_added)

    @staticmethod
    def _sort_key_for_item_added(path_and_value):
        elements = _path_to_elements(path_and_value[0])
        # Example elements: [(4.3, 'GET'), ('b', 'GETATTR'), ('a3', 'GET')]
        # We only care about the values in the elements not how to get the values.
        return [i[0] for i in elements] 

    @staticmethod
    def _sort_comparison(left, right):
        """
        We use sort comparison instead of _sort_key_for_item_added when we run into comparing element types that can not
        be compared with each other, such as None to None. Or integer to string.
        """
        # Example elements: [(4.3, 'GET'), ('b', 'GETATTR'), ('a3', 'GET')]
        # We only care about the values in the elements not how to get the values.
        left_path = [i[0] for i in _path_to_elements(left[0], root_element=None)]
        right_path = [i[0] for i in _path_to_elements(right[0], root_element=None)]
        try:
            if left_path < right_path:
                return -1
            elif left_path > right_path:
                return 1
            else:
                return 0
        except TypeError:
            if len(left_path) > len(right_path):
                left_path = left_path[:len(right_path)]
            elif len(right_path) > len(left_path):
                right_path = right_path[:len(left_path)]
            for l_elem, r_elem in zip(left_path, right_path):
                if type(l_elem) != type(r_elem) or type(l_elem) in None:
                    l_elem = str(l_elem)
                    r_elem = str(r_elem)
                try:
                    if l_elem < r_elem:
                        return -1
                    elif l_elem > r_elem:
                        return 1
                except TypeError:
                    continue
        return 0


    def _do_item_added(self, items, sort=True, insert=False):
        if sort:
            # sorting items by their path so that the items with smaller index
            # are applied first (unless `sort` is `False` so that order of
            # added items is retained, e.g. for dicts).
            try:
                items = sorted(items.items(), key=self._sort_key_for_item_added)
            except TypeError:
                items = sorted(items.items(), key=cmp_to_key(self._sort_comparison))
        else:
            items = items.items()

        for path, new_value in items:
            elem_and_details = self._get_elements_and_details(path)
            if elem_and_details:
                elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details
            else:
                continue  # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198

            # Insert is only true for iterables, make sure it is a valid index.
            if(insert and elem < len(obj)):  # type: ignore
                obj.insert(elem, None)  # type: ignore

            self._set_new_value(parent, parent_to_obj_elem, parent_to_obj_action,
                                obj, elements, path, elem, action, new_value)

    def _do_values_changed(self):
        values_changed = self.diff.get('values_changed')
        if values_changed:
            self._do_values_or_type_changed(values_changed)

    def _do_type_changes(self):
        type_changes = self.diff.get('type_changes')
        if type_changes:
            self._do_values_or_type_changed(type_changes, is_type_change=True)

    def _do_post_process(self):
        if self.post_process_paths_to_convert:
            # Example: We had converted some object to be mutable and now we are converting them back to be immutable.
            # We don't need to check the change because it is not really a change that was part of the original diff.
            self._do_values_or_type_changed(self.post_process_paths_to_convert, is_type_change=True, verify_changes=False)

    def _do_pre_process(self):
        if self._numpy_paths and ('iterable_item_added' in self.diff or 'iterable_item_removed' in self.diff):
            preprocess_paths = dict_()
            for path, type_ in self._numpy_paths.items():  # type: ignore
                preprocess_paths[path] = {'old_type': np_ndarray, 'new_type': list}
                try:
                    type_ = numpy_dtype_string_to_type(type_)
                except Exception as e:
                    self._raise_or_log(NOT_VALID_NUMPY_TYPE.format(e))
                    continue  # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198
                self.post_process_paths_to_convert[path] = {'old_type': list, 'new_type': type_}
            if preprocess_paths:
                self._do_values_or_type_changed(preprocess_paths, is_type_change=True)

    def _get_elements_and_details(self, path):
        try:
            elements = _path_to_elements(path)
            if len(elements) > 1:
                elements_subset = elements[:-2]
                if len(elements_subset) != len(elements):
                    next_element = elements[-2][0]
                    next2_element = elements[-1][0]
                else:
                    next_element = None
                parent = self.get_nested_obj(obj=self, elements=elements_subset, next_element=next_element)
                parent_to_obj_elem, parent_to_obj_action = elements[-2]
                obj = self._get_elem_and_compare_to_old_value(
                    obj=parent, path_for_err_reporting=path, expected_old_value=None,
                    elem=parent_to_obj_elem, action=parent_to_obj_action, next_element=next2_element)  # type: ignore
            else:
                # parent = self
                # obj = self.root
                # parent_to_obj_elem = 'root'
                # parent_to_obj_action = GETATTR
                parent = parent_to_obj_elem = parent_to_obj_action = None
                obj = self
                # obj = self.get_nested_obj(obj=self, elements=elements[:-1])
            elem, action = elements[-1]  # type: ignore
        except Exception as e:
            self._raise_or_log(UNABLE_TO_GET_ITEM_MSG.format(path, e))
            return None
        else:
            if obj is not_found:
                return None
            return elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action

    def _do_values_or_type_changed(self, changes, is_type_change=False, verify_changes=True):
        for path, value in changes.items():
            elem_and_details = self._get_elements_and_details(path)
            if elem_and_details:
                elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details
            else:
                continue  # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198
            expected_old_value = value.get('old_value', not_found)

            current_old_value = self._get_elem_and_compare_to_old_value(
                obj=obj, path_for_err_reporting=path, expected_old_value=expected_old_value, elem=elem, action=action)
            if current_old_value is not_found:
                continue  # pragma: no cover. I have not been able to write a test for this case. But we should still check for it.
            # With type change if we could have originally converted the type from old_value
            # to new_value just by applying the class of the new_value, then we might not include the new_value
            # in the delta dictionary. That is defined in Model.DeltaResult._from_tree_type_changes
            if is_type_change and 'new_value' not in value:
                try:
                    new_type = value['new_type']
                    # in case of Numpy we pass the ndarray plus the dtype in a tuple
                    if new_type in numpy_dtypes:
                        new_value = np_array_factory(current_old_value, new_type)
                    else:
                        new_value = new_type(current_old_value)
                except Exception as e:
                    self._raise_or_log(TYPE_CHANGE_FAIL_MSG.format(obj[elem], value.get('new_type', 'unknown'), e))  # type: ignore
                    continue
            else:
                new_value = value['new_value']

            self._set_new_value(parent, parent_to_obj_elem, parent_to_obj_action,
                                obj, elements, path, elem, action, new_value)

            if verify_changes:
                self._do_verify_changes(path, expected_old_value, current_old_value)

    def _do_item_removed(self, items):
        """
        Handle removing items.
        """
        # Sorting the iterable_item_removed in reverse order based on the paths.
        # So that we delete a bigger index before a smaller index
        try:
            sorted_item = sorted(items.items(), key=self._sort_key_for_item_added, reverse=True)
        except TypeError:
            sorted_item = sorted(items.items(), key=cmp_to_key(self._sort_comparison), reverse=True)
        for path, expected_old_value in sorted_item:
            elem_and_details = self._get_elements_and_details(path)
            if elem_and_details:
                elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details
            else:
                continue  # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198

            look_for_expected_old_value = False
            current_old_value = not_found
            try:
                if action == GET:
                    current_old_value = obj[elem]  # type: ignore
                elif action == GETATTR:
                    current_old_value = getattr(obj, elem)
                look_for_expected_old_value = current_old_value != expected_old_value
            except (KeyError, IndexError, AttributeError, TypeError):
                look_for_expected_old_value = True

            if look_for_expected_old_value and isinstance(obj, list) and not self._iterable_compare_func_was_used:
                # It may return None if it doesn't find it
                elem = self._find_closest_iterable_element_for_index(obj, elem, expected_old_value)
                if elem is not None:
                    current_old_value = expected_old_value
            if current_old_value is not_found or elem is None:
                continue

            self._del_elem(parent, parent_to_obj_elem, parent_to_obj_action,
                           obj, elements, path, elem, action)
            self._do_verify_changes(path, expected_old_value, current_old_value)

    def _find_closest_iterable_element_for_index(self, obj, elem, expected_old_value):
        closest_elem = None
        closest_distance = float('inf')
        for index, value in enumerate(obj):
            dist = abs(index - elem)
            if dist > closest_distance:
                break
            if value == expected_old_value and dist < closest_distance:
                closest_elem = index
                closest_distance = dist
        return closest_elem

    def _do_iterable_opcodes(self):
        _iterable_opcodes = self.diff.get('_iterable_opcodes', {})
        if _iterable_opcodes:
            for path, opcodes in _iterable_opcodes.items():
                transformed = []
                # elements = _path_to_elements(path)
                elem_and_details = self._get_elements_and_details(path)
                if elem_and_details:
                    elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details
                    if parent is None:
                        parent = self
                        obj = self.root
                        parent_to_obj_elem = 'root'
                        parent_to_obj_action = GETATTR
                else:
                    continue  # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198
                # import pytest; pytest.set_trace()
                obj = self.get_nested_obj(obj=self, elements=elements)
                is_obj_tuple = isinstance(obj, tuple)
                for opcode in opcodes:    
                    if opcode.tag == 'replace':
                        # Replace items in list a[i1:i2] with b[j1:j2]
                        transformed.extend(opcode.new_values)
                    elif opcode.tag == 'delete':
                        # Delete items from list a[i1:i2], so we do nothing here
                        continue
                    elif opcode.tag == 'insert':
                        # Insert items from list b[j1:j2] into the new list
                        transformed.extend(opcode.new_values)
                    elif opcode.tag == 'equal':
                        # Items are the same in both lists, so we add them to the result
                        transformed.extend(obj[opcode.t1_from_index:opcode.t1_to_index])  # type: ignore
                if is_obj_tuple:
                    obj = tuple(obj)  # type: ignore
                    # Making sure that the object is re-instated inside the parent especially if it was immutable
                    # and we had to turn it into a mutable one. In such cases the object has a new id.
                    self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
                                                value=obj, action=parent_to_obj_action)
                else:
                    obj[:] = transformed  # type: ignore



                # obj = self.get_nested_obj(obj=self, elements=elements)
                # for


    def _do_iterable_item_removed(self):
        iterable_item_removed = self.diff.get('iterable_item_removed', {})

        iterable_item_moved = self.diff.get('iterable_item_moved')
        if iterable_item_moved:
            # These will get added back during items_added
            removed_dict = {k: v["value"] for k, v in iterable_item_moved.items()}
            iterable_item_removed.update(removed_dict)

        if iterable_item_removed:
            self._do_item_removed(iterable_item_removed)

    def _do_dictionary_item_removed(self):
        dictionary_item_removed = self.diff.get('dictionary_item_removed')
        if dictionary_item_removed:
            self._do_item_removed(dictionary_item_removed)

    def _do_attribute_removed(self):
        attribute_removed = self.diff.get('attribute_removed')
        if attribute_removed:
            self._do_item_removed(attribute_removed)

    def _do_set_item_added(self):
        items = self.diff.get('set_item_added')
        if items:
            self._do_set_or_frozenset_item(items, func='union')

    def _do_set_item_removed(self):
        items = self.diff.get('set_item_removed')
        if items:
            self._do_set_or_frozenset_item(items, func='difference')

    def _do_set_or_frozenset_item(self, items, func):
        for path, value in items.items():
            elements = _path_to_elements(path)
            parent = self.get_nested_obj(obj=self, elements=elements[:-1])
            elem, action = elements[-1]
            obj = self._get_elem_and_compare_to_old_value(
                parent, path_for_err_reporting=path, expected_old_value=None, elem=elem, action=action, forced_old_value=set())
            new_value = getattr(obj, func)(value)
            self._simple_set_elem_value(parent, path_for_err_reporting=path, elem=elem, value=new_value, action=action)

    def _do_ignore_order_get_old(self, obj, remove_indexes_per_path, fixed_indexes_values, path_for_err_reporting):
        """
        A generator that gets the old values in an iterable when the order was supposed to be ignored.
        """
        old_obj_index = -1
        max_len = len(obj) - 1
        while old_obj_index < max_len:
            old_obj_index += 1
            current_old_obj = obj[old_obj_index]
            if current_old_obj in fixed_indexes_values:
                continue
            if old_obj_index in remove_indexes_per_path:
                expected_obj_to_delete = remove_indexes_per_path.pop(old_obj_index)
                if current_old_obj == expected_obj_to_delete:
                    continue
                else:
                    self._raise_or_log(FAIL_TO_REMOVE_ITEM_IGNORE_ORDER_MSG.format(
                        old_obj_index, path_for_err_reporting, expected_obj_to_delete, current_old_obj))
            yield current_old_obj

    def _do_ignore_order(self):
        """

            't1': [5, 1, 1, 1, 6],
            't2': [7, 1, 1, 1, 8],

            'iterable_items_added_at_indexes': {
                'root': {
                    0: 7,
                    4: 8
                }
            },
            'iterable_items_removed_at_indexes': {
                'root': {
                    4: 6,
                    0: 5
                }
            }

        """
        fixed_indexes = self.diff.get('iterable_items_added_at_indexes', dict_())
        remove_indexes = self.diff.get('iterable_items_removed_at_indexes', dict_())
        paths = SetOrdered(fixed_indexes.keys()) | SetOrdered(remove_indexes.keys())
        for path in paths:  # type: ignore
            # In the case of ignore_order reports, we are pointing to the container object.
            # Thus we add a [0] to the elements so we can get the required objects and discard what we don't need.
            elem_and_details = self._get_elements_and_details("{}[0]".format(path))
            if elem_and_details:
                _, parent, parent_to_obj_elem, parent_to_obj_action, obj, _, _ = elem_and_details
            else:
                continue  # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198
            # copying both these dictionaries since we don't want to mutate them.
            fixed_indexes_per_path = fixed_indexes.get(path, dict_()).copy()
            remove_indexes_per_path = remove_indexes.get(path, dict_()).copy()
            fixed_indexes_values = AnySet(fixed_indexes_per_path.values())

            new_obj = []
            # Numpy's NdArray does not like the bool function.
            if isinstance(obj, np_ndarray):
                there_are_old_items = obj.size > 0
            else:
                there_are_old_items = bool(obj)
            old_item_gen = self._do_ignore_order_get_old(
                obj, remove_indexes_per_path, fixed_indexes_values, path_for_err_reporting=path)
            while there_are_old_items or fixed_indexes_per_path:
                new_obj_index = len(new_obj)
                if new_obj_index in fixed_indexes_per_path:
                    new_item = fixed_indexes_per_path.pop(new_obj_index)
                    new_obj.append(new_item)
                elif there_are_old_items:
                    try:
                        new_item = next(old_item_gen)
                    except StopIteration:
                        there_are_old_items = False
                    else:
                        new_obj.append(new_item)
                else:
                    # pop a random item from the fixed_indexes_per_path dictionary
                    self._raise_or_log(INDEXES_NOT_FOUND_WHEN_IGNORE_ORDER.format(fixed_indexes_per_path))
                    new_item = fixed_indexes_per_path.pop(next(iter(fixed_indexes_per_path)))
                    new_obj.append(new_item)

            if isinstance(obj, tuple):
                new_obj = tuple(new_obj)
            # Making sure that the object is re-instated inside the parent especially if it was immutable
            # and we had to turn it into a mutable one. In such cases the object has a new id.
            self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
                                        value=new_obj, action=parent_to_obj_action)

    def _get_reverse_diff(self):
        if not self.bidirectional:
            raise ValueError('Please recreate the delta with bidirectional=True')

        SIMPLE_ACTION_TO_REVERSE = {
            'iterable_item_added': 'iterable_item_removed',
            'iterable_items_added_at_indexes': 'iterable_items_removed_at_indexes',
            'attribute_added': 'attribute_removed',
            'set_item_added': 'set_item_removed',
            'dictionary_item_added': 'dictionary_item_removed',
        }
        # Adding the reverse of the dictionary
        for key in list(SIMPLE_ACTION_TO_REVERSE.keys()):
            SIMPLE_ACTION_TO_REVERSE[SIMPLE_ACTION_TO_REVERSE[key]] = key

        r_diff = {}
        for action, info in self.diff.items():
            reverse_action = SIMPLE_ACTION_TO_REVERSE.get(action)
            if reverse_action:
                r_diff[reverse_action] = info
            elif action == 'values_changed':
                r_diff[action] = {}
                for path, path_info in info.items():
                    reverse_path = path_info['new_path'] if path_info.get('new_path') else path
                    r_diff[action][reverse_path] = {
                        'new_value': path_info['old_value'], 'old_value': path_info['new_value']
                    } 
            elif action == 'type_changes':
                r_diff[action] = {}
                for path, path_info in info.items():
                    reverse_path = path_info['new_path'] if path_info.get('new_path') else path
                    r_diff[action][reverse_path] = {
                        'old_type': path_info['new_type'], 'new_type': path_info['old_type'],
                    }
                    if 'new_value' in path_info:
                        r_diff[action][reverse_path]['old_value'] = path_info['new_value']
                    if 'old_value' in path_info:
                        r_diff[action][reverse_path]['new_value'] = path_info['old_value']
            elif action == 'iterable_item_moved':
                r_diff[action] = {}
                for path, path_info in info.items():
                    old_path = path_info['new_path']
                    r_diff[action][old_path] = {
                        'new_path': path, 'value': path_info['value'],
                    }
            elif action == '_iterable_opcodes':
                r_diff[action] = {}
                for path, op_codes in info.items():
                    r_diff[action][path] = []
                    for op_code in op_codes:
                        tag = op_code.tag
                        tag = {'delete': 'insert', 'insert': 'delete'}.get(tag, tag)
                        new_op_code = Opcode(
                            tag=tag,
                            t1_from_index=op_code.t2_from_index,
                            t1_to_index=op_code.t2_to_index,
                            t2_from_index=op_code.t1_from_index,
                            t2_to_index=op_code.t1_to_index,
                            new_values=op_code.old_values,
                            old_values=op_code.new_values,
                        )
                        r_diff[action][path].append(new_op_code)
        return r_diff

    def dump(self, file):
        """
        Dump into file object
        """
        # Small optimization: Our internal pickle serializer can just take a file object
        # and directly write to it. However if a user defined serializer is passed
        # we want to make it compatible with the expectation that self.serializer(self.diff)
        # will give the user the serialization and then it can be written to
        # a file object when using the dump(file) function.
        param_names_of_serializer = set(self.serializer.__code__.co_varnames)
        if 'file_obj' in param_names_of_serializer:
            self.serializer(self.diff, file_obj=file)
        else:
            file.write(self.dumps())

    def dumps(self):
        """
        Return the serialized representation of the object as a bytes object, instead of writing it to a file.
        """
        return self.serializer(self.diff)

    def to_dict(self):
        return dict(self.diff)

    def _flatten_iterable_opcodes(self, _parse_path):
        """
        Converts op_codes to FlatDeltaRows
        """
        result = []
        for path, op_codes in self.diff['_iterable_opcodes'].items():
            for op_code in op_codes:
                result.append(
                    FlatDeltaRow(
                        path=_parse_path(path),
                        action=OPCODE_TAG_TO_FLAT_DATA_ACTION[op_code.tag],
                        value=op_code.new_values,
                        old_value=op_code.old_values,
                        type=type(op_code.new_values),
                        old_type=type(op_code.old_values),
                        new_path=None,
                        t1_from_index=op_code.t1_from_index,
                        t1_to_index=op_code.t1_to_index,
                        t2_from_index=op_code.t2_from_index,
                        t2_to_index=op_code.t2_to_index,

                    )
                )
        return result

    @staticmethod
    def _get_flat_row(action, info, _parse_path, keys_and_funcs, report_type_changes=True):
        for path, details in info.items():
            row = {'path': _parse_path(path), 'action': action}
            for key, new_key, func in keys_and_funcs:
                if key in details:
                    if func:
                        row[new_key] = func(details[key])
                    else:
                        row[new_key] = details[key]
            if report_type_changes:
                if 'value' in row and 'type' not in row:
                    row['type'] = type(row['value'])
                if 'old_value' in row and 'old_type' not in row:
                    row['old_type'] = type(row['old_value'])
            yield FlatDeltaRow(**row)

    @staticmethod
    def _from_flat_rows(flat_rows_list: List[FlatDeltaRow]):
        flat_dict_list = (i._asdict() for i in flat_rows_list)
        return Delta._from_flat_dicts(flat_dict_list)

    @staticmethod
    def _from_flat_dicts(flat_dict_list):
        """
        Create the delta's diff object from the flat_dict_list
        """
        result = {}
        FLATTENING_NEW_ACTION_MAP = {
            'unordered_iterable_item_added': 'iterable_items_added_at_indexes',
            'unordered_iterable_item_removed': 'iterable_items_removed_at_indexes',
        }
        for flat_dict in flat_dict_list:
            index = None
            action = flat_dict.get("action")
            path = flat_dict.get("path")
            value = flat_dict.get('value')
            new_path = flat_dict.get('new_path')
            old_value = flat_dict.get('old_value', UnkownValueCode)
            if not action:
                raise ValueError("Flat dict need to include the 'action'.")
            if path is None:
                raise ValueError("Flat dict need to include the 'path'.")
            if action in FLATTENING_NEW_ACTION_MAP:
                action = FLATTENING_NEW_ACTION_MAP[action]
                index = path.pop()
            if action in {
                FlatDataAction.attribute_added,
                FlatDataAction.attribute_removed,
            }:
                root_element = ('root', GETATTR)
            else:
                root_element = ('root', GET)
            if isinstance(path, str):
                path_str = path
            else:
                path_str = stringify_path(path, root_element=root_element)  # We need the string path
            if new_path and new_path != path:
                new_path = stringify_path(new_path, root_element=root_element)
            else:
                new_path = None
            if action not in result:
                result[action] = {}
            if action in {
                'iterable_items_added_at_indexes',
                'iterable_items_removed_at_indexes',
            }:
                if path_str not in result[action]:
                    result[action][path_str] = {}
                result[action][path_str][index] = value
            elif action in {
                FlatDataAction.set_item_added,
                FlatDataAction.set_item_removed
            }:
                if path_str not in result[action]:
                    result[action][path_str] = set()
                result[action][path_str].add(value)
            elif action in {
                FlatDataAction.dictionary_item_added,
                FlatDataAction.dictionary_item_removed,
                FlatDataAction.attribute_removed,
                FlatDataAction.attribute_added,
                FlatDataAction.iterable_item_added,
                FlatDataAction.iterable_item_removed,
            }:
                result[action][path_str] = value
            elif action == 'values_changed':
                if old_value == UnkownValueCode:
                    result[action][path_str] = {'new_value': value}
                else:
                    result[action][path_str] = {'new_value': value, 'old_value': old_value}
            elif action == 'type_changes':
                type_ = flat_dict.get('type', UnkownValueCode)
                old_type = flat_dict.get('old_type', UnkownValueCode)

                result[action][path_str] = {'new_value': value}
                for elem, elem_value in [
                    ('new_type', type_),
                    ('old_type', old_type),
                    ('old_value', old_value),
                ]:
                    if elem_value != UnkownValueCode:
                        result[action][path_str][elem] = elem_value
            elif action == FlatDataAction.iterable_item_moved:
                result[action][path_str] = {'value': value}
            elif action in {
                FlatDataAction.iterable_items_inserted,
                FlatDataAction.iterable_items_deleted,
                FlatDataAction.iterable_items_replaced,
                FlatDataAction.iterable_items_equal,
            }:
                if '_iterable_opcodes' not in result:
                    result['_iterable_opcodes'] = {}
                if path_str not in result['_iterable_opcodes']:
                    result['_iterable_opcodes'][path_str] = []
                result['_iterable_opcodes'][path_str].append(
                    Opcode(
                        tag=FLAT_DATA_ACTION_TO_OPCODE_TAG[action],  # type: ignore
                        t1_from_index=flat_dict.get('t1_from_index'),
                        t1_to_index=flat_dict.get('t1_to_index'),
                        t2_from_index=flat_dict.get('t2_from_index'),
                        t2_to_index=flat_dict.get('t2_to_index'),
                        new_values=flat_dict.get('value'),
                        old_values=flat_dict.get('old_value'),
                    )
                )
            if new_path:
                result[action][path_str]['new_path'] = new_path

        return result

    def to_flat_dicts(self, include_action_in_path=False, report_type_changes=True) -> List[FlatDeltaRow]:
        """
        Returns a flat list of actions that is easily machine readable.

        For example:
            {'iterable_item_added': {'root[3]': 5, 'root[2]': 3}}

        Becomes:
            [
                {'path': [3], 'value': 5, 'action': 'iterable_item_added'},
                {'path': [2], 'value': 3, 'action': 'iterable_item_added'},
            ]

        
        **Parameters**

        include_action_in_path : Boolean, default=False
            When False, we translate DeepDiff's paths like root[3].attribute1 into a [3, 'attribute1'].
            When True, we include the action to retrieve the item in the path: [(3, 'GET'), ('attribute1', 'GETATTR')]
            Note that the "action" here is the different than the action reported by to_flat_dicts. The action here is just about the "path" output.

        report_type_changes : Boolean, default=True
            If False, we don't report the type change. Instead we report the value change.

        Example:
            t1 = {"a": None}
            t2 = {"a": 1}

            dump = Delta(DeepDiff(t1, t2)).dumps()
            delta = Delta(dump)
            assert t2 == delta + t1

            flat_result = delta.to_flat_dicts()
            flat_expected = [{'path': ['a'], 'action': 'type_changes', 'value': 1, 'new_type': int, 'old_type': type(None)}]
            assert flat_expected == flat_result

            flat_result2 = delta.to_flat_dicts(report_type_changes=False)
            flat_expected2 = [{'path': ['a'], 'action': 'values_changed', 'value': 1}]

        **List of actions**

        Here are the list of actions that the flat dictionary can return.
            iterable_item_added
            iterable_item_removed
            iterable_item_moved
            values_changed
            type_changes
            set_item_added
            set_item_removed
            dictionary_item_added
            dictionary_item_removed
            attribute_added
            attribute_removed
        """
        return [
            i._asdict() for i in self.to_flat_rows(include_action_in_path=False, report_type_changes=True)
        ]  # type: ignore

    def to_flat_rows(self, include_action_in_path=False, report_type_changes=True) -> List[FlatDeltaRow]:
        """
        Just like to_flat_dicts but returns FlatDeltaRow Named Tuples
        """
        result = []
        if include_action_in_path:
            _parse_path = partial(parse_path, include_actions=True)
        else:
            _parse_path = parse_path
        if report_type_changes:
            keys_and_funcs = [
                ('value', 'value', None),
                ('new_value', 'value', None),
                ('old_value', 'old_value', None),
                ('new_type', 'type', None),
                ('old_type', 'old_type', None),
                ('new_path', 'new_path', _parse_path),
            ]
        else:
            if not self.always_include_values:
                raise ValueError(
                    "When converting to flat dictionaries, if report_type_changes=False and there are type changes, "
                    "you must set the always_include_values=True at the delta object creation. Otherwise there is nothing to include."
                )
            keys_and_funcs = [
                ('value', 'value', None),
                ('new_value', 'value', None),
                ('old_value', 'old_value', None),
                ('new_path', 'new_path', _parse_path),
            ]

        FLATTENING_NEW_ACTION_MAP = {
            'iterable_items_added_at_indexes': 'unordered_iterable_item_added',
            'iterable_items_removed_at_indexes': 'unordered_iterable_item_removed',
        }
        for action, info in self.diff.items():
            if action == '_iterable_opcodes':
                result.extend(self._flatten_iterable_opcodes(_parse_path=_parse_path))
                continue
            if action.startswith('_'):
                continue
            if action in FLATTENING_NEW_ACTION_MAP:
                new_action = FLATTENING_NEW_ACTION_MAP[action]
                for path, index_to_value in info.items():
                    path = _parse_path(path)
                    for index, value in index_to_value.items():
                        path2 = path.copy()
                        if include_action_in_path:
                            path2.append((index, 'GET'))  # type: ignore
                        else:
                            path2.append(index)
                        if report_type_changes:
                            row = FlatDeltaRow(path=path2, value=value, action=new_action, type=type(value))  # type: ignore
                        else:
                            row = FlatDeltaRow(path=path2, value=value, action=new_action)  # type: ignore
                        result.append(row)
            elif action in {'set_item_added', 'set_item_removed'}:
                for path, values in info.items():
                    path = _parse_path(path)
                    for value in values:
                        if report_type_changes:
                            row = FlatDeltaRow(path=path, value=value, action=action, type=type(value))
                        else:
                            row = FlatDeltaRow(path=path, value=value, action=action)
                        result.append(row)
            elif action == 'dictionary_item_added':
                for path, value in info.items():
                    path = _parse_path(path)
                    if isinstance(value, dict) and len(value) == 1:
                        new_key = next(iter(value))
                        path.append(new_key)
                        value = value[new_key]
                    elif isinstance(value, (list, tuple)) and len(value) == 1:
                        value = value[0]
                        path.append(0)  # type: ignore
                        action = 'iterable_item_added'
                    elif isinstance(value, set) and len(value) == 1:
                        value = value.pop()
                        action = 'set_item_added'
                    if report_type_changes:
                        row = FlatDeltaRow(path=path, value=value, action=action, type=type(value))  # type: ignore
                    else:
                        row = FlatDeltaRow(path=path, value=value, action=action)  # type: ignore
                    result.append(row)
            elif action in {
                'dictionary_item_removed', 'iterable_item_added',
                'iterable_item_removed', 'attribute_removed', 'attribute_added'
            }:
                for path, value in info.items():
                    path = _parse_path(path)
                    if report_type_changes:
                        row = FlatDeltaRow(path=path, value=value, action=action, type=type(value))
                    else:
                        row = FlatDeltaRow(path=path, value=value, action=action)
                    result.append(row)
            elif action == 'type_changes':
                if not report_type_changes:
                    action = 'values_changed'

                for row in self._get_flat_row(
                    action=action,
                    info=info,
                    _parse_path=_parse_path,
                    keys_and_funcs=keys_and_funcs,
                    report_type_changes=report_type_changes,
                ):
                    result.append(row)
            else:
                for row in self._get_flat_row(
                    action=action,
                    info=info,
                    _parse_path=_parse_path,
                    keys_and_funcs=keys_and_funcs,
                    report_type_changes=report_type_changes,
                ):
                    result.append(row)
        return result


if __name__ == "__main__":  # pragma: no cover
    import doctest
    doctest.testmod()