update util_geo

pyufunc/util_geo/__init__.py

@@ -14,7 +14,7 @@
                                             get_coordinates_from_geom,
                                             find_k_nearest_points,
                                             )
-from pyufunc.util_geo._coordinate_coversion import (
+from pyufunc.util_geo._coordinate_convert import (
     cvt_wgs84_to_baidu09,
     cvt_wgs84_to_gcj02,
     cvt_gcj02_to_baidu09,
@@ -23,6 +23,8 @@
     cvt_baidu09_to_gcj02,
 )
 
+from pyufunc.util_geo._geo_area import calc_area_from_wkt_geometry
+
 # GMNS: General Modeling Network Specification
 import pyufunc.util_geo._gmns as gmns_geo
 from pyufunc.util_geo._gmns import Node as GMNSNode
@@ -39,6 +41,9 @@
 from pyufunc.util_geo._get_osm_place import get_osm_place
 
 __all__ = [
+    # geo_area
+    'calc_area_from_wkt_geometry',
+
     # geo_circle
     'create_circle_at_point_with_radius',
 
@@ -64,7 +69,7 @@
     # "gmns_read_zone_by_centroid",
     "gmns_read_zone",
 
-    # coordinate conversion
+    # coordinate convert
     "cvt_wgs84_to_baidu09",
     "cvt_wgs84_to_gcj02",
     "cvt_gcj02_to_baidu09",

pyufunc/util_geo/_geo_area.py

@@ -6,3 +6,71 @@
 ##############################################################
 
 
+import os
+from typing import TYPE_CHECKING
+
+from pyufunc.util_magic import requires, import_package
+
+if TYPE_CHECKING:
+    from pyproj import Transformer
+    import shapely
+
+
+@requires("pyproj", "shapely", verbose=False)
+def calc_area_from_wkt_geometry(wkt_geometry: str, unit: str = "sqm", verbose: bool = False) -> float:
+    """
+    Calculate the area of a geometry in WKT format.
+
+    Args:
+        wkt_geometry (str): The geometry in WKT format.
+        transformer (Transformer): The transformer to convert the geometry to the target CRS.
+        unit (str): The unit of the area. Default is "sqm". Options are "sqm", "sqft"
+
+    Example:
+        >>> from pyufunc import calc_area_from_wkt_geometry
+        >>> from pyproj import Transformer
+        >>> wkt_geometry = "POLYGON ((-74.006 40.712, -74.006 40.712, -74.006 40.712, -74.006 40.712))"
+        >>> calc_area_from_wkt_geometry(wkt_geometry, unit="sqm")
+        >>>
+
+    Return (float): The area of the geometry.
+
+    """
+    import_package("shapely", verbose=False)
+    import_package("pyproj", verbose=False)
+    import shapely
+    from pyproj import Transformer
+
+    # TDD
+    if unit not in ["sqm", "sqft"]:
+        raise ValueError("unit must be one of ['sqm', 'sqft']")
+
+    geometry_shapely = shapely.from_wkt(wkt_geometry)
+
+    # Set up a Transformer to convert from WGS 84 to UTM zone 18N (EPSG:32618)
+    transformer = Transformer.from_crs(
+        "EPSG:4326", "EPSG:32618", always_xy=True)
+
+    # Transform the polygon's coordinates to UTM
+    if isinstance(geometry_shapely, shapely.MultiPolygon):
+        transformed_polygons = []
+        for polygon in geometry_shapely.geoms:
+            transformed_coords = [transformer.transform(
+                x, y) for x, y in polygon.exterior.coords]
+            transformed_polygons.append(
+                shapely.Polygon(transformed_coords))
+        transformed_geometry = shapely.MultiPolygon(
+            transformed_polygons)
+    else:
+        transformed_coords = [transformer.transform(
+            x, y) for x, y in geometry_shapely.exterior.coords]
+        transformed_geometry = shapely.Polygon(transformed_coords)
+
+    if unit == "sqm":
+        if verbose:
+            print("Area in sqm:")
+        return transformed_geometry.area
+    else:
+        if verbose:
+            print("Area in sqft:")
+        return transformed_geometry.area * 10.7639104

pyufunc/util_geo/_get_osm_place.py

@@ -355,27 +355,26 @@ def _save_to_cache(self,
         -------
         None
         """
-        if settings["use_cache"]:
-            if not ok:  # pragma: no cover
-                msg = "Did not save to cache because HTTP status code is not OK"
-                if self.verbose:
-                    print(f"  :{msg}")
-            else:
-                # create the folder on the disk if it doesn't already exist
-                cache_folder = Path(settings["cache_folder"])
-                cache_folder.mkdir(parents=True, exist_ok=True)
-
-                # hash the url to make the filename succinct but unique
-                # sha1 digest is 160 bits = 20 bytes = 40 hexadecimal characters
-                checksum = sha1(url.encode("utf-8")).hexdigest()  # noqa: S324
-                cache_filepath = cache_folder / f"{checksum}.json"
-
-                # dump to json, and save to file
-                cache_filepath.write_text(json.dumps(
-                    response_json), encoding="utf-8")
-                msg = f"Saved response to cache file {str(cache_filepath)!r}"
-                if self.verbose:
-                    print(f"  :{msg}")
+        if not settings["use_cache"]:
+            return
+        if ok:
+            # create the folder on the disk if it doesn't already exist
+            cache_folder = Path(settings["cache_folder"])
+            cache_folder.mkdir(parents=True, exist_ok=True)
+
+            # hash the url to make the filename succinct but unique
+            # sha1 digest is 160 bits = 20 bytes = 40 hexadecimal characters
+            checksum = sha1(url.encode("utf-8")).hexdigest()  # noqa: S324
+            cache_filepath = cache_folder / f"{checksum}.json"
+
+            # dump to json, and save to file
+            cache_filepath.write_text(json.dumps(
+                response_json), encoding="utf-8")
+            msg = f"Saved response to cache file {str(cache_filepath)!r}"
+        else:
+            msg = "Did not save to cache because HTTP status code is not OK"
+        if self.verbose:
+            print(f"  :{msg}")
 
     def _url_in_cache(self, url: str) -> Path | None:
         """
@@ -476,9 +475,7 @@ def _get_http_headers(self,
 
         info = {"User-Agent": user_agent, "referer": referer,
                 "Accept-Language": accept_language}
-        headers = dict(requests.utils.default_headers())
-        headers.update(info)
-        return headers
+        return dict(requests.utils.default_headers()) | info
 
     def _resolve_host_via_doh(self, hostname: str) -> str:
         """