enpt_algorithm.py 6.45 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# -*- coding: utf-8 -*-

# enpt_enmapboxapp, A QGIS EnMAPBox plugin providing a GUI for the EnMAP processing tools (EnPT)
#
# Copyright (C) 2019  Daniel Scheffler (GFZ Potsdam, daniel.scheffler@gfz-potsdam.de)
#
# 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 EnPTAlgorithm which is used in case EnPT is installed into QGIS Python environment."""

import os
from pkgutil import find_loader
from glob import glob
30
from pprint import pprint
31
32
33
34
35
36
37
38
39
40
41

from qgis.core import \
    (QgsProcessingContext,
     QgsProcessingFeedback,
     NULL
     )

from ._enpt_alg_base import _EnPTBaseAlgorithm


class EnPTAlgorithm(_EnPTBaseAlgorithm):
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
    @staticmethod
    def _prepare_enpt_environment() -> dict:
        os.environ['PYTHONUNBUFFERED'] = '1'

        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

70
71
72
73
74
75
76
77
78
    def processAlgorithm(self, parameters, context, feedback):
        assert isinstance(parameters, dict)
        assert isinstance(context, QgsProcessingContext)
        assert isinstance(feedback, QgsProcessingFeedback)

        if not find_loader('enpt'):
            raise ImportError("enpt", "EnPT must be installed into the QGIS Python environment "
                                      "when calling 'EnPTAlgorithm'.")

79
80
81
82
83
84
85
86
87
88
89
90
        # replace Enum parameters with corresponding strings
        for n, opts in [
            ('output_format', {0: 'GTiff', 1: 'ENVI'}),
            ('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'}),
        ]:
            parameters[n] = opts[parameters[n]]

91
92
93
94
95
96
97
98
        feedback.pushInfo("The log messages of the EnMAP processing tool are written to the *.log file "
                          "in the specified output folder.")

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

99
100
101
102
103
104
        # 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', '']])
105
        pprint(parameters)
106
107
108
109
110
111
112
113
        print(keyval_str + '\n\n')
        feedback.pushInfo("\nCalling EnPT with the following command:\n"
                          "enpt %s\n\n" % keyval_str)

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

114
        # run EnPT in subprocess that activates the EnPT Anaconda environment
115
116
117
118
119
120
121
122
        # 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(f"enpt {keyval_str}",
                      qgis_feedback=feedback,
                      env=enpt_env
                      )
123
124

        # list output dir
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
        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}