enpt_external_algorithm.py 10.4 KB
Newer Older
1
2
3
4
# -*- coding: utf-8 -*-

# enpt_enmapboxapp, A QGIS EnMAPBox plugin providing a GUI for the EnMAP processing tools (EnPT)
#
5
# Copyright (C) 2018-2021 Daniel Scheffler (GFZ Potsdam, daniel.scheffler@gfz-potsdam.de)
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#
# This software was developed within the context of the EnMAP project supported
# by the DLR Space Administration with funds of the German Federal Ministry of
# Economic Affairs and Energy (on the basis of a decision by the German Bundestag:
# 50 EE 1529) and contributions from DLR, GFZ and OHB System AG.
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program.  If not, see <http://www.gnu.org/licenses/>.

"""This module provides the ExternalEnPTAlgorithm which is used in case EnPT is installed into separate environment."""

import os
28
from subprocess import check_output, CalledProcessError
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from glob import glob
from qgis.core import \
    (QgsProcessingContext,
     QgsProcessingFeedback,
     QgsProcessingParameterFile,
     NULL
     )

from ._enpt_alg_base import _EnPTBaseAlgorithm


class ExternalEnPTAlgorithm(_EnPTBaseAlgorithm):
    # Input parameters
    P_anaconda_root = 'anaconda_root'

    def initAlgorithm(self, configuration=None):
        self.addParameter(QgsProcessingParameterFile(
            name=self.P_anaconda_root,
            description='Anaconda root directory (which contains the EnPT Python environment in a subdirectory)',
            behavior=QgsProcessingParameterFile.Folder,
            defaultValue=self._get_default_anaconda_root(),
            optional=True))

        super().initAlgorithm(configuration=configuration)

    @staticmethod
    def _get_default_anaconda_root():
        if os.getenv('ANACONDA_ROOT') and os.path.exists(os.getenv('ANACONDA_ROOT')):
            return os.getenv('ANACONDA_ROOT')
        elif os.name == 'nt':
            return 'C:\\ProgramData\\Anaconda3'
        else:
            return ''  # FIXME is there a default location in Linux/OSX?

    @staticmethod
    def _locate_EnPT_Anaconda_environment(user_root):
        anaconda_rootdir = None

        if user_root and os.path.exists(user_root):
            anaconda_rootdir = user_root

        elif os.getenv('ANACONDA_ROOT') and os.path.exists(os.getenv('ANACONDA_ROOT')):
            anaconda_rootdir = os.getenv('ANACONDA_ROOT')

        else:
            possPaths = [
                'C:\\ProgramData\\Anaconda3',
                'C:\\Users\\%s\\Anaconda3' % os.getenv('username')
                 ] if os.name == 'nt' else \
                []

            for rootDir in possPaths:
                if os.path.exists(rootDir):
                    anaconda_rootdir = rootDir

        if not anaconda_rootdir:
            raise NotADirectoryError("No valid Anaconda root directory given - "
                                     "neither via the GUI, nor via the 'ANACONDA_ROOT' environment variable.")

        # set ENPT_PYENV_ACTIVATION environment variable
        os.environ['ENPT_PYENV_ACTIVATION'] = \
            os.path.join(anaconda_rootdir, 'Scripts', 'activate.bat') if os.name == 'nt' else \
            os.path.join(anaconda_rootdir, 'bin', 'activate')

        if not os.path.exists(os.getenv('ENPT_PYENV_ACTIVATION')):
            raise FileNotFoundError(os.getenv('ENPT_PYENV_ACTIVATION'))

        return anaconda_rootdir

    @staticmethod
    def _is_enpt_environment_present(anaconda_rootdir):
        return os.path.exists(os.path.join(anaconda_rootdir, 'envs', 'enpt'))

    @staticmethod
    def _locate_enpt_run_script():
        try:
            if os.name == 'nt':
                # Windows
                return check_output('where enpt_run_cmd.bat', shell=True).decode('UTF-8').strip()
                # return "D:\\Daten\\Code\\python\\enpt_enmapboxapp\\bin\\enpt_run_cmd.bat"
            else:
                # Linux / OSX
                return check_output('which enpt_run_cmd.sh', shell=True).decode('UTF-8').strip()
                # return 'enpt_run_cmd.sh '

        except CalledProcessError:
            raise EnvironmentError('The EnPT run script could not be found. Please make sure, that enpt_enmapboxapp '
                                   'is correctly installed into your QGIS Python environment.')

    @staticmethod
119
    def _prepare_enpt_environment() -> dict:
120
        os.environ['PYTHONUNBUFFERED'] = '1'
121
        os.environ['IS_ENPT_GUI_CALL'] = '1'
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

        enpt_env = os.environ.copy()
        enpt_env["PATH"] = ';'.join([i for i in enpt_env["PATH"].split(';') if 'OSGEO' not in i])  # actually not needed
        if "PYTHONHOME" in enpt_env.keys():
            del enpt_env["PYTHONHOME"]
        if "PYTHONPATH" in enpt_env.keys():
            del enpt_env["PYTHONPATH"]

        # FIXME is this needed?
        enpt_env['IPYTHONENABLE'] = 'True'
        enpt_env['PROMPT'] = '$P$G'
        enpt_env['PYTHONDONTWRITEBYTECODE'] = '1'
        enpt_env['PYTHONIOENCODING'] = 'UTF-8'
        enpt_env['TEAMCITY_VERSION'] = 'LOCAL'
        enpt_env['O4W_QT_DOC'] = 'C:/OSGEO4~3/apps/Qt5/doc'
        if 'SESSIONNAME' in enpt_env.keys():
            del enpt_env['SESSIONNAME']

        # import pprint
        # s = pprint.pformat(enpt_env)
        # with open('D:\\env.json', 'w') as fp:
        #     fp.write(s)

        return enpt_env

    def processAlgorithm(self, parameters, context, feedback):
        assert isinstance(parameters, dict)
        assert isinstance(context, QgsProcessingContext)
        assert isinstance(feedback, QgsProcessingFeedback)

        anaconda_root = self._locate_EnPT_Anaconda_environment(parameters[self.P_anaconda_root])
        feedback.pushInfo('Found Anaconda installation at %s.' % anaconda_root)

        if self._is_enpt_environment_present(anaconda_root):
            feedback.pushInfo("The Anaconda installation contains the 'enpt' environment as expected.")
        else:
            feedback.reportError("The Anaconda installation has no environment called 'enpt'. Please follow the EnPT "
                                 "installation instructions to install the EnMAP processing tool backend code "
                                 "(see https://enmap.git-pages.gfz-potsdam.de/GFZ_Tools_EnMAP_BOX/EnPT/doc/"
                                 "installation.html). This is needed to run EnPT from this GUI.")
            return {
                'success': False,
                self.P_OUTPUT_RASTER: '',
                # self.P_OUTPUT_VECTOR: parameters[self.P_OUTPUT_RASTER],
                # self.P_OUTPUT_FILE: parameters[self.P_OUTPUT_RASTER],
                self.P_OUTPUT_FOLDER: ''
            }

        # remove all parameters not to be forwarded to the EnPT CLI
171
172
        parameters = {k: v for k, v in parameters.items()
                      if k not in ['anaconda_root']}
173

174
175
176
        # replace Enum parameters with corresponding strings
        for n, opts in [
            ('output_format', {0: 'GTiff', 1: 'ENVI'}),
177
            ('mode_ac', {0: 'land', 1: 'water', 2: 'combined'}),
178
179
180
181
182
183
184
            ('deadpix_P_algorithm', {0: 'spectral', 1: 'spatial'}),
            ('deadpix_P_interp_spectral', {0: 'linear', 1: 'bilinear', 2: 'cubic', 3: 'spline'}),
            ('deadpix_P_interp_spatial', {0: 'linear', 1: 'bilinear', 2: 'cubic', 3: 'spline'}),
            ('ortho_resampAlg', {0: 'nearest', 1: 'bilinear', 2: 'gauss'}),
            ('vswir_overlap_algorithm', {0: 'order_by_wvl', 1: 'average', 2: 'vnir_only', 3: 'swir_only'}),
            ('target_projection_type', {0: 'UTM', 1: 'Geographic'}),
        ]:
185
186
            if isinstance(parameters[n], int):
                parameters[n] = opts[parameters[n]]
187

188
189
190
191
192
193
194
195
196
        # print parameters and console call to log
        # for key in sorted(parameters):
        #     feedback.pushInfo('{} = {}'.format(key, repr(parameters[key])))
        keyval_str = ' '.join(['--{} {}'.format(key, parameters[key])
                               for key in sorted(parameters)
                               if parameters[key] not in [None, NULL, 'NULL', '']])
        print(parameters)
        print(keyval_str + '\n\n')
        feedback.pushInfo("\nCalling EnPT with the following command:\n"
197
                          "enpt %s\n\n" % keyval_str)
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212

        # prepare environment for subprocess
        enpt_env = self._prepare_enpt_environment()
        path_enpt_runscript = self._locate_enpt_run_script()

        # run EnPT in subprocess that activates the EnPT Anaconda environment
        feedback.pushDebugInfo('Using %s to start EnPT.' % path_enpt_runscript)
        feedback.pushInfo("The log messages of the EnMAP processing tool are written to the *.log file "
                          "in the specified output folder.")

        self._run_cmd("%s %s" % (path_enpt_runscript, keyval_str),
                      qgis_feedback=feedback,
                      env=enpt_env)

        # list output dir
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
        if 'output_dir' in parameters:
            outdir = parameters['output_dir']
            outraster_matches = glob(os.path.join(outdir, '*', '*SPECTRAL_IMAGE.GEOTIFF'))
            outraster = outraster_matches[0] if len(outraster_matches) > 0 else None

            feedback.pushInfo("The output folder '%s' contains:\n" % outdir)
            feedback.pushCommandInfo('\n'.join([os.path.basename(f) for f in os.listdir(outdir)]) + '\n')

            if outraster:
                subdir = os.path.dirname(outraster_matches[0])
                feedback.pushInfo("...where the folder '%s' contains:\n" % os.path.dirname(subdir))
                feedback.pushCommandInfo('\n'.join([os.path.basename(f) for f in os.listdir(subdir)]) + '\n')

            # return outputs
            return {
                'success': True,
                self.P_OUTPUT_RASTER: outraster,
                # self.P_OUTPUT_VECTOR: parameters[self.P_OUTPUT_RASTER],
                # self.P_OUTPUT_FILE: parameters[self.P_OUTPUT_RASTER],
                self.P_OUTPUT_FOLDER: outdir
            }

        else:
            feedback.pushInfo('The output was skipped according to user setting.')

            # return outputs
            return {'success': True}