Python Streamlitを使用した基礎的なWebアプリケーションの作成(2) July, 2025

 前回のBlog記事:https://shikuuk.blogspot.com/2025/06/python-folium-relearning8-from-may.html

 DummyDataをMakeUseしたCodeがAlmostCompleteしたので、NextはJapaneseGovのJapanLandNumericalInformation上にある実在のDamInformationをUseして、DamのSpecificationやLocationを表示するAppにevolutionしたいと思います。

 InAdvanceでTryしてみたところ、StreamlitならではのSubjectやproblemが出てきたので、大変勉強になりました。ExamplyにWriteすると、

  • StreamlitでMakeしたAppは、App上で何かOperationするたびに基本的にCodeをInitialOrderからActivateしていくので、今回のようにWeb上のDataBaseを単純にRequest-GetするCodeをWriteすると、OperationのたびにWeb上のDataBaseをRequestしにいくため、Requestが多すぎてErrorになる。

とか、

  • Likely AboveのReasonで、GeoDataFrameのColumnNameをW01_001といったSimbolicSignから「ダム名称」のようなJapaneseにRewriteする際に、RewriteするCodeの位置を間違うと、一度目はうまくRewriteできるけど、App上でOperateがOccureしたあとの2度目の処理の際にHaveRewrittenのColumnNameをSearchしに行き、すでにColumnNameがRewriteされているのでSearchしようとしているColumnNameがFindOutできずにErrorがHappenする

などなどにEncounterしました。StreamlitのHabitをUnderstandしたうえでCodeをWriteするRequirementがExistであることをRealizedでした。

 So, the Conclusion that I come to, at firstで、Web上からGetするInformationを、App内で取り扱いやすいようにInAdvanceにProseccingしておくことだとThinkしました。

 From Here, FirstでJapan Land Numerical InformationのDamInformationをGetして、App内でUseしやすいGeoDataFrameにProseccingします。

 まずは、幹のPartになるFunctionを作成します。


#@st.cache_resource
def func_create_gdf_data():

# ダウンロードとデータ読み込みのループ
    url = 'http://nlftp.mlit.go.jp/ksj/gml/data/W01/W01-14/W01-14_GML.zip' # 国土数値情報のダムデータを格納しているURLをChoromeのデベロッパーツール活用して確認
    zip_file_name = 'W01-14_GML.zip'
    shp_file_name = 'W01-14-g_Dam.shp'# 国土数値情報のダムデータを一度Downloadしてshpファイルの名称を確認
##    一度LocalPCにDownloadしたファイルを使用する場合
#    zip_file_path = 'W01-14_GML.zip' # app.py と同じディレクトリにある場合、ファイル名だけでOK
#    shp_file_name = 'W01-14-g_Dam.shp'

# データをダウンロードして読み込む
    original_gdf = download_map_data(url, zip_file_name, shp_file_name)

    gdf = func_process_dam_gdf(original_gdf)

    return gdf

# Create the GeoDataFrame
gdf = func_create_gdf_data()

gdf.head(10)

 Firstにある「#@st.cache_resource」が実は肝で、現時点では#を付けて無効化していますが、StreamlitでこのCodeを動かすときにすごくImportantなRoleを果たします。

 まず、MainCodeからfunc_creatr_gdf_data()関数を呼び出して、国土数値情報のURLやZipFileのNameなどを指定し、さらにfunc_load_map_data()関数を呼び出して、そのReturnとなるGeoDataFrameを'original_gdf'として受け取ります。

 func_load_map_data()は次のようなContentです。


# --- 国土数値情報のDamInformationをロードする関数 ---
# @st.cache_resource デコレータはそのまま残します。
# これがないと、アプリが再実行されるたびに毎回解凍処理が走ってしまいます。
# (もし実際の国土数値情報データをロードする場合、このデコレータを適用してください)
# @st.cache_resource

def download_map_data(url, zip_file_name, shp_file_name):
    zip_file_path = 'unzipped_data/' + zip_file_name

# URLからデータをダウンロード
    urllib.request.urlretrieve(url, zip_file_name)

# ZIPファイルを解凍
    with zipfile.ZipFile(zip_file_name, 'r') as zip_ref:
        zip_ref.extractall('unzipped_data')  # 解凍先のディレクトリを指定

# Shapefileを読み込む
#shapefile_path = 'unzipped_data/W05-06_36-g_Stream.shp'
    shp_file_path = 'unzipped_data/' + shp_file_name
    gdf = gpd.read_file(shp_file_path)

    return gdf


 2つのif文以外は基本的に指定したzip_file_pathにあるzipfileを解凍して、asignされたdirectryにshpFileを保存し、そのShpFileからGeoDataFrame'gdf'をmakeして、そのgdfをfunc_create_gdf_data()にReturnするというSimpleなCodeです。
 func_create_gdf_data()に戻ると、func_process_dam_gdf()がCallされます。func_process_dam_gdf()のContentsです。

def func_process_dam_gdf(gdf_raw: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    """
    国土数値情報から読み込んだ生のGeoDataFrameを処理し、
    必要な列を抽出し、日本語名にリネーム・変換して新しいGeoDataFrameを返します。

    Args:
        gdf_raw (gpd.GeoDataFrame): 国土数値情報から直接読み込んだ生のGeoDataFrame。

    Returns:
        gpd.GeoDataFrame: 処理され、日本語の列名と変換された値を持つGeoDataFrame。
                          必須カラムが見つからない場合は、空のGeoDataFrameを返します。
    """
    # 必要な元の列名と新しい列名のマッピングを定義
    column_rename_map = {
        'W01_001': 'ダム名',
        'W01_003': '河川名',
        'W01_004': '水系名',
        'W01_005': 'ダム型式',
        'W01_006': 'ダム目的',
        'W01_007': 'ダム堤高(m)',
        'W01_008': '堤頂長(m)',
        'W01_010': '総貯水量(千m3)',
        'W01_011': 'ダム事業者', # W01_011をダム事業者としてリネーム
        'W01_012': '竣工年',
        'W01_013': '所在地'
    }

    # 抽出する列のリスト (geometryはGeoDataFrameなので自動的に保持される)
    columns_to_extract = list(column_rename_map.keys()) + ['geometry']

    # 必要な列がgdf_rawに存在するかチェック
    missing_cols = [col for col in columns_to_extract if col not in gdf_raw.columns]
    if missing_cols:
        print(f"エラー: 以下の必須カラムがGeoDataFrameに見つかりません: {', '.join(missing_cols)}")
        # 必須カラムがない場合は空のGeoDataFrameを返す
        return gpd.GeoDataFrame(columns=list(column_rename_map.values()) + ['geometry'], crs=gdf_raw.crs)

    # 必要な列のみを抽出し、新しいGeoDataFrameを作成
    gdf_processed = gdf_raw[columns_to_extract].copy()

    # 列名をリネーム
    gdf_processed.rename(columns=column_rename_map, inplace=True)

    # ダム型式 (W01_005) の数値コードを日本語に変換
    if 'ダム型式' in gdf_processed.columns:
        # 数値に変換できないものはNaNにし、Int64型に変換
        gdf_processed['ダム型式'] = pd.to_numeric(gdf_processed['ダム型式'], errors='coerce').astype('Int64')
        # マッピングし、対応しないものは'不明'にする
        gdf_processed['ダム型式'] = gdf_processed['ダム型式'].map(dam_type_map).fillna('不明')

    # ダム目的 (W01_006) の数値コードを日本語に変換 (複数対応)
    if 'ダム目的' in gdf_processed.columns:
        # 'W01_006' 列の値を文字列に変換し、カンマで分割
        # 各コードをマッピングし、有効な目的のみを結合
        def map_multiple_purposes(codes_str):
            if pd.isna(codes_str) or not isinstance(codes_str, str):
                return '不明'

            # カンマで分割し、各コードを数値に変換(エラーは無視)
            codes = [pd.to_numeric(c.strip(), errors='coerce') for c in codes_str.split(',')]

            # 有効な数値コードのみを抽出し、マッピング
            mapped_purposes = []
            for code in codes:
                if pd.notna(code) and int(code) in dam_purpose_map:
                    mapped_purposes.append(dam_purpose_map[int(code)])

            # マッピングされた目的があれば結合、なければ'不明'
            if mapped_purposes:
                return '・'.join(mapped_purposes)
            else:
                return '不明'

        gdf_processed['ダム目的'] = gdf_processed['ダム目的'].astype(str).apply(map_multiple_purposes)

    # ダム事業者 (W01_011) の数値コードを日本語に変換 (新しく追加)
    if 'ダム事業者' in gdf_processed.columns:
        # 数値に変換できないものはNaNにし、Int64型に変換
        gdf_processed['ダム事業者'] = pd.to_numeric(gdf_processed['ダム事業者'], errors='coerce').astype('Int64')
        # マッピングし、対応しないものは'不明'にする
        gdf_processed['ダム事業者'] = gdf_processed['ダム事業者'].map(dam_operator_map).fillna('不明')

    # 都道府県名の抽出 ('所在地'カラムから)
    if '所在地' in gdf_processed.columns:
        gdf_processed['都道府県名'] = gdf_processed['所在地'].astype(str).str.extract(pref_pattern).fillna('')
        # '所在地'カラムはもう不要であれば削除することも可能
        # gdf_processed.drop(columns=['所在地'], inplace=True)
    else:
        print("警告: 「所在地」カラムが見つからないため、「都道府県名」の抽出ができませんでした。")
        gdf_processed['都道府県名'] = '' # 空の都道府県名カラムを作成

    return gdf_processed

 MasterGeminiがしっかりと注釈をつけて、しかも、Errorを回避するif文なども織り込んで的確にGeoDataFrameを処理してくれています。









コメント