[SEDONA-730] Add to_wkt() to sedona.geopandas API

python/sedona/geopandas/geodataframe.py

@@ -670,3 +670,58 @@ def to_parquet(self, path, **kwargs):
         """
         # Use the Spark DataFrame's write method to write to GeoParquet format
         self._internal.spark_frame.write.format("geoparquet").save(path, **kwargs)
+
+    def to_wkt(self, **kwargs):
+        """
+        Encode all geometry columns in the GeoDataFrame to WKT.
+
+        Parameters
+        ----------
+        kwargs
+            rounding_precision
+
+        Returns
+        -------
+        DataFrame
+            geometry columns are encoded to WKT
+        """
+
+        # Create a list of all column expressions for the new dataframe
+        select_expressions = []
+
+        # Read the keyword arguments
+        rounding_precision = kwargs.get("rounding_precision", None)
+
+        # Process geometry columns to get wkts
+        for field in self._internal.spark_frame.schema.fields:
+            col_name = field.name
+            # Skip index column to avoid duplication
+            if col_name == "__index_level_0__" or col_name == "__natural_order__":
+                continue
+
+            if field.dataType.typeName() in {"geometrytype", "binary"}:
+                # Start with the base expression.
+                if field.dataType.typeName() == "binary":
+                    base_expr = f"ST_GeomFromWKB(`{col_name}`)"
+                else:
+                    base_expr = f"`{col_name}`"
+
+                # Chain transformations if the keyword argument is provided.
+                if rounding_precision is not None:
+                    base_expr = f"ST_ReducePrecision({base_expr}, {rounding_precision})"
+
+                wkt_expr = f"ST_AsEWKT({base_expr}) as {col_name}"
+                select_expressions.append(wkt_expr)
+            else:
+                # Keep non-geometry columns as they are
+                select_expressions.append(f"`{col_name}`")
+
+        # Execute the query to get all data in one go
+        result_df = self._internal.spark_frame.selectExpr(*select_expressions)
+
+        # Convert to pandas DataFrame
+        pandas_df = result_df.toPandas()
+
+        # Create a new GeoDataFrame with the result
+        # Note: This avoids the need to manipulate the index columns separately
+        return GeoDataFrame(pandas_df)

python/sedona/geopandas/geoseries.py

@@ -580,6 +580,73 @@ def sjoin(
             **kwargs,
         )
 
+    def to_wkt(self, **kwargs) -> "GeoSeries":
+        """
+        Convert GeoSeries geometries to WKT
+
+        Parameters
+        ----------
+        kwargs
+            rounding_precision
+
+        Returns
+        -------
+        Series
+            WKT representations of the geometries
+
+        Examples
+        --------
+        >>> from shapely.geometry import Point
+        >>> import geopandas as gpd
+        >>> from sedona.geopandas import GeoSeries
+
+        >>> s = GeoSeries([Point(1, 1), Point(2, 2), Point(3, 3)])
+        >>> s
+        0    POINT (1 1)
+        1    POINT (2 2)
+        2    POINT (3 3)
+        dtype: geometry
+
+        >>> s.to_wkt()
+        0    POINT (1 1)
+        1    POINT (2 2)
+        2    POINT (3 3)
+        dtype: object
+
+        """
+
+        # Find the first column with BinaryType or GeometryType
+        first_col = self.get_first_geometry_column()
+
+        # Read the keyword arguments
+        rounding_precision = kwargs.get("rounding_precision", None)
+
+        if self.get_first_geometry_column():
+            data_type = self._internal.spark_frame.schema[first_col].dataType
+
+            # Start with the base expression.
+            if isinstance(data_type, BinaryType):
+                base_expr = f"ST_GeomFromWKB(`{first_col}`)"
+            else:
+                base_expr = f"`{first_col}`"
+
+            # Chain transformations if the keyword argument is provided.
+            if rounding_precision is not None:
+                base_expr = f"ST_ReducePrecision({base_expr}, {rounding_precision})"
+
+            wkt_expr = f"ST_AsEWKT({base_expr}) as wkt"
+
+        sdf = self._internal.spark_frame.selectExpr(wkt_expr)
+        internal = InternalFrame(
+            spark_frame=sdf,
+            index_spark_columns=None,
+            column_labels=[self._column_label],
+            data_spark_columns=[scol_for(sdf, "wkt")],
+            data_fields=[self._internal.data_fields[0]],
+            column_label_names=self._internal.column_label_names,
+        )
+        return _to_geo_series(first_series(PandasOnSparkDataFrame(internal)))
+
     # -----------------------------------------------------------------------------
     # # Utils
     # -----------------------------------------------------------------------------

python/tests/geopandas/test_geodataframe.py

@@ -144,6 +144,58 @@ def test_area(self):
         self.assert_almost_equal(area_values[0], 1.0)
         self.assert_almost_equal(area_values[1], 4.0)
 
+    def test_to_wkt(self):
+        # Create a GeoDataFrame with polygons to test wkt conversion
+        from shapely.geometry import Polygon
+
+        # Create geometries
+        poly1 = Polygon(
+            [(0.091, 0.091), (1.091, 0.091), (1.091, 1.091), (0.091, 1.091)]
+        )
+        poly2 = Polygon(
+            [(0.019, 0.019), (2.019, 0.019), (2.019, 2.019), (0.019, 2.019)]
+        )
+
+        data = {"geometry1": [poly1, poly2], "id": [1, 2], "value": ["a", "b"]}
+
+        df = GeoDataFrame(data)
+
+        # Calculate wkt
+        wkt_df1 = df.to_wkt()
+        wkt_df2 = df.to_wkt(rounding_precision=2)
+
+        # Verify result is a GeoDataFrame
+        assert type(wkt_df1) is GeoDataFrame
+        assert type(wkt_df2) is GeoDataFrame
+
+        # Verify non-geometry columns were preserved
+        assert "id" in wkt_df1.columns
+        assert "value" in wkt_df1.columns
+        assert "id" in wkt_df2.columns
+        assert "value" in wkt_df2.columns
+
+        wkt_values1 = wkt_df1["geometry1"].to_list()
+        wkt_values2 = wkt_df2["geometry1"].to_list()
+        assert len(wkt_values1) == 2
+        assert len(wkt_values2) == 2
+
+        assert (
+            wkt_values1[0]
+            == "POLYGON ((0.091 0.091, 1.091 0.091, 1.091 1.091, 0.091 1.091, 0.091 0.091))"
+        )
+        assert (
+            wkt_values1[1]
+            == "POLYGON ((0.019 0.019, 2.019 0.019, 2.019 2.019, 0.019 2.019, 0.019 0.019))"
+        )
+        assert (
+            wkt_values2[0]
+            == "POLYGON ((0.09 1.09, 1.09 1.09, 1.09 0.09, 0.09 0.09, 0.09 1.09))"
+        )
+        assert (
+            wkt_values2[1]
+            == "POLYGON ((0.02 2.02, 2.02 2.02, 2.02 0.02, 0.02 0.02, 0.02 2.02))"
+        )
+
     def test_buffer(self):
         # Create a GeoDataFrame with geometries to test buffer operation
         from shapely.geometry import Polygon, Point

python/tests/geopandas/test_geoseries.py

@@ -27,7 +27,6 @@
 from tests.test_base import TestBase
 import pyspark.pandas as ps
 
-
 class TestSeries(TestBase):
     def setup_method(self):
         self.tempdir = tempfile.mkdtemp()
@@ -103,12 +102,25 @@ def test_buffer_then_geoparquet(self):
         self.g1.buffer(0.2).to_parquet(temp_file_path)
         assert os.path.exists(temp_file_path)
 
+    def test_to_wkt(self):
+        wkt = self.g1.to_wkt()
+        assert wkt is not None
+        assert type(wkt) is GeoSeries
+        assert wkt.count() is 2
+        assert wkt.dtype == object
+        assert wkt.values[0] == "POLYGON ((0 0, 1 0, 1 1, 0 0))"
+
+        wkt = self.g1.to_wkt(reduce_precision=2)
+        assert wkt is not None
+        assert type(wkt) is GeoSeries
+        assert wkt.count() is 2
+        assert wkt.dtype == object
+        assert wkt.values[0] == "POLYGON ((0 0, 1 0, 1 1, 0 0))"
 
 # -----------------------------------------------------------------------------
 # # Utils
 # -----------------------------------------------------------------------------
 
-
 def check_geoseries(s):
     assert isinstance(s, GeoSeries)
     assert isinstance(s.geometry, GeoSeries)