ClassificationTool.py

# -*- coding: utf-8 -*-
"""
/***************************************************************************
 ClassificationTool
                                 A QGIS plugin
 Classification of remote sensing images
 Generated by Plugin Builder: http://g-sherman.github.io/Qgis-Plugin-Builder/
                              -------------------
        begin                : 2019-05-24
        git sha              : $Format:%H$
        copyright            : (C) 2019 by Steven Hill
        email                : steven.hill@uni-wuerzburg.de
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
"""
from PyQt5.QtCore import (
    Qt,
    QSettings,
    QTranslator,
    qVersion,
    QCoreApplication,
    QThread,
    QTextStream,
    QFile,
)
from PyQt5.QtGui import *
from PyQt5.QtWidgets import QAction, QFileDialog, QListWidgetItem, QToolTip
from qgis.core import *
from qgis.core import Qgis, QgsMessageLog
import os
from time import gmtime, strftime, time
from . import resources

# Initialize Qt resources from file resources.py
# Import the code for the dialog
from .ClassificationTool_dialog import ClassificationToolDialog
from .worker import Worker


class ClassificationTool:
    """QGIS Plugin Implementation."""

    def __init__(self, iface):
        """Constructor.

        :param iface: An interface instance that will be passed to this class
            which provides the hook by which you can manipulate the QGIS
            application at run time.
        :type iface: QgsInterface
        """

        # Save reference to the QGIS interface
        self.iface = iface
        # initialize plugin directory
        self.plugin_dir = os.path.dirname(__file__)
        # initialize locale
        locale = QSettings().value("locale/userLocale")[0:2]
        locale_path = os.path.join(
            self.plugin_dir, "i18n", "ClassificationTool_{}.qm".format(locale)
        )

        if os.path.exists(locale_path):
            self.translator = QTranslator()
            self.translator.load(locale_path)

            if qVersion() > "4.3.3":
                QCoreApplication.installTranslator(self.translator)

        # Create the dialog (after translation) and keep reference
        self.dlg = ClassificationToolDialog()

        # Declare instance attributes
        self.actions = []
        self.menu = self.tr(u"&Supervised Classification")
        # TODO: We are going to let the user set this up in a future iteration
        self.toolbar = self.iface.addToolBar(u"ClassificationTool")
        self.toolbar.setObjectName(u"ClassificationTool")

    # noinspection PyMethodMayBeStatic
    def tr(self, message):
        """Get the translation for a string using Qt translation API.

        We implement this ourselves since we do not inherit QObject.

        :param message: String for translation.
        :type message: str, QString

        :returns: Translated version of message.
        :rtype: QString
        """
        # noinspection PyTypeChecker,PyArgumentList,PyCallByClass
        return QCoreApplication.translate("ClassificationTool", message)

    def add_action(
        self,
        icon_path,
        text,
        callback,
        enabled_flag=True,
        add_to_menu=True,
        add_to_toolbar=True,
        status_tip=None,
        whats_this=None,
        parent=None,
    ):
        """Add a toolbar icon to the toolbar.

        :param icon_path: Path to the icon for this action. Can be a resource
            path (e.g. ':/plugins/foo/bar.png') or a normal file system path.
        :type icon_path: str

        :param text: Text that should be shown in menu items for this action.
        :type text: str

        :param callback: Function to be called when the action is triggered.
        :type callback: function

        :param enabled_flag: A flag indicating if the action should be enabled
            by default. Defaults to True.
        :type enabled_flag: bool

        :param add_to_menu: Flag indicating whether the action should also
            be added to the menu. Defaults to True.
        :type add_to_menu: bool

        :param add_to_toolbar: Flag indicating whether the action should also
            be added to the toolbar. Defaults to True.
        :type add_to_toolbar: bool

        :param status_tip: Optional text to show in a popup when mouse pointer
            hovers over the action.
        :type status_tip: str

        :param parent: Parent widget for the new action. Defaults None.
        :type parent: QWidget

        :param whats_this: Optional text to show in the status bar when the
            mouse pointer hovers over the action.

        :returns: The action that was created. Note that the action is also
            added to self.actions list.
        :rtype: QAction
        """
        icon = QIcon(icon_path)
        action = QAction(icon, text, parent)
        action.triggered.connect(callback)
        action.setEnabled(enabled_flag)

        if status_tip is not None:
            action.setStatusTip(status_tip)

        if whats_this is not None:
            action.setWhatsThis(whats_this)

        if add_to_toolbar:
            # Adds plugin icon to Plugins toolbar
            self.iface.addToolBarIcon(action)

        if add_to_menu:
            self.iface.addPluginToVectorMenu(self.menu, action)

        self.actions.append(action)

        return action

    def initGui(self):
        """Create the menu entries and toolbar icons inside the QGIS GUI."""
        icon_path = ":/plugins/ClassificationTool/icon.png"
        self.add_action(
            icon_path,
            text=self.tr(u"ClassificationTool"),
            callback=self.run,
            parent=self.iface.mainWindow(),
        )

        self.dlg.tb_inVector.clicked.connect(self.openVector)
        self.dlg.tb_inRaster.clicked.connect(self.openRaster)
        self.dlg.tb_outRaster.clicked.connect(self.saveRaster)
        self.loadVectors()
        self.loadRasters()
        self.getClassifier()
        self.getOptions()
        self.dlg.bt_run.clicked.connect(self.process)
        self.dlg.sb_train.valueChanged.connect(self.update_spinbox_test)
        self.dlg.sb_test.valueChanged.connect(self.update_spinbox_train)

        # Tooltips
        QToolTip.setFont(QFont("SansSerif", 10))
        self.dlg.cb_class.setToolTip("Classification method")
        self.dlg.cb_inRaster.setToolTip("Raster layer / stack to be classified")
        self.dlg.cb_inVector.setToolTip("Shapefile with desired classes")
        self.dlg.cb_field.setToolTip("Field name containing the response category")
        self.dlg.le_outRaster.setToolTip("Output path for the classified raster image")

    def startWorker(self):

        self.dlg.progressBar.setRange(0, 0)
        worker = Worker(
            self.inRaster,
            self.inVector,
            self.outRaster,
            self.field,
            self.classifier,
            self.model_params,
            self.split_params,
            self.tiles,
            self.accass,
            self.sb_max_pix,
            self.tr,
        )

        self.worker = worker
        # start the worker in a new thread
        thread = self.thread = QThread()
        worker.moveToThread(thread)

        self.iface.messageBar().pushMessage(
            "INFO", "Classification started", level=Qgis.Info
        )

        worker.signals.finished.connect(self.workerFinished)
        worker.signals.error.connect(self.workerError)
        worker.signals.status.connect(self.updateStatus)
        thread.started.connect(worker.run)

        thread.start()

    def workerFinished(self):
        self.worker.deleteLater()
        self.thread.quit()
        self.thread.wait()
        self.thread.deleteLater()
        self.addLayers()
        self.dlg.progressBar.setRange(0, 1)
        # self.iface.messageBar().pushMessage("INFO", "Classification finished", level=Qgis.Info)

    def workerError(self, current_time, exceptionString):
        item = QListWidgetItem(current_time + " : ERROR: " + exceptionString)
        item.setForeground(Qt.red)
        self.dlg.process_list.addItem(item)

        enditem = QListWidgetItem(current_time + "Process aborted...")
        self.dlg.process_list.addItem(enditem)

        self.worker.deleteLater()
        self.thread.quit()
        self.thread.wait()
        self.thread.deleteLater()
        self.dlg.progressBar.setRange(0, 1)

    def updateStatus(self, current_time, status_text):
        self.dlg.process_list.addItem(current_time + " : " + status_text)

    def loadVectors(self):
        """Load vectors for QGIS table of contents"""
        self.dlg.cb_inVector.clear()
        self.dlg.cb_field.clear()
        layers = [layer for layer in QgsProject.instance().mapLayers().values()]
        vector_layers = []
        for layer in layers:
            if layer.type() == QgsMapLayer.VectorLayer:
                vector_layers.append(layer.name())
        self.dlg.cb_inVector.addItems(vector_layers)
        self.dlg.cb_inVector.currentIndexChanged.connect(self.field_select)

    def loadRasters(self):
        """Load rasters for QGIS table of contents"""
        self.dlg.cb_inRaster.clear()
        layers = [layer for layer in QgsProject.instance().mapLayers().values()]
        raster_layers = []
        for layer in layers:
            if layer.type() == QgsMapLayer.RasterLayer:
                raster_layers.append(layer.name())
        self.dlg.cb_inRaster.addItems(raster_layers)

    def change_layers(self):
        self.dlg.cb_inVector.clear()
        layers = self.iface.legendInterface().layers()
        layer_list = [layer.name() for layer in layers]
        self.dlg.cb_inVector.addItems(layer_list)

    def openVector(self):
        """Open vector from file dialog"""
        inFile = str(
            QFileDialog.getOpenFileName(
                caption="Open shapefile", filter="Shapefiles (*.shp)"
            )[0]
        )
        if inFile is not None:
            self.iface.addVectorLayer(
                inFile, str.split(os.path.basename(inFile), ".")[0], "ogr"
            )
            self.loadVectors()

    def openRaster(self):
        """Open raster from file dialog"""
        inFile = str(
            QFileDialog.getOpenFileName(
                caption="Open raster", filter="GeoTiff (*.tif)"
            )[0]
        )
        if inFile is not None:
            self.iface.addRasterLayer(
                inFile, str.split(os.path.basename(inFile), ".")[0]
            )
            self.loadRasters()

    def saveRaster(self):
        """Get the save file name for the clipped raster from a file dialog"""
        outFile = str(
            QFileDialog.getSaveFileName(
                caption="Save clipped raster as", filter="GeoTiff (*.tif)"
            )[0]
        )
        self.setRasterLine(outFile)

    def setRasterLine(self, text):
        self.dlg.le_outRaster.setText(text)

    def getVectorLayer(self):
        layer = None
        layername = self.dlg.cb_inVector.currentText()
        for lyr in QgsProject.instance().mapLayers().values():
            if lyr.name() == layername:
                layer = lyr
                break
        return layer

    def getRasterLayer(self):
        """Gets vector layer specified in combo box"""
        layer = None
        layername = self.dlg.cb_inRaster.currentText()
        for lyr in QgsProject.instance().mapLayers().values():
            if lyr.name() == layername:
                layer = lyr
                break
        return layer

    def field_select(self, index):
        # get list of all vector layers in QGIS
        layers = [layer for layer in QgsProject.instance().mapLayers().values()]
        listLayers = [
            layer for layer in layers if layer.type() == QgsMapLayer.VectorLayer
        ]

        # get name of selected layer
        provider = listLayers[index].dataProvider()
        fields = provider.fields()
        listFieldNames = [field.name() for field in fields]

        # clear the combo box comboFieldList
        self.dlg.cb_field.clear()

        # add all these field names to combo box comboFieldList
        self.dlg.cb_field.addItems(listFieldNames)

    def getClassifier(self):
        classifier_opt = ["RandomForest", "KNearestNeighbor", "SVC"]
        self.dlg.cb_class.addItems(classifier_opt)

    def update_spinbox_train(self):
        self.dlg.sb_train.setValue(100 - self.dlg.sb_test.value())

    def update_spinbox_test(self):
        self.dlg.sb_test.setValue(100 - self.dlg.sb_train.value())

    def getOptions(self):
        weights = ["uniform", "distance"]
        algorithm = ["auto", "ball_tree", "kd_tree", "brute"]
        metrics = [
            "euclidean",
            "minkowski",
            "manhattan",
            "chebyshev",
            "seuclidean",
            "mahalanobis",
            "wminkowski",
            "hamming",
            "canberra",
            "braycurtis",
            "matching",
            "jaccard",
            "dice",
            "kulsinski",
            "rogerstanimoto",
            "russellrao",
            "sokalmichener",
            "sokalsneath",
            "haversine",
        ]
        self.dlg.sb_kn_n.setValue(5)
        self.dlg.sb_kn_n.setToolTip("Number of neighbors to use")
        self.dlg.cb_kn_weights.addItems(weights)
        self.dlg.cb_kn_weights.setToolTip("Weight function used in prediction")
        self.dlg.cb_kn_metric.addItems(metrics)
        self.dlg.cb_kn_metric.setToolTip("Distance metric to use for the tree")
        self.dlg.cb_kn_algo.addItems(algorithm)
        self.dlg.cb_kn_algo.setToolTip(
            "Algorithm used to compute the nearest neighbors"
        )
        self.dlg.sb_kn_leafsize.setValue(30)
        self.dlg.sb_kn_leafsize.setToolTip("Leaf size passed to BallTree or KDTree")
        self.dlg.sb_kn_pvalue.setValue(2)
        self.dlg.sb_kn_pvalue.setToolTip("Power parameter for the Minkowski metric")
        self.dlg.sb_kn_njobs.setValue(-1)
        self.dlg.sb_kn_njobs.setToolTip(
            "The number of parallel jobs to run for neighbors search"
        )
        bootstrap = ["True", "False"]
        max_features = ["auto", "sqrt", "log", "None"]
        criterion = ["gini", "entropy"]
        oob_score = ["True", "False"]
        self.dlg.sb_rf_nestimator.setValue(10)
        self.dlg.sb_rf_nestimator.setToolTip("The number of trees in the forest")
        self.dlg.cb_rf_criterion.addItems(criterion)
        self.dlg.cb_rf_criterion.setToolTip(
            "Function to measure the quality of a split"
        )
        self.dlg.cb_rf_bootstrap.addItems(bootstrap)
        self.dlg.cb_rf_bootstrap.setToolTip(
            "Whether bootstrap samples are used when building trees"
        )
        self.dlg.sb_rf_min_sample.setValue(2.0)
        self.dlg.sb_rf_min_sample.setToolTip(
            "The minimum number of samples required to split an internal node"
        )
        self.dlg.sb_rf_min_samples_leaf.setValue(1.0)
        self.dlg.sb_rf_min_samples_leaf.setToolTip(
            "The minimum number of samples required to be at a leaf node"
        )
        self.dlg.sb_rf_min_weight_fraction_leaf.setValue(0.0)
        self.dlg.sb_rf_min_weight_fraction_leaf.setToolTip(
            "The minimum weighted fraction of the sum total of weights (of all the input samples) required to be at a leaf node"
        )
        self.dlg.cb_rf_max_features.addItems(max_features)
        self.dlg.cb_rf_max_features.setToolTip("")
        self.dlg.le_rf_max_leaf_nodes.setText("None")
        self.dlg.le_rf_max_leaf_nodes.setToolTip(
            "Grow trees with max_leaf_nodes in best-first fashion"
        )
        self.dlg.sb_rf_min_impurity_decrease.setValue(0.0)
        self.dlg.sb_rf_min_impurity_decrease.setToolTip(
            "A node will be split if this split induces a decrease of the impurity greater than or equal to this value"
        )
        self.dlg.le_rf_max_depth.setText("None")
        self.dlg.le_rf_max_depth.setToolTip("The maximum depth of the tree")
        self.dlg.sb_rf_njobs.setValue(-1)
        self.dlg.sb_rf_njobs.setToolTip(
            "The number of jobs to run in parallel for both fit and predict"
        )
        kernel = ["rbf", "linear", "poly", "sigmoid", "precomputed"]
        shrink = ["True", "False"]
        dcs = ["ovo", "ovr"]
        class_weights = ["None", "balanced"]
        prob = ["True", "False"]
        self.dlg.sb_sv_c.setValue(1.0)
        self.dlg.sb_sv_c.setToolTip("Penalty parameter C of the error term")
        self.dlg.cb_sv_kernel.addItems(kernel)
        self.dlg.cb_sv_kernel.setToolTip(
            "Specifies the kernel type to be used in the algorithm"
        )
        self.dlg.sb_sv_degree.setValue(3)
        self.dlg.sb_sv_degree.setToolTip(
            "Degree of the polynomial kernel function (‘poly’)"
        )
        self.dlg.sb_sv_coef.setValue(0.0)
        self.dlg.sb_sv_coef.setToolTip("Independent term in kernel function.")
        self.dlg.cb_sv_shrink.addItems(shrink)
        self.dlg.cb_sv_shrink.setToolTip("Whether to use the shrinking heuristic")
        self.dlg.cb_sv_dcs.addItems(dcs)
        self.dlg.cb_sv_dcs.setToolTip(
            "Whether to return a one-vs-rest (‘ovr’) decision function of shape (n_samples, n_classes) as all other classifiers, or the original one-vs-one (‘ovo’) decision function of libsvm which has shape (n_samples, n_classes * (n_classes - 1) / 2)"
        )
        self.dlg.cb_sv_classweight.addItems(class_weights)
        self.dlg.cb_sv_classweight.setToolTip(
            "Set the parameter C of class i to class_weight[i]*C for SVC. If not given, all classes are supposed to have weight one. The “balanced” mode uses the values of y to automatically adjust weights inversely proportional to class frequencies in the input data"
        )
        self.dlg.sb_sv_maxiter.setValue(-1)
        self.dlg.sb_sv_maxiter.setToolTip(
            "Hard limit on iterations within solver, or -1 for no limit."
        )
        self.dlg.sb_sv_tol.setValue(0.001)
        self.dlg.sb_sv_tol.setToolTip("Tolerance for stopping criterion")
        self.dlg.cb_sv_probability.addItems(prob)
        self.dlg.cb_sv_probability.setToolTip(
            "Whether to enable probability estimates. This must be enabled prior to calling fit, and will slow down that method."
        )
        self.dlg.sb_sv_cache_size.setValue(200)
        self.dlg.sb_sv_cache_size.setToolTip(
            "Specify the size of the kernel cache (in MB)"
        )
        self.dlg.le_sv_randstate.setText("None")
        self.dlg.le_sv_randstate.setToolTip(
            "The seed of the pseudo random number generator used when shuffling the data for probability estimates"
        )

        self.dlg.sb_tiles.setValue(1)
        self.dlg.sb_tiles.setToolTip(
            "Split raster into multiple tiles"
        )
        self.dlg.sb_max_pix.setValue(-1)
        self.dlg.sb_max_pix.setToolTip(
            "Number of pixels sampled per polygon (-1 = all pixels)"
        )
        self.dlg.check_acc.setToolTip(
            "Calculate accuracy assessment"
        )
        self.dlg.sb_train.setToolTip(
            "Percentage of data used for training"
        )
        self.dlg.sb_test.setToolTip(
            "Percentage of data used for testing"
        )

        self.dlg.check_strat.setToolTip(
                "Split data  in a stratified fashion"
            )

    def addLayers(self):
        # self.iface.addVectorLayer(self.outRaster, str.split(os.path.basename(self.outRaster),".")[0], "ogr")
        self.iface.addRasterLayer(
            self.outRaster, str.split(os.path.basename(self.outRaster), ".")[0]
        )

    def setVariables(self):
        """Get and set all variables from UI"""
        self.inVector = self.getVectorLayer()
        self.inRaster = self.getRasterLayer()
        self.outRaster = self.dlg.le_outRaster.text()
        self.field = self.dlg.cb_field.currentText()
        self.classifier = self.dlg.cb_class.currentText()
        self.accass = self.dlg.check_acc.isChecked()
        self.tiles = self.dlg.sb_tiles.value()
        self.sb_max_pix = self.dlg.sb_max_pix.value()

        self.split_params = {
            "test_size": self.dlg.sb_train.value(),
            "stratify": self.dlg.check_strat.isChecked(),
        }

        if self.dlg.cb_class.currentText() == "KNearestNeighbor":
            self.model_params = {
                "n_neighbors": self.dlg.sb_kn_n.value(),
                "weights": self.dlg.cb_kn_weights.currentText(),
                "algorithm": self.dlg.cb_kn_algo.currentText(),
                "metric": self.dlg.cb_kn_metric.currentText(),
                "leaf_size": self.dlg.sb_kn_leafsize.value(),
                "p": self.dlg.sb_kn_pvalue.value(),
                "n_jobs": self.dlg.sb_kn_njobs.value(),
            }

        if self.classifier == "RandomForest":

            if self.dlg.le_rf_max_depth.text() != "None":
                self.max_depth = int(self.dlg.le_rf_max_depth.text())
            else:
                self.max_depth = None

            if self.dlg.le_rf_max_leaf_nodes.text() != "None":
                self.max_leaf_nodes = int(self.dlg.le_rf_max_leaf_nodes.text())
            else:
                self.max_leaf_nodes = None

            if self.dlg.cb_rf_bootstrap.currentText() == "True":
                self.bootstrap = True
            if self.dlg.cb_rf_bootstrap.currentText() == "False":
                self.bootstrap = False

            self.model_params = {
                "n_estimators": self.dlg.sb_rf_nestimator.value(),
                "criterion": self.dlg.cb_rf_criterion.currentText(),
                "min_samples_split": self.dlg.sb_rf_min_sample.value(),
                "min_samples_leaf": self.dlg.sb_rf_min_samples_leaf.value(),
                "min_weight_fraction_leaf": self.dlg.sb_rf_min_weight_fraction_leaf.value(),
                "max_features": self.dlg.cb_rf_max_features.currentText(),
                "max_depth": self.max_depth,
                "max_leaf_nodes": self.max_leaf_nodes,
                "min_impurity_decrease": self.dlg.sb_rf_min_impurity_decrease.value(),
                "bootstrap": self.bootstrap,
                "n_jobs": self.dlg.sb_rf_njobs.value(),
            }

        if self.dlg.cb_class.currentText() == "SVC":
            if self.dlg.le_sv_randstate.text() != "None":
                self.randstate = int(self.dlg.le_sv_randstate.text())
            else:
                self.randstate = None

            if self.dlg.cb_sv_classweight.currentText() != "None":
                self.classweight = int(self.cb_sv_classweight.currentText())
            else:
                self.classweight = None

            if self.dlg.cb_sv_shrink.currentText() == "True":
                self.shrink = True
            if self.dlg.cb_sv_shrink.currentText() == "False":
                self.shrink = False

            if self.dlg.cb_sv_probability.currentText() == "True":
                self.probability = True
            if self.dlg.cb_sv_probability.currentText() == "False":
                self.probability = False

            ##TODO:all parameter
            self.model_params = {
                "C": self.dlg.sb_sv_c.value(),
                "kernel": self.dlg.cb_sv_kernel.currentText(),
                "degree": self.dlg.sb_sv_degree.value(),
                "coef0": self.dlg.sb_sv_coef.value(),
                "shrinking": self.shrink,
                "probability": self.probability,
                "tol": self.dlg.sb_sv_tol.value(),
                "cache_size": self.dlg.sb_sv_cache_size.value(),
                "class_weight": self.classweight,
                "max_iter": self.dlg.sb_sv_maxiter.value(),
                "decision_function_shape": self.dlg.cb_sv_dcs.currentText(),
                "random_state": self.randstate,
            }

    def unload(self):
        """Removes the plugin menu item and icon from QGIS GUI."""
        for action in self.actions:
            self.iface.removePluginMenu(self.tr(u"&Classification Tool"), action)
            self.iface.removeToolBarIcon(action)
        # remove the toolbar
        del self.toolbar

    def run(self):

        self.dlg.show()
        self.loadVectors()
        self.loadRasters()
        self.dlg.cb_inVector.currentIndexChanged.connect(self.field_select)

    def process(self):
        self.setVariables()
        # self.dlg.progressBar.setRange(0,0)

        self.startWorker()
        # self.dlg.progressBar.setRange(0,1)