Added changes to upload KML files

Signed-off-by: Omkar Mestry <omkar.mestry@here.com>

changes/features/pr.49.md

@@ -0,0 +1 @@
+Added feature to upload data from `kml` file to the space.

tests/space/test_space_objects.py

@@ -629,3 +629,13 @@ def test_spatial_search_geometry_divided(large_data_space):
 
     assert len(feature_read) == 7459
     assert feature_read[0]["type"] == "Feature"
+
+
+@pytest.mark.skipif(not XYZ_TOKEN, reason="No token found.")
+def test_add_features_kml(empty_space):
+    """Test uploading kml file to the space."""
+    space = empty_space
+    kml_file = Path(__file__).parents[1] / "data" / "test.kml"
+    space.add_features_kml(kml_file, features_size=500)
+    stats = space.get_statistics()
+    assert stats["count"]["value"] == 243

xyzspaces/spaces.py

@@ -947,3 +947,46 @@ def add_features_gpx(
                         features_size=features_size,
                         chunk_size=chunk_size,
                     )
+
+    def add_features_kml(
+        self, path: str, features_size: int = 2000, chunk_size: int = 1
+    ):
+        """
+        To upload data from kml file to a space
+
+        :param path: Path to kml file
+        :param features_size: An int representing a number of features to upload at
+            a time.
+        :param chunk_size: Number of chunks for each process to handle. The default value
+            is 1, for a large number of features please use `chunk_size` greater than 1.
+        """
+        gpd.io.file.fiona.drvsupport.supported_drivers["KML"] = "rw"
+
+        gdf = gpd.read_file(path, driver="KML")
+
+        geometry = gdf.geometry
+        flattened_geometry = []
+
+        flattened_gdf = gpd.GeoDataFrame()
+
+        for geom in geometry:
+            if geom.type in [
+                "GeometryCollection",
+                "MultiPoint",
+                "MultiLineString",
+                "MultiPolygon",
+            ]:
+                for subgeom in geom:
+                    flattened_geometry.append(subgeom)
+            else:
+                flattened_geometry.append(geom)
+
+        flattened_gdf.geometry = flattened_geometry
+
+        with tempfile.NamedTemporaryFile() as temp:
+            flattened_gdf.to_file(temp.name, driver="GeoJSON")
+            self.add_features_geojson(
+                path=temp.name,
+                features_size=features_size,
+                chunk_size=chunk_size,
+            )