From 20bd7318fbc88e845a6731c1becf0437e99fe1b1 Mon Sep 17 00:00:00 2001 From: xyluo25 Date: Tue, 29 Oct 2024 12:55:27 -0700 Subject: [PATCH] update util_geo --- pyufunc/util_geo/__init__.py | 9 ++- ...te_coversion.py => _coordinate_convert.py} | 0 pyufunc/util_geo/_geo_area.py | 68 +++++++++++++++++++ pyufunc/util_geo/_get_osm_place.py | 45 ++++++------ 4 files changed, 96 insertions(+), 26 deletions(-) rename pyufunc/util_geo/{_coordinate_coversion.py => _coordinate_convert.py} (100%) diff --git a/pyufunc/util_geo/__init__.py b/pyufunc/util_geo/__init__.py index a59e41f..d4cabf7 100644 --- a/pyufunc/util_geo/__init__.py +++ b/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", diff --git a/pyufunc/util_geo/_coordinate_coversion.py b/pyufunc/util_geo/_coordinate_convert.py similarity index 100% rename from pyufunc/util_geo/_coordinate_coversion.py rename to pyufunc/util_geo/_coordinate_convert.py diff --git a/pyufunc/util_geo/_geo_area.py b/pyufunc/util_geo/_geo_area.py index caecd64..0fa9b2a 100644 --- a/pyufunc/util_geo/_geo_area.py +++ b/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 diff --git a/pyufunc/util_geo/_get_osm_place.py b/pyufunc/util_geo/_get_osm_place.py index fd7beae..87360ec 100644 --- a/pyufunc/util_geo/_get_osm_place.py +++ b/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: """