app.py 文件源码

def create_warped_vrt(self, id):
    raster_path = os.path.abspath(os.path.join(IMAGERY_PATH, id, 'index.tif'))
    vrt_path = os.path.abspath(os.path.join(IMAGERY_PATH, id, 'index.vrt'))
    meta = get_metadata(id)
    approximate_zoom = meta['meta']['approximateZoom']

    # create a warped VRT to reproject on the fly
    gdalwarp = [
        'gdalwarp',
        raster_path,
        vrt_path,
        '-r', 'cubic',
        '-t_srs', 'epsg:3857',
        '-overwrite',
        '-of', 'VRT',
        '-te', '-20037508.34', '-20037508.34', '20037508.34', '20037508.34',
        '-ts', str(2 ** approximate_zoom * 256), str(2 ** approximate_zoom * 256),
    ]

    # add an alpha band (for NODATA) if one wasn't already included
    if meta['meta']['bandCount'] < 4:
        gdalwarp.append('-dstalpha')

    started_at = datetime.utcnow()

    self.update_state(state='RUNNING',
                      meta={
                        'name': 'warped-vrt',
                        'started_at': started_at.isoformat(),
                        'status': 'Creating warped VRT'
                      })

    try:
        returncode = subprocess.call(gdalwarp, timeout=TASK_TIMEOUT)
    except subprocess.TimeoutExpired as e:
        raise Exception(json.dumps({
            'name': 'warped-vrt',
            'started_at': started_at.isoformat(),
            'command': ' '.join(gdalwarp),
            'status': 'Timed out'
        }))

    if returncode != 0:
        raise Exception(json.dumps({
            'name': 'warped-vrt',
            'started_at': started_at.isoformat(),
            'command': ' '.join(gdalwarp),
            'return_code': returncode,
            'status': 'Failed'
        }))

    return {
        'completed_at': datetime.utcnow().isoformat(),
        'started_at': started_at,
        'status': 'Warped VRT creation completed'
    }