Geospatial queries: Using Python to search cities

Daniel Ancuta é um engenheiro de software com vários anos de experiência no uso de diferentes tecnologias. Ele é um grande fã de "The Zen of Python", que tenta aplicar não apenas em seu código, mas também em sua vida pessoal. Você pode encontrá-lo no Twitter: @daniel_ancuta

Consultas geoespaciais: Usando Python para pesquisar cidades

As informações de geolocalização são usadas todos os dias em quase todos os aspectos de nossa interação com os computadores. Seja em um site que deseja nos enviar notificações personalizadas com base na localização, em mapas que nos mostram a rota mais curta possível ou apenas em tarefas executadas em segundo plano que verificam os lugares que visitamos.

Hoje, gostaria de apresentar a você consultas geoespaciais que são usados no Couchbase. Consultas geoespaciais permitem que você pesquise documentos com base em sua localização geográfica.

Juntos, escreveremos uma ferramenta em Python que usa consultas geoespaciais com API REST do Couchbase e Pesquisa de texto completo do Couchbaseque nos ajudará a pesquisar um banco de dados de cidades.

Pré-requisitos

Dependências

Neste artigo, usei Couchbase Enterprise Edition 5.1.0 build 5552 e Python 3.6.4.

Para executar os snippets deste artigo, você deve instalar o Couchbase 2.3 (estou usando o 2.3.4) via pip.

Couchbase

Criar um bucket de cidades
Crie uma pesquisa de cidades com o campo geográfico do tipo geopoint. Você pode ler sobre isso na seção Inserção de um campo filho parte da documentação.

Ele deve se parecer com a imagem abaixo:

Preenchimento do Couchbase com dados

Antes de mais nada, precisamos ter dados para nosso exercício. Para isso, usaremos um banco de dados de cidades de geonames.org.

O GeoNames contém dois bancos de dados principais: lista de cidades e lista de códigos postais.

Todos estão agrupados por país com as informações correspondentes, como nome, coordenadas, população, fuso horário, código do país e assim por diante. Ambos estão no formato CSV.

Para fins deste exercício, usaremos a lista de cidades. Eu usei PL.zip mas sinta-se à vontade para escolher o que preferir da lista lista de cidades.

Modelo de dados

A classe City será nossa representação de uma única cidade que usaremos em todo o aplicativo. Ao encapsulá-la em um modelo, unificamos a API e não precisamos depender de fontes de dados de terceiros (por exemplo, arquivo CSV) que podem mudar.

A maioria dos nossos snippets está localizada (até que seja dito o contrário) no arquivo core.py. Portanto, lembre-se apenas de atualizá-lo (especialmente ao adicionar novas importações) e não substituir todo o conteúdo.

# core.py
class City:
   def __init__(self, geonameid, feature_class, name, population, lat, lon):
       self.geonameid = geonameid
       self.feature_class = feature_class
       self.name = name
       self.population = population
       self.lat = lat
       self.lon = lon

   @classmethod
   def from_csv_row(cls, row):
       return cls(row[0], row[7], row[1], row[12], row[4], row[5])

# core.py

classe Cidade:

def __init__(autônomo, geonameídeo, feature_class, nome, população, lat, solitário):

autônomo.geonameídeo = geonameídeo

autônomo.feature_class = feature_class

autônomo.nome = nome

autônomo.população = população

autônomo.lat = lat

autônomo.solitário = solitário

@método de classe

def from_csv_row(cls, fila):

retorno cls(fila[0], fila[7], fila[1], fila[12], fila[4], fila[5])

Iterador de CSV para processar cidades

Como temos uma classe de modelo, é hora de preparar um iterador que nos ajudará a ler as cidades do arquivo CSV.

# core.py
import csv
from collections import Iterator

class CitiesCsvIterator(Iterator):
   def __init__(self, path):
       self._path = path
       self._fp = None
       self._csv_reader = None

   def __enter__(self):
       self._fp = open(self._path, 'r')
       self._csv_reader = csv.reader(self._fp, delimiter='\t')

       return self

   def __exit__(self, exc_type, exc_val, exc_tb):
       self._fp.close()

   def __next__(self):
       return City.from_csv_row(next(self._csv_reader))

# core.py

importação csv

de coleções importação Iterador

classe CidadesCsvIterador(Iterador):

def __init__(autônomo, caminho):

autônomo._caminho = caminho

autônomo._fp = Nenhum

autônomo._csv_reader = Nenhum

def __enter__(autônomo):

autônomo._fp = aberto(autônomo._caminho, 'r')

autônomo._csv_reader = csv.leitor(autônomo._fp, delimitador='\t')

retorno autônomo

def __exit__(autônomo, exc_type, exc_val, exc_tb):

autônomo._fp.próximo()

def __próximo__(autônomo):

retorno Cidade.from_csv_row(próxima(autônomo._csv_reader))

Inserir cidades no bucket do Couchbase

Unificamos a maneira de representar uma cidade e temos um iterador que lê essas informações do arquivo csv.

É hora de colocar esses dados em nossa fonte de dados principal, Couchbase.

# core.py
import logging
import sys
from couchbase.cluster import Cluster, PasswordAuthenticator


logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler(sys.stdout))


def get_bucket(username, password, connection_string='couchbase://localhost'):
   cluster = Cluster(connection_string)
   authenticator = PasswordAuthenticator(username, password)
   cluster.authenticate(authenticator)

   return cluster.open_bucket('cities')


class CitiesService:
   def __init__(self, bucket):
       self._bucket = bucket

   def load_from_csv(self, path):
       with CitiesCsvIterator(path) as cities_iterator:
           for city in cities_iterator:
               if city.feature_class not in ('PPL', 'PPLA', 'PPLA2', 'PPLA3',
                                             'PPLA4', 'PPLC'):
                   continue

               logger.info(f'Inserting {city.geonameid}')
               self._bucket.upsert(
                   city.geonameid,
                   {
                       'name': city.name,
                       'feature_class': city.feature_class,
                       'population': city.population,
                       'geo': {'lat': float(city.lat),
                               'lon': float(city.lon)}
                   }
               )

# core.py

importação registro

importação sistema

de couchbase.agrupamento importação Aglomerado, PasswordAuthenticator

registrador = registro.getLogger()

registrador.setLevel(registro.DEBUG)

registrador.addHandler(registro.Gerenciador de fluxo(sistema.saída))

def get_bucket(nome de usuário, senha, connection_string='couchbase://localhost'):

agrupamento = Aglomerado(connection_string)

autenticador = PasswordAuthenticator(nome de usuário, senha)

agrupamento.autenticar(autenticador)

retorno agrupamento.open_bucket('cidades')

classe CidadesServiço:

def __init__(autônomo, balde):

autônomo._bucket = balde

def load_from_csv(autônomo, caminho):

com CidadesCsvIterador(caminho) como iterador de cidades:

para cidade em iterador de cidades:

se cidade.feature_class não em ("PPL, 'PPLA', "PPLA2, "PPLA3,

'PPLA4', "PPLC):

continuar

registrador.informações(f'Inserindo {city.geonameid}')

autônomo._bucket.upsert(

cidade.geonameídeo,

{

"nome: cidade.nome,

'feature_class': cidade.feature_class,

"população: cidade.população,

'geo': {'lat': flutuante(cidade.lat),

'lon': flutuante(cidade.solitário)}

}

)

Para verificar se tudo o que escrevemos até agora está funcionando, vamos carregar o conteúdo CSV no Couchbase.

# core.py

bucket = get_bucket('admin', 'test123456')
cities_service = CitiesService(bucket)
cities_service.load_from_csv('~/directory-with-cities/PL/PL.txt', bucket)

# core.py

balde = get_bucket('admin', 'test123456')

cidades_serviço = CidadesServiço(balde)

cidades_serviço.load_from_csv('~/diretório-com-cidades/PL/PL.txt', balde)

Nesse ponto, você deverá ter cidades carregadas no seu bucket do Couchbase. O tempo que isso leva depende do país que você escolheu.

Pesquisar cidades

Temos nosso bucket pronto com dados, então é hora de voltar ao CitiesService e preparar alguns métodos que nos ajudariam a pesquisar cidades.

Mas antes de começarmos, precisamos modificar um pouco a classe City, adicionando o seguinte método:

# core.py

@classmethod
def from_couchbase_dict(cls, row):
fields = row['fields']

return cls(row['id'],
fields['feature_class'],
fields['name'],
fields['population'],
fields['geo'][1],
fields['geo'][0])

# core.py

@método de classe

def from_couchbase_dict(cls, fila):

campos = fila['campos']

retorno cls(fila['id'],

campos['feature_class'],

campos["nome],

campos["população],

campos['geo'][1],

campos['geo'][0])

Essa é uma lista de métodos que implementaremos no CitiesService:

get_by_name(name, limit=10), retorna cidades por seus nomes
get_by_coordinates(lat, lon), retorna a cidade por coordenadas
get_nearest_to_city(city, distance='10', unit='km', limit=10), retorna a cidade mais próxima

get_by_name

# core.py
from couchbase.fulltext import TermQuery

INDEX_NAME = 'cities'

def get_by_name(self, name, limit=10):
result = self._bucket.search(self.INDEX_NAME,
TermQuery(name.lower(), field='name'),
limit=limit,
fields='*')

for c_city in result:
yield City.from_couchbase_dict(c_city)

# core.py

de couchbase.texto completo importação TermQuery

ÍNDICE_NOME = 'cidades'

def get_by_name(autônomo, nome, limite=10):

resultado = autônomo._bucket.pesquisa(autônomo.INDEX_NAME,

TermQuery(nome.inferior(), campo="nome),

limite=limite,

campos='*')

para c_cidade em resultado:

rendimento Cidade.from_couchbase_dict(c_cidade)

get_by_coordinates

# core.py
from couchbase.fulltext import GeoDistanceQuery

INDEX_NAME = 'cities'

def get_by_coordinates(self, lat, lon):
result = self._bucket.search(self.INDEX_NAME,
GeoDistanceQuery('1km', (lon, lat)),
fields='*')

for c_city in result:
yield City.from_couchbase_dict(c_city)

# core.py

de couchbase.texto completo importação Consulta de distância geográfica

ÍNDICE_NOME = 'cidades'

def get_by_coordinates(autônomo, lat, solitário):

resultado = autônomo._bucket.pesquisa(autônomo.INDEX_NAME,

Consulta de distância geográfica('1km', (solitário, lat)),

campos='*')

para c_cidade em resultado:

rendimento Cidade.from_couchbase_dict(c_cidade)

get_nearest_to_city

# core.py
from couchbase.fulltext import RawQuery, SortRaw

INDEX_NAME = 'cities'

def get_nearest_to_city(self, city, distance='10', unit='km', limit=10):
query = RawQuery({
'location': {
'lon': city.lon,
'lat': city.lat
},
'distance': str(distance) + unit,
'field': 'geo'

})
sort = SortRaw([{
'by': 'geo_distance',
'field': 'geo',
'unit': unit,
'location': {
'lon': city.lon,
'lat': city.lat
}
}])

result = self._bucket.search(self.INDEX_NAME,
query,
sort=sort,
fields='*',
limit=limit)

for c_city in result:
yield City.from_couchbase_dict(c_city)

# core.py

de couchbase.texto completo importação RawQuery, SortRaw

ÍNDICE_NOME = 'cidades'

def get_nearest_to_city(autônomo, cidade, distância='10', unidade='km', limite=10):

consulta = RawQuery({

'localização': {

'lon': cidade.solitário,

'lat': cidade.lat

"distância: str(distância) + unidade,

'campo': 'geo'

})

classificar = SortRaw([{

"por: 'geo_distance' (distância geográfica),

'campo': 'geo',

"unidade: unidade,

'localização': {

'lon': cidade.solitário,

'lat': cidade.lat

}

}])

resultado = autônomo._bucket.pesquisa(autônomo.INDEX_NAME,

consulta,

classificar=classificar,

campos='*',

limite=limite)

para c_cidade em resultado:

rendimento Cidade.from_couchbase_dict(c_cidade)

Como você pode notar neste exemplo, usamos as classes RawQuery e SortRaw. Infelizmente, a API couchbase-python-client não funciona corretamente com o Couchbase e as pesquisas geográficas mais recentes.

Métodos de chamada

Como agora temos todos os métodos prontos, podemos chamá-lo!

# core.py

bucket = get_bucket('admin', 'test123456')

cities_service = CitiesService(bucket)
# cities_service.load_from_csv('/my-path/PL/PL.txt')

print('get_by_name')
cities = cities_service.get_by_name('Poznań')
for city in cities:
print(city.__dict__)

print('get_by_coordinates')
cities = cities_service.get_by_coordinates(52.40691997632544,
16.929929926276657)
for city in cities:
print(city.__dict__)

print('get_nearest_to_city')
cities = cities_service.get_nearest_to_city(city)
for city in cities:
print(city.__dict__)

# core.py

balde = get_bucket('admin', 'test123456')

cidades_serviço = CidadesServiço(balde)

# cities_service.load_from_csv('/my-path/PL/PL.txt')

impressão('get_by_name')

cidades = cidades_serviço.get_by_name("Poznań)

para cidade em cidades:

impressão(cidade.__dict__)

impressão('get_by_coordinates')

cidades = cidades_serviço.get_by_coordinates(52.40691997632544,

16.929929926276657)

para cidade em cidades:

impressão(cidade.__dict__)

impressão('get_nearest_to_city')

cidades = cidades_serviço.get_nearest_to_city(cidade)

para cidade em cidades:

impressão(cidade.__dict__)

O que fazer daqui para frente?

Acredito que essa introdução permitirá que você trabalhe em algo mais avançado.

Há algumas coisas que você pode fazer:

Talvez usar uma ferramenta CLI ou uma API REST para fornecer esses dados... Melhorar a forma como carregamos os dados, pois pode não ser muito eficiente se quisermos carregar TODAS as cidades de TODOS os países.

Você pode encontrar o código completo do core.py em github gist.

Se você tiver alguma dúvida, não hesite em me enviar um tweet @daniel_ancuta.

Esta postagem faz parte do Programa de redação comunitária

Laura Czajkowski, gerente da comunidade de desenvolvedores, Couchbase

Compartilhe este artigo

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

Consultas geoespaciais: Usando Python para pesquisar cidades

Consultas geoespaciais: Usando Python para pesquisar cidades

Pré-requisitos

Dependências

Couchbase

Preenchimento do Couchbase com dados

Modelo de dados

Iterador de CSV para processar cidades

Inserir cidades no bucket do Couchbase

Pesquisar cidades

get_by_name

get_by_coordinates

get_nearest_to_city

Métodos de chamada

O que fazer daqui para frente?

Receba atualizações do blog do Couchbase em sua caixa de entrada

Autor

Postado por Laura Czajkowski, gerente da comunidade de desenvolvedores, Couchbase

Deixe um comentário Cancelar resposta

Pronto para começar a usar o Couchbase Capella?

Iniciar a construção

Use o Capella gratuitamente

Entre em contato