Python osgeo.gdal 模块，SetConfigOption() 实例源码

def convert_and_import(xml_file):
    QgsProject.instance().clear()
    with tempfile.NamedTemporaryFile(delete=True) as f:
        out_f = f.name
    config_file = os.path.join(os.path.dirname(__file__), "gmlasconf.xml")
    gdal.SetConfigOption("OGR_SQLITE_SYNCHRONOUS", "OFF")
    ds = gdal.OpenEx("GMLAS:{}".format(xml_file), open_options=['EXPOSE_METADATA_LAYERS=YES', 'CONFIG_FILE={}'.format(config_file)])
    srs = osr.SpatialReference()
    qgs_srs = QgsCoordinateReferenceSystem("EPSG:4326")
    srs.ImportFromWkt(qgs_srs.toWkt())
    params = {
        'destNameOrDestDS': out_f
        , 'srcDS': ds
        , 'format': "SQLite"
        , 'accessMode': "overwrite"
        , 'datasetCreationOptions': ['SPATIALITE=YES']
        , 'options' : ['-forceNullable', '-skipfailures']
        #, 'srcSRS': srs
        #, 'dstSRS': srs
        , 'geometryType': 'CONVERT_TO_LINEAR'
        , 'reproject': False
    }
    # call gdal to convert
    gdal.VectorTranslate(**params)
    # fix geometry types
    ds = None
    # populate the qgis project
    import_in_qgis(out_f, "SQLite")

    layers = []
    for lid in sorted(QgsProject.instance().mapLayers().keys()):
        vl = QgsProject.instance().mapLayer(lid)
        layers.append((vl.name(), vl.wkbType()))
    rels = []
    relations = QgsProject.instance().relationManager().relations()
    for relid in sorted(relations.keys()):
        rel = relations[relid]
        p = rel.fieldPairs()
        rels.append((rel.id()[0:3], rel.referencingLayer().name(), list(p.keys())[0], rel.referencedLayer().name(), list(p.values())[0]))

    return sorted(layers), sorted(rels)

def __exit__(self, type, value, tb):
        # restore previous settings
        if self.http_proxy is not None:
            os.environ['http_proxy'] = self.http_proxy
            os.environ['http_proxy'] = self.http_proxy
        else:
            os.environ.pop('http_proxy', None)
            os.environ.pop('https_proxy', None)
        if self.no_proxy is not None:
            os.environ['no_proxy'] = self.no_proxy
        else:
            os.environ.pop('no_proxy', None)
        gdal.SetConfigOption("GDAL_HTTP_PROXY", self.gdal_http_proxy)
        gdal.SetConfigOption("GDAL_HTTP_PROXYUSERPWD", self.gdal_http_proxyuserpwd)

def accept(self):
        gdal.SetConfigOption("OGR_SQLITE_SYNCHRONOUS", "OFF")
        with NamedTemporaryFile(mode="w+t", suffix='.sqlite', delete=True) as out:
            temp_datasource_path = out.name
        try:
            self.translate(self.reproject_params(temp_datasource_path))
            self.translate(self.export_params(temp_datasource_path))
        except InputError as e:
            e.show()

        return QDialog.accept()

def on_convertButton_clicked(self):
        gdal.SetConfigOption("OGR_SQLITE_SYNCHRONOUS", "OFF")
        gdal.SetConfigOption('GDAL_HTTP_UNSAFESSL', 'YES')

        dest = self.databaseWidget.datasource_name()
        if dest == '' and self.databaseWidget.format() == "SQLite":
            with tempfile.NamedTemporaryFile(suffix='.sqlite') as tmp:
                dest = tmp.name

        if dest.startswith('PG:'):
            schema = self.databaseWidget.schema()
        else:
            schema = None

        try:
            QApplication.setOverrideCursor(Qt.WaitCursor)
            self.translate(self.import_params(dest))
            import_in_qgis(dest, self.databaseWidget.format(), schema)

        except InputError as e:
            e.show()
        except RuntimeError as e:
            QMessageBox.warning(None,
                                plugin_name(),
                                e.args[0])
        finally:
            QApplication.restoreOverrideCursor()

def __enter__(self):
        # keep previous config
        self.http_proxy = os.environ.get("http_proxy")
        self.https_proxy = os.environ.get("https_proxy")
        self.no_proxy = os.environ.get("no_proxy")
        self.gdal_http_proxy = gdal.GetConfigOption("GDAL_HTTP_PROXY")
        self.gdal_http_proxyuserpwd = gdal.GetConfigOption("GDAL_HTTP_PROXYUSERPWD")

        # apply QGIS proxy settings
        settings = QSettings()
        enabled = bool(settings.value("proxy/proxyEnabled", False))
        type = settings.value("proxy/proxyType", "")
        host = settings.value("proxy/proxyHost", "")
        port = settings.value("proxy/proxyPort", "")
        user = settings.value("proxy/proxyUser", "")
        password = settings.value("proxy/proxyPassword", "")
        excludes = settings.value("proxy/proxyExcludedUrls", "")
        if hasattr(excludes, 'isNull') and excludes.isNull():
            excludes = []
        else:
            excludes = excludes.split("|")

        http_proxy = ""
        no_proxy = ""
        gdal_http_proxy = ""
        gdal_http_proxyuserpwd = ""

        if enabled:
            if type == "HttpProxy":
                credentials = ""
                if user != "":
                    credentials = "{}:{}@".format(user, password)
                http_proxy = "http://{}{}:{}".format(credentials, host, port)
                no_proxy = ",".join(excludes)

            os.environ["http_proxy"] = http_proxy
            os.environ["https_proxy"] = http_proxy
            os.environ["no_proxy"] = no_proxy

            gdal_http_proxy = "{}:{}".format(host, port)
            gdal.SetConfigOption('GDAL_HTTP_PROXY', gdal_http_proxy)
            if user != "":
                gdal_http_proxyuserpwd = "{}:{}".format(user, password)
                gdal.SetConfigOption('GDAL_HTTP_PROXYUSERPWD', gdal_http_proxyuserpwd)

def validate(args):
    if 'url' not in args:
        return json.dumps({'status': 'failure', 'error': 'url missing'}), 400, \
               { "Content-Type": "application/json" }

    remove_tmpfile = False
    url = args.get('url')
    if 'local_filename' in args:
        ds = gdal.OpenEx(args['local_filename'], allowed_drivers = ['GTiff'])
    else:

        use_vsicurl = args.get('use_vsicurl', 'true')
        if use_vsicurl.lower() not in ('true', 'false'):
            return json.dumps({'status': 'failure', 'error': 'invalid value for use_vsicurl option. Expected true or false'}), 400, { "Content-Type": "application/json" }
        use_vsicurl = use_vsicurl.lower() == 'true'

        gdal.SetConfigOption('GDAL_DISABLE_READDIR_ON_OPEN', 'EMPTY_DIR')
        if use_vsicurl:
            ds = gdal.OpenEx('/vsicurl/' + url, allowed_drivers = ['GTiff'])
            if ds is None:
                f = gdal.VSIFOpenL('/vsicurl/' + url, 'rb')
                if f is None:
                    return json.dumps({'status': 'failure', 'error': 'Cannot download %s' % url}), 400, { "Content-Type": "application/json" }
                data = gdal.VSIFReadL(1,1,f)
                gdal.VSIFCloseL(f)
                if len(data) == 0:
                    error_msg = 'Cannot download %s' % url
                    gdal_error_msg = gdal.GetLastErrorMsg()
                    if gdal_error_msg == '':
                        gdal_error_msg = gdal.VSIGetLastErrorMsg()
                    if gdal_error_msg != '':
                        error_msg += ': '+ gdal_error_msg
                    return json.dumps({'status': 'failure', 'error': error_msg}), 400, { "Content-Type": "application/json" }
        else:
            try:
                r = requests.get(url)
            except Exception, e:
                return json.dumps({'status': 'failure', 'error': 'Cannot download %s' % url}), 400, { "Content-Type": "application/json" }

            remove_tmpfile = True
            f = open(tmpfilename, 'wb')
            f.write(r.content)
            f.close()
            ds = gdal.OpenEx(tmpfilename, allowed_drivers = ['GTiff'])

    if ds is None:
        return json.dumps({'status': 'failure', 'error': '%s is not a GTiff file' % url}), 400, { "Content-Type": "application/json" }
    errors, details = validate_cloud_optimized_geotiff.validate(ds)
    info = gdal.Info(ds, format = 'json')
    if 'local_filename' in args or remove_tmpfile:
        del info['files']
    info['description'] = url
    ds = None
    if remove_tmpfile:
        os.unlink(tmpfilename)

    if len(errors) == 0:
        return json.dumps({'status': 'success', 'gdal_info' : info, 'details': details}), 200, { "Content-Type": "application/json" }
    else:
        return json.dumps({'status': 'failure', 'gdal_info' : info, 'details': details, 'validation_errors': errors}), 400, { "Content-Type": "application/json" }