Pythonやろうぜ!!

前回、地図上に周辺の飲食店の情報を載せようって言ったけど、その前に今回はその店舗情報を取得する方法について紹介するね

これもまた、何かAPIを使うの？

飲食店の情報を取得できるAPIには「Google Places API」や「Yahoo!ローカルサーチAPI」とかがあるんだけど、これはユーザー登録が必要だったり、利用制限があるから今回は使わずに進めるよ

え！？じゃあどうするの？？まさか自分で店舗情報を回収するの！？

半分正解だね！
「OpenStreetMap」の中には全国の飲食店情報が収録されているデータがあるから、これを使う方針で進めるよ！

全国の飲食店情報って...すごい量じゃない？？

それが、全国の飲食店って全部で大体100万件くらいなんだよね

なんかもっとたくさんあるイメージだったけどそれくらいなんだ...！！

だから、「OpenStreetMap」でダウンロードするファイルのデータ量も2GBほどで、意外と個人でも管理できるレベルなんだよねー

OpenStreetMapの店舗情報をダウンロードできるサイト「Geofabrik」のアジア圏のページ

とはいえ、実際に使おうとするには大きいデータ量だよね
「pbf」っていう見たことない拡張子になってるし...

そこで、このデータの中から必要な情報だけを抽出して、Webページで扱いやすくするために「データベース」にして利用する方法を紹介するよ！

データベース...？？

「データベース」は、情報を整理して効率的に管理するためのシステムだよ

てことは、エクセルとかJSONもデータベース？

まぁ感覚的には近いし、それで今回やりたいことも出来るんだけど、データベースはより専門的なシステムで、大量のデータを効率的に処理・管理することができるんだよね

確かに、100万件もあるデータを一気に処理しようとしたら大変そうだもんね

データベースにも色々種類があるんだけど、今回はその中で最も一般的な「リレーショナルデータベース」を使っていくよ

よく分からないけど、ひとまずは了解！

んで、このデータベースっていうのを扱うのに使うライブラリが「sqlite3」っていう標準ライブラリだよ

標準ライブラリってことは、pipでインストールしなくても良いんだね！

そゆこと
んで、早速sqlite3を使うんだけど、そのためには「SQL」っていうプログラミング言語の知識も必要になっちゃうから、今回はとりあえず下の関数をそのまま真似して使おうか

import sqlite3
import pandas as pd
from geopy.distance import geodesic

#
# 前回のプログラムのメインのプログラムの上に以下の関数を追加しよう
#
# データベースから周辺の店を探す
def search_restaurants(lat, lon, radius_km=1.0):
    try:
        conn = sqlite3.connect("restaurants.db")
        # 検索を速くするための大まかな絞り込み（緯度経度の差）
        # 距離オブジェクトの作成
        d = geodesic(kilometers=1.0)

        # 各方位の地点を計算
        # destination(始点, 方位角)  方位角: 0=北, 90=東, 180=南, 270=西
        north = d.destination((lat, lon), 0)
        south = d.destination((lat, lon), 180)
        east = d.destination((lat, lon), 90)
        west = d.destination((lat, lon), 270)
        
        query = f"""
        SELECT * FROM restaurants 
        WHERE lat BETWEEN {south.latitude} AND {north.latitude}
          AND lon BETWEEN {west.longitude} AND {east.longitude}
        """

        df = pd.read_sql_query(query, conn)
        conn.close()
        
        # 正確な距離でフィルタリング
        results = []
        for _, row in df.iterrows():
            dist = geodesic((lat, lon), (row['lat'], row['lon'])).km
            if dist <= radius_km:
                res_dict = row.to_dict()
                res_dict['distance_km'] = dist
                results.append(res_dict)
        
        return sorted(results, key=lambda x: x['distance_km'])
    except Exception as e:
        st.error(f"データベースエラー: {e}")
        return []

この関数は、調べたい場所の緯度経度と、どの範囲まで探すかを指定すると、データベースからその範囲内の店舗の配列を返してくれるよ

この関数の「query」っていう変数がSQLの部分なんだね！
プログラムは作れないけど、英語が分かっていれば何となく意味は分かるかも！

sqlite3は基本的に、「sqlite3.connect()」で扱いたいデータベースを指定して、クエリって呼ばれる文字列で欲しいデータを指定して読み込んで、「conn.close()」で閉じるって流れになるよ

今回は中身を読み込む時に「Pandas」を利用することで、取り出した時点でデータフレームとして扱えるんだ

Pandasは前にエクセルのデータを読み込むのに使ったけど、データベースでも使うんだね

むしろ、こっちの方がメインの使い方だよ

ところで、「restaurants.db」はどこにあるの？

今回はひとまず、事前に用意したデータベースがあるから、下のリンクから圧縮ファイルをダウンロードして、カレントディレクトリに解凍して使ってみてね

2026/4/2時点の飲食店情報をまとまたデータベース「restaurants.db」の圧縮ファイル

ダウンロードして解凍したよ！

じゃあ次はメインのプログラムでマップを表示する部分を次のプログラムに書き換えよう

# メインのプログラムの表示部分を以下のように書き換えよう
m = folium.Map(location=[lat, lon], zoom_start=15)
restaurants = search_restaurants(lat, lon)
st_folium(m, width=700, height=500, returned_objects=[])

st.write(f"「{target_place}」の周辺に {len(restaurants)} 件の飲食店が見つかりました。")

# 結果をテーブルで表示
if restaurants:
    df_results = pd.DataFrame(restaurants)
    st.dataframe(df_results[['name', 'amenity', 'distance_km']], use_container_width=True)

おおー！ちゃんと一覧が表示されたね！

もし最新のデータベースを扱いたい場合は、ここからダウンロードして、下のプログラムを実行するとデータベースを作ってくれるよ！

import sqlite3
import pandas as pd
import osmium

PBF_FILE = "japan-260402.osm.pbf" # https://download.geofabrik.de/asia.html からダウンロードしたファイルを指定
DB_FILE = "restaurants.db"

# OpenStreetMapのPOIを抽出するクラス
class RestaurantHandler(osmium.SimpleHandler):
    def __init__(self):
        super(RestaurantHandler, self).__init__()
        self.restaurants = []

    def process_element(self, element, elem_type):
        tags = element.tags
        amenity = tags.get('amenity')
        if amenity in ['restaurant', 'cafe', 'fast_food']:
            name = tags.get('name')
            if name:
                lat = element.location.lat if elem_type == 'node' else getattr(element, 'center', lambda: None)().lat if hasattr(element, 'center') else None
                lon = element.location.lon if elem_type == 'node' else getattr(element, 'center', lambda: None)().lon if hasattr(element, 'center') else None
                
                if lat and lon:
                    self.restaurants.append({
                        'id': element.id,
                        'name': name,
                        'amenity': amenity,
                        'cuisine': tags.get('cuisine', ''),
                        'lat': lat,
                        'lon': lon
                    })

    def node(self, n):
        self.process_element(n, 'node')

def main():
    print(f"{PBF_FILE} を読み込んでいます...")
    
    handler = RestaurantHandler()
    
    try:
        print("POIを抽出しています。数分かかる場合があります...")
        handler.apply_file(PBF_FILE)
    except Exception as e:
        print(f"PBFの読み込みに失敗しました: {e}")
        return

    if not handler.restaurants:
        print("条件に合うPOIが見つかりませんでした。")
        return

    print(f"抽出されたPOIの数: {len(handler.restaurants)}")
    
    df = pd.DataFrame(handler.restaurants)
    
    print(f"{DB_FILE} に {len(df)} 件の飲食店を保存しています...")
    
    # SQLiteに接続
    conn = sqlite3.connect(DB_FILE)
    
    # pandas dataframeをsqliteに保存
    df.to_sql("restaurants", conn, if_exists="replace", index=False)
    
    # 空間検索用のインデックスを作成
    cursor = conn.cursor()
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_lat ON restaurants(lat);")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_lon ON restaurants(lon);")
    conn.commit()
    
    conn.close()
    
    print("データの抽出と保存が完了しました！")

if __name__ == "__main__":
    main()

ちなみに、結構時間がかかるし、メモリを凄く使うから他の作業がやりにくい可能性があるから動かす時は気を付けてね

分かった！！
ひとまずは、練習の間は事前に用意してくれてるので十分かな

この時点での全体のプログラム

import sqlite3
import folium
from streamlit_folium import st_folium
import streamlit as st
import pandas as pd
from geopy.geocoders import Nominatim
from geopy.distance import geodesic

# 土地名から緯度経度の取得
def get_coordinates(address):
    geolocator = Nominatim(user_agent="my_geocoding_app")
    geocoder_result = geolocator.geocode(address, timeout=5.0)
    try:
        return float(geocoder_result.latitude), float(geocoder_result.longitude)
    except Exception as e:
        print(f"接続エラー: {e}")
    return None, None

# データベースから周辺の店を探す
def search_restaurants(lat, lon, radius_km=1.0):
    try:
        conn = sqlite3.connect("restaurants.db")
        # 検索を速くするための大まかな絞り込み（緯度経度の差）
        # 距離オブジェクトの作成
        d = geodesic(kilometers=1.0)

        # 各方位の地点を計算
        # destination(始点, 方位角)  方位角: 0=北, 90=東, 180=南, 270=西
        north = d.destination((lat, lon), 0)
        south = d.destination((lat, lon), 180)
        east = d.destination((lat, lon), 90)
        west = d.destination((lat, lon), 270)
        
        query = f"""
        SELECT * FROM restaurants 
        WHERE lat BETWEEN {south.latitude} AND {north.latitude}
          AND lon BETWEEN {west.longitude} AND {east.longitude}
        """

        df = pd.read_sql_query(query, conn)
        conn.close()
        
        # 正確な距離でフィルタリング
        results = []
        for _, row in df.iterrows():
            dist = geodesic((lat, lon), (row['lat'], row['lon'])).km
            if dist <= radius_km:
                res_dict = row.to_dict()
                res_dict['distance_km'] = dist
                results.append(res_dict)
        
        return sorted(results, key=lambda x: x['distance_km'])
    except Exception as e:
        st.error(f"データベースエラー: {e}")
        return []


# 1. st.form を使って、入力中の勝手な更新を防ぐ
with st.form("search_form"):
    target_place = st.text_input("調べたい場所を入力してください", "渋谷駅")
    submitted = st.form_submit_button("検索する")

# 2. ボタンが押された時に検索を実行し、結果を session_state に保存する
if submitted:
    if target_place:
        with st.spinner("検索中..."):
            lat, lon = get_coordinates(target_place)
            if lat and lon:
                m = folium.Map(location=[lat, lon], zoom_start=15)
                restaurants = search_restaurants(lat, lon)
                st_folium(m, width=700, height=500, returned_objects=[])

                st.write(f"「{target_place}」の周辺に {len(restaurants)} 件の飲食店が見つかりました。")
                
                # 結果をテーブルで表示
                if restaurants:
                    df_results = pd.DataFrame(restaurants)
                    st.dataframe(df_results[['name', 'amenity', 'distance_km']], use_container_width=True)
                
            else:
                st.error("指定された場所が見つかりませんでした。")
    else:
        st.error("場所を入力してください。")

＜＜ 2.地図を表示しよう 4.地図にピンを立てよう＞＞