QGISプラグイン開発

qgis

プラグイン開発の基礎を押さえたら、より実践的な機能を実装しましょう。

QGISプラグイン開発【実践編】

この記事では、実務で役立つプラグインの実装パターンを解説します。

処理ツールの実装

Processing Provider

QGISの処理ツールボックスに独自ツールを追加できます。

プロバイダークラス:

python
from qgis.core import QgsProcessingProvider

class MyProvider(QgsProcessingProvider):
    def loadAlgorithms(self):
        self.addAlgorithm(MyProcessingAlgorithm())

    def id(self):
        return 'myplugin'

    def name(self):
        return 'My Plugin'

    def icon(self):
        return QgsProcessingProvider.icon(self)

アルゴリズムクラス:

python
from qgis.core import (
    QgsProcessingAlgorithm,
    QgsProcessingParameterVectorLayer,
    QgsProcessingParameterNumber,
    QgsProcessingParameterFeatureSink
)

class MyProcessingAlgorithm(QgsProcessingAlgorithm):
    INPUT = 'INPUT'
    DISTANCE = 'DISTANCE'
    OUTPUT = 'OUTPUT'

    def name(self):
        return 'mybuffer'

    def displayName(self):
        return 'マイバッファ'

    def group(self):
        return 'ベクター処理'

    def groupId(self):
        return 'vector'

    def initAlgorithm(self, config=None):
        self.addParameter(
            QgsProcessingParameterVectorLayer(
                self.INPUT,
                '入力レイヤ'
            )
        )
        self.addParameter(
            QgsProcessingParameterNumber(
                self.DISTANCE,
                'バッファ距離',
                defaultValue=100
            )
        )
        self.addParameter(
            QgsProcessingParameterFeatureSink(
                self.OUTPUT,
                '出力レイヤ'
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        source = self.parameterAsVectorLayer(parameters, self.INPUT, context)
        distance = self.parameterAsDouble(parameters, self.DISTANCE, context)

        (sink, dest_id) = self.parameterAsSink(
            parameters, self.OUTPUT, context,
            source.fields(), source.wkbType(), source.crs()
        )

        total = source.featureCount()
        for current, feature in enumerate(source.getFeatures()):
            if feedback.isCanceled():
                break

            # バッファ処理
            buffered = feature.geometry().buffer(distance, 5)
            new_feature = QgsFeature()
            new_feature.setGeometry(buffered)
            new_feature.setAttributes(feature.attributes())
            sink.addFeature(new_feature)

            feedback.setProgress(int(current / total * 100))

        return {self.OUTPUT: dest_id}

    def createInstance(self):
        return MyProcessingAlgorithm()

プログレスバーの実装

ダイアログ内のプログレスバー

python
from qgis.PyQt.QtWidgets import QProgressDialog
from qgis.PyQt.QtCore import Qt

def run_with_progress(self):
    layer = self.iface.activeLayer()
    total = layer.featureCount()

    # プログレスダイアログを作成
    progress = QProgressDialog("処理中...", "キャンセル", 0, total)
    progress.setWindowModality(Qt.WindowModal)
    progress.show()

    for i, feature in enumerate(layer.getFeatures()):
        if progress.wasCanceled():
            break

        # 処理
        self.process_feature(feature)

        progress.setValue(i + 1)

    progress.close()

メッセージバーのプログレス

python
from qgis.core import Qgis
from qgis.PyQt.QtWidgets import QProgressBar

def run_with_messagebar_progress(self):
    # プログレスバーを作成
    progressBar = QProgressBar()
    progressBar.setMaximum(100)

    # メッセージバーに追加
    messageBar = self.iface.messageBar().createMessage("処理中...")
    messageBar.layout().addWidget(progressBar)
    self.iface.messageBar().pushWidget(messageBar, Qgis.Info)

    # 処理
    for i in range(100):
        # 処理を実行
        progressBar.setValue(i + 1)

    # 完了
    self.iface.messageBar().clearWidgets()
    self.iface.messageBar().pushMessage("完了", "処理が終了しました", Qgis.Success)

設定の保存と読み込み

QgsSettings

python
from qgis.core import QgsSettings

class MyPlugin:
    def __init__(self, iface):
        self.iface = iface
        self.settings = QgsSettings()

    def save_settings(self):
        # 設定を保存
        self.settings.setValue("myplugin/last_path", "/path/to/file")
        self.settings.setValue("myplugin/buffer_distance", 100)
        self.settings.setValue("myplugin/enabled", True)

    def load_settings(self):
        # 設定を読み込み（デフォルト値付き）
        path = self.settings.value("myplugin/last_path", "")
        distance = self.settings.value("myplugin/buffer_distance", 50, type=int)
        enabled = self.settings.value("myplugin/enabled", False, type=bool)
        return path, distance, enabled

プロジェクト固有の設定

python
from qgis.core import QgsProject

# プロジェクトに設定を保存
project = QgsProject.instance()
project.writeEntry("myplugin", "setting_name", "value")

# 読み込み
value, ok = project.readEntry("myplugin", "setting_name", "default")

外部ライブラリの利用

標準ライブラリ

python
import os
import json
import csv
from datetime import datetime

pandas（インストール済みの場合）

python
try:
    import pandas as pd
    HAS_PANDAS = True
except ImportError:
    HAS_PANDAS = False

def process_with_pandas(self):
    if not HAS_PANDAS:
        QMessageBox.warning(None, "エラー", "pandasがインストールされていません")
        return

    # pandasを使った処理
    df = pd.read_csv("data.csv")

requestsでAPI連携

python
import requests

def fetch_api_data(self, url):
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        QgsMessageLog.logMessage(f"API error: {e}", "MyPlugin", Qgis.Warning)
        return None

エラーハンドリング

基本的なエラー処理

python
from qgis.core import QgsMessageLog, Qgis

def run(self):
    try:
        self.do_processing()
    except FileNotFoundError as e:
        QMessageBox.critical(None, "エラー", f"ファイルが見つかりません: {e}")
    except ValueError as e:
        QMessageBox.warning(None, "警告", f"値が不正です: {e}")
    except Exception as e:
        QgsMessageLog.logMessage(f"予期しないエラー: {e}", "MyPlugin", Qgis.Critical)
        QMessageBox.critical(None, "エラー", "予期しないエラーが発生しました。ログを確認してください。")

入力検証

python
def validate_input(self):
    layer = self.dialog.layerComboBox.currentLayer()
    if layer is None:
        QMessageBox.warning(None, "警告", "レイヤを選択してください")
        return False

    distance = self.dialog.distanceSpinBox.value()
    if distance <= 0:
        QMessageBox.warning(None, "警告", "距離は正の数を入力してください")
        return False

    return True

マルチスレッド処理

QgsTask

長時間の処理はバックグラウンドで実行:

python
from qgis.core import QgsTask, QgsApplication

class MyTask(QgsTask):
    def __init__(self, description, layer):
        super().__init__(description, QgsTask.CanCancel)
        self.layer = layer
        self.result = None

    def run(self):
        total = self.layer.featureCount()
        for i, feature in enumerate(self.layer.getFeatures()):
            if self.isCanceled():
                return False

            # 処理
            self.process_feature(feature)

            self.setProgress(i / total * 100)

        return True

    def finished(self, result):
        if result:
            iface.messageBar().pushMessage("完了", "処理が終了しました", Qgis.Success)
        else:
            iface.messageBar().pushMessage("中断", "処理がキャンセルされました", Qgis.Warning)

    def process_feature(self, feature):
        # 実際の処理
        pass

# タスクを実行
task = MyTask("処理中", layer)
QgsApplication.taskManager().addTask(task)

実践的なプラグイン例

CSV→GISレイヤ変換プラグイン

python
import csv
from qgis.core import (
    QgsVectorLayer, QgsFeature, QgsGeometry, QgsPointXY,
    QgsField, QgsProject
)
from qgis.PyQt.QtCore import QVariant

class CsvToLayerPlugin:
    def convert_csv_to_layer(self, csv_path, x_field, y_field, crs="EPSG:4326"):
        # メモリレイヤを作成
        layer = QgsVectorLayer(f"Point?crs={crs}", "csv_layer", "memory")
        provider = layer.dataProvider()

        # CSVを読み込んでフィールドを取得
        with open(csv_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            fieldnames = reader.fieldnames

            # フィールドを追加
            fields = [QgsField(name, QVariant.String) for name in fieldnames]
            provider.addAttributes(fields)
            layer.updateFields()

            # フィーチャを追加
            f.seek(0)
            next(reader)  # ヘッダーをスキップ

            for row in reader:
                try:
                    x = float(row[x_field])
                    y = float(row[y_field])

                    feature = QgsFeature()
                    feature.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(x, y)))
                    feature.setAttributes([row[name] for name in fieldnames])
                    provider.addFeature(feature)
                except (ValueError, KeyError) as e:
                    continue

        # プロジェクトに追加
        QgsProject.instance().addMapLayer(layer)
        return layer

レイヤ情報エクスポートプラグイン

python
import json

class LayerInfoExporter:
    def export_layer_info(self, layer, output_path):
        info = {
            "name": layer.name(),
            "crs": layer.crs().authid(),
            "feature_count": layer.featureCount(),
            "geometry_type": layer.geometryType(),
            "extent": {
                "xmin": layer.extent().xMinimum(),
                "ymin": layer.extent().yMinimum(),
                "xmax": layer.extent().xMaximum(),
                "ymax": layer.extent().yMaximum()
            },
            "fields": [
                {
                    "name": field.name(),
                    "type": field.typeName()
                }
                for field in layer.fields()
            ]
        }

        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(info, f, indent=2, ensure_ascii=False)

        return info

開発のベストプラクティス

コード構成

my_plugin/
├── __init__.py
├── my_plugin.py          # メインクラス
├── dialogs/
│   ├── __init__.py
│   └── main_dialog.py    # ダイアログクラス
├── processing/
│   ├── __init__.py
│   └── algorithms.py     # 処理アルゴリズム
├── utils/
│   ├── __init__.py
│   └── helpers.py        # ユーティリティ
├── resources/
│   ├── ui/
│   │   └── main_dialog.ui
│   └── icons/
│       └── icon.png
├── metadata.txt
└── README.md

テスト

python
# test_plugin.py
import unittest

class TestMyPlugin(unittest.TestCase):
    def test_buffer_calculation(self):
        # テストケース
        result = calculate_buffer(100)
        self.assertEqual(result, expected)

if __name__ == '__main__':
    unittest.main()

まとめ

実践的なポイント

ポイント

  • Processing Providerで処理ツールを実装
  • プログレスバーで進捗を表示
  • QgsSettingsで設定を永続化
  • 適切なエラーハンドリング
  • 長時間処理はQgsTaskで

次のステップ

おすすめ

  • 公式ドキュメントを参照
  • 既存プラグインのソースを読む
  • 小さなプラグインから始める

関連記事

QGISプラグイン開発入門【基礎編】
PyQGIS入門
QGIS開発ガイド

お問い合わせ

QGISプラグイン開発についてのご相談は、お気軽にお問い合わせください。

  • プラグイン開発相談
  • カスタマイズ支援
  • GIS開発コンサルティング

お問い合わせはこちら

最終更新: 2025年1月