【実践】PythonとGoogle Maps APIで学ぶ経路システム開発入門

少し前から興味のあったGoogle Map APIを少し本格的に使用してみたいと考えたので、PythonGoogle Maps APIを使用して、 現在地から最寄りの避難所への経路URLを生成するシステムを構築みたという体験記になります。

まあ、学び直しということで🫡


1. 開発環境の前提条件

本内容では以下の環境を前提としています。

環境確認コマンド

以下で環境を確認してください。

# WSLバージョン確認
PS> wsl --version
# Pythonバージョン確認
$ python3 --version

# pipバージョン確認
$ pip3 --version

2. プロジェクトのセットアップ

2.1 プロジェクトディレクトリの作成

WSLのディストリビューションを起動し、ターミナルを開いて、プロジェクトディレクトリを作成します。

# ホームディレクトリに移動
$ cd ~

# プロジェクト用のディレクトリを作成
$ mkdir evacuation-route-system
$ cd evacuation-route-system
# 結果データ保存用のディレクトリを作成
$ mkdir outputs

2.2 Python仮想環境(venv)の作成

プロジェクトの仮想環境をvenvを使用して、依存関係を管理します。

# 仮想環境を作成(venvという名前で作成)
$ python3 -m venv venv

# 仮想環境を有効化
$ source venv/bin/activate

# プロンプトに(venv)が表示されることを確認する
# 例: (venv) username@hostname:~/evacuation-route-system$

# 仮想環境内のPythonを確認
$ which python
# 出力の文字列にvenvのパスが含まれているかを確認
# /home/username/evacuation-route-system/venv/bin/python

# 念のためにpipをアップグレード
$ pip install --upgrade pip

# 仮想環境が正しく動作することを確認
$ python --version
$ pip --version

※ 仮想環境を終了する場合はdeactivate ※ 次回作業時は必ずsource venv/bin/activateで仮想環境を有効化すること!


3. Google Cloud Platform(GCP)の設定

3.1 GCPアカウント

  1. Googleアカウントの準備

  2. 既存のGoogleアカウントを使用するか、新規作成します。

  3. https://accounts.google.co にアクセス

accounts.google.com

  1. GCPコンソールへのアクセス

  2. (https://console.cloud.google.com)https://console.cloud.google.com にアクセス

  3. Googleアカウントでログイン

console.cloud.google.com

  1. 無料トライアルの有効化

    • 初回利用時はクレジットが付与される無料トライアルを有効化可能
    • クレジットカード情報の登録が必要(自動課金はされません)

私はすでにGCPアカウントを持っているので、ここでは省略します。

3.2 プロジェクトの作成

新規プロジェクトの作成

GCPコンソール上の手順は以下の通り

画面上部のプロジェクト名のボタンをクリック(現在選択されているプロジェクト名が表示されていますが、新規作成するので表示名は何でも大丈夫です)

開いたダイアログの上部にある【新しいプロジェクト】をクリック

【プロジェクト名】のテキストボックスに以下のように入力します。プロジェクトIDには自動生成されたものを使用しています。

【プロジェクト名】 evacuation-route-system

【作成】ボタンをクリック

プロジェクトIDの確認

作成後に、プロジェクトセレクタで新しく作成したプロジェクトに切り替え、【ダッシュボード】をクリックし、以下のように設定されていることを確認してください。

  • プロジェクト名:evacuation-route-system
  • プロジェクトID:evacuation-route-system-xxxxx

3.3 必要なAPIの有効化

APIライブラリ経由での有効化(推奨)

左メニューまたはプロジェクトの画面で【APIとサービス】ボタンをクリックし、

遷移した画面の【ライブラリ】をクリックします。

検索ボックスで各APIを検索(文字列を入力)

今回有効化するAPI

  1. Directions API … 経路情報の取得
  2. Geocoding API … 住所から座標への変換
  3. Maps Static API … 静的地図画像の生成

APIの詳細画面で【有効にする】ボタンをクリック

3.4 APIキーの作成と制限

APIキーの作成を行います。【APIとサービス】の画面で、左メニューの【鍵と認証情報】をクリックし、【APIを有効にする】タブをクリックし、【APIを有効にする】ボタンをクリックします。

以下のようなダイアログが表示されたら、【後で】をクリックしてください。今回はテストなので制限をしないことにします。

作成されたAPIキーをコピーし安全に保管します。

左のメニューの【鍵と認証情報】を選択し、【鍵を表示します】をクリックすると再度表示可能です。

すべてのAPIを使用するのではないので、作成したAPIキーには機能制限をかけるのが良いでしょう。【APIの制限】で上記3つのAPIを選択し、【保存】ボタンをクリックします。

⚠️ウルトラ 重要

  • APIキーは安全な場所に保管してください
  • APIキーを含むファイルはGitHubに直接アップロードは禁止です。(キーの含まれるファイルは.gitignoreにファイル名を追加してください)

4. 基本実装

4.1 必要なライブラリのインストール

仮想環境が有効化されていることを確認してから、以降を実行してください。

# 必要なライブラリをインストール
$ pip install requests
$ pip install python-dotenv
# requirements.txtの作成
$ pip freeze > requirements.txt

4.2 設定ファイルの作成

APIキーは.envファイルを使用して管理を行います。

.env

# Google Maps API Key
GOOGLE_MAPS_API_KEY=ここに取得したAPIキーを貼り付け(ダブルクオートは不要)

# 以下はサンプルプログラムで使用する値
# デフォルト位置の設定(名古屋駅の座標)
# 重要:以下の値を変更しないでください
DEFAULT_LOCATION_LAT=35.170694
DEFAULT_LOCATION_LNG=136.881637

.envファイルの読み込みテストを行うために、以下の内容でconfig.pyファイルを作成します。

config.py

import os
from dotenv import load_dotenv

# .envファイルを読み込み
load_dotenv()

# APIキーを環境変数から取得
GOOGLE_MAPS_API_KEY = os.getenv('GOOGLE_MAPS_API_KEY')

# デフォルト設定(名古屋駅)
DEFAULT_LOCATION = {
    'lat': float(os.getenv('DEFAULT_LOCATION_LAT', 35.170694)),
    'lng': float(os.getenv('DEFAULT_LOCATION_LNG', 136.881637))
}

# API設定の確認
if not GOOGLE_MAPS_API_KEY:
    raise ValueError("GOOGLE_MAPS_API_KEYが設定されていません。.envファイルを確認してください。")

print("設定ファイルが正常に読み込まれました。")

4.3 基本クラスの実装とデータ管理

今回は避難所データをCSVからJSONに変換し、Pythonクラスで管理します。

CSVデータは以下のURLからダウンロードしてください。

避難所データのサンプル

このデータは国土地理院の以下のサイトの情報に準拠して作成しました。

hinanmap.gsi.go.jp

データは以下のようにしています。

施設・場所名,住所,洪水,崖崩れ、土石流及び地滑り,高潮,地震,津波,大規模な火事,内水氾濫,火山現象,指定避難所との住所同一,緯度,経度
蟹江川排水機場,愛知県蟹江町大字蟹江本町地先,,,,,1,,,,,35.111092,136.792548
中央卸売市場北部市場,愛知県豊山町大字豊場字八反107,1,,1,,,,1,,1,35.24188372,136.905289
いろは公園,愛知県名古屋市港区いろは町2,,,,1,,,,,,35.10387,136.87351
…(以下略)…

では、このCSVファイルをもとにデータ管理用のクラスを作成します。 以下のソースコード@dataclassを使用していますが、Python 3.7以上の機能ですが、非常に便利なのでぜひ活用してください。

models.py

"""
データモデルの定義
CSVデータ構造に基づく設計
"""
from dataclasses import dataclass
from typing import Optional, Dict

@dataclass
class Location:
    """位置情報を表すクラス"""
    lat: float # 緯度
    lng: float # 経度
    name: Optional[str] = None # 場所の名前
    address: Optional[str] = None # 住所
    
    def to_string(self) -> str:
        """Google Maps URL用の文字列表現"""
        if self.address:
            return self.address
        if self.name:
            return self.name
        return f"{self.lat},{self.lng}"
    
    def to_coords(self) -> str:
        """座標の文字列表現"""
        return f"{self.lat},{self.lng}"

@dataclass
class EvacuationCenter:
    """
    避難所情報を管理するクラス
    実際のCSVデータに基づく構造
    """
    name: str # 施設・場所名
    location: Location # 位置情報
    disaster_support: Dict = None # 災害対応情報
    is_designated_shelter: bool = False # 指定避難所フラグ
    
    def __str__(self):
        return f"{self.name}"
    
    def __post_init__(self):
        """初期化後の処理"""
        if self.disaster_support is None:
            self.disaster_support = {}

@dataclass
class Route:
    """経路情報を保存するクラス"""
    origin: Location  # 出発地
    destination: Location  # 目的地
    distance: int  # 距離(メートル)
    duration: int  # 所要時間(秒)
    distance_text: str  # 距離の表示用文字列
    duration_text: str  # 時間の表示用文字列
    travel_mode: str  # 移動手段
    google_maps_url: str = ""  # Google Maps URL
    
    def __str__(self):
        return f"経路: {self.distance_text}, {self.duration_text}"

では、このmodel.pyを使用して、CSVデータをJSONに変換するプログラムを作成します。

csv_to_json_converter.py

"""
CSVファイルから名古屋駅周辺の避難所データをJSONファイルに変換
"""
import csv
import json
import math
from typing import List, Dict

def calculate_distance(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
    """2点間の距離を計算(km)"""
    R = 6371  # 地球の半径(km)
    
    lat1, lon1 = math.radians(lat1), math.radians(lng1)
    lat2, lon2 = math.radians(lat2), math.radians(lng2)
    
    dlat = lat2 - lat1
    dlon = lon2 - lon1
    
    a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
    c = 2 * math.asin(math.sqrt(a))
    
    return R * c

def convert_csv_to_json(csv_file: str, output_file: str, center_lat: float, center_lng: float, max_distance: float = 5.0):
    """
    CSVファイルから中心地点の近くの避難所をJSONに変換
    """
    evacuation_centers = []
    
    with open(csv_file, 'r', encoding='utf-8-sig') as f:
        reader = csv.DictReader(f)
        
        for row in reader:
            # 緯度経度が空の場合はスキップ
            if not row['緯度'] or not row['経度']:
                continue
            
            try:
                lat = float(row['緯度'])
                lng = float(row['経度'])
            except ValueError:
                continue
            
            # 中心地点からの距離を計算
            distance = calculate_distance(center_lat, center_lng, lat, lng)
            
            # 指定距離以内の避難所のみ選択
            if distance <= max_distance:
                # 実際のデータのみを含む避難所データ
                center_data = {
                    "name": row['施設・場所名'],
                    "location": {
                        "lat": lat,
                        "lng": lng,
                        "address": row['住所']
                    },
                    "distance_from_center": round(distance, 2),
                    "disaster_support": {
                        "洪水": row.get('洪水', '') == '1',
                        "崖崩れ_土石流": row.get('崖崩れ、土石流及び地滑り', '') == '1',
                        "高潮": row.get('高潮', '') == '1',
                        "地震": row.get('地震', '') == '1',
                        "津波": row.get('津波', '') == '1',
                        "大規模火事": row.get('大規模な火事', '') == '1',
                        "内水氾濫": row.get('内水氾濫', '') == '1',
                        "火山": row.get('火山現象', '') == '1'
                    },
                    "is_designated_shelter": row.get('指定避難所との住所同一', '') == '1'
                }
                evacuation_centers.append(center_data)
    
    # 距離でソート
    evacuation_centers.sort(key=lambda x: x['distance_from_center'])
    
    # 上位20件を選択
    evacuation_centers = evacuation_centers[:20]
    
    # 主要な場所の座標も追加
    locations_data = {
        "evacuation_centers": evacuation_centers,
        "test_locations": [
            {
                "name": "名古屋駅",
                "lat": 35.170694,
                "lng": 136.881637,
                "address": "愛知県名古屋市中村区名駅1丁目1-4"
            },
            {
                "name": "栄駅",
                "lat": 35.170032,
                "lng": 136.908687,
                "address": "愛知県名古屋市中区栄3丁目"
            },
            {
                "name": "金山駅",
                "lat": 35.143015,
                "lng": 136.901612,
                "address": "愛知県名古屋市中区金山1丁目"
            }
        ]
    }
    
    # JSONファイルに保存
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(locations_data, f, ensure_ascii=False, indent=2)
    
    print(f"{len(evacuation_centers)}件の避難所データを {output_file} に保存しました")
    print(f"中心地点(名古屋駅)から{max_distance}km以内の避難所を選択")
    
    # 対応災害の統計
    disaster_count = {
        "洪水": 0,
        "地震": 0,
        "津波": 0,
        "大規模火事": 0
    }
    
    for center in evacuation_centers:
        for disaster, supported in center['disaster_support'].items():
            if disaster in disaster_count and supported:
                disaster_count[disaster] += 1
    
    print("\n災害対応統計:")
    for disaster, count in disaster_count.items():
        print(f"  {disaster}対応: {count}件")
    
    # 指定避難所の数を表示
    designated_count = sum(1 for c in evacuation_centers if c['is_designated_shelter'])
    print(f"\n指定避難所: {designated_count}件")

if __name__ == "__main__":
    # 名古屋駅の座標
    NAGOYA_STATION_LAT = 35.170694
    NAGOYA_STATION_LNG = 136.881637
    
    # 変換実行(カレントディレクトリの避難所.csvを使用)
    convert_csv_to_json(
        csv_file="避難所.csv",  # カレントディレクトリのCSVファイル
        output_file="nagoya_evacuation_centers.json",
        center_lat=NAGOYA_STATION_LAT,
        center_lng=NAGOYA_STATION_LNG,
        max_distance=5.0  # 5km以内
    )

実行結果

JSONファイルを生成したら、次にこのJSONファイルを読み込んでデータを管理作成します。

data_loader.py

"""
JSONファイルから避難所データを読み込むモジュール
"""
import json
import os
from typing import List, Optional
from models import Location, EvacuationCenter

class DataLoader:
    """避難所データと位置情報を管理するクラス"""
    
    def __init__(self, json_file: str = "nagoya_evacuation_centers.json"):
        """
        Args:
            json_file: 避難所データのJSONファイルパス
        """
        self.json_file = json_file
        self.evacuation_centers = []
        self.test_locations = []
        self.raw_evacuation_data = []  # 生のJSONデータを保持
        self.load_data()
    
    def load_data(self) -> None:
        """JSONファイルからデータを読み込む"""
        if not os.path.exists(self.json_file):
            print(f"警告: {self.json_file} が見つかりません。")
            print("csv_to_json_converter.py を実行してJSONファイルを生成してください。")
            return
        
        try:
            with open(self.json_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            # 生データを保持
            self.raw_evacuation_data = data.get('evacuation_centers', [])
            
            # 避難所データを読み込み
            for center_data in self.raw_evacuation_data:
                location = Location(
                    lat=center_data['location']['lat'],
                    lng=center_data['location']['lng'],
                    name=center_data['name'],
                    address=center_data['location']['address']
                )
                
                # 実データを持つEvacuationCenterを作成
                evacuation_center = EvacuationCenter(
                    name=center_data['name'],
                    location=location,
                    disaster_support=center_data.get('disaster_support', {}),
                    is_designated_shelter=center_data.get('is_designated_shelter', False)
                )
                # distance_from_centerは属性として追加
                evacuation_center.distance_from_center = center_data.get('distance_from_center', 0)
                
                self.evacuation_centers.append(evacuation_center)
            
            # テスト用位置データを読み込み
            for loc_data in data.get('test_locations', []):
                location = Location(
                    lat=loc_data['lat'],
                    lng=loc_data['lng'],
                    name=loc_data['name'],
                    address=loc_data.get('address', '')
                )
                self.test_locations.append(location)
            
            print(f"{len(self.evacuation_centers)}件の避難所データを読み込みました")
            print(f"テスト地点: {', '.join([loc.name for loc in self.test_locations])}")
            
        except Exception as e:
            print(f"エラー: JSONファイルの読み込みに失敗しました: {e}")
    
    def get_evacuation_centers(self) -> List[EvacuationCenter]:
        """避難所リストを取得"""
        return self.evacuation_centers
    
    def get_test_locations(self) -> List[Location]:
        """テスト用位置リストを取得"""
        return self.test_locations
    
    def print_evacuation_info(self):
        """避難所情報を実データのみで表示"""
        if not self.evacuation_centers:
            print("避難所データが読み込まれていません")
            return
        
        print("\n=== 避難所情報 ===")
        
        # 指定避難所の数を表示
        designated_count = sum(1 for c in self.evacuation_centers if c.is_designated_shelter)
        print(f"\n指定避難所: {designated_count}件 / 全{len(self.evacuation_centers)}件")
        
        # 災害対応統計
        disaster_stats = {
            "洪水": 0,
            "地震": 0,
            "津波": 0,
            "大規模火事": 0
        }
        
        for center in self.evacuation_centers:
            support = center.disaster_support
            if support.get('洪水'): disaster_stats['洪水'] += 1
            if support.get('地震'): disaster_stats['地震'] += 1
            if support.get('津波'): disaster_stats['津波'] += 1
            if support.get('大規模火事'): disaster_stats['大規模火事'] += 1
        
        print("\n災害対応統計:")
        for disaster, count in disaster_stats.items():
            print(f"  {disaster}対応: {count}件")
        
        # 最初の3件の避難所を表示
        print("\n最寄りの避難所(上位3件):")
        for i, center in enumerate(self.evacuation_centers[:3], 1):
            print(f"\n{i}. {center.name}")
            print(f"   住所: {center.location.address}")
            if hasattr(center, 'distance_from_center'):
                print(f"   名古屋駅から: {center.distance_from_center}km")
            
            # 対応災害を表示
            supported_disasters = [d for d, v in center.disaster_support.items() if v]
            if supported_disasters:
                print(f"   対応災害: {', '.join(supported_disasters)}")
            
            if center.is_designated_shelter:
                print("   ※指定避難所")

# グローバルインスタンスとして初期化
data_loader = DataLoader()

# エクスポート
SAMPLE_EVACUATION_CENTERS = data_loader.get_evacuation_centers()
TEST_LOCATIONS = data_loader.get_test_locations()

# テスト実行
if __name__ == "__main__":
    print("\n=== データローダーテスト ===")
    
    loader = DataLoader()
    
    # データの表示
    loader.print_evacuation_info()

プログラムの使用方法

# CSVからJSONを生成(初回およびデータ更新時のみ使用)
$ python csv_to_json_converter.py

# データローダーのテスト
$ python data_loader.py

実行結果


5. API連携機能の実装

5.1 Google Mapsで使用可能なURL生成

以下のプログラムでは、Google MapsのURLを生成する関数を実装します。クリックするだけでGoogle Mapsで経路を表示ができます。

google_maps_url.py

import urllib.parse
from typing import List, Optional
from models import Location

def generate_google_maps_url(
    origin: Location,
    destination: Location,
    mode: str = "walking",
    waypoints: Optional[List[Location]] = None
) -> str:
    """
    Google Maps URLを生成
    
    Args:
        origin: 出発地
        destination: 目的地
        mode: 移動手段 (driving, walking, bicycling, transit)
        waypoints: 経由地リスト(オプション)
    
    Returns:
        Google Maps URL
    """
    base_url = "https://www.google.com/maps/dir/"
    
    # URLパラメータの構築(座標を使用して正確な位置を指定)
    params = {
        'api': '1',
        'origin': origin.to_coords(), # 座標を使用
        'destination': destination.to_coords(), # 座標を使用
    }
    
    # 移動手段の設定
    mode_map = {
        'driving': 'driving',
        'walking': 'walking',
        'bicycling': 'bicycling',
        'transit': 'transit'
    }
    params['travelmode'] = mode_map.get(mode.lower(), 'walking')
    
    # 経由地がある場合
    if waypoints:
        waypoint_coords = [wp.to_coords() for wp in waypoints]
        params['waypoints'] = '|'.join(waypoint_coords)
    
    # URLエンコード
    query_string = urllib.parse.urlencode(params)
    
    return f"{base_url}?{query_string}"

def generate_static_map_url(
    origin: Location,
    destination: Location,
    api_key: str,
    size: str = "600x400",
    zoom: int = 14
) -> str:
    """
    Google Static Maps APIのURLを生成
    
    Args:
        origin: 出発地
        destination: 目的地
        api_key: Google Maps APIキー
        size: 画像サイズ
        zoom: ズームレベル
    
    Returns:
        Static Maps URL
    """
    base_url = "https://maps.googleapis.com/maps/api/staticmap"
    
    # パラメータ構築
    params = {
        'size': size,
        'zoom': zoom,
        'language': 'ja',
        'key': api_key,
        'markers': [
            f"color:blue|label:S|{origin.to_coords()}",
            f"color:red|label:G|{destination.to_coords()}"
        ]
    }
    
    # マーカーパラメータを構築
    markers_param = '&'.join([f"markers={marker}" for marker in params['markers']])
    
    # クエリ文字列を構築
    query_string = f"size={params['size']}&zoom={params['zoom']}&language={params['language']}&key={params['key']}&{markers_param}"
    
    return f"{base_url}?{query_string}"


# テスト用コード
if __name__ == "__main__":
    from data_loader import TEST_LOCATIONS, SAMPLE_EVACUATION_CENTERS
    import urllib.parse
    
    if TEST_LOCATIONS and SAMPLE_EVACUATION_CENTERS:
        origin = TEST_LOCATIONS[0]  # 名古屋駅
        destination = SAMPLE_EVACUATION_CENTERS[0].location  # 最寄りの避難所
        
        print("=== Google Maps URL生成テスト ===")
        print(f"出発地: {origin.name} ({origin.lat}, {origin.lng})")
        print(f"目的地: {destination.name} ({destination.lat}, {destination.lng})")
        print()
        
        # URL生成テスト
        url = generate_google_maps_url(origin, destination, mode="walking")
        print("生成されたURL(エンコード済み):")
        print(url)
        print()
        
        # URLパラメータを分解して確認
        parsed = urllib.parse.urlparse(url)
        params = urllib.parse.parse_qs(parsed.query)
        print("URLパラメータの内容:")
        for key, value in params.items():
            print(f"  {key}: {value[0]}")
        print()
        
        # 異なる移動手段でテスト
        print("=== 各移動手段でのURL ===")
        for mode in ["walking", "driving", "bicycling", "transit"]:
            url = generate_google_maps_url(origin, destination, mode=mode)
            print(f"{mode}モード:")
            print(f"  {url}")
            print()
    else:
        print("警告: nagoya_evacuation_centers.json が見つかりません。")
        print("まず csv_to_json_converter.py を実行してください。")

実行結果

生成された経路のURLの表示

5.2 Geocoding APIの実装

続いては、住所から座標、座標から住所を取得するためのGeocoding APIのクラスを実装します。

geocoding.py

import requests
from typing import Optional
from models import Location
import config

class Geocoder:
    """Geocoding APIを使用して住所と座標を相互変換"""
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or config.GOOGLE_MAPS_API_KEY
        self.base_url = "https://maps.googleapis.com/maps/api/geocode/json"
    
    def address_to_location(
        self, 
        address: str,
        region: str = "jp",  # 日本地域を優先
        bounds: str = None   # 検索範囲の制限(オプション)
    ) -> Optional[Location]:
        """
        住所から座標を取得
        
        Args:
            address: 住所文字列
            region: 地域コード(jp=日本)
            bounds: 検索範囲 "lat,lng|lat,lng" 形式(オプション)
                    例:"35.0,136.7|35.3,137.0" (名古屋周辺)
        
        Returns:
            Location オブジェクト、失敗時はNone
        """
        params = {
            'address': address,
            'key': self.api_key,
            'language': 'ja',
            'region': region  # 日本の結果を優先
        }
        
        # 検索範囲を限定(名古屋周辺など)
        if bounds:
            params['bounds'] = bounds
        
        try:
            response = requests.get(self.base_url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            if data['status'] == 'OK' and data['results']:
                result = data['results'][0]
                location = result['geometry']['location']
                
                # 複数の候補がある場合は警告
                if len(data['results']) > 1:
                    print(f"警告: '{address}' に対して{len(data['results'])}件の候補が見つかりました")
                    print(f"  使用: {result['formatted_address']}")
                
                return Location(
                    lat=location['lat'],
                    lng=location['lng'],
                    name=address,
                    address=result['formatted_address']
                )
            else:
                print(f"Geocoding API エラー: {data['status']}")
                if 'error_message' in data:
                    print(f"エラーメッセージ: {data['error_message']}")
                if data['status'] == 'ZERO_RESULTS':
                    print(f"'{address}' に一致する場所が見つかりませんでした")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"ネットワークエラー: {e}")
            return None
    
    def location_to_address(
        self, 
        lat: float, 
        lng: float,
        result_type: str = None  # 結果タイプの指定(オプション)
    ) -> Optional[str]:
        """
        座標から住所を取得(逆ジオコーディング)
        
        Args:
            lat: 緯度
            lng: 経度
            result_type: 取得する住所タイプ(street_address, route, locality等)
        
        Returns:
            住所文字列、失敗時はNone
        """
        params = {
            'latlng': f"{lat},{lng}",
            'key': self.api_key,
            'language': 'ja',
            'region': 'jp'
        }
        
        if result_type:
            params['result_type'] = result_type
        
        try:
            response = requests.get(self.base_url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            if data['status'] == 'OK' and data['results']:
                # 最も詳細な住所を返す
                return data['results'][0]['formatted_address']
            else:
                print(f"Reverse Geocoding API エラー: {data['status']}")
                if data['status'] == 'ZERO_RESULTS':
                    print(f"座標({lat}, {lng})に対応する住所が見つかりませんでした")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"ネットワークエラー: {e}")
            return None

# テスト用コード
if __name__ == "__main__":
    geocoder = Geocoder()
    
    # 住所から座標を取得
    print("=== 住所から座標への変換テスト ===")
    test_addresses = [
        "名古屋駅",
        "愛知県名古屋市中村区名駅1-1-4",
        "栄駅",
        "金山駅"
    ]
    
    for address in test_addresses:
        print(f"\n検索: {address}")
        # 名古屋周辺に限定して検索
        location = geocoder.address_to_location(
            address, 
            bounds="35.0,136.7|35.3,137.0"  # 名古屋周辺
        )
        if location:
            print(f"  座標: ({location.lat}, {location.lng})")
            print(f"  正式住所: {location.address}")
    
    # 座標から住所を取得
    print("\n=== 座標から住所への変換テスト ===")
    test_coords = [
        (35.170694, 136.881637),  # 名古屋駅
        (35.170032, 136.908687)   # 栄駅
    ]
    
    for lat, lng in test_coords:
        address = geocoder.location_to_address(lat, lng)
        if address:
            print(f"座標: ({lat}, {lng})")
            print(f"  住所: {address}")
        print()

実行結果

5.3 Directions APIの実装

次は、経路情報を取得するためのDirections APIのクラスを実装します。

directions.py

# directions.py
import requests
from typing import Optional, List
from models import Location, Route
import config

class DirectionsAPI:
    """Google Directions APIのラッパークラス"""
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or config.GOOGLE_MAPS_API_KEY
        self.base_url = "https://maps.googleapis.com/maps/api/directions/json"
    
    def get_route(
        self,
        origin: Location,
        destination: Location,
        mode: str = "walking",
        alternatives: bool = False,
        region: str = "jp"
    ) -> Optional[Route]:
        """
        Directions APIで経路を取得
        
        Args:
            origin: 出発地
            destination: 目的地
            mode: 移動手段 (driving, walking, bicycling, transit)
            alternatives: 代替ルートを取得するか
            region: 地域コード(jp=日本)
        
        Returns:
            Route オブジェクト、失敗時はNone
        """
        params = {
            'origin': origin.to_coords(),        # 座標を使用
            'destination': destination.to_coords(), # 座標を使用
            'mode': mode.lower(),
            'alternatives': str(alternatives).lower(),
            'key': self.api_key,
            'language': 'ja',
            'units': 'metric',
            'region': region
        }
        
        try:
            response = requests.get(self.base_url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            if data['status'] == 'OK' and data['routes']:
                route_data = data['routes'][0]
                leg = route_data['legs'][0]
                
                # Google Maps URLを生成
                from google_maps_url import generate_google_maps_url
                maps_url = generate_google_maps_url(origin, destination, mode)
                
                return Route(
                    origin=origin,
                    destination=destination,
                    distance=leg['distance']['value'],
                    duration=leg['duration']['value'],
                    distance_text=leg['distance']['text'],
                    duration_text=leg['duration']['text'],
                    travel_mode=mode,
                    google_maps_url=maps_url
                )
            else:
                print(f"Directions API エラー: {data['status']}")
                if data['status'] == 'ZERO_RESULTS':
                    print("ルートが見つかりませんでした")
                    print(f"  起点: {origin.name or origin.to_coords()}")
                    print(f"  終点: {destination.name or destination.to_coords()}")
                elif data['status'] == 'OVER_QUERY_LIMIT':
                    print("APIの利用制限を超えました")
                elif data['status'] == 'REQUEST_DENIED':
                    print("リクエストが拒否されました。APIキーを確認してください")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"ネットワークエラー: {e}")
            return None
    
    def get_multiple_routes(
        self,
        origin: Location,
        destinations: List[Location],
        mode: str = "walking"
    ) -> List[Optional[Route]]:
        """
        複数の目的地への経路を一括取得
        
        Args:
            origin: 出発地
            destinations: 目的地リスト
            mode: 移動手段
        
        Returns:
            Route オブジェクトのリスト
        """
        routes = []
        
        for destination in destinations:
            route = self.get_route(origin, destination, mode)
            routes.append(route)
        
        return routes


# テスト用コード
if __name__ == "__main__":
    from data_loader import TEST_LOCATIONS, SAMPLE_EVACUATION_CENTERS
    
    if TEST_LOCATIONS and SAMPLE_EVACUATION_CENTERS:
        api = DirectionsAPI()
        
        origin = TEST_LOCATIONS[0]  # 名古屋駅
        destination = SAMPLE_EVACUATION_CENTERS[0].location
        
        print("=== 単一経路の取得テスト ===")
        for mode in ["walking", "driving"]:
            print(f"\n{mode}モード:")
            route = api.get_route(origin, destination, mode=mode)
            
            if route:
                print(f"  出発地: {origin.name}")
                print(f"  目的地: {destination.name}")
                print(f"  距離: {route.distance_text}")
                print(f"  時間: {route.duration_text}")
                print(f"  URL: {route.google_maps_url}")
        
        print("\n=== 複数経路の取得テスト ===")
        destinations = [center.location for center in SAMPLE_EVACUATION_CENTERS[:3]]
        routes = api.get_multiple_routes(origin, destinations, mode="walking")
        
        for i, route in enumerate(routes):
            if route:
                print(f"\n経路{i+1}: {destinations[i].name}")
                print(f"  距離: {route.distance_text}")
                print(f"  時間: {route.duration_text}")
    else:
        print("警告: nagoya_evacuation_centers.json が見つかりません。")
        print("まず csv_to_json_converter.py を実行してください。")

実行結果

生成された経路のURLの表示


6. 避難所検索機能の実装

evacuation_route_finder.py

import math
from typing import List, Tuple, Optional
import json
from datetime import datetime

from models import Location, EvacuationCenter, Route
from geocoding import Geocoder
from directions import DirectionsAPI
from google_maps_url import generate_google_maps_url
from data_loader import SAMPLE_EVACUATION_CENTERS  # data_loaderを使用
import config

class EvacuationRouteFinder:
    """避難所検索と経路生成のメインクラス"""
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or config.GOOGLE_MAPS_API_KEY
        self.geocoder = Geocoder(self.api_key)
        self.directions_api = DirectionsAPI(self.api_key)
        self.evacuation_centers = SAMPLE_EVACUATION_CENTERS
    
    def calculate_distance(self, loc1: Location, loc2: Location) -> float:
        """2点間の直線距離を計算(km)"""
        R = 6371  # 地球の半径(km)
        
        lat1, lon1 = math.radians(loc1.lat), math.radians(loc1.lng)
        lat2, lon2 = math.radians(loc2.lat), math.radians(loc2.lng)
        
        dlat = lat2 - lat1
        dlon = lon2 - lon1
        
        a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
        c = 2 * math.asin(math.sqrt(a))
        
        return R * c
    
    def find_nearest_centers(
        self,
        origin: Location,
        limit: int = 3
    ) -> List[Tuple[EvacuationCenter, float]]:
        """最寄りの避難所を探す"""
        centers_with_distance = []
        
        for center in self.evacuation_centers:
            distance = self.calculate_distance(origin, center.location)
            centers_with_distance.append((center, distance))
        
        # 距離でソート
        centers_with_distance.sort(key=lambda x: x[1])
        
        return centers_with_distance[:limit]
    
    def get_location_from_address(self, address: str) -> Optional[Location]:
        """住所から位置情報を取得"""
        return self.geocoder.address_to_location(address)
    
    def generate_evacuation_report(
        self,
        origin: Location,
        limit: int = 3,
        modes: List[str] = None
    ) -> dict:
        """避難経路レポートを生成"""
        if modes is None:
            modes = ["walking", "driving"]
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'origin': {
                'lat': origin.lat,
                'lng': origin.lng,
                'name': origin.name or f"座標({origin.lat}, {origin.lng})",
                'address': origin.address
            },
            'evacuation_centers': []
        }
        
        # 最寄りの避難所を取得
        nearest_centers = self.find_nearest_centers(origin, limit)
        
        print(f"\n{'='*60}")
        print(f"現在地: {origin.name or origin.address or f'({origin.lat}, {origin.lng})'}")
        print(f"{'='*60}")
        
        # デバッグ情報を追加
        print(f"\n最寄りの避難所(直線距離):")
        for center, dist in nearest_centers:
            print(f"  - {center.name}: {dist:.2f}km")
            print(f"    座標: ({center.location.lat}, {center.location.lng})")
        
        for center, straight_distance in nearest_centers:
            print(f"\n{center.name}への経路を計算中...")
            
            center_info = {
                'name': center.name,
                'location': {
                    'lat': center.location.lat,
                    'lng': center.location.lng,
                    'address': center.location.address
                },
                'straight_distance_km': round(straight_distance, 2),
                'disaster_support': center.disaster_support if hasattr(center, 'disaster_support') else {},
                'is_designated_shelter': center.is_designated_shelter if hasattr(center, 'is_designated_shelter') else False,
                'routes': {}
            }
            
            # 各移動手段での経路を取得
            for mode in modes:
                route = self.directions_api.get_route(
                    origin,
                    center.location,
                    mode=mode
                )
                
                if route:
                    center_info['routes'][mode] = {
                        'distance': route.distance_text,
                        'duration': route.duration_text,
                        'google_maps_url': route.google_maps_url
                    }
                    print(f"  {mode}: {route.distance_text}, {route.duration_text}")
            
            report['evacuation_centers'].append(center_info)
        
        return report
    
    def print_report(self, report: dict):
        """レポートを見やすく出力"""
        print(f"\n{'='*60}")
        print("避難経路レポート")
        print(f"{'='*60}")
        print(f"作成日時: {report['timestamp']}")
        print(f"現在地: {report['origin']['name']}")
        
        for i, center in enumerate(report['evacuation_centers'], 1):
            print(f"\n{'-'*60}")
            print(f"【{i}. {center['name']}】")
            print(f"  住所: {center['location']['address']}")
            print(f"  直線距離: 約{center['straight_distance_km']}km")
            
            # 災害対応情報を表示
            if center.get('disaster_support'):
                supported = [d for d, v in center['disaster_support'].items() if v]
                if supported:
                    print(f"  対応災害: {', '.join(supported)}")
            
            if center.get('is_designated_shelter'):
                print("  ※指定避難所")
            
            if 'walking' in center['routes']:
                route = center['routes']['walking']
                print(f"\n  徒歩:")
                print(f"    距離: {route['distance']}")
                print(f"    時間: {route['duration']}")
                print(f"    地図URL:")
                print(f"    {route['google_maps_url']}")
            
            if 'driving' in center['routes']:
                route = center['routes']['driving']
                print(f"\n  車:")
                print(f"    距離: {route['distance']}")
                print(f"    時間: {route['duration']}")
                print(f"    地図URL:")
                print(f"    {route['google_maps_url']}")
        
        print(f"\n{'='*60}")
    
    def save_report(self, report: dict, filename: str = None):
        """レポートをJSONファイルとして保存"""
        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"outputs/evacuation_report_{timestamp}.json"
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        
        print(f"レポートを {filename} に保存しました")
        return filename
    
    def find_evacuation_route(
        self,
        address: str = None,
        lat: float = None,
        lng: float = None,
        limit: int = 3
    ) -> dict:
        """避難経路を検索するメインメソッド"""
        # 現在地を特定
        if address:
            print(f"住所から現在地を検索中: {address}")
            origin = self.get_location_from_address(address)
            if not origin:
                print("エラー: 住所が見つかりませんでした")
                return None
        elif lat is not None and lng is not None:
            origin = Location(lat, lng, f"座標({lat}, {lng})")
        else:
            # デフォルト位置(名古屋駅)
            origin = Location(
                config.DEFAULT_LOCATION['lat'],
                config.DEFAULT_LOCATION['lng'],
                "デフォルト位置(名古屋駅)"
            )
        
        # レポート生成
        report = self.generate_evacuation_report(origin, limit)
        
        # 結果を表示
        self.print_report(report)
        
        # ファイルに保存
        self.save_report(report)
        
        return report

def main():
    """メイン処理"""
    print("避難経路検索システム")
    print("-" * 60)
    
    # 避難所データの確認
    if not SAMPLE_EVACUATION_CENTERS:
        print("エラー: 避難所データが読み込まれていません")
        print("csv_to_json_converter.py を実行してJSONファイルを生成してください")
        return
    
    print(f"{len(SAMPLE_EVACUATION_CENTERS)}件の避難所データを使用")
    
    # 避難経路検索システムの初期化
    finder = EvacuationRouteFinder()
    
    # ユーザー入力を取得
    print("\n現在地を入力してください:")
    print("1. 住所で検索")
    print("2. 座標で検索")
    print("3. 名古屋駅を使用")
    
    choice = input("\n選択 (1-3): ").strip()
    
    if choice == "1":
        address = input("住所を入力: ").strip()
        if address:
            finder.find_evacuation_route(address=address)
        else:
            print("住所が入力されませんでした")
    
    elif choice == "2":
        try:
            lat = float(input("緯度を入力: "))
            lng = float(input("経度を入力: "))
            finder.find_evacuation_route(lat=lat, lng=lng)
        except ValueError:
            print("無効な座標です")
    
    else:
        # デフォルト位置である名古屋駅を使用
        finder.find_evacuation_route()
    
    print("\n" + "="*60)
    print("検索完了!上記のURLをブラウザで開くと経路が表示されます")
    print("="*60)

if __name__ == "__main__":
    main()

7. 実行とテスト

システムの実行

# 仮想環境が有効化されていることを確認
# $ source venv/bin/activate
# CSVからJSONを生成(初回のみ)
# $ python csv_to_json_converter.py

# メインプログラムを実行
$ python evacuation_route_finder.py

動作確認

正常に動作している場合、以下のような出力になります:

選択肢3(デフォルト位置)を選んだ場合の正しい出力例

最寄りの避難所(直線距離):
  - 則武コミュニティセンター: 0.42km
  - 牧野小学校: 0.46km
  - 新明コミュニティセンター: 0.53km

選択肢1の実行結果

生成された経路のURLの表示

対話モードでの実行

$ python

>>> from evacuation_route_finder import EvacuationRouteFinder
>>> finder = EvacuationRouteFinder()
>>> 
>>> # 名古屋駅から検索
>>> finder.find_evacuation_route(address="名古屋駅")
>>> 
>>> # 座標から検索
>>> finder.find_evacuation_route(lat=35.170694, lng=136.881637)

おわりに

これで、名古屋駅周辺の実際の避難所データを使用した経路生成のプログラムが完成しました! さすがにいつもの倍の分量だったので大変でした🤩

補足資料

作成したシステムの概要

  • 実際のCSVデータから避難所情報を読み込み
  • 災害対応情報や指定避難所の情報を含む
  • Google Maps URLを生成(座標ベースで正確)
  • JSONファイルでデータを管理

使用するデータの内容

  • 施設・場所名 … 実際の避難所名
  • 住所 … 実際の住所
  • 座標 … 実際の緯度・経度
  • 災害対応情報 … 洪水、地震津波、大規模火事など
  • 指定避難所フラグ … 指定避難所かどうか
  • 名古屋駅からの距離 … 0.42km〜5km以内の避難所

プロジェクトのツリー構造

evacuation-route-system/
├── venv/                       # 仮想環境
├── outputs/                    # 出力ファイル
│   └── evacuation_report_*.json
├── .env                        # 環境変数(APIキー)
├── config.py                   # 設定ファイル
├── models.py                   # データモデル
├── csv_to_json_converter.py    # CSV→JSON変換
├── nagoya_evacuation_centers.json  # 避難所データ
├── data_loader.py              # データローダー
├── google_maps_url.py          # URL生成
├── geocoding.py                # Geocoding API
├── directions.py               # Directions API
├── evacuation_route_finder.py  # メインプログラム
├── 避難所.csv                  # 避難所の元データ
└── requirements.txt            # 依存パッケージ