# Copyright Crown and Cartopy Contributors
#
# This file is part of Cartopy and is released under the BSD 3-clause license.
# See LICENSE in the root of the repository for full licensing details.

import hashlib
import os
import types
import warnings

import numpy as np
from numpy.testing import assert_array_almost_equal as assert_arr_almost
import pytest
import shapely.geometry as sgeom

from cartopy import config
import cartopy.crs as ccrs
import cartopy.io.img_tiles as cimgt
import cartopy.io.ogc_clients as ogc


RESOLUTION = (30, 30)

#: Maps Google tile coordinates to native mercator coordinates as defined
#: by https://goo.gl/pgJi.
KNOWN_EXTENTS = {(0, 0, 0): (-20037508.342789244, 20037508.342789244,
                             -20037508.342789244, 20037508.342789244),
                 (2, 0, 2): (0., 10018754.17139462,
                             10018754.17139462, 20037508.342789244),
                 (0, 2, 2): (-20037508.342789244, -10018754.171394622,
                             -10018754.171394622, 0),
                 (2, 2, 2): (0, 10018754.17139462,
                             -10018754.171394622, 0),
                 (8, 9, 4): (0, 2504688.542848654,
                             -5009377.085697312, -2504688.542848654),
                 }


def GOOGLE_IMAGE_URL_REPLACEMENT(self, tile):
    # TODO: This is a hack to replace the Google image URL with a static image.
    #       This service has been deprecated by Google, so we need to replace it
    #       See https://developers.google.com/chart/image for the notice
    pytest.xfail(reason="Google has deprecated the tile API used in this test")

    x, y, z = tile
    return (f'https://chart.googleapis.com/chart?chst=d_text_outline&'
            f'chs=256x256&chf=bg,s,00000055&chld=FFFFFF|16|h|000000|b||||'
            f'Google:%20%20({x},{y})|Zoom%20{z}||||||'
            f'____________________________')


def test_google_tile_styles():
    """
    Tests that setting the Google Maps tile style works as expected.

    This is essentially just assures information is properly propagated through
    the class structure.
    """
    reference_url = ("https://mts0.google.com/vt/lyrs={style}@177000000&hl=en"
                     "&src=api&x=1&y=2&z=3&s=G")
    tile = ["1", "2", "3"]

    # Default is street.
    gt = cimgt.GoogleTiles()
    url = gt._image_url(tile)
    assert reference_url.format(style="m") == url

    # Street
    gt = cimgt.GoogleTiles(style="street")
    url = gt._image_url(tile)
    assert reference_url.format(style="m") == url

    # Satellite
    gt = cimgt.GoogleTiles(style="satellite")
    url = gt._image_url(tile)
    assert reference_url.format(style="s") == url

    # Terrain
    gt = cimgt.GoogleTiles(style="terrain")
    url = gt._image_url(tile)
    assert reference_url.format(style="t") == url

    # Streets only
    gt = cimgt.GoogleTiles(style="only_streets")
    url = gt._image_url(tile)
    assert reference_url.format(style="h") == url

    # Exception is raised if unknown style is passed.
    with pytest.raises(ValueError):
        cimgt.GoogleTiles(style="random_style")


def test_google_wts():
    gt = cimgt.GoogleTiles()

    ll_target_domain = sgeom.box(-15, 50, 0, 60)
    multi_poly = gt.crs.project_geometry(ll_target_domain, ccrs.PlateCarree())
    target_domain = multi_poly.geoms[0]

    with pytest.raises(AssertionError):
        list(gt.find_images(target_domain, -1))
    assert (tuple(gt.find_images(target_domain, 0)) ==
            ((0, 0, 0),))
    assert (tuple(gt.find_images(target_domain, 2)) ==
            ((1, 1, 2), (2, 1, 2)))

    assert (list(gt.subtiles((0, 0, 0))) ==
            [(0, 0, 1), (0, 1, 1), (1, 0, 1), (1, 1, 1)])
    assert (list(gt.subtiles((1, 0, 1))) ==
            [(2, 0, 2), (2, 1, 2), (3, 0, 2), (3, 1, 2)])

    with pytest.raises(AssertionError):
        gt.tileextent((0, 1, 0))

    assert_arr_almost(gt.tileextent((0, 0, 0)), KNOWN_EXTENTS[(0, 0, 0)])
    assert_arr_almost(gt.tileextent((2, 0, 2)), KNOWN_EXTENTS[(2, 0, 2)])
    assert_arr_almost(gt.tileextent((0, 2, 2)), KNOWN_EXTENTS[(0, 2, 2)])
    assert_arr_almost(gt.tileextent((2, 2, 2)), KNOWN_EXTENTS[(2, 2, 2)])
    assert_arr_almost(gt.tileextent((8, 9, 4)), KNOWN_EXTENTS[(8, 9, 4)])


def test_tile_bbox_y0_at_south_pole():
    tms = cimgt.MapQuestOpenAerial()

    # Check the y0_at_north_pole keywords returns the appropriate bounds.
    assert_arr_almost(tms.tile_bbox(8, 6, 4, y0_at_north_pole=False),
                      np.array(KNOWN_EXTENTS[(8, 9, 4)]).reshape([2, 2]))


def test_tile_find_images():
    gt = cimgt.GoogleTiles()
    # Test the find_images method on a GoogleTiles instance.
    ll_target_domain = sgeom.box(-10, 50, 10, 60)
    multi_poly = gt.crs.project_geometry(ll_target_domain, ccrs.PlateCarree())
    target_domain = multi_poly.geoms[0]

    assert (list(gt.find_images(target_domain, 4)) ==
            [(7, 4, 4), (7, 5, 4), (8, 4, 4), (8, 5, 4)])


@pytest.mark.network
def test_image_for_domain():
    gt = cimgt.GoogleTiles()
    gt._image_url = types.MethodType(GOOGLE_IMAGE_URL_REPLACEMENT, gt)

    ll_target_domain = sgeom.box(-10, 50, 10, 60)
    multi_poly = gt.crs.project_geometry(ll_target_domain, ccrs.PlateCarree())
    target_domain = multi_poly.geoms[0]

    _, extent, _ = gt.image_for_domain(target_domain, 6)

    ll_extent = ccrs.Geodetic().transform_points(gt.crs,
                                                 np.array(extent[:2]),
                                                 np.array(extent[2:]))
    assert_arr_almost(ll_extent[:, :2],
                      [[-11.25, 48.92249926],
                       [11.25, 61.60639637]])


def test_quadtree_wts():
    qt = cimgt.QuadtreeTiles()

    ll_target_domain = sgeom.box(-15, 50, 0, 60)
    multi_poly = qt.crs.project_geometry(ll_target_domain, ccrs.PlateCarree())
    target_domain = multi_poly.geoms[0]

    with pytest.raises(ValueError):
        list(qt.find_images(target_domain, 0))

    assert qt.tms_to_quadkey((1, 1, 1)) == '1'
    assert qt.quadkey_to_tms('1') == (1, 1, 1)

    assert qt.tms_to_quadkey((8, 9, 4)) == '1220'
    assert qt.quadkey_to_tms('1220') == (8, 9, 4)

    assert tuple(qt.find_images(target_domain, 1)) == ('0', '1')
    assert tuple(qt.find_images(target_domain, 2)) == ('03', '12')

    assert list(qt.subtiles('0')) == ['00', '01', '02', '03']
    assert list(qt.subtiles('11')) == ['110', '111', '112', '113']

    with pytest.raises(ValueError):
        qt.tileextent('4')

    assert_arr_almost(qt.tileextent(''), KNOWN_EXTENTS[(0, 0, 0)])
    assert_arr_almost(qt.tileextent(qt.tms_to_quadkey((2, 0, 2), google=True)),
                      KNOWN_EXTENTS[(2, 0, 2)])
    assert_arr_almost(qt.tileextent(qt.tms_to_quadkey((0, 2, 2), google=True)),
                      KNOWN_EXTENTS[(0, 2, 2)])
    assert_arr_almost(qt.tileextent(qt.tms_to_quadkey((2, 0, 2), google=True)),
                      KNOWN_EXTENTS[(2, 0, 2)])
    assert_arr_almost(qt.tileextent(qt.tms_to_quadkey((2, 2, 2), google=True)),
                      KNOWN_EXTENTS[(2, 2, 2)])
    assert_arr_almost(qt.tileextent(qt.tms_to_quadkey((8, 9, 4), google=True)),
                      KNOWN_EXTENTS[(8, 9, 4)])


def test_mapbox_tiles_api_url():
    token = 'foo'
    map_name = 'bar'
    tile = [0, 1, 2]
    exp_url = ('https://api.mapbox.com/styles/v1/mapbox/bar/tiles'
               '/2/0/1?access_token=foo')

    mapbox_sample = cimgt.MapboxTiles(token, map_name)
    url_str = mapbox_sample._image_url(tile)
    assert url_str == exp_url


def test_mapbox_style_tiles_api_url():
    token = 'foo'
    username = 'baz'
    map_id = 'bar'
    tile = [0, 1, 2]
    exp_url = ('https://api.mapbox.com/styles/v1/'
               'baz/bar/tiles/256/2/0/1'
               '?access_token=foo')

    mapbox_sample = cimgt.MapboxStyleTiles(token, username, map_id)
    url_str = mapbox_sample._image_url(tile)
    assert url_str == exp_url


@pytest.mark.parametrize("style,extension,resolution", [
    ("alidade_smooth", "png", ""),
    ("alidade_smooth", "png", "@2x"),
    ("stamen_watercolor", "jpg", "")])
def test_stadia_maps_tiles_api_url(style, extension, resolution):
    apikey = 'foo'
    tile = [0, 1, 2]
    exp_url = ('http://tiles.stadiamaps.com/tiles/'
               f'{style}/2/0/1{resolution}.{extension}'
               '?api_key=foo')

    sample = cimgt.StadiaMapsTiles(apikey, style=style, resolution=resolution)
    url_str = sample._image_url(tile)
    assert url_str == exp_url


def test_ordnance_survey_tile_styles():
    """
    Tests that setting the Ordnance Survey tile style works as expected.

    This is essentially just assures information is properly propagated through
    the class structure.
    """
    dummy_apikey = "None"

    ref_url = "https://api.os.uk/maps/raster/v1/zxy/" \
              "{layer}/{z}/{x}/{y}.png?key=None"
    tile = ["1", "2", "3"]

    # Default is Road_3857.
    ordsurvey = cimgt.OrdnanceSurvey(dummy_apikey)
    url = ordsurvey._image_url(tile)
    assert url == ref_url.format(layer="Road_3857",
                                 z=tile[2], y=tile[1], x=tile[0])

    for layer in ("Road_3857", "Light_3857", "Outdoor_3857",
                  "Road", "Light", "Outdoor"):
        ordsurvey = cimgt.OrdnanceSurvey(dummy_apikey, layer=layer)
        url = ordsurvey._image_url(tile)
        layer = layer if layer.endswith("_3857") else layer + "_3857"
        assert url == ref_url.format(layer=layer,
                                     z=tile[2], y=tile[1], x=tile[0])

    # Exception is raised if unknown style is passed.
    with pytest.raises(ValueError):
        cimgt.OrdnanceSurvey(dummy_apikey, layer="random_style")


@pytest.mark.network
def test_ordnance_survey_get_image():
    # In order to test fetching map images from OS
    # an API key needs to be provided
    try:
        api_key = os.environ['ORDNANCE_SURVEY_API_KEY']
    except KeyError:
        pytest.skip('ORDNANCE_SURVEY_API_KEY environment variable is unset.')

    os1 = cimgt.OrdnanceSurvey(api_key, layer="Outdoor_3857")
    os2 = cimgt.OrdnanceSurvey(api_key, layer="Light_3857")

    tile = (500, 300, 10)

    img1, extent1, _ = os1.get_image(tile)
    img2, extent2, _ = os2.get_image(tile)

    # Different images for different layers
    assert img1 != img2

    # The extent is the same though
    assert extent1 == extent2


def test_azuremaps_tiles_api_url():
    subscription_key = 'foo'
    tileset_id = 'bar'
    tile = [0, 1, 2]

    exp_url = ('https://atlas.microsoft.com/map/tile?api-version=2.0'
               '&tilesetId=bar&x=0&y=1&zoom=2&subscription-key=foo')

    az_maps_sample = cimgt.AzureMapsTiles(subscription_key, tileset_id)
    url_str = az_maps_sample._image_url(tile)
    assert url_str == exp_url


@pytest.mark.network
def test_azuremaps_get_image():
    # In order to test fetching map images from Azure Maps
    # an API key needs to be provided
    try:
        api_key = os.environ['AZURE_MAPS_SUBSCRIPTION_KEY']
    except KeyError:
        pytest.skip('AZURE_MAPS_SUBSCRIPTION_KEY environment variable '
                    'is unset.')

    am1 = cimgt.AzureMapsTiles(api_key, tileset_id="microsoft.imagery")
    am2 = cimgt.AzureMapsTiles(api_key, tileset_id="microsoft.base.road")

    tile = (500, 300, 10)

    img1, extent1, _ = am1.get_image(tile)
    img2, extent2, _ = am2.get_image(tile)

    # Different images for different layers
    assert img1 != img2

    # The extent is the same though
    assert extent1 == extent2


@pytest.mark.network
@pytest.mark.parametrize('cache_dir', ["tmpdir", True, False])
@pytest.mark.skipif(not ogc._OWSLIB_AVAILABLE, reason='OWSLib is unavailable.')
def test_wmts_cache(cache_dir, tmp_path):
    if cache_dir == "tmpdir":
        tmpdir_str = str(tmp_path)
    else:
        tmpdir_str = cache_dir

    if cache_dir is True:
        config["cache_dir"] = str(tmp_path)

    # URI = 'https://map1c.vis.earthdata.nasa.gov/wmts-geo/wmts.cgi'
    # layer_name = 'VIIRS_CityLights_2012'
    URI = 'https://basemap.nationalmap.gov/arcgis/rest/services/USGSImageryOnly/MapServer/WMTS/1.0.0/WMTSCapabilities.xml'
    layer_name='USGSImageryOnly'
    projection = ccrs.PlateCarree()

    # Fetch tiles and save them in the cache
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter('always')
        source = ogc.WMTSRasterSource(URI, layer_name, cache=tmpdir_str)
    extent = [-10, 10, 40, 60]
    located_image, = source.fetch_raster(projection, extent,
                                         RESOLUTION)

    # Do not check the result if the cache is disabled
    if cache_dir is False:
        assert source.cache_path is None
        return

    # Check that the warning is properly raised (only when cache is True)
    if cache_dir is True:
        assert len(w) == 1
    else:
        assert len(w) == 0

    # Define expected results
    x_y_f_h = [
        (1, 1, '1_1.npy', '0de548bd47e4579ae0500da6ceeb08e7'),
        (1, 2, '1_2.npy', '4beebcd3e4408af5accb440d7b4c8933'),
    ]

    # Check the results
    cache_dir_res = source.cache_path / "WMTSRasterSource"
    files = list(cache_dir_res.iterdir())
    hashes = {
        f:
            hashlib.md5(
                np.load(cache_dir_res / f, allow_pickle=True).data
            ).hexdigest()
        for f in files
    }
    assert sorted(files) == [cache_dir_res / f for x, y, f, h in x_y_f_h]
    assert set(files) == set([cache_dir_res / c for c in source.cache])

    assert sorted(hashes.values()) == sorted(
        h for x, y, f, h in x_y_f_h
    )

    # Update images in cache (all white)
    for f in files:
        filename = cache_dir_res / f
        img = np.load(filename, allow_pickle=True)
        img.fill(255)
        np.save(filename, img, allow_pickle=True)

    wmts_cache = ogc.WMTSRasterSource(URI, layer_name, cache=tmpdir_str)
    located_image_cache, = wmts_cache.fetch_raster(projection, extent,
                                         RESOLUTION)

    # Check that the new fetch_raster() call used cached images
    assert wmts_cache.cache == set([cache_dir_res / c for c in source.cache])
    assert (np.array(located_image_cache.image) == 255).all()


@pytest.mark.network
@pytest.mark.parametrize('cache_dir', ["tmpdir", True, False])
def test_cache(cache_dir, tmp_path):
    if cache_dir == "tmpdir":
        tmpdir_str = str(tmp_path)
    else:
        tmpdir_str = cache_dir

    if cache_dir is True:
        config["cache_dir"] = str(tmp_path)

    # Fetch tiles and save them in the cache
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter('always')
        gt = cimgt.GoogleTiles(cache=tmpdir_str)
    gt._image_url = types.MethodType(GOOGLE_IMAGE_URL_REPLACEMENT, gt)

    ll_target_domain = sgeom.box(-10, 50, 10, 60)
    multi_poly = gt.crs.project_geometry(ll_target_domain, ccrs.PlateCarree())
    target_domain = multi_poly.geoms[0]

    img_init, _, _ = gt.image_for_domain(target_domain, 6)

    # Do not check the result if the cache is disabled
    if cache_dir is False:
        assert gt.cache_path is None
        return

    # Check that the warning is properly raised (only when cache is True)
    if cache_dir is True:
        assert len(w) == 1
    else:
        assert len(w) == 0

    # Define expected results
    x_y_z_f_h = [
        (30, 18, 6, '30_18_6.npy', '545db25f1aa348ad85e1f437fd0db0d9'),
        (30, 19, 6, '30_19_6.npy', '10355add0674bfa33f673ea27a6d1206'),
        (30, 20, 6, '30_20_6.npy', 'ab3e7f2ed8d71977ac176094973695ae'),
        (30, 21, 6, '30_21_6.npy', '3e8947b93a6ffa07f22cfea4042a4740'),
        (31, 18, 6, '31_18_6.npy', 'd0fa58b9146aa99b273eb75256b328cc'),
        (31, 19, 6, '31_19_6.npy', '9255bd0cd22736bd2c25a9087bd47b20'),
        (31, 20, 6, '31_20_6.npy', 'ac0f7e32bdf8edb50d1dccf3ec0ef446'),
        (31, 21, 6, '31_21_6.npy', 'f36b8cc1825bf267b2daead837facae9'),
        (32, 18, 6, '32_18_6.npy', '9f4ddd90cd1ae76ef2bbc8f0252ead91'),
        (32, 19, 6, '32_19_6.npy', 'a995803578bb94ecfca8563754717196'),
        (32, 20, 6, '32_20_6.npy', 'def9e71d77fd6007c77c2a14dfae858f'),
        (32, 21, 6, '32_21_6.npy', 'a3d7935037019ec58ae78f60e6fb924e'),
        (33, 18, 6, '33_18_6.npy', '4e51e32da73fb99229817dcd7b7e1f4f'),
        (33, 19, 6, '33_19_6.npy', 'b9b5057fa012c5788cbbe1e18c9bb512'),
        (33, 20, 6, '33_20_6.npy', 'b55a7c0a8d86167df496732f85bddcf9'),
        (33, 21, 6, '33_21_6.npy', '4208ba897c460e9bb0d2469552e127ff')
    ]

    # Check the results
    cache_dir_res = gt.cache_path / "GoogleTiles"
    files = list(cache_dir_res.iterdir())
    hashes = {
        f:
            hashlib.md5(
                np.load(cache_dir_res / f, allow_pickle=True).data
            ).hexdigest()
        for f in files
    }
    assert sorted(files) == [cache_dir_res / f for x, y, z, f, h in x_y_z_f_h]
    assert set(files) == gt.cache

    assert sorted(hashes.values()) == sorted(
        h for x, y, z, f, h in x_y_z_f_h
    )

    # Update images in cache (all white)
    for f in files:
        filename = cache_dir_res / f
        img = np.load(filename, allow_pickle=True)
        img.fill(255)
        np.save(filename, img, allow_pickle=True)

    gt_cache = cimgt.GoogleTiles(cache=tmpdir_str)
    gt_cache._image_url = types.MethodType(
        GOOGLE_IMAGE_URL_REPLACEMENT, gt_cache)
    img_cache, _, _ = gt_cache.image_for_domain(target_domain, 6)

    # Check that the new image_for_domain() call used cached images
    assert gt_cache.cache == gt.cache
    assert (img_cache == 255).all()