Geospatial queries: Using Python to search cities

다니엘 안쿠타는 다년간 다양한 기술을 사용해 온 소프트웨어 엔지니어입니다. 그는 '파이썬의 선'의 열렬한 팬으로, 코드뿐만 아니라 사생활에도 적용하려고 노력합니다. 트위터에서 그를 만날 수 있습니다: 다니엘_안쿠타

지리공간 쿼리: Python을 사용하여 도시 검색하기

위치 정보는 우리가 컴퓨터와 상호작용하는 거의 모든 측면에서 매일 사용됩니다. 위치를 기반으로 맞춤형 알림을 보내려는 웹사이트, 최단 경로를 보여주는 지도, 방문한 장소를 확인하는 백그라운드에서 실행 중인 작업 등 모든 곳에서 위치 정보가 사용됩니다.

오늘 소개해드릴 내용은 다음과 같습니다. 지리공간 쿼리 를 사용합니다. 지리공간 쿼리 를 사용하면 지리적 위치를 기준으로 문서를 검색할 수 있습니다.

함께 파이썬에서 다음을 사용하는 도구를 작성해 보겠습니다. 지리공간 쿼리 와 함께 Couchbase REST API 그리고 카우치베이스 전체 텍스트 검색를 클릭하면 도시 데이터베이스를 검색하는 데 도움이 됩니다.

전제 조건

종속성

이 문서에서는 다음을 사용했습니다. 카우치베이스 엔터프라이즈 에디션 5.1.0 빌드 5552 및 Python 3.6.4.

이 문서의 스니펫을 실행하려면 pip를 통해 Couchbase 2.3(저는 2.3.4를 사용 중입니다)을 설치해야 합니다.

카우치베이스

도시 버킷 만들기
지오포인트 유형의 지리적 필드를 사용하여 도시 검색을 만듭니다. 이에 대한 자세한 내용은 하위 필드 삽입 문서의 일부입니다.

아래 이미지와 같이 표시되어야 합니다:

Couchbase에 데이터 채우기

우선, 운동에 필요한 데이터가 필요합니다. 이를 위해 다음의 도시 데이터베이스를 사용할 것입니다. geonames.org.

지오네임에는 도시 목록과 우편번호 목록이라는 두 가지 주요 데이터베이스가 있습니다.

모두 국가별로 이름, 좌표, 인구, 시간대, 국가 코드 등의 해당 정보가 포함된 국가별로 그룹화되어 있습니다. 둘 다 CSV 형식입니다.

이 연습에서는 도시 목록을 사용하겠습니다. 저는 PL.zip 에서 원하는 것을 자유롭게 선택할 수 있습니다. 도시 목록.

데이터 모델

도시 클래스는 전체 애플리케이션에서 사용할 단일 도시의 표현이 될 것입니다. 이를 모델로 캡슐화함으로써 API를 통합하고 변경될 수 있는 타사 데이터 소스(예: CSV 파일)에 의존할 필요가 없습니다.

대부분의 스니펫은 (달리 명시되지 않는 한) core.py 파일에 있습니다. 따라서 전체 내용을 덮어쓰지 말고 업데이트(특히 새 가져오기를 추가할 때)하는 것을 잊지 마세요.

# 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

클래스 도시:

def __init__(self, geonameid, 기능_클래스, 이름, 인구, 위도, lon):

self.geonameid = geonameid

self.기능_클래스 = 기능_클래스

self.이름 = 이름

self.인구 = 인구

self.위도 = 위도

self.lon = lon

@클래스 메서드

def FROM_CSV_ROW(cls, 행):

반환 cls(행[0], 행[7], 행[1], 행[12], 행[4], 행[5])

도시를 처리하는 CSV 반복기

모델 클래스가 있으므로 이제 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

가져오기 csv

에서 컬렉션 가져오기 이터레이터

클래스 CitiesCsvIterator(이터레이터):

def __init__(self, 경로):

self._path = 경로

self._fp = 없음

self._csv_reader = 없음

def __enter__(self):

self._fp = 열기(self._path, 'r')

self._csv_reader = csv.reader(self._fp, 구분 기호='\t')

반환 self

def __exit__(self, exc_type, exc_val, exc_tb):

self._fp.닫기()

def 다음__(self):

반환 도시.FROM_CSV_ROW(다음(self._csv_reader))

카우치베이스 버킷에 도시 삽입하기

도시를 표현하는 방식을 통합했으며, csv 파일에서 도시를 읽는 이터레이터가 있습니다.

이제 이 데이터를 기본 데이터 소스에 넣을 차례입니다, 카우치베이스.

# 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

가져오기 로깅

가져오기 sys

에서 카우치베이스.클러스터 가져오기 클러스터, 비밀번호 인증기

logger = 로깅.getLogger()

logger.setLevel(로깅.DEBUG)

logger.추가 핸들러(로깅.스트림 핸들러(sys.stdout))

def get_bucket(사용자 이름, 비밀번호, 연결 문자열='couchbase://localhost'):

클러스터 = 클러스터(연결 문자열)

인증자 = 비밀번호 인증기(사용자 이름, 비밀번호)

클러스터.인증(인증자)

반환 클러스터.open_bucket('cities')

클래스 도시 서비스:

def __init__(self, 버킷):

self._버킷 = 버킷

def load_from_csv(self, 경로):

와 함께 CitiesCsvIterator(경로) as 도시의_이터레이터:

에 대한 도시 in 도시의_이터레이터:

만약 도시.기능_클래스 not in ('PPL', 'PPLA', 'PPLA2', 'PPLA3',

'PPLA4', 'PPLC'):

계속

logger.정보(f'{city.geonameid} 삽입')

self._버킷.업서트(

도시.geonameid,

{

'name': 도시.이름,

'feature_class': 도시.기능_클래스,

'인구': 도시.인구,

'geo': {'lat': float(도시.위도),

'lon': float(도시.lon)}

}

)

지금까지 작성한 모든 내용이 제대로 작동하는지 확인하기 위해 CSV 콘텐츠를 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

버킷 = get_bucket('admin', 'test123456')

cities_service = 도시 서비스(버킷)

cities_service.load_from_csv('~/directory-with-cities/PL/PL.txt', 버킷)

이 시점에서 Couchbase 버킷에 도시가 로드되어 있어야 합니다. 소요 시간은 선택한 국가에 따라 다릅니다.

도시 검색

데이터로 버킷을 준비했으니 이제 CitiesService로 돌아와서 도시 검색에 도움이 될 몇 가지 방법을 준비할 차례입니다.

하지만 시작하기 전에 다음 메서드를 추가하여 City 클래스를 약간 수정해야 합니다:

# 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

@클래스 메서드

def FROM_COUCHBASE_DICT(cls, 행):

필드 = 행['fields']

반환 cls(행['id'],

필드['feature_class'],

필드['name'],

필드['인구'],

필드['geo'][1],

필드['geo'][0])

이것이 CitiesService에서 구현할 방법의 목록입니다:

get_by_name(name, limit=10)은 이름별로 도시를 반환합니다.
get_by_coordinates(lat, long), 좌표로 도시를 반환합니다.
get_nearest_to_city(city, distance='10', unit='km', limit=10)는 가장 가까운 도시를 반환합니다.

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

에서 카우치베이스.전체 텍스트 가져오기 용어 쿼리

INDEX_이름 = 'cities'

def get_by_name(self, 이름, limit=10):

결과 = self._버킷.검색(self.INDEX_NAME,

용어 쿼리(이름.lower(), 필드='name'),

limit=limit,

필드='*')

에 대한 c_city in 결과:

수익률 도시.FROM_COUCHBASE_DICT(c_city)

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

에서 카우치베이스.전체 텍스트 가져오기 지리적 거리 쿼리

INDEX_이름 = 'cities'

def get_by_coordinates(self, 위도, lon):

결과 = self._버킷.검색(self.INDEX_NAME,

지리적 거리 쿼리('1km', (lon, 위도)),

필드='*')

에 대한 c_city in 결과:

수익률 도시.FROM_COUCHBASE_DICT(c_city)

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

에서 카우치베이스.전체 텍스트 가져오기 RawQuery, SortRaw

INDEX_이름 = 'cities'

def get_nearest_to_city(self, 도시, 거리='10', 단위='km', limit=10):

쿼리 = RawQuery({

'위치': {

'lon': 도시.lon,

'lat': 도시.위도

'거리': str(거리) + 단위,

'field': 'geo'

})

정렬 = SortRaw([{

'by': 'geo_distance',

'field': 'geo',

'unit': 단위,

'위치': {

'lon': 도시.lon,

'lat': 도시.위도

}

}])

결과 = self._버킷.검색(self.INDEX_NAME,

쿼리,

정렬=정렬,

필드='*',

limit=limit)

에 대한 c_city in 결과:

수익률 도시.FROM_COUCHBASE_DICT(c_city)

이 예제에서 알 수 있듯이 RawQuery와 SortRaw 클래스를 사용했습니다. 안타깝게도 couchbase-python-client API는 최신 Couchbase 및 지역 검색에서 제대로 작동하지 않습니다.

호출 방법

이제 모든 메소드가 준비되었으므로 호출할 수 있습니다!

# 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

버킷 = get_bucket('admin', 'test123456')

cities_service = 도시 서비스(버킷)

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

인쇄('get_by_name')

도시 = cities_service.get_by_name('포즈난')

에 대한 도시 in 도시:

인쇄(도시.__dict__)

인쇄('get_by_coordinates')

도시 = cities_service.get_by_coordinates(52.40691997632544,

16.929929926276657)

에 대한 도시 in 도시:

인쇄(도시.__dict__)

인쇄('get_nearest_to_city')

도시 = cities_service.get_nearest_to_city(도시)

에 대한 도시 in 도시:

인쇄(도시.__dict__)

이제 어디로 가야 하나요?

이 소개를 통해 더 발전된 작업을 할 수 있을 것이라 믿습니다.

몇 가지 할 수 있는 일이 있습니다:

모든 국가의 모든 도시를 로드하려면 성능이 좋지 않을 수 있으므로 데이터를 로드하는 방식을 개선해야 합니다.

core.py의 전체 코드는 다음에서 찾을 수 있습니다. 깃허브 요점.

궁금한 점이 있으면 주저하지 말고 트위터로 문의하세요. 다니엘_안쿠타.

이 게시물은 커뮤니티 글쓰기 프로그램

Laura Czajkowski, 개발자 커뮤니티 관리자, 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

모든 게시물

지리공간 쿼리: Python을 사용하여 도시 검색하기

지리공간 쿼리: Python을 사용하여 도시 검색하기

전제 조건

종속성

카우치베이스

Couchbase에 데이터 채우기

데이터 모델

도시를 처리하는 CSV 반복기

카우치베이스 버킷에 도시 삽입하기

도시 검색

get_by_name

get_by_coordinates

get_nearest_to_city

호출 방법

이제 어디로 가야 하나요?

작성자

게시자 Laura Czajkowski, 개발자 커뮤니티 관리자, Couchbase

댓글 남기기 응답 취소