diff --git a/Ai_tottle/ai_tottle_api.py b/Ai_tottle/ai_tottle_api.py index 9e2f6ef..c4040bc 100644 --- a/Ai_tottle/ai_tottle_api.py +++ b/Ai_tottle/ai_tottle_api.py @@ -12,7 +12,6 @@ from queue import Queue import gdal2tiles as gdal2tiles from map_find import map_process_images from yolo_train import auto_train -from map_cut import process_tiling from cv_video_counter import start_video_session,switch_model_session,stop_video_session,stream_sessions import torch from yolo_photo import map_process_images_with_progress # 引入你的处理函数 diff --git a/Ai_tottle/map_cut.py b/Ai_tottle/map_cut.py deleted file mode 100644 index 674d8a7..0000000 --- a/Ai_tottle/map_cut.py +++ /dev/null @@ -1,172 +0,0 @@ -import os -import tempfile -import uuid -import aiohttp -import yaml -from minio import Minio -from minio.error import S3Error -import subprocess -import sys -import json -from osgeo import gdal, osr -from miniohelp import load_config,upload_file - -# 异步下载文件函数 -async def download_file(url: str, output_path: str) -> bool: - try: - async with aiohttp.ClientSession() as session: - async with session.get(url) as response: - if response.status == 200: - with open(output_path, 'wb') as f: - f.write(await response.read()) - return True - else: - print(f"下载失败: {url}") - return False - except Exception as e: - print(f"下载错误: {e}") - return False - -# 上传文件到 MinIO -def upload_to_minio(local_file: str, bucket: str, object_name: str,minio_client: Minio): - try: - minio_client.fput_object(bucket, object_name, local_file) - except S3Error as e: - print(f"上传失败: {local_file} -> {object_name} 错误: {e}") - -def generate_tiles_from_tif_cli( - input_tif: str, - output_dir: str, - zoom_levels: str = "0-18", - processes: int = 4, - verbose: bool = True -): - # Step 1: 调用 gdal2tiles - args = [ - sys.executable, - "-m", "osgeo_utils.gdal2tiles", - "-z", zoom_levels, - "--processes", str(processes) - ] - if verbose: - args.append("-v") - args += [input_tif, output_dir] - print(f"Running: {' '.join(args)}") - subprocess.run(args, check=True) - - # Step 2: 提取地理信息 - ds = gdal.Open(input_tif) - gt = ds.GetGeoTransform() - width = ds.RasterXSize - height = ds.RasterYSize - - minx = gt[0] - maxy = gt[3] - pixel_width = gt[1] - pixel_height = gt[5] - maxx = minx + width * pixel_width - miny = maxy + height * pixel_height - - # 尝试解析坐标系 EPSG 编码 - srs = osr.SpatialReference(wkt=ds.GetProjection()) - try: - proj_epsg = int(srs.GetAttrValue("AUTHORITY", 1)) - except: - proj_epsg = 4326 # fallback 默认值 - - # Step 3: 构造 JSON 数据 - bounds = { - "west": minx, - "south": miny, - "east": maxx, - "north": maxy - } - - json_data = { - "bounds": bounds, - "contentType": "image/png", - "generatetool": "custom_script@yourdomain.com/gdal2tiles", - "latLonBounds": bounds, # 和 bounds 相同 - "maxzoom": int(zoom_levels.split("-")[-1]), - "minzoom": int(zoom_levels.split("-")[0]), - "proj": proj_epsg, - "tilesize": 256, - "tiletrans": "google", - "type": "image", - "ziped": False - } - - # Step 4: 保存为 JSON 文件 - json_path = output_dir.rstrip("\\/") + ".json" - with open(json_path, 'w', encoding='utf-8') as f: - json.dump(json_data, f, indent=2) - - print(f"瓦片生成成功，元数据已保存到: {json_path}") - -# 主处理函数 -async def process_tiling(tif_url: str = None, prj_url: str = None, zoom_levels: str = "1-18", bucket: str = "300",config: str = "222.212.85.86") -> dict: - # 创建唯一任务ID - task_id = str(uuid.uuid4()) - temp_dir = tempfile.mkdtemp() - tif_path = os.path.join(temp_dir, f"{task_id}.tif") - prj_path = os.path.join(temp_dir, f"{task_id}.prj") if prj_url else None - output_dir = os.path.join(temp_dir, f"tiles_{task_id}") - os.makedirs(output_dir, exist_ok=True) - - # 下载 TIF - if tif_url: - tif_ok = await download_file(tif_url, tif_path) - if not tif_ok: - raise Exception("TIF 下载失败") - - # 下载 PRJ（如果有） - if prj_url: - prj_ok = await download_file(prj_url, prj_path) - if prj_ok: - # GDAL 要求 PRJ 文件与 TIF 同名（不同后缀） - fixed_prj_path = tif_path.replace('.tif', '.prj') - os.rename(prj_path, fixed_prj_path) - else: - print("PRJ 下载失败，继续处理...") - - # 切片 - generate_tiles_from_tif_cli(tif_path, output_dir, zoom_levels=zoom_levels, processes=4, verbose=True) - - # 上传结果到 MinIO - minio_base_path = f"maps/{task_id}/" - for root, _, files in os.walk(output_dir): - for file in files: - local_path = os.path.join(root, file) - rel_path = os.path.relpath(local_path, output_dir) - object_name = os.path.join(minio_base_path, rel_path).replace("\\", "/") - upload_to_minio(local_path, bucket, object_name) - - minio_info = load_config(config) - # 清洗 endpoint，去掉 http:// 或 https:// 前缀 - endpoint = minio_info["MinIOEndpoint"].replace("http://", "").replace("https://", "") - - # 返回构建的瓦片URL信息 - base_url = f"http://{endpoint}//{bucket}/{minio_base_path}" - min_zoom, max_zoom = zoom_levels.split("-") - - return { - "tile_url": { - "base_url": base_url, - "format": "{z}/{x}/{y}.png", - "preview_url": f"{base_url}openlayers.html", - "min_zoom": min_zoom, - "max_zoom": max_zoom, - "tilejson_url": f"{base_url}tilemapresource.xml" - } - } - -# 示例调用 -if __name__ == "__main__": - # tif_url = "https://your-server.com/sample.tif" - # prj_url = "https://your-server.com/sample.prj" # 可选 - # result = asyncio.run(process_tiling(tif_url, prj_url, zoom_levels="5-15")) - # print(result) - - input_tif = r"e:/new.tif" - output_dir = r"D:/work/AI_Python/cut" - generate_tiles_from_tif_cli(input_tif, output_dir, zoom_levels="5-15", processes=4, verbose=True)