LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@yamamorisobain日鉄ソリューションズ株式会社

生成AIを活用した地形データの補間処理（アレフガルド in マインクラフト Part 13）

Posted at

はじめに

昨年のアドベントカレンダーでマインクラフト上にアレフガルドを生成する取り組みについて紹介し、魔の島の一部の生成について投稿しました1。本校では、その後の取り組みについてご報告します。1月には魔の島の残りの部分、2月にはラダトーム周辺の再現に取り組んでいました。これらの取り組みについては、技術的に新しい内容を含まないので、Noteに進捗を投稿しています2。今回は、ラダトーム北部のガライヤ地方について、ドラゴンクエストビルダーズを用いた地形測量から得られた知見と、生成AIを活用したデータ補間処理についてお伝えします。

ガライヤ地方の測量と発見

ドラゴンクエストビルダーズで、ラダトームから北方に位置するガライヤ地方の測量を行った結果、メルキド地方と同様に、ガライヤ地方もラダトーム地方と比較して地形が縮小・間引きされていることが判明しました。ゲーム内で再現されている地形は、原作の雰囲気は保ちつつも簡略化されていたのです。

生成AIを用いたデータ補間方法の検討

間引きされた地形データから本来の大きさの地形を復元するためには、欠損したデータを適切に補間する処理が必要です。データ分析やモデリングの専門知識がない私にとって、これは大きな課題でした。

そこで思い出したのが、「生成AIは穴埋め問題を解くのが得意」という話です。ならば、この課題にも生成AIの力を借りてみようと考えました。

今回の補間処理では、Claude Pro と ChatGPT Plus の2種類の生成AIを使用しました。入力データとしては、縮尺を元に戻すために空行を追加した地形データのCSVファイルを用意し、それぞれのAIに以下のように補間を依頼しました。ガライヤ地方は海岸線のデータしか測量できていなかったため、全体を測量済みのメルキド地方のデータを利用しています。

  • 縮尺を元に戻すために空行を挿入した地形データのCSVファイルを入力
  • 欠損したデータを適切に補間することを依頼

image.png
Claude Proへの依頼

image.png
ChatGPT Plusへの依頼

私自身はデータ補間の方法論については素人ですが、生成AIに依頼することで様々な補間手法を試すことができました。各AIは世の中に存在する補間手法を次々と提案し、それらをコードとして実装してくれました。
提案された手法の例：

  • バイキュービック補間
  • クリギング（Kriging）法
  • 最近傍補間
  • スプライン補間
  • 逆距離加重法（IDW）

これらの手法を実際に適用するソースコードを生成AIに出力してもらい、実際に保管してみた結果を比較し、最も自然に感じられる補間結果を得られる手法を選ぶことができました。

実際に採用したデータ補間手法について

最終的に採用したのは、KDTreeを使用した逆距離加重法（IDW）による補間です。この手法では、欠損データの周囲にある既知のデータポイントから、距離に反比例した重み付けで補間値を計算します。距離が近いデータポイントほど大きな影響を与え、遠いポイントほど影響が小さくなるため、地形の連続性を保ちながら自然な補間が可能になるそうです（これも生成AIからの受け売りです）。

実装したPythonコードの解説

Appendixに掲載したevaluateData2.pyは、この逆距離加重法を実装したPythonスクリプトです。主要な機能は以下の通りです：

  1. データの前処理：CSVファイルから地形データを読み込み、欠損値の位置を特定します。
  2. KDTreeの構築：既知のデータポイントからKDTree構造を作成し、効率的な最近傍探索を可能にします。
  3. IDW補間の実行：各欠損ポイントについて、指定数の最近傍点を見つけ、距離の逆数を重みとした加重平均を計算します。
  4. 後処理：補間結果を整数に丸めたり、最小値を設定したりして、適切な地形データに変換します。
  5. 結果の可視化：補間前後のデータをヒートマップで表示し、補間効果を視覚的に確認できます。

特に重要なのはidw_interpolation関数で、以下のように動作します：

def idw_interpolation(known_points, known_values, missing_points, n_neighbors=50, power=2):
    # KDTreeを構築して効率的に最近傍点を探索
    tree = KDTree(known_points)
    
    # 各欠損点について、指定数の最近傍点を検索
    distances, indices = tree.query(missing_points, k=n_neighbors)
    
    # 距離が0の場合の対処
    distances[distances == 0] = 1e-10
    
    # 逆距離の重みを計算（距離のべき乗に反比例）
    weights = 1.0 / (distances ** power)
    
    # 重み付き平均を計算して補間値を求める
    weights_sum = np.sum(weights, axis=1)
    interpolated_values = np.zeros(len(missing_points))
    
    for i in range(len(missing_points)):
        values_i = known_values[indices[i]]
        weights_i = weights[i]
        # 重み付き平均を計算
        interpolated_values[i] = np.sum(values_i * weights_i) / weights_sum[i]
    
    return interpolated_values

このアルゴリズムでは、パラメータとして「近傍点の数（n_neighbors）」と「距離の重み（power）」を調整することで、補間の特性を変えることができます。重みのパラメータが大きいほど、近い点の影響がより強くなるそうです。

補間結果の可視化

補間の効果を確認するため、visualize_interpolation関数で補間前後のデータを4つのグラフで可視化しています：
kdtree_idw_interpolation.png

  1. 左上：欠損値の分布（白い部分が欠損データ）
    ここでは300行目付近に水平な欠損ラインがあることがわかります。

  2. 右上：元のデータ（欠損値あり）
    オリジナルの地形データで、欠損部分は色が表示されていません。
    緑や黄色の部分が高地で、紫や濃い青の部分が低地です。

  3. 左下：補間後のデータ
    KDTreeとIDW法による補間後の地形データです。
    欠損していた部分にも自然な地形の連続性が保たれています。

  4. 右下：補間による値の追加（欠損値があった場所のみ）
    欠損していた部分にどのような値が補間されたかを示しています。
    青や白の部分は周囲の地形と調和した標高値が補間されています。

この可視化から、KDTree IDW法による補間が、地形の連続性と特徴を保ちながら自然な補間を実現していることがわかります。特に300行目付近の水平な欠損ラインが、周囲の地形データから適切に補間されており、違和感のない地形になっていることが確認できます。

生成AIを活用してみた所感

今回、地形データの補間に生成AIを活用してみて、以下に述べるような所感が得られました。これまでもちょっとした調べものに使ったりはしていましたが、改めて生成AIが持つ潜在的な可能性を実感させられました。

試行錯誤の高速化と結果

生成AIを活用することで、専門知識がなくても様々な手法を試すことができました。各手法のコードを実装してもらい、結果を比較して最適な方法を選ぶという試行錯誤のプロセスが大幅に高速化されました。
独学で対応しようとすれば、方法論の学習から実装まで何か月もかかるであろう作業を、数時間程度で片づけることができたのです。

生成AIの利用価値

この経験を通して、生成AIを活用することの価値を実感しました：

  1. 知識のデモクラタイゼーション
    インターネットが「世界中の情報にローコストでアクセスする手段」を提供したように、生成AIは「世界中の知識をローコストで活用する手段」を提供してくれます。専門家でなくても、その分野の知識や方法論を引き出し、すぐに活用できる点が革命的と感じられました。
  2. 複数のAIによるセカンドオピニオン
    Claude ProとChatGPT Plusの両方に相談することで、異なるアプローチを比較検討できました。「Claudeはこう言っているけど、あなた(ChatGPT)はどう思う？」というように、AI同士の見解を比較・検討することで、より良い解決策を見つけられます。

Claude ProとGPT Plusの比較

両方のAIに同様の依頼をした結果、それぞれ異なる特徴が見えてきました。
※地形データの補間に関する１サンプルからの印象なので、全体の傾向を表しているわけではないかもしれませんが。

Claude Pro:

  • チューニング可能なパラメータを対話的に変更できるコードを提供
  • デフォルト値を設定してくれるため、初心者でも使いやすい
  • 補間前後の地形データをヒートマップで可視化する機能を自発的に追加
  • 全体的に「気が利いている」コードを作ってくれた3印象

ChatGPT Plus:

  • 依頼した内容を素直に実行

  • データが欠損したファイルを渡し補間を依頼すると、コードではなく補間結果を直接返すこともある
    image.png

  • 再現性が必要ない一度きりの処理でよい場合にはこのアプローチが便利な場合もありそう

まとめ

ドラゴンクエストの世界（アレフガルド）をマインクラフト上に再現するという個人的なプロジェクトから、生成AIの強力な活用法を学ぶことができました。専門知識がなくても、AIの力を借りることで複雑な問題に取り組み、効率的に解決できることを実感しています。
地形データの補間という専門的な処理においても、生成AIは様々な手法を提案し、それらを実装したコードを提供してくれました。その結果、本来なら何か月もかかるであろう作業を数時間で完了させることがでました。
最終的に採用したKDTreeを使用した逆距離加重法（IDW）による補間は、地形の連続性と特性を保ちながら自然な補間を実現しました。可視化の結果から、この手法が欠損データを効果的に補完し、元の地形との整合性を保ちながら地形を復元できていることが確認できました。
最後に、複数の生成AIを比較検討することで、より多角的な視点を得られることもわかりました。それぞれのAIが持つ異なる特性を理解し、目的に応じて使い分けることで、より効果的に問題解決ができそうです。
生成AIは既に私たちの創作活動や問題解決の強力な助っ人となることを体感できました。今後も、こうしたツールをうまく活用しながら、アレフガルドの生成に取り組んでいきたいと思います。

Appendix 実際の処理に使用したコード

evaluateData2.py ``` python import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib import cm from scipy.spatial import KDTree import argparse import os import sys from tqdm import tqdm

def idw_interpolation(known_points, known_values, missing_points, n_neighbors=50, power=2):
"""
逆距離加重法（IDW）による補間を行う

Parameters:
-----------
known_points : numpy.ndarray
    既知のデータポイントの座標 (N, 2)
known_values : numpy.ndarray
    既知のデータポイントの値 (N,)
missing_points : numpy.ndarray
    欠損データの座標 (M, 2)
n_neighbors : int, optional
    各点の補間に使用する近傍点の数
power : float, optional
    距離の重み付けに使用する累乗値（大きいほど近い点の影響が強くなる）
    
Returns:
--------
numpy.ndarray
    補間された値 (M,)
"""
# KDTreeを構築
tree = KDTree(known_points)

# 各欠損点について、n_neighbors個の最近傍点を検索
distances, indices = tree.query(missing_points, k=n_neighbors)

# 距離が0の場合（既知点と欠損点が完全に重なる場合）の対処
# 小さな値で置き換える
distances[distances == 0] = 1e-10

# 逆距離の重みを計算
weights = 1.0 / (distances ** power)

# 重み付き平均を計算
weights_sum = np.sum(weights, axis=1)
interpolated_values = np.zeros(len(missing_points))

for i in range(len(missing_points)):
    values_i = known_values[indices[i]]
    weights_i = weights[i]
    # 重み付き平均を計算
    interpolated_values[i] = np.sum(values_i * weights_i) / weights_sum[i]

return interpolated_values

def interpolate_mesh_data(input_file, output_file, n_neighbors=50, power=2,
integer_output=True, min_value=0, visualize=False):
"""
KDTreeと逆距離加重法を用いてメッシュデータの欠損値を補完する

Parameters:
-----------
input_file : str
    入力CSVファイルのパス
output_file : str
    出力CSVファイルのパス
n_neighbors : int, optional
    各点の補間に使用する近傍点の数
power : float, optional
    IDW補間における距離の重み（大きいほど近い点の影響が強くなる）
integer_output : bool, optional
    True の場合、補完値を整数に丸める
min_value : float, optional
    補完値の最小値
visualize : bool, optional
    True の場合、欠損値の分布と補完結果を可視化する
    
Returns:
--------
pandas.DataFrame
    欠損値が補完されたデータフレーム
"""
print(f"ファイル '{input_file}' を読み込んでいます...")

try:
    # CSVファイルを読み込む
    df = pd.read_csv(input_file, header=None)
except Exception as e:
    print(f"CSVファイルの読み込みに失敗しました: {str(e)}")
    try:
        # エンコーディングを指定して再試行
        df = pd.read_csv(input_file, header=None, encoding='shift-jis')
    except Exception as e2:
        print(f"異なるエンコーディングでも失敗しました: {str(e2)}")
        return None

print(f"データサイズ: {df.shape[0]}行 x {df.shape[1]}列")

# 欠損値の位置を特定
missing_mask = df.isna()
missing_count = missing_mask.sum().sum()
total_cells = df.size
missing_percentage = (missing_count / total_cells) * 100

print(f"欠損値の数: {missing_count}/{total_cells} ({missing_percentage:.2f}%)")

if missing_count == 0:
    print("欠損データはありません。処理を終了します。")
    return df

# 2次元メッシュデータを補間用の形式に変換
rows, cols = df.shape
x, y = np.meshgrid(np.arange(cols), np.arange(rows))

# メッシュグリッドを1次元配列に変換
x_flat = x.flatten()
y_flat = y.flatten()
z_flat = df.values.flatten()

# 既知のデータポイントと欠損ポイントを分離
known_mask = ~np.isnan(z_flat)
known_points = np.column_stack((x_flat[known_mask], y_flat[known_mask]))
known_values = z_flat[known_mask]

missing_mask_flat = np.isnan(z_flat)
missing_points = np.column_stack((x_flat[missing_mask_flat], y_flat[missing_mask_flat]))

print(f"補間に使用する既知のデータポイント数: {len(known_points)}")
print(f"補間する欠損ポイント数: {len(missing_points)}")

# 補間の実行
print(f"KDTreeを使用して近傍{n_neighbors}点から逆距離加重補間を実行中...")

# numpyの警告を抑制（0での除算など）
with np.errstate(divide='ignore', invalid='ignore'):
    try:
        # 欠損点が多い場合はプログレスバーを表示
        if len(missing_points) > 1000:
            # バッチ処理で補間を実行（メモリ使用量の削減とプログレス表示のため）
            batch_size = 1000
            interpolated_values = np.zeros(len(missing_points))
            
            for i in tqdm(range(0, len(missing_points), batch_size), desc="IDW補間を実行中"):
                batch_end = min(i + batch_size, len(missing_points))
                batch_points = missing_points[i:batch_end]
                batch_values = idw_interpolation(
                    known_points, known_values, batch_points, 
                    n_neighbors=n_neighbors, power=power
                )
                interpolated_values[i:batch_end] = batch_values
        else:
            # 一度に処理
            interpolated_values = idw_interpolation(
                known_points, known_values, missing_points, 
                n_neighbors=n_neighbors, power=power
            )
            
        # 整数出力と最小値の制約を適用
        if integer_output:
            interpolated_values = np.round(interpolated_values).astype(int)
        
        interpolated_values = np.maximum(interpolated_values, min_value)
        
        # 元のデータフレームに補間値を埋め込む
        result_df = df.copy()
        for i, (x_pos, y_pos) in enumerate(missing_points):
            result_df.iloc[int(y_pos), int(x_pos)] = interpolated_values[i]
            
        print("IDW補間が完了しました。")
        
    except Exception as e:
        print(f"補間に失敗しました: {str(e)}")
        # エラー発生時のフォールバック処理
        print("単純な補間方法を試行します...")
        result_df = df.copy()
        # 行方向と列方向の両方で前後の値を伝播させて補間
        result_df = result_df.fillna(method='ffill').fillna(method='bfill')
        result_df = result_df.fillna(method='ffill', axis=1).fillna(method='bfill', axis=1)
        
        # それでも残る欠損値を最小値で埋める
        if result_df.isna().sum().sum() > 0:
            result_df = result_df.fillna(min_value)

# 欠損値がすべて補完されたか確認
still_missing = result_df.isna().sum().sum()
if still_missing > 0:
    print(f"警告: {still_missing}個のデータポイントは補間できませんでした。最小値で埋めます。")
    result_df = result_df.fillna(min_value)
    
# 可視化（必要な場合）
if visualize:
    visualize_interpolation(df, result_df, missing_mask, n_neighbors, power)
    
# 結果を保存
result_df.to_csv(output_file, index=False, header=False)
print(f"補完済みデータを '{output_file}' に保存しました")
    
return result_df

def visualize_interpolation(original_df, interpolated_df, missing_mask, n_neighbors, power):
"""
補間前後のデータを可視化する

Parameters:
-----------
original_df : pandas.DataFrame
    元のデータフレーム
interpolated_df : pandas.DataFrame
    補間後のデータフレーム
missing_mask : pandas.DataFrame
    欠損値のマスク（Trueが欠損値）
n_neighbors : int
    補間に使用した近傍点の数
power : float
    IDW補間における距離の重み
"""
plt.figure(figsize=(15, 10))

# 欠損値の分布を表示
plt.subplot(221)
plt.imshow(missing_mask, cmap='binary', interpolation='none')
plt.colorbar(label='欠損値の位置')
plt.title('欠損値の分布（白が欠損値）')
plt.xlabel('列 (X)')
plt.ylabel('行 (Y)')

# 元のデータを表示
plt.subplot(222)
plt.imshow(original_df, cmap='viridis', interpolation='none')
plt.colorbar(label='値')
plt.title('元のデータ（欠損値あり）')
plt.xlabel('列 (X)')
plt.ylabel('行 (Y)')

# 補間後のデータを表示
plt.subplot(223)
plt.imshow(interpolated_df, cmap='viridis', interpolation='none')
plt.colorbar(label='値')
plt.title(f'補間後のデータ（KDTree IDW, 近傍{n_neighbors}点, 距離重み={power}）')
plt.xlabel('列 (X)')
plt.ylabel('行 (Y)')

# 差分を表示
# 元のデータの欠損値を0で埋めて差分計算を可能にする
original_filled = original_df.fillna(0)
diff = interpolated_df - original_filled
# 差分が0の部分（元の値がある部分）をマスク
diff_masked = diff.copy()
diff_masked[~missing_mask] = np.nan

plt.subplot(224)
plt.imshow(diff_masked, cmap='coolwarm', interpolation='none')
plt.colorbar(label='差分値')
plt.title('補間による値の追加（欠損値があった場所のみ）')
plt.xlabel('列 (X)')
plt.ylabel('行 (Y)')

plt.tight_layout()
plt.savefig('kdtree_idw_interpolation.png', dpi=300, bbox_inches='tight')
print("可視化結果を 'kdtree_idw_interpolation.png' に保存しました")
plt.close()

def get_user_input():
"""対話モードでパラメータを設定する"""
print("===== KDTreeと逆距離加重法を用いたメッシュデータ欠損値補完ツール =====")

# 入力ファイルの指定
while True:
    input_file = input("入力CSVファイルのパスを入力してください: ")
    if os.path.exists(input_file):
        break
    else:
        print(f"エラー: ファイル '{input_file}' が見つかりません。正しいパスを入力してください。")

# 出力ファイルの指定
default_output = os.path.splitext(input_file)[0] + "_IDW補完済み.csv"
output_file = input(f"出力CSVファイルのパス [デフォルト: {default_output}]: ")
if not output_file:
    output_file = default_output

# 近傍点数の指定
n_neighbors_str = input("補間に使用する近傍点の数を入力してください [デフォルト: 50]: ")
n_neighbors = 50 if not n_neighbors_str else int(n_neighbors_str)

# 距離重みの指定
power_str = input("逆距離の重み（距離のべき乗）を入力してください [デフォルト: 2]: ")
power = 2 if not power_str else float(power_str)

# 整数出力とするかどうか
integer_output = input("補完値を整数に丸めますか? (y/n) [デフォルト: y]: ").lower() != 'n'

# 最小値の設定
min_value_str = input("補完値の最小値を指定してください [デフォルト: 0]: ")
min_value = 0 if not min_value_str else float(min_value_str)

# 可視化するかどうか
visualize = input("補間前後のデータを可視化しますか? (y/n) [デフォルト: y]: ").lower() != 'n'

return {
    'input_file': input_file,
    'output_file': output_file,
    'n_neighbors': n_neighbors,
    'power': power,
    'integer_output': integer_output,
    'min_value': min_value,
    'visualize': visualize
}

def main():
parser = argparse.ArgumentParser(description='KDTreeと逆距離加重法を用いたメッシュデータの欠損値補完ツール')
parser.add_argument('-i', '--input', help='入力CSVファイルのパス')
parser.add_argument('-o', '--output', help='出力CSVファイルのパス')
parser.add_argument('-n', '--neighbors', type=int, default=50, help='補間に使用する近傍点の数 (デフォルト: 50)')
parser.add_argument('-p', '--power', type=float, default=2.0, help='逆距離の重み（距離のべき乗） (デフォルト: 2.0)')
parser.add_argument('--integer', action='store_true', help='補完値を整数に丸める')
parser.add_argument('--min-value', type=float, default=0, help='補完値の最小値 (デフォルト: 0)')
parser.add_argument('-v', '--visualize', action='store_true', help='補間前後のデータを可視化する')
parser.add_argument('--interactive', action='store_true', help='対話モードで実行する')

args = parser.parse_args()

# 対話モードかコマンドライン引数モードかを決定
if args.interactive or (not args.input):
    # 対話モード
    params = get_user_input()
else:
    # コマンドライン引数モード
    if not os.path.exists(args.input):
        print(f"エラー: 入力ファイル '{args.input}' が見つかりません。")
        sys.exit(1)
        
    output_file = args.output
    if not output_file:
        # 出力ファイル名が指定されていない場合、デフォルト名を生成
        output_file = os.path.splitext(args.input)[0] + "_IDW補完済み.csv"
        
    params = {
        'input_file': args.input,
        'output_file': output_file,
        'n_neighbors': args.neighbors,
        'power': args.power,
        'integer_output': args.integer,
        'min_value': args.min_value,
        'visualize': args.visualize
    }

# 処理の実行
try:
    interpolate_mesh_data(**params)
    print("処理が完了しました。")
except Exception as e:
    print(f"エラー: 処理中に問題が発生しました: {str(e)}")
    import traceback
    traceback.print_exc()
    sys.exit(1)

if name == "main":
main()

</details>

  1. https://qiita.com/yamamorisoba/items/c37088064e75ee305908 ↩

  2. https://note.com/yamamorisoba/n/n5fb808de7763
    https://note.com/yamamorisoba/n/n6eb9dafd9eb1 ↩

  3. この処理を行う時にはこういうモノが欲しくなるよね、という部分を明示的に指定しなくても生成してくれる ↩

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?