matcher.py

# Copyright 2021 Joschua Gandert
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.

from difflib import SequenceMatcher
from operator import itemgetter
from typing import Generic, TypeVar, Mapping, Callable, Any, Union, List, \
    Sequence, Optional, Tuple


T = TypeVar('T')
Real = Union[int, float]
AttributeGetter = Callable[[T], Any]
AttributeGetterToWeight = Mapping[AttributeGetter, Real]

SUPPORTED_NUMBER_TYPES = (int, float)


class _MatchData:
    def __init__(self, a_idx, a_value, b_size):
        self.a_idx = a_idx
        self.a_value = a_value
        self.b_idx_to_similarity = [0 for _ in range(b_size)]

        self.best_b_idx = -1
        self.best_b_similarity = float('-inf')
        self.second_best_b_idx = -1
        self.second_best_b_similarity = float('-inf')
        self.found_best = False

        self.continue_attr_index = 0

        # rarely needed since in a lot of cases there won't be conflicts
        self._sorted_b_similarity_with_idx_pairs = None
        self.is_fully_measured = False

    def add_similarity(self, b_idx, similarity_part):
        self.b_idx_to_similarity[b_idx] += similarity_part
        b_idx_total_similarity = self.b_idx_to_similarity[b_idx]

        if b_idx_total_similarity > self.best_b_similarity:
            self.second_best_b_idx = self.best_b_idx
            self.second_best_b_similarity = self.best_b_similarity

            self.best_b_idx = b_idx
            self.best_b_similarity = b_idx_total_similarity
        elif b_idx_total_similarity > self.second_best_b_similarity:
            self.second_best_b_idx = b_idx
            self.second_best_b_similarity = b_idx_total_similarity

    def set_to_fully_measured(self):
        self.is_fully_measured = True

        # We sort so last = best as calling pop() on a list is O(1)
        # We negate the index here, so that the sorting is correct if the
        # similarities of two b's are the same (lowest index last)
        self._sorted_b_similarity_with_idx_pairs = sorted(
                ((s, -i) for i, s in enumerate(self.b_idx_to_similarity)))

    def replace_best(self):
        if not self._sorted_b_similarity_with_idx_pairs:
            return

        self._sorted_b_similarity_with_idx_pairs.pop()
        if not self._sorted_b_similarity_with_idx_pairs:
            # None left, which means all others a's have better values
            # than this one. As a result, this a will stay alone :(
            self.best_b_idx = -1
            self.best_b_similarity = float('-inf')
            return

        sim, idx = self._sorted_b_similarity_with_idx_pairs[-1]
        self.best_b_idx = -idx
        self.best_b_similarity = sim


class ObjectListMatcher(Generic[T]):
    """
    Utility class that compares the objects of two lists (all the same type), and finds
    the best matches. What is considered a good match depends on the supplied mapping,
    which maps "attribute getter" functions to their weight.

    Weights are normalized, so you can use values > 1.

    A specific getter has to return objects of only one of these types:
    - Real number (e.g. int or float) - compared by 1 - abs(delta) / max_delta
    - Other non-Sequence - converted to str which is a Sequence and then..
    - Sequence - compared using difflib.SequenceMatcher

    The problem this solves is called the "Assignment problem".

    Note that this class will not necessarily call every attribute function and thus
    not use its weight in the calculation, as it will avoid doing unnecessary
    calculations when the current best match for an element is undefeatable.

    For example, let's say we have an attribute x with weight 0.8 and an attribute y
    with weight 0.2. If, after checking the similarity on attribute x with all items,
    the highest similarity is 0.8 (it was equal to the compared item) and the second
    highest has a similarity of 0.5, then we won't check attribute y at all, as the
    second best cannot possibly win with the remaining weight (0.5 + 0.2 < 0.8)."""

    _matcher: SequenceMatcher
    _attr_with_weight: List[Tuple[Callable[[T], Any], float]]
    similarity_matrix: List[List[float]]
    _b_idx_to_a_match_data: List[Optional[_MatchData]]

    def __init__(self, attr_to_weight: AttributeGetterToWeight):
        self.update_attr_to_weights(attr_to_weight)
        self._matcher = SequenceMatcher()
        self.should_store_similarity_matrix = False
        """Whether or not to store the similarity_matrix."""

        self.similarity_matrix = []
        """This will only be populated if should_store_similarity_matrix is True, and
        then every time get_indices is called.
        It's a matrix that has the same number of elements as a_items. Each of those
        lists has the same number of elements as b_items. Those elements represent how
        similar an a item is to each b item. So matrix[a_idx][b_idx] gives you the
        similarity between the a item and b item at those indices."""

        self.minimum_similarity = 0
        """A value in the range [0, 1] that defines what similarity is required to be a
        valid match. Setting this to a reasonable value will speed up the calculation
        as low-quality matches will be discarded before a match conflict can even arise. 
        This has to be carefully chosen with regard to the weights, as this class will
        not necessarily incorporate all (but the largest) weights in the score.
        So, if you set this to 0.8, but your largest weight is 0.5, it's possible that
        a match will almost never be discarded. This can be fine, of course."""

        # This has not been tested heavily, so there might be some bugs!
        self.lock_in_if_similarity_first_above = 1  # FIXME: test this more
        """If the similarity is above this value, we lock it in as the best match.
        Improves performance by removing alternatives from all following elements."""

    @classmethod
    def for_sequence(cls, weights: Sequence[Real]):
        """Creates a matcher where itemgetter(n) is mapped to weight_list[n]"""
        attr_to_weight = {itemgetter(n): w for n, w in enumerate(weights)}
        return ObjectListMatcher(attr_to_weight)

    @classmethod
    def for_one_attr(cls, attr: AttributeGetter):
        """Creates a matcher where a single attribute of the objects are compared."""
        return ObjectListMatcher({attr: 1})

    @classmethod
    def of_identity(cls):
        """Creates a matcher where only the objects themselves are compared."""
        return cls.for_one_attr((lambda i: i))

    def update_attr_to_weights(self, attr_to_weight: AttributeGetterToWeight):
        if not attr_to_weight:
            raise ValueError("there must be at least one weight")

        # normalize
        weight_sum = sum(attr_to_weight.values())

        self._attr_with_weight = []
        for attr, weight in attr_to_weight.items():
            if weight <= 0:
                raise ValueError("weights <= 0 are not allowed")

            self._attr_with_weight.append((attr, weight / weight_sum))

        # from largest to smallest weight
        self._attr_with_weight.sort(key=lambda i: i[1], reverse=True)

        weight_left = 1.0
        self._attr_to_weight_left = []
        for _, weight in self._attr_with_weight:
            self._attr_to_weight_left.append(weight_left)
            weight_left -= weight

    def get_indices(self, a_items: List[T], b_items: List[T]) -> List[int]:
        """
        Returns the indices of b ordered so that they match elements in a.

        Size of a_items and b_items can differ. If there are more a_items than
        there are b_items, -1 is used if no match could be assigned to an a.
        As a result, the returned list always has the size of a_items.

        In terms of performance, it's preferable to supply the smaller list as a_items.
        """
        if not b_items:
            return [-1 for _ in a_items]

        self._b_items = list(enumerate(b_items))
        b_size = len(b_items)

        self._b_idx_to_a_match_data = [None for _ in range(b_size)]
        self._a_idx_to_match_data = []

        for a_idx, a in enumerate(a_items):
            match_data = _MatchData(a_idx, a, b_size)
            self._a_idx_to_match_data.append(match_data)

            self._measure_similarity_to_find_best_b_match(match_data)

            if match_data.best_b_similarity < self.minimum_similarity:
                match_data.best_b_idx = -1
            elif match_data.found_best:
                self._b_idx_to_a_match_data[match_data.best_b_idx] = match_data
            else:
                self._handle_conflicts_if_any(match_data)
            pass

        result = [md.best_b_idx for md in self._a_idx_to_match_data]

        if self.should_store_similarity_matrix:
            matrix = [md.b_idx_to_similarity for md in
                      self._a_idx_to_match_data]
            self.similarity_matrix = matrix

        # cleanup
        del self._b_items
        del self._b_idx_to_a_match_data
        del self._a_idx_to_match_data

        return result

    def _handle_conflicts_if_any(self, a1_match_data):
        while True:
            best_b_idx = a1_match_data.best_b_idx
            if best_b_idx == -1:
                # a1 has no matches left or could not find a match
                return

            a2_match_data = self._b_idx_to_a_match_data[best_b_idx]
            if a2_match_data is None:
                self._b_idx_to_a_match_data[best_b_idx] = a1_match_data
                return

            # We have found a conflict and will now solve it.
            a1_match_data = self._get_worse_match_data(a1_match_data,
                                                       a2_match_data)

    def _get_worse_match_data(self, a1_match_data, a2_match_data):
        # b is matched to a previous a (a2), so we have to find a better match
        self._finish_similarity_measures(a1_match_data)
        
        if a2_match_data.found_best:
            a1_match_data.replace_best()
            # still need to find a better match for a1

            return a1_match_data
        
        self._finish_similarity_measures(a2_match_data)

        # As the index of a1 will almost always be larger than a2, we use <=
        # here, since in case they're equal in terms of similarity, we want to
        # give some weight to the current order of b (index 0 preferred to 1).
        if a1_match_data.best_b_similarity <= a2_match_data.best_b_similarity:
            a1_match_data.replace_best()

            # still need to find a better match for a1
            return a1_match_data
        else:
            # a1 is better so replace a2 in map
            best_b_idx = a1_match_data.best_b_idx
            self._b_idx_to_a_match_data[best_b_idx] = a1_match_data

            a2_match_data.replace_best()

            # we need to find a new match for a2
            return a2_match_data

    def _finish_similarity_measures(self, a_match_data):
        """Only if this was called, is match data full measured and only then
        can match_data.replace_best() be called."""
        if a_match_data.is_fully_measured:
            return

        # figure out total similarity (without stopping) if we didn't
        self._measure_similarity_to_find_best_b_match(a_match_data)
        a_match_data.set_to_fully_measured()

    def _measure_similarity_to_find_best_b_match(self, a_match_data):
        continue_attr_idx = a_match_data.continue_attr_index
        can_stop = continue_attr_idx == 0
        attr_size = len(self._attr_with_weight)

        for attr_idx in range(continue_attr_idx, attr_size):
            if a_match_data.found_best:
                a_match_data.continue_attr_index = attr_size
                return

            # stop if one has more similarity than is possible for the rest
            if can_stop and self._is_max_similarity_undefeatable(attr_idx,
                                                                 a_match_data):
                a_match_data.continue_attr_index = attr_idx
                return

            self._measure_similarity_for_attr(attr_idx, a_match_data)

        a_match_data.continue_attr_index = attr_size

    def _is_max_similarity_undefeatable(self, attr_idx, a_match_data):
        # Must only be called at the start of the loop.
        # We need a second best, so we can check if they have any chance.
        if a_match_data.second_best_b_idx < 0:
            return False

        # Any similarity has to be in [0, 1], so the following is the maximum
        # similarity that could be achieved with the remaining attr / weights
        optimal_second_best_similarity = self._attr_to_weight_left[attr_idx]

        # Now we add the second best similarity, since we want to know if it's
        # even possible for the second best to win against the current best
        optimal_second_best_similarity += a_match_data.second_best_b_similarity

        return optimal_second_best_similarity < a_match_data.best_b_similarity

    def _measure_similarity_for_attr(self, attr_idx, a_match_data):
        get_attr, weight = self._attr_with_weight[attr_idx]
        a_attr = get_attr(a_match_data.a_value)

        # isinstance does not work with Union, so we need to use a tuple here
        if isinstance(a_attr, SUPPORTED_NUMBER_TYPES):
            self._add_number_similarity(a_match_data, a_attr, get_attr, weight)
            return

        if not isinstance(a_attr, Sequence):
            a_attr = str(a_attr)

            def get_attr(obj, original_get_attr=get_attr):
                return str(original_get_attr(obj))

            # replace it so we don't have to do this again for this attr_idx
            self._attr_with_weight[attr_idx] = (get_attr, weight)

        self._add_sequence_similarity(a_match_data, a_attr, get_attr, weight)

    def _add_number_similarity(self, a_match_data, a_attr, get_attr, weight):
        deltas = [abs(get_attr(b) - a_attr) for _, b in self._b_items]
        max_delta = max(deltas)
        if max_delta == 0:
            if a_match_data.best_b_idx == -1:
                # We try to set a match here manually to avoid the situation where
                # nothing is matched to a and -1 returned just because all are the same.
                # In all other cases MatchData's add_similarity takes care of this.
                for b_idx, match_data in enumerate(
                        self._b_idx_to_a_match_data):
                    # find the first b index that's not matched to an a element
                    if match_data is None:
                        a_match_data.best_b_idx = b_idx
                        a_match_data.best_b_similarity = 0
                        break

            # no delta, so all the same and no change in similarity
            return  # also let's avoid division by zero

        for b_idx, delta in enumerate(deltas):
            # if delta is small relative to max, similarity is higher
            # for example, if delta is 0, b_attr is the same as a_attr
            similarity = weight * (1 - delta / max_delta)
            a_match_data.add_similarity(b_idx, similarity)

    def _add_sequence_similarity(self, a_match_data, a_attr, get_attr, weight):
        matcher = self._matcher
        if a_match_data.found_best:
            return

        # set_seq2 is used here as it caches information (contrary to 1)
        matcher.set_seq2(a_attr)

        lock_in_min = self.lock_in_if_similarity_first_above

        b_items = self._b_items

        for i in range(len(b_items)):
            b_idx, b = b_items[i]
            b_attr = get_attr(b)

            matcher.set_seq1(b_attr)

            a_match_data.add_similarity(b_idx, weight * matcher.ratio())

            if (a_match_data.best_b_idx == b_idx and  #
                    a_match_data.best_b_similarity > lock_in_min and
                    a_attr and b_attr):
                del b_items[i]
                a_match_data.found_best = True
                return