wzy-warehouse
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
"""
|
|
地图缩放策略模块。支持 5 种缩放模式。
|
|
"""
|
|
from qgis.core import (
|
|
QgsPointXY, QgsGeometry, QgsRectangle,
|
|
QgsCoordinateTransform, QgsCoordinateReferenceSystem,
|
|
)
|
|
|
|
from app.config.paths import get_logger
|
|
|
|
logger = get_logger("qgis.zoom")
|
|
|
|
ZOOM_RULES = {
|
|
"11": "pan_to_center",
|
|
"12": "by_layer",
|
|
"13": "layer_intersect",
|
|
"14": "center_distance",
|
|
"15": "layer_merged",
|
|
}
|
|
|
|
|
|
class MapZoom:
|
|
def __init__(self, project, layout, map_item):
|
|
self.project = project
|
|
self.layout = layout
|
|
self.map_item = map_item
|
|
|
|
def execute(self, rule: str, data: dict) -> None:
|
|
"""执行缩放操作"""
|
|
method_name = ZOOM_RULES.get(rule, "pan_to_center")
|
|
method = getattr(self, method_name)
|
|
logger.info(f"缩放: {method_name}, 参数: {data}")
|
|
method(data)
|
|
|
|
def _make_transform(self, source_crs=None) -> QgsCoordinateTransform:
|
|
qct = QgsCoordinateTransform()
|
|
qct.setDestinationCrs(self.map_item.crs())
|
|
qct.setSourceCrs(
|
|
source_crs if source_crs
|
|
else QgsCoordinateReferenceSystem("EPSG:4326")
|
|
)
|
|
return qct
|
|
|
|
def pan_to_center(self, data: dict) -> None:
|
|
qct = self._make_transform()
|
|
point = QgsPointXY(float(data["X"]), float(data["Y"]))
|
|
geom = QgsGeometry.fromWkt(point.asWkt())
|
|
geom.transform(qct, QgsCoordinateTransform.ForwardTransform)
|
|
center = geom.asPoint()
|
|
qr = QgsRectangle.fromCenterAndSize(
|
|
center, self.map_item.extent().width(), self.map_item.extent().height()
|
|
)
|
|
self.map_item.zoomToExtent(qr)
|
|
|
|
def center_distance(self, data: dict) -> None:
|
|
qct = self._make_transform()
|
|
point = QgsPointXY(float(data["X"]), float(data["Y"]))
|
|
geom = QgsGeometry.fromWkt(point.asWkt())
|
|
geom.transform(qct, QgsCoordinateTransform.ForwardTransform)
|
|
box = geom.buffer(float(data["value"]) / 1000, 100).boundingBox()
|
|
self.map_item.zoomToExtent(box.buffered(box.width() * 0.1))
|
|
|
|
def by_layer(self, data: dict) -> None:
|
|
layers = self.project.mapLayersByName(data["value"])
|
|
if not layers:
|
|
logger.warning(f"图层不存在: {data['value']}")
|
|
return
|
|
layer = layers[0]
|
|
if layer.featureCount() == 0:
|
|
self.pan_to_center(data)
|
|
return
|
|
qct = self._make_transform(layer.crs())
|
|
geom = QgsGeometry.fromWkt(layer.extent().asWktPolygon())
|
|
geom.transform(qct, QgsCoordinateTransform.ForwardTransform)
|
|
box = geom.boundingBox()
|
|
self.map_item.zoomToExtent(box.buffered(box.width() * 0.1))
|
|
|
|
def layer_merged(self, data: dict) -> None:
|
|
names = data["value"].split(",")
|
|
combined = None
|
|
for name in names:
|
|
layers = self.project.mapLayersByName(name.strip())
|
|
if not layers:
|
|
continue
|
|
layer = layers[0]
|
|
qct = self._make_transform(layer.crs())
|
|
geom = QgsGeometry.fromWkt(layer.extent().asWktPolygon())
|
|
geom.transform(qct, QgsCoordinateTransform.ForwardTransform)
|
|
box = geom.boundingBox()
|
|
if combined is None:
|
|
combined = box
|
|
else:
|
|
combined.combineExtentWith(box)
|
|
if combined:
|
|
self.map_item.zoomToExtent(combined.buffered(combined.width() * 0.1))
|
|
|
|
def layer_intersect(self, data: dict) -> None:
|
|
names = data["value"].split(",")
|
|
intersection = None
|
|
for name in names:
|
|
layers = self.project.mapLayersByName(name.strip())
|
|
if not layers:
|
|
continue
|
|
layer = layers[0]
|
|
qct = self._make_transform(layer.crs())
|
|
layer.selectAll()
|
|
geom = None
|
|
for fid in layer.selectedFeatureIds():
|
|
if geom is None:
|
|
geom = layer.getGeometry(fid)
|
|
else:
|
|
geom = geom.combine(layer.getGeometry(fid))
|
|
if geom:
|
|
qgm = QgsGeometry.fromWkt(geom.asWkt())
|
|
qgm.transform(qct, QgsCoordinateTransform.ForwardTransform)
|
|
box = qgm.boundingBox()
|
|
if intersection is None:
|
|
intersection = box
|
|
else:
|
|
intersection = intersection.intersection(box)
|
|
if intersection:
|
|
bbox = intersection.boundingBox()
|
|
self.map_item.zoomToExtent(bbox.buffered(bbox.width() * 0.1)) |