Skip to content

Commit

Permalink
feat: Run mypy in strict mode
Browse files Browse the repository at this point in the history
Can't pull out image type because of
<python/mypy#7178>.
  • Loading branch information
l0b0 authored and kodiakhq[bot] committed Jan 16, 2023
1 parent 1465b6d commit b6449b7
Show file tree
Hide file tree
Showing 2 changed files with 51 additions and 17 deletions.
Expand Up @@ -100,13 +100,43 @@
"\n",
"# suppress warnings - raised by external rasterio library\n",
"import warnings\n",
"from datetime import date\n",
"from typing import List, Optional, Tuple, TypedDict\n",
"\n",
"import geopandas as gpd\n",
"import matplotlib.pyplot as plt\n",
"import numpy as np\n",
"import numpy.typing as npt\n",
"import rasterio\n",
"from affine import Affine\n",
"from rasterio.crs import CRS\n",
"from rasterio.plot import show\n",
"from rio_tiler.io import COGReader\n",
"from satsearch import Search\n",
"from satstac import ItemCollection\n",
"from shapely.geometry import shape\n",
"from typing_extensions import NotRequired\n",
"\n",
"if sys.version_info >= (3, 9):\n",
" from typing import Annotated\n",
"else:\n",
" from typing_extensions import Annotated\n",
"\n",
"\n",
"class Image(TypedDict):\n",
" \"\"\"Image info being passed around this code.\"\"\"\n",
"\n",
" date: date\n",
" src_crs: str\n",
" rgb: npt.NDArray[np.float64]\n",
" mndwi: float\n",
" transform_window: Affine\n",
" image_id: str\n",
" water_mask: NotRequired[gpd.GeoDataFrame]\n",
" water: NotRequired[gpd.GeoDataFrame]\n",
" water_area: int\n",
" area: NotRequired[int]\n",
"\n",
"\n",
"warnings.filterwarnings(\"ignore\")\n",
"\n",
Expand Down Expand Up @@ -221,7 +251,13 @@
"metadata": {},
"outputs": [],
"source": [
"def range_request(image_url, crs, bbox, width=None, height=None):\n",
"def range_request(\n",
" image_url: str,\n",
" crs: CRS,\n",
" bbox: Annotated[List[float], 4],\n",
" width: Optional[int] = None,\n",
" height: Optional[int] = None,\n",
") -> Tuple[npt.NDArray[np.float64], Affine]:\n",
" \"\"\"Request and read just the required pixels from the COG.\"\"\"\n",
" with COGReader(image_url) as image:\n",
" img = image.part(bbox, width=width, dst_crs=crs, height=height, nodata=-9999)\n",
Expand All @@ -242,7 +278,7 @@
"metadata": {},
"outputs": [],
"source": [
"def image_search(bbox, date_range):\n",
"def image_search(bbox: Annotated[List[float], 4], date_range: str) -> ItemCollection:\n",
" \"\"\"Use SatSearch to find all Sentinel-2 images that meet our criteria.\"\"\"\n",
" # Note, we are not querying cloud cover and accepting\n",
" # all images irrelevant of cloud metadata.\n",
Expand Down Expand Up @@ -276,25 +312,25 @@
"metadata": {},
"outputs": [],
"source": [
"def export_raster(image, image_tpye=\"rgb\"):\n",
"def export_raster(image: Image) -> None:\n",
" \"\"\"Export GeoTiffs for use in a GIS.\"\"\"\n",
" raster_output = os.path.join(output_directory, f\"{image['image_id']}.tif\")\n",
" number_of_bands, height, width = image[image_tpye].shape\n",
" number_of_bands, height, width = image[\"rgb\"].shape\n",
"\n",
" profile = {\n",
" \"driver\": \"GTiff\",\n",
" \"count\": number_of_bands,\n",
" \"height\": height,\n",
" \"width\": width,\n",
" \"crs\": image[\"src_crs\"],\n",
" \"dtype\": image[image_tpye].dtype,\n",
" \"dtype\": image[\"rgb\"].dtype,\n",
" \"transform\": image[\"transform_window\"],\n",
" \"nodata\": -9999,\n",
" \"photometric\": \"RGB\",\n",
" }\n",
"\n",
" with rasterio.open(raster_output, \"w\", **profile) as dst:\n",
" dst.write(image[image_tpye])"
" dst.write(image[\"rgb\"])"
]
},
{
Expand All @@ -317,8 +353,7 @@
"metadata": {},
"outputs": [],
"source": [
"images = []\n",
"\n",
"images: List[Image] = []\n",
"# Iterate over all observations meeting our search criteria\n",
"items = image_search(bbox, date_range)\n",
"for item in items:\n",
Expand All @@ -327,7 +362,7 @@
" swir = item.asset(\"swir16\")[\"href\"]\n",
" rgb = item.asset(\"visual\")[\"href\"]\n",
" scl = item.asset(\"SCL\")[\"href\"]\n",
" date = item.date.strftime(\"%d/%m/%Y\")\n",
" date_ = item.date.strftime(\"%d/%m/%Y\")\n",
" crs = f\"EPSG:{item.properties['proj:epsg']}\"\n",
"\n",
" # Streamed pixels within bbox\n",
Expand Down Expand Up @@ -355,14 +390,12 @@
" # Store the data for further processing\n",
" images.append(\n",
" {\n",
" \"date\": date,\n",
" \"date\": date_,\n",
" \"src_crs\": crs,\n",
" \"rgb\": rgb_subset,\n",
" \"mndwi\": mndwi_subset,\n",
" \"transform_window\": transform_window,\n",
" \"image_id\": item.id,\n",
" \"water_mask\": None,\n",
" \"water\": None,\n",
" \"water_area\": 0,\n",
" }\n",
" )"
Expand Down Expand Up @@ -459,14 +492,11 @@
"metadata": {},
"outputs": [],
"source": [
"import geopandas as gpd\n",
"from shapely.geometry import shape\n",
"\n",
"sieve_threshold = 10\n",
"\n",
"for image in images:\n",
" # Mask out all values below the mndwi threshold (these are not water)\n",
" water_mask = np.ma.masked_less(image[\"mndwi\"], mndwi_threshold)\n",
" water_mask = np.ma.masked_less(image[\"mndwi\"], mndwi_threshold) # type: ignore[no-untyped-call]\n",
"\n",
" # Extract the polygons from the mask\n",
" water_polygons = rasterio.features.shapes(\n",
Expand Down
6 changes: 5 additions & 1 deletion pyproject.toml
Expand Up @@ -3,15 +3,19 @@ profile = "black"

[tool.mypy]
show_error_codes = true
strict = true

[[tool.mypy.overrides]]
module = [
"affine", # https://github.com/rasterio/affine/issues/74
"geopandas", # https://github.com/geopandas/geopandas/issues/1974
"matplotlib", # https://github.com/matplotlib/matplotlib/issues/20504
"matplotlib.pyplot", # https://github.com/matplotlib/matplotlib/issues/20504
"rasterio", # https://github.com/rasterio/rasterio/issues/2322
"rasterio.crs", # https://github.com/rasterio/rasterio/issues/2322
"rasterio.plot", # https://github.com/rasterio/rasterio/issues/2322
"satsearch", # Seems to be dead
"satsearch", # https://github.com/sat-utils/sat-search/issues/131
"satstac", # https://github.com/sat-utils/sat-stac/issues/72
"shapely.geometry", # https://github.com/shapely/shapely/issues/721
]
ignore_missing_imports = true

0 comments on commit b6449b7

Please sign in to comment.