[refactor] refactor tms2geotiff functions

src/io/tms2geotiff.py

@@ -35,13 +35,12 @@ import math
 import re
 import time
 from typing import Tuple, Callable
-import PIL
 from PIL import Image
 
 from src import app_logger
 from src.utilities.constants import EARTH_EQUATORIAL_RADIUS, RETRY_DOWNLOAD, TIMEOUT_DOWNLOAD, TILE_SIZE, \
     CALLBACK_INTERVAL_DOWNLOAD
-from src.utilities.type_hints import PIL_Image
+from src.utilities.type_hints import PIL_Image, tuple_float, tuple_float_any, list_int, tuple_int
 
 Image.MAX_IMAGE_PIXELS = None
 
@@ -63,21 +62,21 @@ SESSION.headers.update({
 re_coords_split = re.compile('[ ,;]+')
 
 
-def from4326_to3857(lat, lon):
-    x_tile = math.radians(lon) * EARTH_EQUATORIAL_RADIUS
-    y_tile = math.log(math.tan(math.radians(45 + lat / 2.0))) * EARTH_EQUATORIAL_RADIUS
+def _from4326_to3857(lat: float, lon: float) -> tuple_float or tuple_float_any:
+    x_tile: float = math.radians(lon) * EARTH_EQUATORIAL_RADIUS
+    y_tile: float = math.log(math.tan(math.radians(45 + lat / 2.0))) * EARTH_EQUATORIAL_RADIUS
     return x_tile, y_tile
 
 
-def deg2num(lat, lon, zoom):
+def _deg2num(lat: float, lon: float, zoom: int):
     n = 2 ** zoom
     x_tile = ((lon + 180) / 360 * n)
     y_tile = (1 - math.asinh(math.tan(math.radians(lat))) / math.pi) * n / 2
     return x_tile, y_tile
 
 
-def is_empty(im):
-    extrema = im.getextrema()
+def _is_empty(image):
+    extrema = image.getextrema()
     if len(extrema) >= 3:
         if len(extrema) > 3 and extrema[-1] == (0, 0):
             return True
@@ -85,39 +84,36 @@ def is_empty(im):
             if ext != (0, 0):
                 return False
         return True
-    else:
-        return extrema[0] == (0, 0)
+    return extrema[0] == (0, 0)
 
 
-def paste_tile(big_im, base_size, tile, corner_xy, bbox):
+def _paste_tile(big_image: PIL_Image or None, base_size: list_int, tile: bytes, corner_xy: tuple_int, bbox: tuple_int):
     if tile is None:
-        return big_im
-    im = Image.open(io.BytesIO(tile))
-    mode = 'RGB' if im.mode == 'RGB' else 'RGBA'
-    size = im.size
-    if big_im is None:
-        base_size[0] = size[0]
-        base_size[1] = size[1]
-        new_im = Image.new(mode, (
-            size[0] * (bbox[2] - bbox[0]), size[1] * (bbox[3] - bbox[1])))
-    else:
-        new_im = big_im
-
-    dx = abs(corner_xy[0] - bbox[0])
-    dy = abs(corner_xy[1] - bbox[1])
-    xy0 = (size[0] * dx, size[1] * dy)
-    if mode == 'RGB':
-        new_im.paste(im, xy0)
-    else:
-        if im.mode != mode:
-            im = im.convert(mode)
-        if not is_empty(im):
-            new_im.paste(im, xy0)
-    im.close()
-    return new_im
-
-
-def get_tile(url):
+        return big_image
+    with Image.open(io.BytesIO(tile)) as tmp_image:
+        mode = 'RGB' if tmp_image.mode == 'RGB' else 'RGBA'
+        size = tmp_image.size
+        new_image = big_image
+        if big_image is None:
+            base_size[0] = size[0]
+            base_size[1] = size[1]
+            new_image = Image.new(mode, (
+                size[0] * (bbox[2] - bbox[0]), size[1] * (bbox[3] - bbox[1])))
+
+        dx = abs(corner_xy[0] - bbox[0])
+        dy = abs(corner_xy[1] - bbox[1])
+        xy0 = (size[0] * dx, size[1] * dy)
+        if mode == 'RGB':
+            new_image.paste(tmp_image, xy0)
+        else:
+            if tmp_image.mode != mode:
+                tmp_image = tmp_image.convert(mode)
+            if not _is_empty(tmp_image):
+                new_image.paste(tmp_image, xy0)
+        return new_image
+
+
+def _get_tile(url: str) -> bytes or None:
     retry = RETRY_DOWNLOAD
     while 1:
         try:
@@ -135,14 +131,15 @@ def get_tile(url):
     return r.content
 
 
-def print_progress(progress, total, done=False):
+def log_progress(progress, total, done=False):
+    """log the progress download"""
     if done:
         app_logger.info('Downloaded image %d/%d, %.2f%%' % (progress, total, progress * 100 / total))
 
 
 def download_extent(
         source: str, lat0: float, lon0: float, lat1: float, lon1: float, zoom: int,
-        save_image: bool = True, progress_callback: Callable = print_progress,
+        save_image: bool = True, progress_callback: Callable = log_progress,
         callback_interval: float = CALLBACK_INTERVAL_DOWNLOAD
 ) -> Tuple[PIL_Image, Tuple[float]] or Tuple[None]:
     """
@@ -162,8 +159,8 @@ def download_extent(
     Returns:
         parsed request input
     """
-    x0, y0 = deg2num(lat0, lon0, zoom)
-    x1, y1 = deg2num(lat1, lon1, zoom)
+    x0, y0 = _deg2num(lat0, lon0, zoom)
+    x1, y1 = _deg2num(lat1, lon1, zoom)
     if x0 > x1:
         x0, x1 = x1, x0
     if y0 > y1:
@@ -181,13 +178,13 @@ def download_extent(
     cancelled = False
     with concurrent.futures.ThreadPoolExecutor(5) as executor:
         for x, y in corners:
-            future = executor.submit(get_tile, source.format(z=zoom, x=x, y=y))
+            future = executor.submit(_get_tile, source.format(z=zoom, x=x, y=y))
             futures[future] = (x, y)
         bbox = (math.floor(x0), math.floor(y0), math.ceil(x1), math.ceil(y1))
-        big_im = None
+        big_image = None
         base_size = [TILE_SIZE, TILE_SIZE]
-        big_im, cancelled, done_num = run_future_tile_download(
-            base_size, bbox, big_im, callback_interval, cancelled, done_num, futures, last_callback, last_done_num,
+        big_image, cancelled, done_num = _run_future_tile_download(
+            base_size, bbox, big_image, callback_interval, cancelled, done_num, futures, last_callback, last_done_num,
             progress_callback, save_image, total_num
         )
     if cancelled:
@@ -203,19 +200,19 @@ def download_extent(
     y2 = round(base_size[1] * y_frac)
     img_w = round(base_size[0] * (x1 - x0))
     img_h = round(base_size[1] * (y1 - y0))
-    final_image = big_im.crop((x2, y2, x2 + img_w, y2 + img_h))
+    final_image = big_image.crop((x2, y2, x2 + img_w, y2 + img_h))
     if final_image.mode == 'RGBA' and final_image.getextrema()[3] == (255, 255):
         final_image = final_image.convert('RGB')
-    big_im.close()
-    xp0, yp0 = from4326_to3857(lat0, lon0)
-    xp1, yp1 = from4326_to3857(lat1, lon1)
+    big_image.close()
+    xp0, yp0 = _from4326_to3857(lat0, lon0)
+    xp1, yp1 = _from4326_to3857(lat1, lon1)
     p_width = abs(xp1 - xp0) / final_image.size[0]
     p_height = abs(yp1 - yp0) / final_image.size[1]
     matrix = min(xp0, xp1), p_width, 0, max(yp0, yp1), 0, -p_height
     return final_image, matrix
 
 
-def run_future_tile_download(
+def _run_future_tile_download(
         base_size, bbox, big_im, callback_interval, cancelled, done_num, futures, last_callback, last_done_num,
         progress_callback, save_image, total_num
 ):
@@ -228,7 +225,7 @@ def run_future_tile_download(
             img_data = fut.result()
             xy = futures[fut]
             if save_image:
-                big_im = paste_tile(big_im, base_size, img_data, xy, bbox)
+                big_im = _paste_tile(big_im, base_size, img_data, xy, bbox)
             del futures[fut]
             done_num += 1
         if time.monotonic() > last_callback + callback_interval: