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

def create_mask_from_vector(vector_data_path, cols, rows, geo_transform, projection, target_value=1,
                            output_fname='', dataset_format='MEM'):
    """
    Rasterize the given vector (wrapper for gdal.RasterizeLayer). Return a gdal.Dataset.
    :param vector_data_path: Path to a shapefile
    :param cols: Number of columns of the result
    :param rows: Number of rows of the result
    :param geo_transform: Returned value of gdal.Dataset.GetGeoTransform (coefficients for
                          transforming between pixel/line (P,L) raster space, and projection
                          coordinates (Xp,Yp) space.
    :param projection: Projection definition string (Returned by gdal.Dataset.GetProjectionRef)
    :param target_value: Pixel value for the pixels. Must be a valid gdal.GDT_UInt16 value.
    :param output_fname: If the dataset_format is GeoTIFF, this is the output file name
    :param dataset_format: The gdal.Dataset driver name. [default: MEM]
    """
    data_source = gdal.OpenEx(vector_data_path, gdal.OF_VECTOR)
    if data_source is None:
        report_and_exit("File read failed: %s", vector_data_path)
    layer = data_source.GetLayer(0)
    driver = gdal.GetDriverByName(dataset_format)
    target_ds = driver.Create(output_fname, cols, rows, 1, gdal.GDT_UInt16)
    target_ds.SetGeoTransform(geo_transform)
    target_ds.SetProjection(projection)
    gdal.RasterizeLayer(target_ds, [1], layer, burn_values=[target_value])
    return target_ds

def gmlas_datasource(self):
        gmlasconf = self.gmlas_config()
        datasourceFile = self.gmlPathLineEdit.text()
        if datasourceFile == '':
            raise InputError(self.tr("You must select a input file or URL"))
        isXsd = datasourceFile.endswith(".xsd")
        isUrl = datasourceFile.startswith("http")
        swapCoordinates = self.swapCoordinatesCombo.currentText()
        driverConnection = ""
        openOptions = ['EXPOSE_METADATA_LAYERS=YES', 'CONFIG_FILE={}'.format(gmlasconf)]

        openOptions.append('SWAP_COORDINATES={}'.format(swapCoordinates))

        if isXsd:
            driverConnection = "GMLAS:"
            openOptions.append('XSD={}'.format(datasourceFile))
        elif isUrl:
            driverConnection = "GMLAS:/vsicurl_streaming/{}".format(datasourceFile)
        else:
            driverConnection = "GMLAS:{}".format(datasourceFile)

        with qgis_proxy_settings():
            return gdal.OpenEx(driverConnection,
                               open_options=openOptions)

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 src_datasource(self):
        options = ['LIST_ALL_TABLES=YES']
        options.append('SCHEMAS={}'.format(self.databaseWidget.schema()))
        datasource = gdal.OpenEx(self.databaseWidget.datasource_name(),
                           gdal.OF_VECTOR,
                           open_options=options)
        return datasource

def export_params(self, temp_datasource_path):
        temp_datasource = gdal.OpenEx(temp_datasource_path,
                                      open_options=['LIST_ALL_TABLES=YES'])
        params = {
            'destNameOrDestDS': self.dst_datasource_name(),
            'srcDS': temp_datasource,
            'format': 'GMLAS',
            'datasetCreationOptions': self.dataset_creation_options()
        }
        return params

    #@pyqtSlot()
    #def on_exportButton_clicked(self):

def open_vector(filename, driver=None):
    """Open vector file, return gdal.Dataset and OGR.Layer

        .. warning:: dataset and layer have to live in the same context,
            if dataset is deleted all layer references will get lost

        .. versionadded:: 0.12.0

    Parameters
    ----------
    filename : string
        vector file name
    driver : string
        gdal driver string

    Returns
    -------
    dataset : gdal.Dataset
        dataset
    layer : ogr.Layer
        layer
    """
    dataset = gdal.OpenEx(filename)

    if driver:
        gdal.GetDriverByName(driver)

    layer = dataset.GetLayer()

    return dataset, layer

def open_raster(filename, driver=None):
    """Open raster file, return gdal.Dataset

        .. versionadded:: 0.6.0

    Parameters
    ----------
    filename : string
        raster file name
    driver : string
        gdal driver string

    Returns
    -------
    dataset : gdal.Dataset
        dataset
    """

    dataset = gdal.OpenEx(filename)

    if driver:
        gdal.GetDriverByName(driver)

    return dataset

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" }