[GH-3017] Implement rotate

python/sedona/spark/geopandas/base.py

@@ -1362,6 +1362,56 @@ def segmentize(self, max_segment_length):
     # def transform(self, transformation, include_z=False):
     #     raise NotImplementedError("This method is not implemented yet.")
 
+    def rotate(self, angle, origin="center", use_radians=False):
+        """Return a ``GeoSeries`` with rotated geometries.
+
+        See http://shapely.readthedocs.io/en/latest/manual.html#shapely.affinity.rotate
+        for details.
+
+        Parameters
+        ----------
+        angle : float
+            The angle of rotation can be specified in either degrees (default)
+            or radians by setting use_radians=True. Positive angles are
+            counter-clockwise and negative are clockwise rotations.
+        origin : string, Point, or tuple (x, y)
+            The point of origin can be a keyword 'center' for the bounding box
+            center (default), 'centroid' for the geometry's centroid, a Point
+            object or a coordinate tuple (x, y).
+        use_radians : boolean
+            Whether to interpret the angle of rotation as degrees or radians
+
+        Examples
+        --------
+        >>> from shapely.geometry import Point, LineString, Polygon
+        >>> s = geopandas.GeoSeries(
+        ...     [
+        ...         Point(1, 1),
+        ...         LineString([(1, -1), (1, 0)]),
+        ...         Polygon([(3, -1), (4, 0), (3, 1)]),
+        ...     ]
+        ... )
+        >>> s
+        0                         POINT (1 1)
+        1              LINESTRING (1 -1, 1 0)
+        2    POLYGON ((3 -1, 4 0, 3 1, 3 -1))
+        dtype: geometry
+
+        >>> s.rotate(90)
+        0                                          POINT (1 1)
+        1                      LINESTRING (1.5 -0.5, 0.5 -0.5)
+        2    POLYGON ((4.5 -0.5, 3.5 0.5, 2.5 -0.5, 4.5 -0.5))
+        dtype: geometry
+
+        >>> s.rotate(90, origin=(0, 0))
+        0                       POINT (-1 1)
+        1              LINESTRING (1 1, 0 1)
+        2    POLYGON ((1 3, 0 4, -1 3, 1 3))
+        dtype: geometry
+
+        """
+        return _delegate_to_geometry_column("rotate", self, angle, origin, use_radians)
+
     def force_2d(self):
         """Force the dimensionality of a geometry to 2D.
 

python/sedona/spark/geopandas/geoseries.py

@@ -1160,6 +1160,36 @@ def transform(self, transformation, include_z=False):
         # Implementation of the abstract method.
         raise NotImplementedError("This method is not implemented yet.")
 
+    def rotate(self, angle, origin="center", use_radians=False) -> "GeoSeries":
+        import math
+
+        if not use_radians:
+            angle = angle * math.pi / 180.0
+        if isinstance(origin, str):
+            if origin == "center":
+                origin_x = stf.ST_X(stf.ST_Centroid(stf.ST_Envelope(self.spark.column)))
+                origin_y = stf.ST_Y(stf.ST_Centroid(stf.ST_Envelope(self.spark.column)))
+                spark_expr = stf.ST_Rotate(self.spark.column, angle, origin_x, origin_y)
+            elif origin == "centroid":
+                origin_x = stf.ST_X(stf.ST_Centroid(self.spark.column))
+                origin_y = stf.ST_Y(stf.ST_Centroid(self.spark.column))
+                spark_expr = stf.ST_Rotate(self.spark.column, angle, origin_x, origin_y)
+            else:
+                raise ValueError(
+                    f"origin must be 'center', 'centroid', a Point, or (x, y) tuple, got {origin!r}"
+                )
+        elif isinstance(origin, tuple) and len(origin) == 2:
+            spark_expr = stf.ST_Rotate(
+                self.spark.column, angle, float(origin[0]), float(origin[1])
+            )
+        elif isinstance(origin, shapely.geometry.Point):
+            spark_expr = stf.ST_Rotate(self.spark.column, angle, origin.x, origin.y)
+        else:
+            raise TypeError(
+                f"origin must be 'center', 'centroid', a Point, or (x, y) tuple"
+            )
+        return self._query_geometry_column(spark_expr, returns_geom=True)
+
     def force_2d(self) -> "GeoSeries":
         spark_expr = stf.ST_Force_2D(self.spark.column)
         return self._query_geometry_column(spark_expr, returns_geom=True)

python/tests/geopandas/test_geoseries.py

@@ -1847,6 +1847,44 @@ def test_segmentize(self):
     def test_transform(self):
         pass
 
+    def test_rotate(self):
+        geoms = [
+            Point(1, 1),
+            LineString([(1, -1), (1, 0)]),
+            Polygon([(3, -1), (4, 0), (3, 1)]),
+            None,
+        ]
+        s = GeoSeries(geoms)
+        gpd_s = gpd.GeoSeries(geoms)
+
+        # Test default (degrees, origin='center')
+        self.check_sgpd_equals_gpd(s.rotate(90), gpd_s.rotate(90))
+
+        # Test with explicit origin tuple
+        self.check_sgpd_equals_gpd(
+            s.rotate(90, origin=(0, 0)), gpd_s.rotate(90, origin=(0, 0))
+        )
+
+        # Test use_radians
+        import math
+
+        self.check_sgpd_equals_gpd(
+            s.rotate(math.pi / 2, use_radians=True),
+            gpd_s.rotate(math.pi / 2, use_radians=True),
+        )
+
+        # Test origin='centroid'
+        self.check_sgpd_equals_gpd(
+            s.rotate(45, origin="centroid"), gpd_s.rotate(45, origin="centroid")
+        )
+
+        # Test GeoDataFrame works too
+        self.check_sgpd_equals_gpd(s.to_geoframe().rotate(90), gpd_s.rotate(90))
+
+        # Test invalid origin
+        with pytest.raises((ValueError, TypeError)):
+            s.rotate(90, origin="invalid")
+
     def test_force_2d(self):
         s = sgpd.GeoSeries(
             [

python/tests/geopandas/test_match_geopandas_series.py

@@ -956,6 +956,17 @@ def test_segmentize(self):
     def test_transform(self):
         pass
 
+    def test_rotate(self):
+        for geom in self.geoms:
+            # Sedona converts empty geometries to None, skip them
+            if not gpd.GeoSeries(geom).is_empty.any():
+                sgpd_result = GeoSeries(geom).rotate(45)
+                gpd_result = gpd.GeoSeries(geom).rotate(45)
+                self.check_sgpd_equals_gpd(sgpd_result, gpd_result)
+                sgpd_result = GeoSeries(geom).rotate(45, origin=(0, 0))
+                gpd_result = gpd.GeoSeries(geom).rotate(45, origin=(0, 0))
+                self.check_sgpd_equals_gpd(sgpd_result, gpd_result)
+
     def test_force_2d(self):
         # force_2d was added from geopandas 1.0.0
         if parse_version(gpd.__version__) < parse_version("1.0.0"):