鹿と列車の衝突を調べる
「ゆるWeb勉強会@札幌 2022年アドベントカレンダー」の15日目の記事ということにしました。
成果物 (Streamlit)
イメージ
実装
要素技術
- Twitter API
- 鉄道駅の緯度経度情報
- PyDeck
背景
- JR北海道では、鹿(エゾシカ)をはじめとする野生動物との接触事故が頻繁に起きています。 [1]
- ところで私はリモートワークで、まだ今の上司と直接会ったことがありません。近々会いにいらっしゃることになりました。
- 私は1回の特急乗車中に2回鹿ヒットを記録したことがあります。ボスの道中の鹿ヒット率に思いを馳せるため、過去の事故の位置をプロットしてみました。
データソース
JR北海道公式ページやTwitterからは見つけられませんでした。列車運行情報ページには情報が載りますが、リアルタイムに出ては消えていき、過去を遡れないようです。
今回はJR北海道 在来線運行情報【非公式】(@JRHBot) のツイートを活用させていただきました。おそらく上記列車運行情報が情報元と思われますが、それを過去に遡れます。
実装
過去の運行情報をTwitter APIで取得
Twitter APIを使って、運行情報ツイートを取得します。
APIキーの確保
利用登録の方法はたくさん情報がありますので省略します。
4つのキーが得られますので、StreamlitのSecretとして登録します（参考）。
ローカル開発時は、.streamlit/secrets.toml というファイル名のテキストファイルに記述します。（即刻.gitignoreに入れてください！）
[twitter]
api_key = "..."
api_secret_key = "..."
access_token = "..."
access_token_secret_key = "..."
ツイートの取得
Twitter APIのドキュメント: https://developer.twitter.com/ja/docs
tweepy を使用しました。
pip install tweepy
上でSecretに設定したAPIキーを参照しながら、ツイートを取得します。
一度に多くの結果を得る際はCursorが便利です: https://kurozumi.github.io/tweepy/cursor_tutorial.html
from dataclasses import dataclass
import tweepy
import streamlit as st
@dataclass(frozen=True)
class Tweet:
id: str
created_at: str
text: str
@st.experimental_memo
def get_tweets_by_timeline() -> list[Tweet]:
auth = tweepy.OAuthHandler(st.secrets.twitter.api_key, st.secrets.twitter.api_secret_key)
auth.set_access_token(st.secrets.twitter.access_token, st.secrets.twitter.access_token_secret_key)
api = tweepy.API(auth)
result = []
for status in tweepy.Cursor(api.user_timeline, id="JRHbot").items(5000):
result.append(Tweet(id=status.id, created_at=status.created_at.isoformat(), text=status.text))
return result
@st.experimental_memo により、WebAPIに何度もアクセスするのを防いでキャッシュします [2]。Streamlit開発の鉄則として、無駄なAPIアクセスを防ぐため真っ先にデータ取得メソッドの仕様を固めるのが良いです。複雑な処理は極力そこには入れず、後段に譲ります。
ちなみに、はじめに試したのはタイムラインの取得(user_timeline)ではなく、ツイートの検索(search_tweets)のほうでした。コードを示します。
@st.experimental_memo
def get_tweets_by_search() -> list[Tweet]:
auth = tweepy.OAuthHandler(st.secrets.twitter.api_key, st.secrets.twitter.api_secret_key)
auth.set_access_token(st.secrets.twitter.access_token, st.secrets.twitter.access_token_secret_key)
api = tweepy.API(auth)
result = []
for status in tweepy.Cursor(api.search_tweets, q="接触 (from:JRHbot)").items(1000):
result.append(Tweet(id=status.id, created_at=status.created_at.isoformat(), text=status.text))
return result
ただし、検索だと標準では7日前までしか遡れないようで、断念しました。鹿の衝突はかなり多いので、普通にタイムラインを遡るだけでも次々に見つけられます。https://developer.twitter.com/en/docs/twitter-api/v1/tweets/search/overview
取得したデータの例
ツイート日時は、tweepyで返るdatetimeだとpickableでないためキャッシュ不能とのことでstrにしています。
tweets = get_tweets_by_timeline()
print(tweets)
[Tweet(id="1603676472870178816", created_at="2022-12-16T10:26:36.234012", text="【鹿との接触による列車への影響について】\n本日（１２／１６）、石北線 伊香牛～愛別駅間において、旭川 １６時３１分発 上川行き 普通列車が鹿と接触した影響により、一部列車に遅れが発生しています。\nhttps://www3.jrhokkaido.co.jp/webunkou/"),
...]
参照したツイート: https://twitter.com/JRHbot/status/1603676472870178816
駅名から緯度経度を得る
ツイートを取得できました。障害があった区間（AA駅～BB駅）が本文にあります。
地図にプロットするには、その駅の緯度経度を求めます。今回は鉄道駅LODの情報を参照させて頂きました。
ここからデータを取得するには、以下の方法が便利です。路線単位でGeoJSONをダウンロードすると、駅ごとの緯度経度が一括で得られます。今回は北海道新幹線以外の13路線分を1つ1つダウンロードしました[3]。
得られるGeoJSONは以下のような形式で、"features"以下に、同じ駅名で2回ずつ出現します。geometry.typeがLineStringのものとPointのものがあります。
GeoJSONの例 (函館本線)
{
"type":"FeatureCollection",
"features":[
{
"properties":{
"name":"函館",
"uri":"https://uedayou.net/jrslod/北海道旅客鉄道/函館本線/函館",
"color":"2CB431"
},
"type":"Feature",
"geometry":{
"type":"LineString",
"coordinates":[
[
140.72572,
41.77366
],
[
140.72597,
41.77372
],
...
]
}
},
{
"properties":{
"name":"函館",
"uri":"https://uedayou.net/jrslod/北海道旅客鉄道/函館本線/函館"
},
"type":"Feature",
"geometry":{
"type":"Point",
"coordinates":[
140.72655,
41.77402
]
}
},
{
"properties":{
"name":"五稜郭",
...
駅の位置はPointの方のみを選んでいけば良さそうです。GeoJSON力に自信が無いので、早々にばらして必要なデータだけjqで集めてみます。CSVとして出力しました。
cat 北海道旅客鉄道函館本線.geojson | \
jq -r '.features[] | select(.geometry.type=="Point") | [.properties.name, .geometry.coordinates[1], .geometry.coordinates[0]] | @csv'> station_locations.csv
"函館",41.77402,140.72655
"五稜郭",41.80234,140.73386
"桔梗",41.84748,140.72246
"大中山",41.86412,140.71397
"七飯",41.88775,140.68786
"新函館北斗",41.904844,140.648815
...
このCSVを駅名から座標への変換データベースとします。Pythonで読み込んで辞書として保持します。上の手順を全路線分行ってCSVを連結し、ヘッダ行を手で追加した前提です: station,lat,lon
import csv
from dataclasses import dataclass
@dataclass(frozen=True)
class Location:
lat: float
lon: float
def get_station_locations() -> dict[str, Location]:
with open("station_locations.csv", "r", encoding="utf-8-sig", newline="") as file:
csv_reader = csv.DictReader(file)
result = {}
for row in csv_reader:
result[row["station"]] = Location(float(row["lat"]), float(row["lon"]))
return result
ツイート（運行情報）文字列から、駅と動物を抽出
石北線 伊香牛～愛別駅間において、旭川 １６時３１分発 上川行き 普通列車が鹿と接触した影響により、一部列車に遅れが発生しています。
ツイート本文からこのような文字列を得られるので、駅の区間やぶつかった動物を抽出していきます。
ここは泥臭くがんばった結果をそのまま貼って終わります。
from dataclasses import dataclass
import re
@dataclass(frozen=True)
class Appearance:
date: str # 日付 YYYY-mm-dd
sections: list[tuple[str, str]] # [("AA駅", "BB駅"), ...)]
reason: str # object+述語 (鹿と衝突)
object: str # 何とぶつかった？
train: str # 列車種別
def extract_appearance(tweets: Iterable[Tweet]) -> Iterable[Appearance]:
for tweet in tweets:
for text in re.split("ならびに", tweet.text):
text = re.sub("および|及び|、", " & ", text)
text = re.sub("間と", "間 & ", text)
if match := re.search(r"(?P<train>貨物列車|列車|特急\S+|快速\S*?|はこだてライナー)(が|は.+?(で|において))(?P<reason>(?P<object>\S+)(と|を|の)(接触|衝突|衝撃|発見|巻き込んで))", text):
reason, object, train = match.group("reason", "object", "train")
else:
continue
train = "普通列車" if train == "列車" else train
sections: list[tuple[str, str]] = []
for match in re.finditer(r"(?P<st1>\S+?)駅?(～|\~|\-|－|(?<!ビ)ー(?!ル))(?P<st2>\S+?)[駅席]?(間で|間にて|間?(付近)?において|間の.+?踏切)", text):
sections.append(match.group("st1", "st2"))
if not sections:
for match in re.finditer(r"(?P<st>\S+?)(駅構内|駅?付近において)", text):
sections.append((match.group("st"), ""))
if not sections:
continue
date_str = datetime.fromisoformat(tweet.created_at).strftime("%Y-%m-%d")
yield Appearance(date_str sections, reason, object, train)
解釈が難しかったものの例
「ならびに」で2件まとめて示されている例です。
1回の運行中に2回ヒットしたと思しき例です。
文章表現は都度微妙に異なることがあります。「巻き込んで」とはどんな感じだったのでしょうか・・・。
「AA駅～BB駅」という区間だけでなく、特定の駅ピンポイントで示されることもあります。ぶつかるのは旅客列車だけではありません。
対応しなかった例
路線名のみで具体的な駅が不明なケースが相当数ありました。今回はプロットに適さないとして除外しました。
地図にプロット
前回の記事ではstreamlit-foriumを使いましたが、今回はst.pydeck_chartにします。pydeck (deck.gl) を利用します。
これまでに集めたデータをpandasのDataFrameにまとめて、pydeckに与えます。雑な実装です。
-
get_position=["lon", "lat"]で、各アイコンの緯度・経度情報をDataFrameの列名で与えます（順番注意）。 -
tooltip={"text": "{text}"}は、text列をツールチップ（マウスカーソルが乗ったときの表示）として指定しています。 -
ICON_DATAにより、アイコンとして表示する画像を与えます。以下のコードでは固定値ですが、完成品ではぶつかった物（鹿・熊・鳥・車等々）により分岐させてみました。 -
initial_view_state=で初期状態の地図の視点を決められます。今回は2次元的に見るため真上から見ています(pitch=0)が、pitchを大きくすると斜めからの視点になります。(60くらいが限界？)
ICON_DATA = {
"url": "https://raw.githubusercontent.com/shimat/deer_appearance/ec75280793daa24d76b2ed591ca1125bfd1877a6/image/animal_deer.png",
"width": 305,
"height": 400,
"anchorY": 400,
}
station_locations = get_station_locations()
tweets = get_tweets()
appearances = list(extract_appearance(tweets))
rows = []
for a in appearances:
for s in a.sections:
if s[1] == "":
lat, lon = station_locations[s[0]].to_tuple()
place = f"{s[0]}駅"
else:
lat, lon = Location.midpoint(station_locations[s[0]], station_locations[s[1]]).to_tuple()
place = f"{s[0]} ～ {s[1]} 駅間"
text = f"{place}\n{a.date} {a.train}\n{a.reason}"
rows.append((lat, lon, text, a.date, ICON_DATA))
data = pd.DataFrame(
rows,
columns=["lat", "lon", "text", "date", "icon_data"])
st.pydeck_chart(pdk.Deck(
map_style="light",
initial_view_state=pdk.ViewState(
latitude=43.6,
longitude=142.7,
zoom=6,
pitch=0,
),
layers=[
pdk.Layer(
"IconLayer",
data=data,
get_icon="icon_data",
get_size=4,
size_scale=15,
get_position=["lon", "lat"],
pickable=True,
),
],
tooltip={"text": "{text}"}
))
2駅間の区間で示される場合は、その間の地点で代表させることにしました。地理情報的にはたぶん雑な計算です。
@dataclass(frozen=True)
class Location:
# (略)
@staticmethod
def midpoint(a: "Location", b: "Location") -> "Location":
new_lat = (a.lat + b.lat) / 2
new_lon = (a.lon + b.lon) / 2
return Location(new_lat, new_lon)
なお現実には、ツイート(元のJR北海道の情報源?)にはたまに誤記があって、情報抽出や座標解決に失敗したりします。ちまちまこちらで対応するしかないですね。
例:) 崎守席 -> 崎守駅？
おわりに
今一度結果の地図を眺めながら感想を述べておわりにします。 (Streamlit)
集められたツイートからの集計期間が297日で、プロットできたのが445件です。取り漏らしたケースを含めるとおそらく、1日に2件弱の衝突があるペースと言えます。多いですね。
まず室蘭～苫小牧が目立ちます。私が2回ヒットを記録したのもこの区間[4]で、運行本数が多いのは理由としてありそうです。線形がまっすぐでスピードが出る区間なので、見つけてもブレーキが間に合わない、といったこともあるのでしょうか。
道北（名寄以北）や道東（釧路・根室）は、よく報道でも取り上げられている印象で、運行の少なさの割にとても多いですね。そのほかもどこでもよくぶつかっています。お疲れ様です・・・(to人間)＆かわいそうですね(to鹿)。
札幌はどちらかというと乗用車など鹿ではない事故が記録されていますが、一番近くて白石～厚別駅間で鹿ヒットがあり、全く他人事ではありません。
十勝の真ん中や、旭川から名寄にかけてなど、空白地帯があります。山が遠く、一面畑や田んぼの平地の中を進むので、鹿が出にくいのでしょうか。
鹿以外に、熊(ヒグマ)とのヒットがざっと10件くらい(2～3%くらい)あり、他にも鳥、小動物、乗用車、倒木といったものも検出しました。JR北海道の資料によると、鹿は乗務員さんが人力でどけるそうで[5]、熊は専用のクレーンで吊ってどけるようです。なかなか図のインパクトがあるので、ぜひリンクからご一読ください。
Discussion