File size: 5,878 Bytes
0914710 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 |
import unittest
import numpy as np
from samgis_core.utilities.utilities import hash_calculate
from samgis import app_logger
from samgis.io.tms2geotiff import download_extent
from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER
input_bbox = [[39.036252959636606, 15.040283203125002], [38.302869955150044, 13.634033203125002]]
class TestTms2geotiff(unittest.TestCase):
# def test_download_extent_simple_source(self):
# from rasterio import Affine
# from xyzservices import TileProvider
# from tests.local_tiles_http_server import LocalTilesHttpServer
#
# listen_port = 8000
#
# with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
# pt0, pt1 = input_bbox
# zoom = 10
#
# n_lat = pt0[0]
# e_lng = pt0[1]
# s_lat = pt1[0]
# w_lng = pt1[1]
#
# source = TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE, attribution="")
# img, matrix = download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=source)
# app_logger.info(f"# DOWNLOAD ENDED, shape: {img.shape} #")
# np_img = np.ascontiguousarray(img)
# output_hash = hash_calculate(np_img)
# assert output_hash == b'UmbkwbPJpRT1XXcLnLUapUDP320w7YhS/AmT3H7u+b4='
# assert Affine.to_gdal(matrix) == (
# 1517657.1966021745, 152.8740565703525, 0.0, 4726942.266183584, 0.0, -152.87405657034955)
def test_download_extent_source_with_parameter(self):
from rasterio import Affine
from xyzservices import TileProvider
from tests.local_tiles_http_server import LocalTilesHttpServer
listen_port = 8000
with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
pt0, pt1 = input_bbox
zoom = 10
n_lat = pt0[0]
e_lng = pt0[1]
s_lat = pt1[0]
w_lng = pt1[1]
local_url = "http://localhost:8000/{parameter}/{z}/{x}/{y}.png"
download_extent_args_no_parameter = {"name": "local_tile_provider", "url": LOCAL_URL_TILE, "attribution": ""}
download_extent_args = {
"no_parameter": download_extent_args_no_parameter,
"with_parameter": {"url": local_url, "parameter": "lambda_handler", **download_extent_args_no_parameter}
}
for _args_names, _args in download_extent_args.items():
app_logger.info(f"args_names:{_args_names}.")
source = TileProvider(**_args)
img, matrix = download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=source)
app_logger.info(f"# DOWNLOAD ENDED, shape: {img.shape} #")
np_img = np.ascontiguousarray(img)
output_hash = hash_calculate(np_img)
assert output_hash == b'UmbkwbPJpRT1XXcLnLUapUDP320w7YhS/AmT3H7u+b4='
assert Affine.to_gdal(matrix) == (
1517657.1966021745, 152.8740565703525, 0.0, 4726942.266183584, 0.0, -152.87405657034955)
def test_download_extent_source_with_parameter_key_error(self):
from xyzservices import TileProvider
with self.assertRaises(KeyError):
try:
pt0, pt1 = input_bbox
zoom = 10
n_lat = pt0[0]
e_lng = pt0[1]
s_lat = pt1[0]
w_lng = pt1[1]
local_url_tile2 = "http://localhost:8000/{parameter}/{z}/{x}/{y}.png"
source = TileProvider(name="local_tile_provider", url=local_url_tile2, attribution="")
download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=source)
except KeyError as ke:
assert str(ke) == "'parameter'"
raise ke
def test_download_extent_io_error1(self):
with self.assertRaises(Exception):
try:
pt0, pt1 = input_bbox
zoom = 10
n_lat = pt0[0]
e_lng = pt0[1]
s_lat = pt1[0]
w_lng = pt1[1]
download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=f"http://{LOCAL_URL_TILE}")
print("exception not raised")
except ConnectionError as ioe1:
app_logger.error(f"ioe1:{ioe1}.")
msg0 = "HTTPConnectionPool(host='localhost', port=8000): Max retries exceeded with url: /lambda_handler"
msg1 = "Caused by NewConnectionError"
msg2 = ": Failed to establish a new connection: [Errno 61] Connection refused'))"
assert msg0 in str(ioe1)
assert msg1 in str(ioe1)
assert msg2 in str(ioe1)
raise ioe1
def test_download_extent_io_error2(self):
from requests import HTTPError
from tests.local_tiles_http_server import LocalTilesHttpServer
listen_port = 8000
with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
pt0, pt1 = input_bbox
zoom = 10
with self.assertRaises(HTTPError):
try:
n_lat = pt0[0]
e_lng = pt0[1]
s_lat = pt1[0]
w_lng = pt1[1]
download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom,
source=LOCAL_URL_TILE + "_not_found_raster!")
except HTTPError as http_e:
app_logger.error(f"ae:{http_e}.")
assert "Tile URL resulted in a 404 error. Double-check your tile url:" in str(http_e)
raise http_e
|