def test_write_keywords(self):
     """Test writing keywords with no sublayer, no existing file."""
     keywords = DKI_KEYWORDS
     filename = self.make_temp_file()
     write_keywords(keywords, filename)
     # Read back contents and check against control dataset
     control_keywords = read_keywords(SIMPLE_PATH, all_blocks=True)
     actual_keywords = read_keywords(filename, all_blocks=True)
     msg = 'Expected:\n%s\nGot:\n%s\n' % (control_keywords, actual_keywords)
     assert control_keywords == actual_keywords, msg

    def test_write_keywords_singlesublayer(self):
        """Test writing keywords for single sublayer, no existing file."""

        filename = self.make_temp_file()
        write_keywords(OSM_KEYWORDS, filename=filename, sublayer='osm')
        # read back contents and check against control dataset
        control_path = test_data_path('other', 'expected_singlelayer.keywords')
        control_keywords = read_keywords(control_path, all_blocks=True)
        actual_keywords = read_keywords(filename, all_blocks=True)
        msg = 'Expected:\n%s\nGot:\n%s\n' % (control_keywords, actual_keywords)
        assert control_keywords == actual_keywords, msg

    def test_write_keywords_singlesublayer(self):
        """Test writing keywords for single sublayer, no existing file."""

        filename = self.make_temp_file()
        write_keywords(OSM_KEYWORDS, filename=filename, sublayer="osm")
        # read back contents and check against control dataset
        control_path = os.path.abspath(os.path.join(UNITDATA, "other", "expected_singlelayer.keywords"))
        control_keywords = read_keywords(control_path, all_blocks=True)
        actual_keywords = read_keywords(filename, all_blocks=True)
        msg = "Expected:\n%s\nGot:\n%s\n" % (control_keywords, actual_keywords)
        assert control_keywords == actual_keywords, msg

    def test_write_keywords_multisublayer(self):
        """Test writing keywords for named sublayer, no existing file."""

        filename = self.make_temp_file()
        write_keywords(DKI_KEYWORDS, filename=filename, sublayer="dki")
        write_keywords(OSM_KEYWORDS, filename=filename, sublayer="osm")
        # Read back contents and check against control dataset
        control_path = test_data_path("other", "expected_multilayer.keywords")
        control_keywords = read_keywords(control_path, all_blocks=True)
        actual_keywords = read_keywords(filename, all_blocks=True)
        msg = "Expected:\n%s\nGot:\n%s\n" % (control_keywords, actual_keywords)
        assert control_keywords == actual_keywords, msg

    def test_write_keywords_multisublayer(self):
        """Test writing keywords for named sublayer, no existing file."""

        filename = self.make_temp_file()
        write_keywords(DKI_KEYWORDS, filename=filename, sublayer='dki')
        write_keywords(OSM_KEYWORDS, filename=filename, sublayer='osm')
        # Read back contents and check against control dataset
        control_path = os.path.abspath(
            os.path.join(UNITDATA, 'other', 'expected_multilayer.keywords'))
        control_keywords = read_keywords(control_path, all_blocks=True)
        actual_keywords = read_keywords(filename, all_blocks=True)
        msg = 'Expected:\n%s\nGot:\n%s\n' % (control_keywords, actual_keywords)
        assert control_keywords == actual_keywords, msg

 def test_read_keywords_for_sublayer(self):
     """Test reading keywords for specific sublayer."""
     keywords = read_keywords(KEYWORD_PATH, sublayer='osm_flood')
     expected_keywords = OSM_KEYWORDS
     msg = 'Expected:\n%s\nGot:\n%s\n' % (expected_keywords, keywords)
     self.assertEquals(keywords, expected_keywords, msg)
     LOGGER.debug(keywords)

 def test_read_keywords_simple(self):
     """Test reading keywords from keywords file with no sublayers"""
     keywords = read_keywords(SIMPLE_PATH)
     expected_keywords = DKI_KEYWORDS
     msg = 'Expected:\n%s\nGot:\n%s\n' % (expected_keywords, keywords)
     self.assertEquals(keywords, expected_keywords, msg)
     LOGGER.debug(keywords)

 def test_read_all_keywords(self):
     """Test reading all keywords for all layers"""
     keywords = read_keywords(KEYWORD_PATH, all_blocks=True)
     expected_keywords = {"osm_buildings": DKI_KEYWORDS, "osm_flood": OSM_KEYWORDS}
     msg = "Expected:\n%s\nGot:\n%s\n" % (expected_keywords, keywords)
     self.assertEquals(keywords, expected_keywords, msg)
     LOGGER.debug(keywords)

 def test_read_keywords(self):
     """Test reading keywords - get first block from kwds as default."""
     keywords = read_keywords(KEYWORD_PATH)
     expected_keywords = DKI_KEYWORDS
     msg = 'Expected:\n%s\nGot:\n%s\n' % (expected_keywords, keywords)
     self.assertEquals(keywords, expected_keywords, msg)
     LOGGER.debug(keywords)

 def testSqliteWriting(self):
     """Test that writing a dataset to sqlite works."""
     keywords = read_keywords(SHP_BASE + '.keywords')
     layer = Vector(data=SHP_BASE + '.shp', keywords=keywords)
     test_dir = temp_dir(sub_dir='test')
     test_file = unique_filename(suffix='.sqlite', dir=test_dir)
     layer.write_to_file(test_file, sublayer='foo')

 def test_sqlite_writing(self):
     """Test that writing a dataset to sqlite works."""
     keywords = read_keywords(SHP_BASE + ".keywords")
     layer = Vector(data=SHP_BASE + ".shp", keywords=keywords)
     test_dir = temp_dir(sub_dir="test")
     test_file = unique_filename(suffix=".sqlite", dir=test_dir)
     layer.write_to_file(test_file, sublayer="foo")
     self.assertTrue(os.path.exists(test_file))

 def test_shapefile_loading(self):
     """Test that loading a dataset with no sublayers works."""
     keywords = read_keywords(SHP_BASE + ".keywords")
     layer = Vector(data=SHP_BASE + ".shp", keywords=keywords)
     msg = "Expected layer to be a polygon layer, got a %s" % layer.geometry_type
     self.assertTrue(layer.is_polygon_data, msg)
     count = len(layer)
     self.assertEqual(count, 250, "Expected 250 features, got %s" % count)

 def test_sublayer_loading(self):
     """Test if we can load sublayers."""
     keywords = read_keywords(KEYWORD_PATH, EXPOSURE_SUBLAYER_NAME)
     layer = Vector(data=SQLITE_PATH, keywords=keywords, sublayer=EXPOSURE_SUBLAYER_NAME)
     msg = "Expected layer to be a polygon layer, got a %s" % layer.geometry_type
     self.assertTrue(layer.is_polygon_data, msg)
     count = len(layer)
     self.assertEqual(count, 250, "Expected 250 features, got %s" % count)

 def testShpLoading(self):
     """Test that loading a dataset with no sublayers works."""
     keywords = read_keywords(SHP_BASE + '.keywords')
     layer = Vector(data=SHP_BASE + '.shp', keywords=keywords)
     msg = ('Expected layer to be a polygon layer, got a %s' %
            layer.geometry_type)
     assert layer.is_polygon_data, msg
     count = len(layer)
     assert count == 250, 'Expected 250 features, got %s' % count

def read_file_keywords(layer_path, keyword=None):
    """Get metadata from the keywords file associated with a local
     file in the file system.

    .. note:: Requires a str representing a file path instance
              as parameter As opposed to read_keywords_from_layer which
              takes a inasafe file object as parameter.

    .. seealso:: read_keywords_from_layer

    :param: layer_path: a string representing a path to a layer
           (e.g. '/tmp/foo.shp', '/tmp/foo.tif')
    :type layer_path: str

    :param keyword: optional - the metadata keyword to retrieve e.g. 'title'
    :type keyword: str

    :return: A string containing the retrieved value for the keyword if
             the keyword argument is specified, otherwise the
             complete keywords dictionary is returned.

    :raises: KeywordNotFoundError, NoKeywordsFoundError, InvalidParameterError

    Note:
        * KeywordNotFoundError - occurs when the keyword is not recognised.
        * NoKeywordsFoundError - occurs when no keyword file exists.
        * InvalidParameterError - occurs when the layer does not exist.
    """
    # check the source layer path is valid
    if not os.path.isfile(layer_path):
        message = tr('Cannot get keywords from a non-existent file. File '
                     '%s does not exist.' % layer_path)
        raise InvalidParameterError(message)

    # check there really is a keywords file for this layer
    # priority for iso path first
    keyword_file_path = os.path.splitext(layer_path)[0]
    keyword_file_path += '.keywords'
    keyword_iso_path = os.path.splitext(layer_path)[0]
    keyword_iso_path += '.xml'
    if not os.path.isfile(keyword_file_path)\
            and not os.path.isfile(keyword_iso_path):
        message = tr('No keywords file found for %s' % keyword_file_path)
        raise NoKeywordsFoundError(message)
    elif os.path.isfile(keyword_file_path) \
            and not os.path.isfile(keyword_iso_path):
        # switch to .keywords file if iso xml file didn't exist
        keyword_iso_path = keyword_file_path

    # now get the requested keyword using the inasafe library
    try:
        dictionary = read_keywords(keyword_iso_path)
    except Exception, e:
        message = tr(
            'Keyword retrieval failed for %s (%s) \n %s' % (
                keyword_file_path, keyword, str(e)))
        raise KeywordNotFoundError(message)

 def testSublayerLoading(self):
     keywords = read_keywords(KEYWORD_PATH, EXPOSURE_SUBLAYER_NAME)
     layer = Vector(data=SQLITE_PATH, keywords=keywords,
                    sublayer=EXPOSURE_SUBLAYER_NAME)
     msg = ('Expected layer to be a polygon layer, got a %s' %
            layer.geometry_type)
     assert layer.is_polygon_data, msg
     count = len(layer)
     assert count == 250, 'Expected 250 features, got %s' % count

    def read_from_file(self, filename):

        # Open data file for reading
        # File must be kept open, otherwise GDAL methods segfault.
        fid = self.fid = gdal.Open(filename, gdal.GA_ReadOnly)
        if fid is None:
            msg = 'Could not open file %s' % filename
            raise Exception(msg)

        # Record raster metadata from file
        basename, ext = os.path.splitext(filename)

        # If file is ASCII, check that projection is around.
        # GDAL does not check this nicely, so it is worth an
        # error message
        if ext == '.asc':
            try:
                open(basename + '.prj')
            except IOError:
                msg = ('Projection file not found for %s. You must supply '
                       'a projection file with extension .prj' % filename)
                raise RuntimeError(msg)

        # Look for any keywords
        self.keywords = read_keywords(basename + '.keywords')

        # Determine name
        if 'title' in self.keywords:
            rastername = self.keywords['title']
        else:
            # Use basename without leading directories as name
            rastername = os.path.split(basename)[-1]

        self.name = rastername
        self.filename = filename

        self.projection = Projection(self.fid.GetProjection())
        self.geotransform = self.fid.GetGeoTransform()
        self.columns = fid.RasterXSize
        self.rows = fid.RasterYSize
        self.number_of_bands = fid.RasterCount

        # Assume that file contains all data in one band
        msg = 'Only one raster band currently allowed'
        if self.number_of_bands > 1:
            msg = ('WARNING: Number of bands in %s are %i. '
                   'Only the first band will currently be '
                   'used.' % (filename, self.number_of_bands))
            # FIXME(Ole): Let us use python warnings here
            raise Exception(msg)

        # Get first band.
        band = self.band = fid.GetRasterBand(1)
        if band is None:
            msg = 'Could not read raster band from %s' % filename
            raise Exception(msg)

    def test_qgis_vector_layer_loading(self):
        """Test that reading from QgsVectorLayer works."""
        keywords = read_keywords(KEYWORD_PATH, EXPOSURE_SUBLAYER_NAME)
        if QGIS_IS_AVAILABLE:
            qgis_layer = QgsVectorLayer(SHP_BASE + ".shp", "test", "ogr")

            layer = Vector(data=qgis_layer, keywords=keywords)
            msg = "Expected layer to be a polygon layer, got a %s" % layer.geometry_type
            self.assertTrue(layer.is_polygon_data, msg)
            count = len(layer)
            self.assertEqual(count, 250, "Expected 250 features, got %s" % count)

    def test_convert_to_qgis_vector_layer(self):
        """Test that converting to QgsVectorLayer works."""
        if QGIS_IS_AVAILABLE:
            # Create vector layer
            keywords = read_keywords(SHP_BASE + ".keywords")
            layer = Vector(data=SHP_BASE + ".shp", keywords=keywords)

            # Convert to QgsVectorLayer
            qgis_layer = layer.as_qgis_native()
            provider = qgis_layer.dataProvider()
            count = provider.featureCount()
            self.assertEqual(count, 250, "Expected 250 features, got %s" % count)

    def test_qgis_vector_layer_loading(self):
        """Test that reading from QgsVectorLayer works."""
        keywords = read_keywords(KEYWORD_PATH, EXPOSURE_SUBLAYER_NAME)
        if qgis_imported:
            qgis_layer = QgsVectorLayer(SHP_BASE + '.shp', 'test', 'ogr')

            layer = Vector(data=qgis_layer, keywords=keywords)
            msg = ('Expected layer to be a polygon layer, got a %s' %
                   layer.geometry_type)
            self.assertTrue(layer.is_polygon_data, msg)
            count = len(layer)
            self.assertEqual(
                count, 250, 'Expected 250 features, got %s' % count)