文学フリマ38に出す本に載せられなかった乗換経路検索プログラム公開

概要

仏印フラスコというサークル名で文学フリマ38に出店します。

題材として乗換案内APIの利用を取り上げ、そもそもAPIとはなんぞやというところから初心者が実際のコードを組んで動かすところまでの一連の流れを本にして出す予定です。

文学なのにプログラミングの話でいいのか？という葛藤はありましたが、公式HPに

文学フリマとは、文学作品の展示即売会です。
出店者が「自分が〈文学〉と信じるもの」を自らの手で販売します。
作品の内容は、小説・物語・詩・俳句・短歌・ノンフィクション・エッセイほか、評論・研究書など多岐にわたり、対象年齢やジャンルも実にさまざまです。

とのことなのでこれは広義の文学です。

↓サークルページです
仏印フラスコ

新書サイズの本にプログラムの全体を掲載することはできなかったので、こちらに公開します。

①駅すぱあとAPIを利用してユーザーが入力した駅情報から経路探索を行い、②特急列車や新幹線を利用した場合の費用対効果を計算し、③ターミナルに表示する流れになっています。

駅情報取得プログラムは、kindleにて駅すぱあとの提供者であるヴァル研究所様が公開しているAPIキーを入力することで動作確認ができます。

はじめての駅すぱあとWebサービス

もっと整理できそうですし細かいバグ取りまで終わっていないのですが、締め切りもあり一通りやりたいことができたのでこれでいったん完成としました。

ご興味おありの方はぜひスペースまでお越しください。
お待ちしております。

駅情報取得プログラム

import requests
import urllib.parse
import pandas as pd

# ユーザーから駅名を入力してもらう
station_name = input("駅名を入力してください: ")

# APIのURLを構築する
base_url = "https://api.ekispert.jp/v1/json/station/light"
api_key = "LE_XuW67vBhQfSWd"
params = {
    "key": api_key,
    "name": station_name
}
URL = f"{base_url}?{urllib.parse.urlencode(params)}"

# APIにリクエストを送信し、結果をresultに格納する
res = requests.get(URL)
result = res.json()

# URLと検索結果を表示
print(URL)
print()
print(result)

経路探索プログラム全体

import json
import requests
import urllib.parse
import pandas as pd

# エンドポイントのURL構築に必要な種々パラメータ
base_url = "https://api.ekispert.jp/v1/json/station/light"
api_key_free = "無料APIキー"
api_key = "有料APIキー"
transfer_type = "train"

# 駅情報取得
def get_station_info(station_name):
    params = {"key": api_key_free, "name": station_name, "type": transfer_type}
    url = f"{base_url}?{urllib.parse.urlencode(params)}"
    res = requests.get(url)
    result = res.json()
    points = result.get("ResultSet", {}).get("Point", [])
    if not isinstance(points, list):
        points = [points]
    return pd.DataFrame([{**point["Station"]} for point in points])

# 経路探索
def search_route(selected_departure_code, selected_arrival_code, shinkansen_usage, limitedExpress_usage, bus_usage, plane_usage):
    search_url = "https://api.ekispert.jp/v1/json/search/course/plain"
    search_params = {
        "key": api_key,
        "from": selected_departure_code,
        "to": selected_arrival_code,
        "bus": bus_usage,
        "shinkansen": shinkansen_usage,
        "limitedExpress": limitedExpress_usage,
        "plane": plane_usage
    }
    search_url = f"{search_url}?{urllib.parse.urlencode(search_params)}"
    search_res = requests.get(search_url)
    if search_res.status_code == 200:
        return search_res.json().get("ResultSet", {}).get("Course", [])
    else:
        print(f"Error: HTTP status code {search_res.status_code}")
        return []

# レスポンスから必要な情報を抽出
def extract_route_information(course):
    serialize_data = course.get("SerializeData", "")
    stations = course.get("Route", {}).get("Point", [])
    railways = course.get("Route", {}).get("Line", [])
    if not isinstance(stations, list):
        stations = [stations]
    if not isinstance(railways, list):
        railways = [railways]
    railway_name = [railway["Name"] for railway in railways]

    search_url_passstation = "https://api.ekispert.jp/v1/json/course/passStation"
    search_params_passstation = {"key": api_key, "serializeData": serialize_data}
    search_url_passstation = f"{search_url_passstation}?{urllib.parse.urlencode(search_params_passstation)}"
    search_res_passstation = requests.get(search_url_passstation)
    if search_res_passstation.status_code == 200:
        pass_stations = search_res_passstation.json().get("ResultSet", {}).get("Point", [])
        if not isinstance(pass_stations, list):
            pass_stations = [pass_stations]
        transfer_stations = [pass_station["Station"]["Name"] for pass_station in pass_stations]
    else:
        print(f"Error: HTTP status code {search_res_passstation.status_code}")
        transfer_stations = []

    time_on_board = int(course.get("Route", {}).get("timeOnBoard", 0))
    time_other = int(course.get("Route", {}).get("timeOther", 0))
    total_time = time_on_board + time_other

    fares = course.get("Price", [])
    fare_info = {}
    fare_ticket = 0
    fare_ticket_special = 0
    for fare in fares:
        if fare.get("kind") == "FareSummary":
            fare_name = fare["Oneway"]
            fare_amount = int(fare["Oneway"])
            fare_info[fare_name] = fare_amount
            fare_ticket += fare_amount
        if fare.get("Name") in ("座席未指定料金(事前購入時)", "指定席"):
            fare_name_special = fare["Oneway"]
            fare_amount_special = int(fare["Oneway"])
            fare_info[fare_name_special] = fare_amount_special
            fare_ticket_special += fare_amount_special

    return {
        "RailwayName": railway_name,
        "TotalTime": total_time,
        "TimeOnBoard": time_on_board,
        "TimeOther": time_other,
        "TicketPrice": fare_ticket,
        "SpecialTicketPrice": fare_ticket_special,
        "total_fare": fare_ticket + fare_ticket_special,
        "transfer_stations": transfer_stations
    }

# 入力欄の表示
def select_station(choices):
    print("選択肢:")
    for i, choice in enumerate(choices):
        print(f"{i+1}. {choice}")
    choice = int(input("番号を選択してください: "))
    selected_station = choices[choice - 1]
    return selected_station


# 出発駅名と到着駅名の入力
departure_station = input("出発駅名を入力してください: ")
arrival_station = input("到着駅名を入力してください: ")

# 駅情報の取得
departure_df = get_station_info(departure_station)
arrival_df = get_station_info(arrival_station)

# 出発駅と到着駅の選択
selected_departure_station = select_station(departure_df['Name'].tolist())
selected_arrival_station = select_station(arrival_df['Name'].tolist())

# 選択された出発駅と到着駅のコードを取得
selected_departure_code = departure_df.loc[departure_df['Name'] == selected_departure_station, 'code'].iloc[0]
selected_arrival_code = arrival_df.loc[arrival_df['Name'] == selected_arrival_station, 'code'].iloc[0]

# 経路検索1回目
routes = search_route(selected_departure_code, selected_arrival_code, "false", "false", "false", "false")

# 結果をデータフレームに変換
result_df = pd.DataFrame([extract_route_information(route) for route in routes])

# result_dfにshinkansen_usage, limitedExpress_usage列を追加し検索条件を記録
result_df["新幹線"] = "false"
result_df["特急"] = "false"

# 経路検索2回目
routes2 = search_route(selected_departure_code, selected_arrival_code, "false", "true", "false", "false")

# 結果をresult_dfに追加
result_df2 = pd.DataFrame([extract_route_information(route) for route in routes2])
result_df2 = result_df2.sort_values("TotalTime")
result_df2["新幹線"] = "false"
result_df2["特急"] = "true"
result_df = pd.concat([result_df, result_df2], ignore_index=True)

# 経路検索3回目
routes3 = search_route(selected_departure_code, selected_arrival_code, "true", "true", "false", "true")

# 結果をresult_dfに追加
result_df3 = pd.DataFrame([extract_route_information(route) for route in routes3])
result_df3 = result_df3.sort_values("TotalTime")
result_df3["新幹線"] = "true"
result_df3["特急"] = "true"
result_df = pd.concat([result_df, result_df3], ignore_index=True)

# result_dfからtotal_fareが同じものを一つにまとめる
result_df = result_df.drop_duplicates(subset="total_fare")

# result_dfから運賃の最小値を取得
min_fare = result_df["total_fare"].min()

# result_dfから運賃が最安の時の所要時間を取得
min_fare_time = result_df[result_df["total_fare"] == result_df["total_fare"].min()]["TotalTime"].iloc[0]

# 所要時間短縮の費用対効果を計算
result_df["FareDiff"] = result_df["total_fare"] - min_fare
result_df["TimeDiff"] = min_fare_time - result_df["TotalTime"] 
result_df["PricePerMin"] = result_df["FareDiff"] / result_df["TimeDiff"]
result_df = result_df.round({"PricePerMin": 2})

# 結果表示
result_df = pd.DataFrame(result_df, columns=["TotalTime", "total_fare", "PricePerMin", "TicketPrice", "SpecialTicketPrice", "新幹線", "特急", "RailwayName"])

print("")
print(result_df.sort_values("TotalTime"))