Streamlitで国交省直轄工事のダッシュボード作ってみた

本投稿は、
国土交通データプラットフォーム Advent Calendar 2023 19日目の記事です。

---

国土交通省のデータプラットフォームでは、国土交通省が発注している工事の基本情報が、APIを通じて提供されています。このAPIを使って、工事名、緯度・経度、工事の概要、工事種別、発注者、受注者、完成年度などの情報を取得して、地図やグラフにして表示してみようと思います。

ダッシュボードのイメージ

実際に作ったアプリケーションは、Streamlitに公開しています。

---

直轄工事の基本情報とは

国交省が道路や河川等の工事を発注しますが、その工事を受注した事業者は工事の完成時に、図面や３Dモデルなどの成果物を発注者に納品する必要があります。成果品は、国交省が定める納品要領「工事完成図書の電子納品等要領（R5.3）」に従って作成され、成果品に含まれる「工事管理ファイル（Index_c.xml）」には、工事の属性情報（工事名や工事内容、緯度経度など）が含まれています。

工事完成図書の電子納品等要領の 工事管理項目（抜粋

国土交通省のデータプラットフォーム（以下、国土交通DPF）では、工事成果品が集められた「電子納品保管管理システム」と連携されており、成果品の一部（工事管理ファイルとIFCモデル）が公開されています。

国土交通DPFで検索した工事管理ファイルのデータを表示

国土交通DPFのAPIを使ってデータ取得する

まず、国土交通DPFから直轄工事の工事管理ファイルデータをAPIを使用して取得します。
例えば、データセットが「工事管理ファイル」かつ都道府県が「千葉県」を検索するクエリは、以下のようになります。

# GraphQL query
query {
    search(
        term: ""
        first: 0
        size: 10000
        attributeFilter: {
            AND: [
                { attributeName: "DPF:dataset_id", is: "cals_construction" },
                { attributeName: "DPF:prefecture_name", is: "千葉県" },
            ]
        }
    ) {
        totalNumber
        searchResults {
            id
            title
            metadata
        }
    }
}

{{ attributeName: "DPF:dataset_id", is: "cals_construction" }} で、データセット「工事管理ファイル」を指定しています。
{{ attributeName: "DPF:prefecture_name", is: "{prefecture}" }},で、都道府県を指定しています。

国土交通DPFにログインして、GraphiQLでクエリを実行して、正しく結果が得られることを確認してみます。
検索結果として、検索結果のヒット数totalNumber　と、各データのID, title, metadata を返します。

GraphiQLでクエリ実行

StreamlitでWebアプリにする

では、StreamlitでWebアプリケーションにしてみましょう。

必要なモジュールをインポートする

Streamlitの他に、集計やグラフを作成するために必要なライブラリもインポートします。

import os
import requests
import pandas as pd
import streamlit as st
import matplotlib.pyplot as plt
import japanize_matplotlib
import altair as alt
import plotly.express as px

サイドバーで都道府県を選択する

Streamlitのサイドバーで都道府県を選択するセレクトボックスを表示する。

def render_search():
    st.sidebar.header("Search")
     # 都道府県のセレクトボックスを表示し、選択された都道府県をセッションステートに保存
    selected_prefecture = st.sidebar.selectbox(
        "Prefecture",
        ("北海道", "青森県", "岩手県", ..., "鹿児島県", "沖縄県"),
        index=("北海道", "青森県", "岩手県", ..., "鹿児島県", "沖縄県").index(st.session_state['selected_prefecture']),
    )  
    if st.sidebar.button("Search"):
        handler_search(selected_prefecture)

サイドバーのセレクトボックス

選択した都道府県の工事管理ファイルをAPIで取得する

国土交通DPFのAPIから特定の条件でデータを検索し、結果を取得する関数を定義します。特定の都道府県の工事情報を取得します。
検索結果のmetadata から、ID、工事名、年度、緯度・経度などの必要な項目を抽出しています。

def handler_search(prefecture):
    api_endpoint = "https://www.mlit-data.jp/api/v1/" # API Endpoint
    api_key = st.secrets["https://www.mlit-data.jp/api/v1/"]["api_key"] # API Key
    # GraphQL query
    query = f"""
    query {{
        search(
            term: ""
            first: 0
            size: 10000
            attributeFilter: {{
                AND: [
                    {{ attributeName: "DPF:dataset_id", is: "cals_construction" }},
                    {{ attributeName: "DPF:prefecture_name", is: "{prefecture}" }},
                ]
            }}
        ) {{
            totalNumber
            searchResults {{
                id
                title
                metadata
            }}
        }}
    }}"""

    # Make the HTTP request
    headers = {
            "Content-Type": "application/json",
            "apikey": api_key,
    }

    response = requests.post(api_endpoint, json={"query": query}, headers=headers)
    response.raise_for_status()

    # Parse the response
    response_body = response.json()
    total_number = response_body["data"]["search"]["totalNumber"]
    search_results = response_body["data"]["search"]["searchResults"]

    # Extract the relevant metadata from the search results
    metadata = []
    for result in search_results:
        metadata.append({
            "id": result["id"],
            "title": result["title"],
            "year": result["metadata"]["DPF:year"], #年度
            "title": result["metadata"]["DPF:title"], #工事名
            "lat": result["metadata"].get("DPF:latitude", ""), #緯度
            "lon": result["metadata"].get("DPF:longitude", ""), #経度
            "client_cat_sub": result["metadata"]["CALS:client_info"].get("sub_category", ""), #発注者カテゴリ
            "construction_type": result["metadata"]["CALS:construction_name_etc"].get("construction_type", ""), #工事種別
            "construction_name": result["metadata"]["CALS:construction_name_etc"].get("construction_name", ""), #工事名称
            "construction_field": result["metadata"]["CALS:construction_name_etc"].get("construction_field", ""), #工事分野
            "construction_content" : result["metadata"]["CALS:construction_name_etc"].get("construction_content", ""), #工事内容
            "construction_end_date" : result["metadata"]["CALS:construction_name_etc"].get("construction_end_date", ""), #工期終了日

        })
    
    return render_search_results(total_number, metadata, search_results)

集計値と地図、グラフを表示する

これらのデータを、Streamlitで集計値と地図、グラフを表示します。

集計値を表示する

def render_search_results(total_number, metadata, search_results):
	# Create a DataFrame from the metadata
	df = pd.DataFrame(metadata)

	# Display prefecuture name
	st.markdown("### 都道府県：" + st.session_state['selected_prefecture'])

	# Display metrics
	st.metric(label="工事件数", value=total_number)

地図を表示する

# plotting the map
## Convert latitude and longitude to numeric
df['lat'] = pd.to_numeric(df['lat'], errors='coerce')
df['lon'] = pd.to_numeric(df['lon'], errors='coerce')
## Drop rows with NaN values in latitude and longitude
df_map = df.dropna(subset=['lat', 'lon'])
## Plot the map
st.map(df_map)

地図を表示

表を表示する

# Table: Number of construction by year
df_year = df.groupby("year").size().reset_index(name="count")
st.write(df_year)
fig = plt.figure()

グラフを表示する

# Plotting the holizontal bar chart: Number of construction by year and construction field
df_year_field = df.groupby(["year", "construction_field"]).size().reset_index(name="count")
fig = plt.figure()
fig = px.bar(df_year_field, x="year", y="count", color="construction_field", barmode="stack")
st.plotly_chart(fig)

グラフなどの表示イメージ

---

まとめ

Streamlitでお手軽にアプリにすることができました。
今回は都道府県毎の集計にしてみましたが、工事分野や年度などで集計してみるといいかもしれませんね。