Skip to main content

WMTS

Haiyue3/29/24About 15 mingeosciencetile coordinates

Core Code

wmts_main.py
import web
import sys
sys.path.append('.')
import datetime
from utils.oss_util import oss_util
from utils.tile_util import tile_util
from utils.vivid_download import Vivid_WMTS
from utils.vivid_db import vivid_db
from etc import *

accessKey = ''
accessSecretKey = ''
bucketName = ''
oss_obj = oss_util(accessKey, accessSecretKey, bucketName)


urls = (
    '/wmts', 'wmts_ctrl',
    '/WMTS', 'wmts_ctrl',
    '/wmts/(.*)/(.*)/(.*)/(.*)/(.*).jpg', 'wmts_ctrl_resurl',
    '/WMTS/(.*)/(.*)/(.*)/(.*)/(.*).jpg', 'wmts_ctrl_resurl',
)
app = web.application(urls, globals())


def get_oss_key(ilayer, row, col):
    return f'image/{ilayer}/{row}/{col}.jpg'

tile_obj = tile_util()
vivid_obj = Vivid_WMTS("access_key", 'prefix', "account", "password")

vivid_sqlite_obj = vivid_db(wmts_sqlite)

class wmts_tile_service:
    def getTile(self, layer, tilematrix, row, col):
        #return self.getGoogleTile(layer, tilematrix, row, col)
        return self.getVividTile(layer, tilematrix, row, col)
    
    def getVividTile(self, layer, tilematrix, row, col):
        #return self.getTileFromOSS(layer, tilematrix, row, col)
        return self.getTileFromSqlite(tilematrix, row, col)
        
    def getTileFromSqlite(self, tilematrix, row, col):
        return vivid_sqlite_obj.get_tile(int(tilematrix)-1, int(row), int(col))
    
    def getTileFromOSS(self, layer, tilematrix, row, col):
        quad = tile_obj.WMTS2Quad(int(tilematrix)-1, int(row), int(col))
        key = vivid_obj.get_oss_url(quad)
        print(key)
        if oss_obj.is_exists(key):
            print(f'{tilematrix}-{row}-{col}', quad,key, ' found '*10)
            #img = vivid_obj.get(quad)
            return oss_obj.get_obj(key)
        print(quad,key, 'not found')
        return None
    
    def getGoogleTile(self, layer, tilematrix, row, col):
        # Get max row of google wmts
        maxr = 2**int(tilematrix)
        # switch the position
        row = 2**(int(tilematrix)-1) - row
        # Convert the swtched index to google index
        row = int(maxr/4 + row)-1
        return oss_obj.get_obj(get_oss_key(tilematrix, row, col))
        
    def __convertwmts2dbindex__(self, wmts_tilematrix, wmts_r, wmts_c):
        if int(wmts_tilematrix) < 1:
            return None
        level = int(wmts_tilematrix)  #google level
        col = wmts_c                #google column
        maxr = 2**level          #google max row num
        row = (maxr/2-(wmts_r+1)) + maxr / 4          #google row
        
        return int(level), int(row), int(col)
        
    def getCapabilities(self):
        rc = None
        capabilityBase = ''
        content = None
        with open('template/getCapabilities.xml', 'r', encoding='utf-8') as fr:
            content = fr.read()
        return content.replace("127.0.0.1:8045", f"{wmts_ip}:{wmts_port}")

class wmts_service_list_ctrl:
    def __init__(self):
        pass

    def GET(self):
        val = 'webpy get test'
        web.header("Access-Control-Allow-Origin", "*")
        web.header('content-type','text/json')
        return tile_service().getTileDB()
    
    def POST(self):
        val = 'webpy dg_meta'
        web.header("Access-Control-Allow-Origin", "*")
        web.header('content-type','text')
        return val

#tile404 = None
#with open('404.png','rb') as fr:
#    tile404 = fr.read()
    
def get404Tile():
    global tile404
    return None

class wmts_ctrl_resurl:
    def __init__(self):
        pass

    #def GET(self, layer):
    def GET(self, layer, tilematrixset, TileMatrix, TileRow, TileCol):
        val = 'webpy get test'
        web.header("Access-Control-Allow-Origin", "*")
        rc = None
        starttime = datetime.datetime.now()
        layer     = layer
        tilelevel = TileMatrix
        tilecol   = TileCol
        tilerow   = TileRow
        if not(tilelevel is None or layer is None):
            web.header('content-type','image/jpeg')
            rc = wmts_tile_service().getTile(layer, tilelevel, int(tilerow), int(tilecol))
            endtime = datetime.datetime.now()
            pass
        print(rc)
        #if rc is None:
        #    rc = get404Tile()
        if rc is None:
            print(f'{tilelevel}-{tilerow}-{tilecol} not  found {"#"*10}')
            web.notfound()
        else:
            return rc
    def POST(self):
        val = 'webpy dg_meta'
        web.header("Access-Control-Allow-Origin", "*")
        web.header('content-type','text')
        return val

# 获取所有支持的图层列表
class wmts_ctrl:
    def __init__(self):
        pass

    def GET(self):
        val = 'webpy get test'
        web.header("Access-Control-Allow-Origin", "*")
        rc = None
        content = ''
        starttime = datetime.datetime.now()
        self.__kvp__ = dict(web.input())
        service   = self.__getValue__('service')
        layer     = self.__getValue__('layer')
        style     = self.__getValue__('style')
        tilematrixset = self.__getValue__('tilematrixset')
        req       = self.__getValue__('Request')
        version   = self.__getValue__('Version')
        format    = self.__getValue__('Format')
        tileMatrix = self.__getValue__('TileMatrix')
        tilecol   = self.__getValue__('TileCol')
        tilerow   = self.__getValue__('TileRow')
        
        #print(self.__kvp__)
        if not(service is None or req is None):
            if service.upper() == 'WMTS':
                if req.upper() == 'GetCapabilities'.upper():
                    web.header('content-type','text/xml')
                    rc = wmts_tile_service().getCapabilities()
                elif req.upper() == 'GetTile'.upper():
                    web.header('content-type','image/png')
                    print('Get Tile')
                    rc = wmts_tile_service().getTile(layer, tileMatrix, int(tilerow), int(tilecol))
                    endtime = datetime.datetime.now()
                    pass
                elif req.upper() == 'GetFeatureInfo'.upper():
                    pass
        
        #if rc is None:
        #    rc = get404Tile()
        
        #print(rc)
        #if rc is None:
        #    rc = get404Tile()
        if rc is None:
            print(f'{tileMatrix}-{tilerow}-{tilecol} not  found {"#"*10}')
            web.notfound()
        else:
            return rc
    
    def __getCapabilitiesContent__(self):
        rc = None
        capabilityBase = ''
        with open('getCapabilities.xml', 'r') as fr:
            content = fr.read()
            capabilityBase = content
        layerTemplate = ''
        with open('layerTemplate.xml', 'r') as fr:
            layerTemplate = fr.read()
        
        if len(capabilityBase) > 0 and len(layerTemplate)> 0:
            layers = ''
            for l in layers_conf:
                layers += layerTemplate.format(l['id'], l['title'], l['abstract'],TileMatrixSet='{TileMatrixSet}',TileMatrix='{TileMatrix}',TileRow='{TileRow}', TileCol='{TileCol}')
            rc = capabilityBase.format(layers)
        return rc
            
    
    def __getValue__(self, key):
        kvp = self.__kvp__
        rc = None
        for item in kvp.keys():
            if item.upper() == key.upper():
                rc = kvp[item]
                break
        return rc
    
    def POST(self):
        val = 'webpy dg_meta'
        return self.GET()

    def __getimg__(self, layer, level, row, col):
        return wmts_tile_service().getTile(layer, level, row, col)

if __name__=="__main__":
    web.internalerror = web.debugerror
    app.run()

Conf

etc/init
wmts_ip = "127.0.0.1"
wmts_port=89
wmts_sqlite="vivid_tiles.db"
Last Updated
Contributors: Haiyue
Default footer
Copyright © 2025 Haiyue