Python拼接地图瓦片时常见的5个错误及解决方案

import math

def wgs84_to_web_mercator(lon, lat):
    """
    将WGS84经纬度转换为Web墨卡托坐标（米）。
    Args:
        lon: 经度
        lat: 纬度
    Returns:
        (x, y) 单位：米
    """
    x = lon * 20037508.34 / 180.0
    y = math.log(math.tan((90.0 + lat) * math.pi / 360.0)) / (math.pi / 180.0)
    y = y * 20037508.34 / 180.0
    return x, y

def web_mercator_to_wgs84(x, y):
    """
    将Web墨卡托坐标转换回WGS84经纬度。
    Args:
        x: 墨卡托x坐标（米）
        y: 墨卡托y坐标（米）
    Returns:
        (lon, lat)
    """
    lon = x / 20037508.34 * 180.0
    lat = y / 20037508.34 * 180.0
    lat = 180.0 / math.pi * (2 * math.atan(math.exp(lat * math.pi / 180.0)) - math.pi / 2.0)
    return lon, lat

def deg2num(lon_deg, lat_deg, zoom):
    """
    将经纬度转换为特定缩放级别下的瓦片索引。
    Args:
        lon_deg: 经度 (WGS84)
        lat_deg: 纬度 (WGS84)
        zoom: 缩放级别 (0-通常为最大)
    Returns:
        (xtile, ytile)
    """
    lat_rad = math.radians(lat_deg)
    n = 2.0 ** zoom
    # 先换算为0-1范围内的比例
    xtile = (lon_deg + 180.0) / 360.0 * n
    # 注意这里的公式，使用了tan和log
    ytile = (1.0 - math.log(math.tan(lat_rad) + (1 / math.cos(lat_rad))) / math.pi) / 2.0 * n
    # 关键步骤：使用 floor 确保索引正确
    return int(math.floor(xtile)), int(math.floor(ytile))

import matplotlib.pyplot as plt
import numpy as np
from PIL import Image

def get_tile_extent(xtile, ytile, zoom):
    """
    根据瓦片索引计算其地理范围 (Web墨卡托坐标)。
    Args:
        xtile, ytile: 瓦片索引
        zoom: 缩放级别
    Returns:
        extent list: [x_min, x_max, y_min, y_max] in Web Mercator meters.
    """
    # 每个缩放级别的瓦片总数
    n = 2.0 ** zoom
    # 整个世界的墨卡托坐标范围
    world_size = 20037508.34 * 2
    # 单个瓦片的宽度/高度（米）
    tile_size = world_size / n

    # 计算当前瓦片的边界（米）
    x_min = -20037508.34 + xtile * tile_size
    x_max = -20037508.34 + (xtile + 1) * tile_size
    # 注意：墨卡托y轴原点在顶部，向下为正
    y_max = 20037508.34 - ytile * tile_size
    y_min = 20037508.34 - (ytile + 1) * tile_size

    return [x_min, x_max, y_min, y_max]

# 拼接示例
fig, ax = plt.subplots(figsize=(10, 8))
ax.set_aspect('equal') # 保持纵横比，防止地图变形

# 假设我们有三张相邻的瓦片
tiles_to_draw = [(x, y) for x in range(10, 13) for y in range(5, 8)]
zoom = 12

for xtile, ytile in tiles_to_draw:
    # 1. 加载瓦片图像（这里用随机数组模拟）
    tile_path = f'./tiles/{zoom}/{xtile}/{ytile}.png'
    try:
        img = np.array(Image.open(tile_path)) # 或使用 matplotlib.image.imread
    except FileNotFoundError:
        continue # 处理缺失瓦片

    # 2. 计算该瓦片的地理范围
    extent = get_tile_extent(xtile, ytile, zoom)

    # 3. 使用正确的extent和origin显示
    ax.imshow(img, extent=extent, origin='upper', zorder=0)

# 现在，你可以用相同的数据坐标（墨卡托米）在ax上绘制你的业务数据了
# ax.plot([x1, x2], [y1, y2], 'r-', linewidth=2) # 这会精确叠加在地图上

import concurrent.futures
import requests
import os
from urllib.parse import urljoin

def download_single_tile(base_url, xtile, ytile, zoom, save_dir, max_retries=3):
    """下载单张瓦片，包含重试机制"""
    url = urljoin(base_url, f'{zoom}/{xtile}/{ytile}.png')
    save_path = os.path.join(save_dir, f'{zoom}/{xtile}/{ytile}.png')
    os.makedirs(os.path.dirname(save_path), exist_ok=True)

    for attempt in range(max_retries):
        try:
            # 设置超时，避免无限等待
            resp = requests.get(url, timeout=10)
            resp.raise_for_status() # 检查HTTP状态码
            with open(save_path, 'wb') as f:
                f.write(resp.content)
            print(f'成功下载: {save_path}')
            return True
        except (requests.exceptions.RequestException, IOError) as e:
            print(f'下载失败 {url}, 尝试 {attempt+1}/{max_retries}: {e}')
            if attempt == max_retries - 1:
                return False
    return False

def download_tile_batch(tile_list, base_url, save_dir, max_workers=20):
    """并发下载一批瓦片"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_tile = {}
        for xtile, ytile, zoom in tile_list:
            future = executor.submit(download_single_tile, base_url, xtile, ytile, zoom, save_dir)
            future_to_tile[future] = (xtile, ytile, zoom)

        for future in concurrent.futures.as_completed(future_to_tile):
            xtile, ytile, zoom = future_to_tile[future]
            try:
                success = future.result()
                if not success:
                    # 记录失败的任务，后续可能用备用源或留空
                    print(f'瓦片 {zoom}/{xtile}/{ytile} 最终下载失败')
            except Exception as exc:
                print(f'瓦片 {zoom}/{xtile}/{ytile} 生成异常: {exc}')

from PIL import Image

def merge_tiles_vertically(tile_dir, x_range, y_start, y_end, zoom):
    """垂直拼接一列瓦片，减少内存占用"""
    column_images = []
    for y in range(y_start, y_end + 1):
        row_images = []
        for x in range(x_range[0], x_range[1] + 1):
            path = os.path.join(tile_dir, f'{zoom}/{x}/{y}.png')
            if os.path.exists(path):
                row_images.append(Image.open(path))
            else:
                # 用空白瓦片占位
                row_images.append(Image.new('RGB', (256, 256), (200, 200, 200)))
        # 水平拼接一行
        row_merged = Image.new('RGB', (256 * len(row_images), 256))
        x_offset = 0
        for img in row_images:
            row_merged.paste(img, (x_offset, 0))
            x_offset += img.width
        column_images.append(row_merged)

    # 垂直拼接所有行
    total_height = sum(img.height for img in column_images)
    result = Image.new('RGB', (column_images[0].width, total_height))
    y_offset = 0
    for img in column_images:
        result.paste(img, (0, y_offset))
        y_offset += img.height
        img.close() # 及时关闭，释放内存
    return result

class TileSource:
    """封装不同瓦片源的配置"""
    SOURCES = {
        'osm': {
            'url_template': 'https://tile.openstreetmap.org/{z}/{x}/{y}.png',
            'projection': 'epsg:3857',
            'tms': False, # XYZ 方案
            'max_zoom': 19,
            'attribution': '© OpenStreetMap contributors'
        },
        # 示例：假设一个TMS源
        'custom_tms': {
            'url_template': 'http://example.com/{z}/{x}/{y}.png',
            'projection': 'epsg:3857',
            'tms': True, # TMS 方案，y轴原点在底部
            'max_zoom': 18,
        }
    }

    def __init__(self, source_key='osm'):
        self.config = self.SOURCES.get(source_key)
        if not self.config:
            raise ValueError(f'未知的瓦片源: {source_key}')

    def get_tile_url(self, x, y, z):
        """根据配置生成瓦片URL"""
        if self.config['tms']:
            # TMS 转 XYZ (y轴翻转)
            y = (2 ** z - 1) - y
        url = self.config['url_template'].format(x=x, y=y, z=z)
        return url

    def download_tile(self, x, y, z, save_path):
        """下载瓦片，处理可能的404"""
        url = self.get_tile_url(x, y, z)
        try:
            resp = requests.get(url, timeout=5, headers={'User-Agent': 'Your-App/1.0'})
            if resp.status_code == 200:
                with open(save_path, 'wb') as f:
                    f.write(resp.content)
                return True
            elif resp.status_code == 404:
                # 该位置可能无瓦片（如海洋），记录并跳过
                print(f'瓦片不存在 (404): {url}')
                return False
            else:
                print(f'下载失败，状态码 {resp.status_code}: {url}')
                return False
        except requests.exceptions.RequestException as e:
            print(f'网络错误: {e}')
            return False