基于Ganos百行代码实现亿级矢量空间数据在线可视化

ST_AsPng('points_geom', '1_2_1','{"point_size": 5,"line_width": 2,"line_color": "#003399FF","fill_color": "#6699CCCC","background": "#FFFFFF00"}')

gid|geom                                                                                                                                                                                                                                                           |
---|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
  1|MULTIPOLYGON(((-88.066953 34.916114 0,-88.066704 34.916114 0,-88.066704 34.91602 0,-88.066953 34.91602 0,-88.066953 34.916114 0)))                 
  2|MULTIPOLYGON(((-87.924658 34.994797 0,-87.924791 34.99476 0,-87.924817 34.994824 0,-87.924685 34.994861 0,-87.924658 34.994797 0)))                

# -*- coding: utf-8 -*-
# @File : Vector.py
import json
from psycopg2 import pool
from threading import Semaphore
from flask import Flask, jsonify, Response, send_from_directory
import binascii
# 连接参数
CONNECTION = "dbname=postgres user=postgres password=postgres host=YOUR_HOST port=5432"
class ReallyThreadedConnectionPool(pool.ThreadedConnectionPool):
    """
    面向多线程的连接池，提高地图瓦片类高并发场景的响应。
    """
    def __init__(self, minconn, maxconn, *args, **kwargs):
        self._semaphore = Semaphore(maxconn)
        super().__init__(minconn, maxconn, *args, **kwargs)
    def getconn(self, *args, **kwargs):
        self._semaphore.acquire()
        return super().getconn(*args, **kwargs)
    def putconn(self, *args, **kwargs):
        super().putconn(*args, **kwargs)
        self._semaphore.release()
class VectorViewer:
    def __init__(self, connect, table_name, column_name, fid):
        self.table_name = table_name
        self.column_name = column_name
        # 创建一个连接池
        self.connect = ReallyThreadedConnectionPool(5, 10, connect)
        # 约定金字塔表名
        self.pyramid_table = f"{self.table_name}_{self.column_name}"
        self.fid = fid
        self.tileSize = 512
        # self._build_pyramid()
    def _build_pyramid(self):
        """创建金字塔"""
        config = {
            "name": self.pyramid_table,
            "tileSize": self.tileSize
        }
        sql = f"select st_BuildPyramid('{self.table_name}','{self.column_name}','{self.fid}','{json.dumps(config)}')"
        self.poll_query(sql)
    def poll_query(self, query: str):
        pg_connection = self.connect.getconn()
        pg_cursor = pg_connection.cursor()
        pg_cursor.execute(query)
        record = pg_cursor.fetchone()
        pg_connection.commit()
        pg_cursor.close()
        self.connect.putconn(pg_connection)
        if  record is not None:
          return record[0]
class PngViewer(VectorViewer):
    def get_png(self, x, y, z):
        # 默认参数
        config = {
            "point_size": 5,
            "line_width": 2,
            "line_color": "#003399FF",
            "fill_color": "#6699CCCC",
            "background": "#FFFFFF00"
        }
        # 在使用psycpg2时，将二进制数据以16进制字符串的形式传回效率更高
        sql = f"select encode(st_aspng('{self.pyramid_table}','{z}_{x}_{y}','{json.dumps(config)}'),'hex')"
        result = self.poll_query(sql)
        # 只有在使用16进制字符串的形式传回时才需要将其转换回来
        result = binascii.a2b_hex(result)
        return result
class MvtViewer(VectorViewer):
    def get_mvt(self, x, y, z):
        # 在使用psycpg2时，将二进制数据以16进制字符串的形式传回效率更高
        sql = f"select encode(st_tile('{self.pyramid_table}','{z}_{x}_{y}'),'hex')"
        result = self.poll_query(sql)
        # 只有在使用16进制字符串的形式传回时才需要将其转换回来
        result = binascii.a2b_hex(result)
        return result
app = Flask(__name__)
@app.route('/vector')
def vector_demo():
    return send_from_directory("./", "Vector.html")
# 定义表名，字段名称等
pngViewer = PngViewer(CONNECTION, 'usbf', 'geom', 'gid')
@app.route('/vector/png/<int:z>/<int:x>/<int:y>')
def vector_png(z, x, y):
    png = pngViewer.get_png(x, y, z)
    return Response(
        response=png,
        mimetype="image/png"
    )
mvtViewer = MvtViewer(CONNECTION, 'points', 'geom', 'gid')
@app.route('/vector/mvt/<int:z>/<int:x>/<int:y>')
def vector_mvt(z, x, y):
    mvt=mvtViewer.get_mvt(x, y, z)
    return Response(
        response=mvt,
        mimetype="application/vnd.mapbox-vector-tile"
    )
if __name__ == "__main__":
    app.run(port=5000, threaded=True)