sky_starearth_tiler/src/app/mission_jobs/image_slice.py

import os
import shutil

from app.defines import TILE_GRID_TYPE
from app.utils.create_geojsonl import create_default_geojsonl
from config import Config
from starearth import FileSystem, Grid
from starearth.storage.osm_zxy import StorageOSMZXY
from slice import slice
from starearth.utils.general_utils import print_log

config = Config()


def slice_zxy(
        input_file,
        output_path,
        tile_size,
        tile_grid,
        tile_format,
        min_zoom,
        max_zoom,
        epsg='EPSG:4326',
        nodata=0,
        enable_msmt=0,
        render_type="RGB",
        channels=[1, 2, 3],
        merging=2,
):
    print_log('开始切片...')

    # epsg = verify_geotiff(input_file, None)

    filesystem_cache = FileSystem(config.common.CACHE_PATH)

    tmp_tiles = filesystem_cache.path('tiles')

    sliceTiler_type_dict = {TILE_GRID_TYPE.WebMercatorQuad: Grid.WebMercatorQuad,
                            TILE_GRID_TYPE.WGS1984Quad: Grid.WGS1984Quad,
                            TILE_GRID_TYPE.GoogleEarthQuad: Grid.GoogleEarthQuad}

    sliceTiler_type = sliceTiler_type_dict[tile_grid]


    geojsonl = os.path.splitext(input_file)[0] + '.geojsonl'

    if not os.path.exists(str(geojsonl)):
        create_default_geojsonl(input_file, geojsonl, {"date": os.path.splitext(os.path.basename(input_file))[0]})

    slice \
            (
            input_file=input_file,
            tile_grid=sliceTiler_type,
            tile_size=tile_size,
            tmp_tiles=tmp_tiles,
            epsg=epsg,
            nodata=nodata,
            tile_format=tile_format,
            min_zoom=min_zoom,
            max_zoom=max_zoom,
            enable_msmt=enable_msmt,
            channels=channels,
            render_type=render_type,
            lut=None,
        )
    print_log('执行切片完成.')

    base_name = os.path.splitext(os.path.basename(input_file))[0]
    print_log("开始OSM_ZXY存储..")

    storage_osmzxy_obj = StorageOSMZXY(
        tiles_path=os.path.join(str(tmp_tiles), base_name),
        min_zoom=min_zoom,
        max_zoom=max_zoom,
        merging=merging,
        output_path=output_path,
        tile_format=tile_format
    )
    storage_osmzxy_obj.storage()

    # 删除临时目录
    if os.path.exists(os.path.join(str(tmp_tiles), base_name)):
        shutil.rmtree(os.path.join(str(tmp_tiles), base_name))

    print_log("OSM_ZXY存储完毕..")