Geospatial queries: Using Python to search cities

Daniel Ancuta es un ingeniero de software con varios años de experiencia utilizando diferentes tecnologías. Es un gran fan de "El Zen de Python", que intenta aplicar no sólo en su código sino también en su vida privada. Puedes encontrarle en Twitter: @daniel_ancuta

Consultas geoespaciales: Uso de Python para buscar ciudades

La información de geolocalización se utiliza a diario en casi todos los aspectos de nuestra interacción con los ordenadores. Ya sea un sitio web que quiere enviarnos notificaciones personalizadas basadas en la ubicación, mapas que nos muestran la ruta más corta posible o simplemente tareas que se ejecutan en segundo plano y comprueban los lugares que hemos visitado.

Hoy me gustaría presentarles a consultas geoespaciales que se utilizan en Couchbase. Consultas geoespaciales permiten buscar documentos en función de su localización geográfica.

Juntos escribiremos una herramienta en Python que utilice consultas geoespaciales con API REST de Couchbase y Búsqueda de texto completo en Couchbaseque nos ayudará a buscar en una base de datos de ciudades.

Requisitos previos

Dependencias

En este artículo he utilizado Couchbase Edición Enterprise 5.1.0 build 5552 y Python 3.6.4.

Para ejecutar fragmentos de este artículo debe instalar Couchbase 2.3 (estoy usando 2.3.4) a través de pip.

Couchbase

Crear un cubo de ciudades
Cree una búsqueda de ciudades con un campo geográfico de tipo geopunto. Puede leer sobre ello en el Insertar un campo hijo parte de la documentación.

Debería parecerse a la imagen de abajo:

Rellenar Couchbase con datos

En primer lugar, necesitamos datos para nuestro ejercicio. Para ello, utilizaremos una base de datos de ciudades de geonames.org.

GeoNames contiene dos bases de datos principales: lista de ciudades y lista de códigos postales.

Todos están agrupados por países con la información correspondiente como nombre, coordenadas, población, zona horaria, código de país, etc. Ambos están en formato CSV.

Para este ejercicio, utilizaremos la lista de ciudades. He utilizado PL.zip pero siéntase libre de elegir el que prefiera del lista de ciudades.

Modelo de datos

La clase City será nuestra representación de una única ciudad que utilizaremos en toda la aplicación. Al encapsularla en un modelo, unificamos la API y no necesitamos depender de fuentes de datos de terceros (por ejemplo, un archivo CSV) que podrían cambiar.

La mayoría de nuestros fragmentos se encuentran (hasta que se diga lo contrario) en el archivo core.py. Así que recuerda actualizarlo (especialmente cuando añadas nuevas importaciones) y no anular todo el contenido.

# 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

clase Ciudad:

def __init__(auto, geonameid, clase_característica, nombre, población, lat, lon):

auto.geonameid = geonameid

auto.clase_característica = clase_característica

auto.nombre = nombre

auto.población = población

auto.lat = lat

auto.lon = lon

@método de clase

def from_csv_row(cls, fila):

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

Iterador CSV para procesar ciudades

Como ya tenemos una clase modelo, es hora de preparar un iterador que nos ayude a leer las ciudades del fichero 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

importar csv

de colecciones importar Iterador

clase CiudadesCsvIterator(Iterador):

def __init__(auto, ruta):

auto.Camino = ruta

auto._fp = Ninguno

auto._csv_reader = Ninguno

def __enter__(auto):

auto._fp = abra(auto.Camino, 'r')

auto._csv_reader = csv.lector(auto._fp, delimitador='\t')

devolver auto

def __exit__(auto, exc_tipo, exc_val, exc_tb):

auto._fp.cerrar()

def __next__(auto):

devolver Ciudad.from_csv_row(siguiente(auto._csv_reader))

Insertar ciudades en el bucket de Couchbase

Hemos unificado la forma de representar una ciudad, y tenemos un iterador que leería las del archivo csv.

Es hora de poner estos datos en nuestra fuente de datos 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

importar registro

importar sys

de couchbase.grupo importar Grupo, PasswordAuthenticator

registrador = registro.getLogger()

registrador.setLevel(registro.DEBUG)

registrador.addHandler(registro.StreamHandler(sys.stdout))

def get_bucket(nombre de usuario, contraseña, cadena_de_conexión=couchbase://localhost):

grupo = Grupo(cadena_de_conexión)

autentificador = PasswordAuthenticator(nombre de usuario, contraseña)

grupo.autentifique(autentificador)

devolver grupo.open_bucket(ciudades)

clase CiudadesServicio:

def __init__(auto, cubo):

auto._bucket = cubo

def cargar_desde_csv(auto, ruta):

con CiudadesCsvIterator(ruta) como iterador_ciudades:

para ciudad en iterador_ciudades:

si ciudad.clase_característica no en (PPL, PPLA, PPLA2, PPLA3,

PPLA4, PPLC):

continuar

registrador.información(f'Insertando {ciudad.geonameid}')

auto._bucket.upsert(

ciudad.geonameid,

{

nombre: ciudad.nombre,

clase_característica: ciudad.clase_característica,

población: ciudad.población,

geo: {"lat: float(ciudad.lat),

"lon: float(ciudad.lon)}

}

)

Para comprobar si todo lo que hemos escrito hasta ahora funciona, vamos a cargar contenido CSV en 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

cubo = get_bucket('admin', prueba123456)

servicio_ciudades = CiudadesServicio(cubo)

servicio_ciudades.cargar_desde_csv(~/directorio-con-ciudades/PL/PL.txt, cubo)

En este punto deberías tener las ciudades cargadas en tu bucket de Couchbase. El tiempo que tarde dependerá del país que haya elegido.

Buscar ciudades

Ya tenemos nuestro bucket listo con los datos, así que es hora de volver a CitiesService y preparar unos cuantos métodos que nos ayuden en la búsqueda de ciudades.

Pero antes de empezar, tenemos que modificar ligeramente la clase City, añadiendo el siguiente 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 clase

def from_couchbase_dict(cls, fila):

campos = fila[campos]

devolver cls(fila[id],

campos[clase_característica],

campos[nombre],

campos[población],

campos[geo][1],

campos[geo][0])

Esa es la lista de métodos que implementaremos en CitiesService:

get_by_name(name, limit=10), devuelve las ciudades por su nombre
get_by_coordinates(lat, lon), devuelve la ciudad por coordenadas
get_nearest_to_city(ciudad, distancia='10', unidad='km', limit=10), devuelve la ciudad más cercana

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 importar TermQuery

ÍNDICE_NOMBRE = ciudades

def get_by_name(auto, nombre, límite=10):

resultado = auto._bucket.busque en(auto.NOMBRE_DEL_ÍNDICE,

TermQuery(nombre.inferior(), campo=nombre),

límite=límite,

campos='*')

para c_ciudad en resultado:

rendimiento Ciudad.from_couchbase_dict(c_ciudad)

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 importar GeoDistanceQuery

ÍNDICE_NOMBRE = ciudades

def get_by_coordinates(auto, lat, lon):

resultado = auto._bucket.busque en(auto.NOMBRE_DEL_ÍNDICE,

GeoDistanceQuery(1 km, (lon, lat)),

campos='*')

para c_ciudad en resultado:

rendimiento Ciudad.from_couchbase_dict(c_ciudad)

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 importar RawQuery, SortRaw

ÍNDICE_NOMBRE = ciudades

def get_nearest_to_city(auto, ciudad, distancia='10', unidad=km, límite=10):

consulta = RawQuery({

ubicación: {

"lon: ciudad.lon,

"lat: ciudad.lat

distancia: str(distancia) + unidad,

campo: geo

})

ordenar = SortRaw([{

"por: geo_distancia,

campo: geo,

unidad: unidad,

ubicación: {

"lon: ciudad.lon,

"lat: ciudad.lat

}

}])

resultado = auto._bucket.busque en(auto.NOMBRE_DEL_ÍNDICE,

consulta,

ordenar=ordenar,

campos='*',

límite=límite)

para c_ciudad en resultado:

rendimiento Ciudad.from_couchbase_dict(c_ciudad)

Como podrás notar en este ejemplo, usamos las clases RawQuery y SortRaw. Lamentablemente, la API couchbase-python-client no funciona correctamente con las nuevas búsquedas de Couchbase y geo.

Métodos de llamada

Como ya tenemos todos los métodos listos, ¡podemos llamarlo!

# 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

cubo = get_bucket('admin', prueba123456)

servicio_ciudades = CiudadesServicio(cubo)

# cities_service.load_from_csv('/mi-ruta/PL/PL.txt')

imprimir(get_by_name)

ciudades = servicio_ciudades.get_by_name(Poznań)

para ciudad en ciudades:

imprimir(ciudad.__dict__)

imprimir('get_by_coordinates')

ciudades = servicio_ciudades.get_by_coordinates(52.40691997632544,

16.929929926276657)

para ciudad en ciudades:

imprimir(ciudad.__dict__)

imprimir(get_nearest_to_city)

ciudades = servicio_ciudades.get_nearest_to_city(ciudad)

para ciudad en ciudades:

imprimir(ciudad.__dict__)

¿Qué hacer a partir de ahora?

Creo que esta introducción le permitirá trabajar en algo más avanzado.

Hay algunas cosas que puedes hacer:

Tal vez utilizar una herramienta CLI o REST API para servir estos datos... Mejorar la forma en que cargamos los datos, ya que podría no ser super performante si queremos cargar TODAS las ciudades de TODOS los países.

Puedes encontrar el código completo de core.py en gist github.

Si tienes alguna pregunta, no dudes en tuitearme @daniel_ancuta.

Este post forma parte del Programa de escritura comunitaria

Laura Czajkowski, Directora de la Comunidad de Desarrolladores, Couchbase

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

Todos los mensajes

Consultas geoespaciales: Uso de Python para buscar ciudades

Consultas geoespaciales: Uso de Python para buscar ciudades

Requisitos previos

Dependencias

Couchbase

Rellenar Couchbase con datos

Modelo de datos

Iterador CSV para procesar ciudades

Insertar ciudades en el bucket de Couchbase

Buscar ciudades

get_by_name

get_by_coordinates

get_nearest_to_city

Métodos de llamada

¿Qué hacer a partir de ahora?

Autor

Publicado por Laura Czajkowski, Directora de la Comunidad de Desarrolladores, Couchbase

Dejar una respuesta Cancelar respuesta