Caso de uso: Pontos mais próximos, dividir para conquistar

Introdução

Encontrar a melhor solução para um problema muitas vezes exige, além do resultado esperado, que o processo tenha a melhor performance possível.

Neste artigo, irei descrever a solução para um problema clássico, encontrar em um plano com vários pontos, aqueles que estão mais próximos entre si.

Suponha que uma região tenha 50 usuários de uma empresa de transportes e ela precisa descobrir qual par de usuários estão mais próximos entre si.

Uma forma comum de se pensar é calcular todas as combinações de usuários, dois a dois, e escolher a menor distância entre eles.

Porém, esta solução carrega um custo de n², ou seja, o custo de solucionar cresce exponencialmente com o aumento de n.

Se formos pensar em 1 milhão de usuários, estamos falando de fazer 1 TRILHÃO de processamentos. E olha que 1 milhão nem é tanto assim hoje em dia.

Uma outra maneira, mais inteligente, é utilizar um algoritmo de DIVISÃO E CONQUISTA.

Este algoritmo tem a finalidade de dividir o problema em pedaços menores (DIVISÃO) e juntar as soluções encontradas em cada pedaço(CONQUISTA).

O “custo” de processamento dele é bem menor em função de n, chegando a ficar na ordem de nlog(n).

Formato dos dados

Antes de tudo, vamos supor que os dados chegam em formato JSON:

{
	'place_a': ['coord_x', 'coord_y'],
	'place_b': ['coord_x', 'coord_y'],
	'place_c': ['coord_x', 'coord_y'],
	'place_d': ['coord_x', 'coord_y'],
	...
}

Como exemplo, vamos usar estes 20 pontos gerados aleatoriamente:

_dict = {
  "onfsc": [78.4, 78.1],
  "gtbgn": [5.5, 8.4],
  "aseoh": [51.0, 52.1],
  "zzhwe": [89.0, 15.9],
  "trvki": [19.9, 35.3],
  "fpnwj": [28.9, 5.3],
  "zznss": [59.9, 14.5],
  "vzema": [70.7, 69.8],
  "rghob": [27.3, 31.1],
  "xhyub": [63.9, 64.1],
  "budfa": [79.7, 59.6],
  "xbkbj": [32.1, 46.6],
  "uoopk": [46.5, 75.1],
  "aaocg": [29.4, 67.6],
  "wzkfl": [90.7, 24.8],
  "vrnef": [57.8, 19.2],
  "cwyix": [11.1, 14.4],
  "tbpit": [28.6, 45.8],
  "pwnhf": [94.5, 82.3],
  "banvu": [25.7, 25.6]
}

Divisão e Conquista

Para resolver, vamos dividir a solução em 2 etapas. A primeira se baseia em ordenar os pontos do plano em relação a um dos eixos. Por convenção, é o plano X, mas poderia ser o eixo Y. O resultado é o mesmo.

PRIMEIRA ETAPA

Começamos criando a classe Places:

from random import uniform, choice
import string

class Places:

    def __init__(self, places=None, n=None, amplitude=100):
        self.places = places
        if not places:
            assert n, "if you don't give the places, give me a number of places to generate random n places"
            letters = string.ascii_lowercase
            self.places = {''.join(choice(letters) for i in range(5)): [uniform(0,amplitude), uniform(0,amplitude)] for j in range(n)}
        else:
            assert type(places) == dict, "places must be a dict, if you refering to n, please name it"

Na classe há duas opções, ou fornecemos o dicionário com os pontos, ou fornecemos um valor n para a classe gerar aleatóriamente os pontos.

Para podermos ordenar os pontos pela coordenada x, vamos primeiro converter o dicionário para uma lista, que armazene tanto as coordenadas quanto o nome de cada ponto. Adicionamos o método dict_to_list na classe:

def dict_to_list(self, _dict):
    return list(zip(_dict.values(), _dict.keys()))

Os valores ficaram então no formato:

# _list
[
  ([78.4, 78.1], "onfsc" ),
  ([5.5, 8.4], "gtbgn" ),
  ([51.0, 52.1], "aseoh" ),
  ([89.0, 15.9], "zzhwe" ),
  ([19.9, 35.3], "trvki" ),
  ([28.9, 5.3], "fpnwj" ),
  ([59.9, 14.5], "zznss" ),
  ([70.7, 69.8], "vzema" ),
  ([27.3, 31.1], "rghob" ),
  ([63.9, 64.1], "xhyub" ),
  ([79.7, 59.6], "budfa" ),
  ([32.1, 46.6], "xbkbj" ),
  ([46.5, 75.1], "uoopk" ),
  ([29.4, 67.6], "aaocg" ),
  ([90.7, 24.8], "wzkfl" ),
  ([57.8, 19.2], "vrnef" ),
  ([11.1, 14.4], "cwyix" ),
  ([28.6, 45.8], "tbpit" ),
  ([94.5, 82.3], "pwnhf" ),
  ([25.7, 25.6], "banvu" )
]

Para a ordenação usamos um método chamado merge sort. Este método também está relacionado com divisão e conquista.

O merge sort consiste em dividir a lista de pontos até a forma mais simples, que no caso é um ponto por lista. Esta divisão é feita de forma recursiva.

Após esta divisão, vamos para a parte da conquista, que consiste em comparar lista a lista, sempre olhando o primeiro elemento de cada lista e tirando o menor deles.

É como se você tivesse duas pilhas de baralho, que estão ordenadas e você precisa juntá-las. A melhor maneira é você olhar no topo de cada pilha e pegar a menor, sucessivamente, até consumir as duas pilhas de cartas.

Para resolver isto em python:

def merge_list(self, list_a, list_b):
    ordered = []
    while list_a or list_b:
        if len(list_a) and len(list_b):
            if list_a[0][0] < list_b[0][0]:
                ordered.append(list_a.pop(0))
            else:
                ordered.append(list_b.pop(0))
        elif len(list_a):
            ordered.extend(list_a)
            list_a = []
        elif len(list_b):
            ordered.extend(list_b)
            list_b = []
    return ordered

def merge_sort(self, _list):
    ordered = []
    if len(_list) > 1:
        mid = len(_list) // 2
        left = self.merge_sort(_list[0:mid])
        right = self.merge_sort(_list[mid:])
        ordered = self.merge_list(left, right)
        return ordered
    else:
        return _list

A função merge_sort vai dividindo a lista recursivamente até todas terem o tamanho de 1. Em seguida, ela vai mesclando (daí o nome “merge”) de duas em duas listas ordenadas, e gerando uma terceira que é composta das duas primeiras, e também está ordenada.

SEGUNDA ETAPA

A segunda etapa consiste em dividir os pontos, já ordenados por um dos eixos, em suas medianas até ter a forma mais simples do problema, que vamos considerar sendo uma faixa com 3 ou 2 pontos.

Quando cada fatia tiver 2 ou 3 pontos, podemos utilizar a força bruta. Obtendo o par de pontos mais próximos desta fatia.

Para calcular a distância entre dois pontos, usamos distância euclidiana. Essa é a forma mais simples de fazer este cálculo e não serve para coordenadas geográficas como latitude e longitude, mas serve para o nosso exemplo.

from numpy import sqrt

def distance(self, point_1, point_2):
    return sqrt(sum(((point_1[0] - point_2[0])**2, (point_1[1] - point_2[1])**2)))

Para saber o ponto mais próximo usando a força bruta, podemos utilizar:

def brute_force_distance(self, _list):
    """
    format list [((x1, y1), name1), ((x2, y2), name2), ...]
    """
    assert len(_list) <= 3, "Please, don't ask me to do what I can't do"
    results = []
    for i, location in enumerate(_list):
        for j, neighbor in enumerate(_list[i+1:]):
            results.append((self.distance(location[0], neighbor[0]),
                            location[1], neighbor[1]))
    return min(results)

Quando tiver a menor distância de cada um dos lados, vamos chamá-la de δ. Usamos esse delta para verificar se há pontos na divisa que possuem uma distância menor que ele. Em outras palavras, usamos o valor δ para traçar a faixa que selecionará os pontos incluídos na faixa da divisa.

Dentro da faixa, será calculada a distância entre todos os pontos da esquerda e até no máximo os seis mais próximos a cada um deles.

Mas por que 6?

Aí é que está a grande sacada, a qual entendi olhando este arquivo. Vou explicar com a ilustração a seguir:

Suponha que P seja um ponto da esquerda, o máximo de pontos da direita que podem ser mais proximos de P são no máximo 6. Senão, a distância entre eles seria menor que δ, o que não pode ser verdade, pois δ já é a menor distância entre todos os pontos de ambos os lados.

Então, para cada ponto da esquerda precisamos calcular apenas os seis pontos mais próximos.

Conseguindo o par de pontos mais próximos de cada faixa e cada divisa, evoluimos até obter o par mais próximos do plano todo.

Essa evolução ocorre dentro de uma recursividade na função em python:

#list format [[(x,y), name]]
def closest_pair(self, _list=None):  # must be a sorted list
    if not _list:
        _list = self.dict_to_list(self.places)
        _list = self.merge_sort((_list))
    if len(_list) > 3:
        middle = len(_list) // 2
        left = self.closest_pair(_list[:middle])
        right = self.closest_pair(_list[middle:])
        closest = left if min(left[0], right[0]) == left[0] else right
        return self.min_from_middle(middle, _list, closest)
    else:
        return self.brute_force_distance(_list)

def min_from_middle(self, middle, _list, closest):
    middle_x = (_list[middle][0][0] + _list[middle - 1][0][0]) / 2
    upper = middle_x + closest[0]
    lower = middle_x - closest[0]
    points = [item for item in _list if (lower <= item[0][0] <= upper)]
    for i, point in enumerate(points):
        for j, neighbor in enumerate(points[min(i + 1, len(points)): min(len(points),i+6)]):
            dist = self.distance(point[0], neighbor[0])
            if dist < closest[0]:
                closest = dist, point[1], neighbor[1]

    return closest

A função closest_pair vai retornar a distância entre os pontos mais próximos e o nome destes pontos.

>> closest_pair(_list)
(3.5902646142032495, 'tbpit', 'xbkbj')

Plotagem

Podemos aprimorar, adicionando um metodo para plotar os pontos em um gráfico.

Para isso vamos utilizar a incrível biblioteca Plotly.

import plotly.graph_objs as go

def plot(self, show_closest=False):
    x, y = zip(*self.places.values())
    fig = go.Figure()
    fig.add_trace(
        go.Scatter(
            x=x,
            y=y,
            mode='markers'
        )
    )

    if show_closest:
        closest = self.closest_pair()[1:]
        x0 = self.places[closest[0]][0]
        y0 = self.places[closest[0]][1]
        x1 = self.places[closest[1]][0]
        y1 = self.places[closest[1]][1]
        fig.add_trace(
            go.Scatter(
                x=[x0, x1],
                y=[y0, y1],
                mode='markers',
                marker=dict(
                    size=16,
                    color="red",
                )
            )
        )
    fig.show()

Caso usemos show_closest=True no método, os pontos mais próximos ficarão em destaque.

Pronto, finalizamos nossa classe. Agora vamos experimentá-la

places = Places(places=_dict)
places.plot()

E agora, com o par de pontos mais próximo em destaque:

places = Places(places=_dict)
places.plot(True)

Nosso script pode receber a variável n na linha de comando adicionando:

if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("n", help="number of places for generate")
    args = parser.parse_args()
    n = int(args.n)
    places = Places(n=n)

Podemos implementar a variável amplitude e/ou receber um caminho para um arquivo com os pontos em json. Mas isso fica para vocês implementarem.

Versão Final

Nosso script ficou então da seguinte forma:

from random import uniform, choice
import string

from numpy import sqrt
import plotly.graph_objs as go

class Places:

    def __init__(self, places=None, n=None, amplitude=100):
        self.places = places
        if not places:
            assert n, "if you don't give the places, give me a number of places to generate random n places"
            letters = string.ascii_lowercase
            self.places = {''.join(choice(letters) for i in range(5)): [uniform(0,amplitude), uniform(0,amplitude)] for j in range(n)}
        else:
            assert type(places) == dict, "places must be a dict, if you refering to n, please name it"


    def merge_list(self, list_a, list_b):
        ordered = []
        while list_a or list_b:
            if len(list_a) and len(list_b):
                if list_a[0][0] < list_b[0][0]:
                    ordered.append(list_a.pop(0))
                else:
                    ordered.append(list_b.pop(0))
            elif len(list_a):
                ordered.extend(list_a)
                list_a = []
            elif len(list_b):
                ordered.extend(list_b)
                list_b = []
        return ordered


    def merge_sort(self, _list):
        ordered = []
        if len(_list) > 1:
            mid = len(_list) // 2
            left = self.merge_sort(_list[0:mid])
            right = self.merge_sort(_list[mid:])
            ordered = self.merge_list(left, right)
            return ordered
        else:
            return _list

    def distance(self, point_1, point_2):
        return sqrt(sum(((point_1[0] - point_2[0])**2, (point_1[1] - point_2[1])**2)))

    def dict_to_list(self, _dict):
        return list(zip(_dict.values(), _dict.keys()))

    def brute_force_distance(self, _list):
        """
        format list [((x1, y1), name1), ((x2, y2), name2), ...]
        """
        assert len(_list) <= 3, "Please, don't ask me to do what I can't do"
        results = []
        for i, location in enumerate(_list):
            for j, neighbor in enumerate(_list[i+1:]):
                results.append((self.distance(location[0], neighbor[0]),
                                location[1], neighbor[1]))
        return min(results)

    #list format [[(x,y), name]]
    def closest_pair(self, _list=None):  # must be a sorted list
        if not _list:
            _list = self.dict_to_list(self.places)
            _list = self.merge_sort((_list))
        if len(_list) > 3:
            middle = len(_list) // 2
            left = self.closest_pair(_list[:middle])
            right = self.closest_pair(_list[middle:])
            closest = left if min(left[0], right[0]) == left[0] else right
            return self.min_from_middle(middle, _list, closest)
        else:
            return self.brute_force_distance(_list)

    def min_from_middle(self, middle, _list, closest):
        middle_x = (_list[middle][0][0] + _list[middle - 1][0][0]) / 2
        upper = middle_x + closest[0]
        lower = middle_x - closest[0]
        points = [item for item in _list if (lower <= item[0][0] <= upper)]
        for i, point in enumerate(points):
            for j, neighbor in enumerate(points[min(i + 1, len(points)): min(len(points),i+6)]):
                dist = self.distance(point[0], neighbor[0])
                if dist < closest[0]:
                    closest = dist, point[1], neighbor[1]

        return closest

    def plot(self, show_closest=False):
        x, y = zip(*self.places.values())
        fig = go.Figure()
        fig.add_trace(
            go.Scatter(
                x=x,
                y=y,
                mode='markers'
            )
        )

        if show_closest:
            closest = self.closest_pair()[1:]
            x0 = self.places[closest[0]][0]
            y0 = self.places[closest[0]][1]
            x1 = self.places[closest[1]][0]
            y1 = self.places[closest[1]][1]
            fig.add_trace(
                go.Scatter(
                    x=[x0, x1],
                    y=[y0, y1],
                    mode='markers',
                    marker=dict(
                        size=16,
                        color="red",
                    )
                )
            )
        fig.show()

if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("n", help="number of places for generate")
    args = parser.parse_args()
    n = int(args.n)
    places = Places(n=n)

Todo o código está no meu perfil do GitHub.