519 lines
22 KiB
Python
519 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
栅格处理界面: 提供栅格重分类、栅格转矢量和小面积图斑消除的界面操作
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import arcpy
|
|
import traceback
|
|
import multiprocessing
|
|
|
|
from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit,
|
|
QPushButton, QFileDialog, QGridLayout, QCheckBox,
|
|
QGroupBox, QMessageBox, QSpinBox)
|
|
from PyQt6.QtCore import pyqtSignal
|
|
from tools.core.utils import common_utils
|
|
from tools.ui.runners.script_runner import ScriptRunner
|
|
from .config_editor_dialog import ConfigEditorDialogVisual
|
|
from .raster_processing_common import VectorParamsWidget
|
|
from ..components.file_list_group import FileListGroup
|
|
|
|
|
|
class RasterTab(QWidget):
|
|
value_changed = pyqtSignal(str)
|
|
"""栅格处理窗口部件类"""
|
|
|
|
def __init__(self, parent=None):
|
|
super(RasterTab, self).__init__(parent)
|
|
self.main_window = parent
|
|
self.init_ui()
|
|
self.setWindowTitle("栅格处理")
|
|
|
|
def init_ui(self):
|
|
"""初始化用户界面"""
|
|
main_layout = QVBoxLayout(self)
|
|
|
|
# 批量处理区域
|
|
self.batch_mode_group = QGroupBox("重分类并消除小面积图斑")
|
|
batch_layout = QGridLayout()
|
|
|
|
# 配置文件选择 (公用)
|
|
batch_layout.addWidget(QLabel("配置文件:"), 0, 0)
|
|
self.config_file_edit = QLineEdit()
|
|
self.config_file_edit.textChanged.connect(self.on_config_file_changed)
|
|
batch_layout.addWidget(self.config_file_edit, 0, 1)
|
|
self.browse_config_btn = QPushButton("浏览...")
|
|
self.browse_config_btn.clicked.connect(self.browse_config_file)
|
|
batch_layout.addWidget(self.browse_config_btn, 0, 2)
|
|
|
|
# 添加编辑配置文件的按钮 <--- Add this button
|
|
self.edit_config_btn = QPushButton("编辑配置内容...")
|
|
self.edit_config_btn.setEnabled(False)
|
|
self.edit_config_btn.clicked.connect(self.open_config_editor)
|
|
batch_layout.addWidget(self.edit_config_btn, 0, 3)
|
|
|
|
# 输入文件夹
|
|
batch_layout.addWidget(QLabel("栅格文件夹:"), 1, 0)
|
|
self.input_folder_edit = QLineEdit()
|
|
batch_layout.addWidget(self.input_folder_edit, 1, 1, 1, 2)
|
|
self.browse_input_folder_btn = QPushButton("浏览...")
|
|
self.browse_input_folder_btn.clicked.connect(self.browse_input_folder)
|
|
batch_layout.addWidget(self.browse_input_folder_btn, 1, 3)
|
|
|
|
# 批量输出文件夹
|
|
batch_layout.addWidget(QLabel("输出文件夹:"), 2, 0)
|
|
self.batch_output_folder_edit = QLineEdit()
|
|
batch_layout.addWidget(self.batch_output_folder_edit, 2, 1, 1, 2)
|
|
self.browse_batch_output_btn = QPushButton("浏览...")
|
|
self.browse_batch_output_btn.clicked.connect(self.browse_batch_output_folder)
|
|
batch_layout.addWidget(self.browse_batch_output_btn, 2, 3)
|
|
|
|
# 选择裁剪面checkbox
|
|
self.clip_checkbox = QCheckBox("启用裁剪")
|
|
self.clip_checkbox.setChecked(False)
|
|
self.clip_checkbox.stateChanged.connect(self.on_clip_checkbox_changed)
|
|
batch_layout.addWidget(self.clip_checkbox, 3, 0)
|
|
self.clip_features_edit = QLineEdit()
|
|
self.clip_features_edit.setVisible(False)
|
|
batch_layout.addWidget(self.clip_features_edit, 3, 1, 1, 2)
|
|
self.clip_features_btn = QPushButton("选择裁剪面")
|
|
self.clip_features_btn.clicked.connect(self.browse_clip_features)
|
|
self.clip_features_btn.setVisible(False)
|
|
batch_layout.addWidget(self.clip_features_btn, 3, 3)
|
|
|
|
self.batch_mode_group.setLayout(batch_layout)
|
|
|
|
# 文件列表显示组
|
|
self.file_list_group = FileListGroup(self, "选择要处理的栅格")
|
|
self.file_list_group.load_files.connect(self.on_load_raster)
|
|
|
|
# --- 处理参数设置 ---
|
|
param_group = QGroupBox("处理参数")
|
|
param_layout = QVBoxLayout(param_group)
|
|
|
|
# 矢量化参数组件 (包含简化和最小面积阈值)
|
|
self.vector_params = VectorParamsWidget()
|
|
param_layout.addWidget(self.vector_params)
|
|
|
|
# 多进程设置
|
|
export_layout = QHBoxLayout()
|
|
mutiprocess_layout = QHBoxLayout()
|
|
mutiprocess_layout.addWidget(QLabel("使用多进程导出："))
|
|
self.mutiprocess_check = QCheckBox()
|
|
self.mutiprocess_check.setChecked(True)
|
|
mutiprocess_layout.addWidget(self.mutiprocess_check)
|
|
mutiprocess_layout.addStretch()
|
|
export_layout.addLayout(mutiprocess_layout)
|
|
|
|
# 进程数量设置
|
|
process_count_layout = QHBoxLayout()
|
|
process_count_layout.addWidget(QLabel("进程数量："))
|
|
self.process_count = QSpinBox()
|
|
self.process_count.setRange(1, multiprocessing.cpu_count())
|
|
self.process_count.setValue(max(1, multiprocessing.cpu_count() - 1)) # 默认使用CPU核心数-1
|
|
self.process_count.setFixedWidth(180)
|
|
process_count_layout.addWidget(self.process_count)
|
|
|
|
export_layout.addLayout(process_count_layout)
|
|
param_layout.addLayout(export_layout)
|
|
|
|
# 操作按钮
|
|
btn_layout = QHBoxLayout()
|
|
self.process_btn = QPushButton("开始处理")
|
|
self.process_btn.clicked.connect(self.on_start_processing)
|
|
|
|
self.cancel_btn = QPushButton("取消")
|
|
# self.cancel_btn.clicked.connect(self.close)
|
|
|
|
btn_layout.addWidget(self.process_btn)
|
|
btn_layout.addWidget(self.cancel_btn)
|
|
|
|
# 添加所有组件到主布局
|
|
main_layout.addWidget(self.batch_mode_group)
|
|
main_layout.addWidget(self.file_list_group)
|
|
main_layout.addWidget(param_group)
|
|
main_layout.addLayout(btn_layout)
|
|
|
|
self.setLayout(main_layout)
|
|
|
|
def browse_input_folder(self):
|
|
"""浏览选择输入文件夹"""
|
|
folder_path = QFileDialog.getExistingDirectory(self, "选择栅格文件夹")
|
|
if folder_path:
|
|
self.input_folder_edit.setText(folder_path)
|
|
self.on_load_raster()
|
|
|
|
def browse_batch_output_folder(self):
|
|
"""浏览选择批量处理输出文件夹"""
|
|
folder_path = QFileDialog.getExistingDirectory(self, "选择批量处理输出文件夹")
|
|
if folder_path:
|
|
if folder_path.endswith(".gdb"):
|
|
QMessageBox.warning(self, "错误", "请选择一个文件夹，而不是一个数据库文件")
|
|
return
|
|
self.batch_output_folder_edit.setText(folder_path)
|
|
|
|
def browse_config_file(self):
|
|
"""浏览选择配置文件"""
|
|
file_path, _ = QFileDialog.getOpenFileName(
|
|
self, "选择配置文件", "",
|
|
"JSON 文件 (*.json);;所有文件 (*)"
|
|
)
|
|
if file_path:
|
|
self.config_file_edit.setText(file_path)
|
|
self.validate_config_file(file_path)
|
|
self.edit_config_btn.setEnabled(True)
|
|
def browse_clip_features(self):
|
|
"""浏览选择裁剪面 (支持 shapefile)"""
|
|
file_path, _ = QFileDialog.getOpenFileName(
|
|
self, "选择裁剪要素 (Shapefile)", "",
|
|
"Shapefile 文件 (*.shp);;所有文件 (*)"
|
|
)
|
|
if not file_path:
|
|
return
|
|
|
|
if file_path.lower().endswith(".shp"):
|
|
self.clip_features_edit.setText(file_path)
|
|
self.log_message(f"已选择 Shapefile: {file_path}")
|
|
|
|
else:
|
|
QMessageBox.warning(self, "选择错误", "不支持的文件类型。请选择 .shp 文件。")
|
|
self.clip_features_edit.clear()
|
|
|
|
def validate_config_file(self, config_file_path):
|
|
"""对文件进行快速检查，以查看它是否类似于预期的配置结构。"""
|
|
if not os.path.exists(config_file_path):
|
|
return False
|
|
|
|
try:
|
|
with open(config_file_path, 'r', encoding='utf-8') as f:
|
|
config = json.load(f)
|
|
|
|
if "export_config" in config and isinstance(config["export_config"], dict):
|
|
if config["export_config"]:
|
|
first_item_value = next(iter(config["export_config"].values()))
|
|
if isinstance(first_item_value, dict) and all(k in first_item_value for k in ["项目名称", "标准等级"]):
|
|
self.log_message(f"配置文件 '{os.path.basename(config_file_path)}' 加载成功。")
|
|
return True
|
|
else:
|
|
self.log_message(f"警告: 配置文件 '{os.path.basename(config_file_path)}' 内容结构可能不匹配预期。")
|
|
return False
|
|
else:
|
|
self.log_message(f"警告: 配置文件 '{os.path.basename(config_file_path)}' 中的 'export_config' 部分为空。")
|
|
return False
|
|
else:
|
|
self.log_message(f"警告: 配置文件 '{os.path.basename(config_file_path)}' 缺少 'export_config' 键或格式不正确。")
|
|
return False
|
|
|
|
except json.JSONDecodeError:
|
|
self.log_message(f"警告: 配置文件 '{os.path.basename(config_file_path)}' 不是有效的 JSON 格式。")
|
|
return False
|
|
except Exception as e:
|
|
self.log_message(f"警告: 检查配置文件 '{os.path.basename(config_file_path)}' 时出错: {str(e)}")
|
|
return False
|
|
|
|
def open_config_editor(self):
|
|
"""打开配置文件编辑器对话框"""
|
|
config_path = self.config_file_edit.text()
|
|
|
|
if not config_path:
|
|
reply = QMessageBox.question(self, "编辑配置", "没有选择配置文件，是否创建一个新配置文件？",
|
|
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No)
|
|
if reply == QMessageBox.StandardButton.No:
|
|
return
|
|
|
|
config_path = ""
|
|
|
|
# 使用配置编辑对话框
|
|
dialog = ConfigEditorDialogVisual(config_path, self)
|
|
|
|
if dialog.exec() == ConfigEditorDialogVisual.DialogCode.Accepted:
|
|
new_path = dialog.get_new_config_path()
|
|
if new_path and new_path != self.config_file_edit.text():
|
|
self.config_file_edit.setText(new_path)
|
|
self.validate_config_file(new_path)
|
|
|
|
def on_config_file_changed(self):
|
|
text = self.config_file_edit.text()
|
|
self.value_changed.emit(text)
|
|
def on_clip_checkbox_changed(self):
|
|
clip_checkbox_state = self.clip_checkbox.isChecked()
|
|
self.clip_features_edit.setVisible(clip_checkbox_state)
|
|
self.clip_features_btn.setVisible(clip_checkbox_state)
|
|
|
|
def on_load_raster(self):
|
|
"""加载图层列表"""
|
|
try:
|
|
data_source_paths = self.input_folder_edit.text()
|
|
|
|
self.file_list_group.load_file_btn.setEnabled(False)
|
|
|
|
# 清空列表
|
|
self.file_list_group.file_list.clear()
|
|
|
|
if not data_source_paths or not arcpy.Exists(data_source_paths):
|
|
self.log_message(f"输入路径不存在或不是有效的ArcGIS工作空间/文件夹: {data_source_paths}")
|
|
return
|
|
|
|
arcpy.env.workspace = data_source_paths
|
|
|
|
rasters = arcpy.ListRasters()
|
|
if rasters:
|
|
for raster in rasters:
|
|
self.file_list_group.file_list.addItem(raster)
|
|
|
|
self.log_message(f"已加载 {self.file_list_group.file_list.count()} 个图层")
|
|
|
|
else:
|
|
self.log_message(f"文件夹 '{os.path.basename(data_source_paths)}' 中没有找到栅格图层。")
|
|
|
|
except Exception as e:
|
|
self.log_message(f"加载图层列表失败: {str(e)}")
|
|
traceback.print_exc()
|
|
finally:
|
|
self.file_list_group.load_file_btn.setEnabled(True)
|
|
|
|
def validate_inputs(self):
|
|
"""验证输入参数"""
|
|
# 验证配置文件路径
|
|
config_path = self.config_file_edit.text()
|
|
if not config_path or not os.path.exists(config_path):
|
|
QMessageBox.warning(self, "输入错误", "请选择有效的配置文件")
|
|
return False
|
|
try:
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
config_data = json.load(f)
|
|
except Exception as e:
|
|
QMessageBox.warning(self, "输入错误", f"无法加载或解析配置文件: {e}")
|
|
return False
|
|
|
|
if "export_config" not in config_data or not isinstance(config_data["export_config"], dict):
|
|
QMessageBox.warning(self, "输入错误", "配置文件的内容无效或缺少 'export_config' 部分。")
|
|
return False
|
|
|
|
standards_dict = config_data["export_config"]
|
|
|
|
if not standards_dict:
|
|
QMessageBox.warning(self, "输入错误", "配置文件中的 'export_config' 部分为空，没有定义任何处理参数。")
|
|
return False
|
|
|
|
# 验证'标准等级'用于配置重分类表
|
|
error_keys_remap = []
|
|
for key, value in standards_dict.items():
|
|
try:
|
|
levels_data = value.get("标准等级", {})
|
|
if not isinstance(levels_data, dict):
|
|
error_keys_remap.append(f"{key} ('标准等级' 格式不正确)")
|
|
continue
|
|
if not common_utils.create_remap_table(levels_data):
|
|
error_keys_remap.append(f"{key} ('标准等级' 内容无效)")
|
|
except Exception:
|
|
error_keys_remap.append(f"{key} (验证出错)")
|
|
|
|
if error_keys_remap:
|
|
QMessageBox.warning(self, "输入错误", "配置文件中部分参数的 '标准等级' 数据无效:\n" + "\n".join(error_keys_remap))
|
|
return False
|
|
|
|
# 验证输入、输出文件夹
|
|
if not self.input_folder_edit.text():
|
|
QMessageBox.warning(self, "输入错误", "请选择输入栅格文件夹")
|
|
return False
|
|
|
|
if not self.batch_output_folder_edit.text() or self.batch_output_folder_edit.text().endswith('.gdb'):
|
|
QMessageBox.warning(self, "输入错误", "请选择批量处理输出文件夹")
|
|
return False
|
|
|
|
if self.clip_checkbox.isChecked():
|
|
clip_path = self.clip_features_edit.text()
|
|
if not clip_path or not arcpy.Exists(clip_path):
|
|
QMessageBox.warning(self, "输入错误", "已启用裁剪，但裁剪要素路径无效或不存在")
|
|
return False
|
|
|
|
# 3. 验证裁剪要素是否可用
|
|
if self.clip_checkbox.isChecked():
|
|
clip_path = self.clip_features_edit.text()
|
|
if not clip_path:
|
|
QMessageBox.warning(self, "输入错误", "已启用裁剪，但裁剪要素路径为空。")
|
|
return False
|
|
|
|
# Use arcpy.Exists and Describe to validate the path regardless of whether it's SHP or GDB/FeatureClass
|
|
if not arcpy.Exists(clip_path):
|
|
QMessageBox.warning(self, "输入错误", f"已启用裁剪，但裁剪要素路径不存在:\n{clip_path}")
|
|
return False
|
|
|
|
# Optional: Check if the path points to a Feature Class (SHP or inside GDB)
|
|
try:
|
|
desc = arcpy.Describe(clip_path)
|
|
if desc.dataType != 'ShapeFile':
|
|
QMessageBox.warning(self, "输入错误", f"已启用裁剪，但选择的路径不是一个要素类:\n{clip_path}")
|
|
return False
|
|
except Exception as e:
|
|
QMessageBox.warning(self, "输入错误", f"无法描述裁剪要素路径，请检查是否为有效要素类:\n{clip_path}\n错误: {e}")
|
|
return False
|
|
|
|
# 验证是否选择文件
|
|
selected_files = [file.text() for file in self.file_list_group.file_list.selectedItems()]
|
|
if not selected_files:
|
|
QMessageBox.warning(self, "输入错误", "请在栅格列表中勾选要处理的文件。")
|
|
return False
|
|
|
|
return True
|
|
|
|
def on_start_processing(self):
|
|
"""开始处理栅格数据"""
|
|
if not self.validate_inputs():
|
|
return
|
|
|
|
try:
|
|
if self.main_window and hasattr(self.main_window, 'save_settings'):
|
|
self.main_window.update_settings()
|
|
self.main_window.save_settings()
|
|
settings_file = self.main_window.settings_file
|
|
|
|
# 创建的导出线程
|
|
self.script_runner = ScriptRunner()
|
|
|
|
# 连接ScriptRunner的信号
|
|
self.script_runner.task_started.connect(self.on_script_started)
|
|
self.script_runner.task_finished.connect(self.on_script_finished)
|
|
self.script_runner.task_error.connect(self.on_script_error)
|
|
self.script_runner.task_log.connect(self.on_script_log)
|
|
self.script_runner.manager_log.connect(self.log_message)
|
|
|
|
# 获取公共参数 (从 VectorParamsWidget 获取)
|
|
selected_files = [file.text() for file in self.file_list_group.file_list.selectedItems()]
|
|
|
|
if self.mutiprocess_check.isChecked():
|
|
self.script_runner.set_max_concurrent(self.process_count.value())
|
|
else:
|
|
self.script_runner.set_max_concurrent(1)
|
|
|
|
# 调用批量处理脚本
|
|
for raster_file in selected_files:
|
|
params = {
|
|
"settings_path": settings_file,
|
|
"input_raster": raster_file,
|
|
}
|
|
|
|
task_id = self.script_runner.run_process_raster(params)
|
|
# if task_id:
|
|
# self.log_message(f"[GUI] 任务 {task_id} 处理 {raster_file} 已添加到列队")
|
|
|
|
except Exception as e:
|
|
error_msg = f"处理过程中出错: {str(e)}"
|
|
QMessageBox.critical(self, "处理错误", error_msg)
|
|
self.log_message(error_msg)
|
|
|
|
def get_raster_settings(self):
|
|
"""保存当前配置到JSON文件"""
|
|
# 从 VectorParamsWidget 获取参数
|
|
vector_params = self.vector_params.get_params()
|
|
|
|
config = {
|
|
"input_folder": self.input_folder_edit.text(),
|
|
"batch_output_folder": self.batch_output_folder_edit.text(),
|
|
"config_file_path": self.config_file_edit.text(),
|
|
"simplify": vector_params["simplify"], # 从 VectorParamsWidget 获取
|
|
"min_area": vector_params["min_area"], # 从 VectorParamsWidget 获取
|
|
"area_unit": vector_params["area_unit"], # 从 VectorParamsWidget 获取
|
|
"clip_features": self.clip_features_edit.text() if self.clip_checkbox.isChecked() else "",
|
|
"clip_enabled": self.clip_checkbox.isChecked()
|
|
}
|
|
|
|
return config
|
|
|
|
def load_settings(self, settings):
|
|
"""从字典加载配置"""
|
|
try:
|
|
# 批处理参数
|
|
self.input_folder_edit.setText(settings.get("input_folder", ""))
|
|
self.batch_output_folder_edit.setText(settings.get("batch_output_folder", ""))
|
|
|
|
# 配置文件路径和加载
|
|
config_file_path = settings.get("config_file_path", "")
|
|
self.config_file_edit.setText(config_file_path)
|
|
self.edit_config_btn.setEnabled(bool(config_file_path) and os.path.exists(config_file_path))
|
|
|
|
if config_file_path:
|
|
self.validate_config_file(config_file_path)
|
|
|
|
clip_enabled = settings.get("clip_enabled", False)
|
|
self.clip_checkbox.setChecked(clip_enabled)
|
|
self.clip_features_edit.setText(settings.get("clip_features", ""))
|
|
self.on_clip_checkbox_changed()
|
|
|
|
# 矢量化参数 (使用 VectorParamsWidget 的 set_params 方法)
|
|
vector_params_config = {
|
|
"simplify": settings.get("simplify", False), # 注意默认值需要与 VectorParamsWidget 匹配
|
|
"min_area": settings.get("min_area", 1000),
|
|
"area_unit": settings.get("area_unit", "平方米")
|
|
}
|
|
self.vector_params.set_params(vector_params_config)
|
|
|
|
if self.input_folder_edit.text():
|
|
self.on_load_raster()
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "加载配置错误", f"加载配置时出错: {str(e)}")
|
|
return False
|
|
|
|
def set_buttons_enabled(self, enabled):
|
|
"""设置按钮是否可用"""
|
|
self.process_btn.setEnabled(enabled)
|
|
|
|
def on_script_started(self, task_id, task_description):
|
|
"""脚本开始执行"""
|
|
self.set_buttons_enabled(False)
|
|
self.log_message(f"{task_id}: 正在运行 - {task_description}")
|
|
|
|
def on_script_finished(self, task_id:str, success:bool, message:str):
|
|
"""脚本执行完成"""
|
|
|
|
try:
|
|
# 将shp文件导入到GDB数据库
|
|
if self.script_runner.get_running_tasks() == []:
|
|
output_path = self.batch_output_folder_edit.text()
|
|
shape_files = []
|
|
for file in os.listdir(output_path):
|
|
if file.endswith("eliminate.shp"):
|
|
shape_files.append(os.path.join(output_path, file))
|
|
if shape_files:
|
|
try:
|
|
path_result = arcpy.management.CreateFileGDB(output_path, "过程数据(消除小图斑).gdb", "CURRENT")
|
|
# 批量导入数据
|
|
arcpy.conversion.FeatureClassToGeodatabase(shape_files, path_result)
|
|
arcpy.management.Delete(shape_files)
|
|
QMessageBox.information(self, "成功", f"数据已保存到{path_result}")
|
|
self.set_buttons_enabled(True)
|
|
except Exception as e:
|
|
self.log_message(f"批量导入数据出错: {str(e)}")
|
|
except Exception as e:
|
|
self.set_buttons_enabled(True)
|
|
self.log_message(f"批量导入数据出错: {str(e)}")
|
|
|
|
def on_script_error(self, error_msg):
|
|
"""脚本执行出错"""
|
|
self.set_buttons_enabled(True)
|
|
self.log_message(f"错误: {error_msg}")
|
|
QMessageBox.critical(self, "错误", error_msg)
|
|
|
|
def on_script_log(self, task_id, message):
|
|
"""脚本输出日志"""
|
|
self.log_message(f"{task_id}: {message}")
|
|
|
|
def log_message(self, message):
|
|
"""日志输出"""
|
|
if self.main_window and hasattr(self.main_window, 'log_signal'):
|
|
self.main_window.log_signal.emit(message)
|
|
else:
|
|
print(message)
|
|
|
|
if __name__ == "__main__":
|
|
from PyQt6.QtWidgets import QApplication
|
|
import sys
|
|
app = QApplication(sys.argv)
|
|
window = RasterTab()
|
|
window.show()
|
|
sys.exit(app.exec()) |