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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
# 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]) |
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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
# 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)) |
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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
# 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)} } ) |
Para comprobar si todo lo que hemos escrito hasta ahora funciona, vamos a cargar contenido CSV en Couchbase.
|
1 2 3 4 5 |
# core.py bucket = get_bucket('admin', 'test123456') cities_service = CitiesService(bucket) cities_service.load_from_csv('~/directory-with-cities/PL/PL.txt', bucket) |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
# 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]) |
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
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
# 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) |
get_by_coordinates
|
1 2 3 4 5 6 7 8 9 10 11 12 |
# 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) |
get_nearest_to_city
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
# 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) |
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!
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
# 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__) |
¿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
