Files
ArcGis_Py/tools/ui/tabs/area_stat_tab.py
missum 6313905356 修改
Co-authored-by: Copilot <copilot@github.com>
2026-04-23 17:30:48 +08:00

572 lines
24 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 tools.core.utils import common_utils
from tools.ui.runners.script_runner import ScriptRunner
from .config_editor_dialog import ConfigEditorDialogVisual
from ..components.file_list_group import FileListGroup
yunnan_dlbm = """
def yunnan_dlbm(dlbm):
if dlbm.startswith("0101"):
return "0101"
elif dlbm.startswith("0102"):
return "0102"
elif dlbm.startswith("0103"):
return "0103"
elif dlbm.startswith("0201"):
return "0201"
elif dlbm.startswith("0202"):
return "0202"
elif dlbm.startswith("0203"):
return "0203"
elif dlbm.startswith("0204"):
return "0204"
else:
return dlbm[:2]
"""
class XlsxToJpgTab(QWidget):
"""栅格处理窗口部件类"""
def __init__(self, parent=None):
super(XlsxToJpgTab, self).__init__(parent)
self.main_window = parent
# 创建的导出线程
self.script_runner = ScriptRunner()
self.connect_signals()
self.init_ui()
self.setWindowTitle("面积统计")
def connect_signals(self):
# 连接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)
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.setEnabled(False)
batch_layout.addWidget(self.config_file_edit, 0, 1,1,2)
# self.browse_config_btn = QPushButton("浏览...")
# self.browse_config_btn.clicked.connect(self.browse_config_file)
# self.browse_config_btn.setEnabled(False)
# 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)
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, 2)
# 批量输出文件夹
batch_layout.addWidget(QLabel("输出文件夹:"), 2, 0)
self.batch_output_folder_edit = QLineEdit()
batch_layout.addWidget(self.batch_output_folder_edit, 2, 1)
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, 2)
# 选择乡镇界线
batch_layout.addWidget(QLabel("乡镇界线:"), 3, 0)
self.xzq_features_edit = QLineEdit()
batch_layout.addWidget(self.xzq_features_edit, 3, 1)
self.xzq_features_btn = QPushButton("浏览")
self.xzq_features_btn.clicked.connect(self.browse_xzq_features)
batch_layout.addWidget(self.xzq_features_btn, 3, 2)
# 选择地类图斑
batch_layout.addWidget(QLabel("地类图斑:"), 4, 0)
self.dltb_features_edit = QLineEdit()
batch_layout.addWidget(self.dltb_features_edit, 4, 1)
self.dltb_features_btn = QPushButton("浏览")
self.dltb_features_btn.clicked.connect(self.browse_dltb_features)
batch_layout.addWidget(self.dltb_features_btn, 4, 2)
# 输入行政区名称
batch_layout.addWidget(QLabel("行政区名称:"), 5, 0)
self.xzqmc_edit = QLineEdit()
batch_layout.addWidget(self.xzqmc_edit, 5, 1, 1, 2)
# 单选框
ext_layout = QHBoxLayout()
ext_layout.addWidget(QLabel("是否同时对行政区和地类进行平差(需要xlsx文件有其他地类的平差面积):"))
self.is_adjust_xzq_and_landuse_checkbox = QCheckBox()
ext_layout.addWidget(self.is_adjust_xzq_and_landuse_checkbox)
ext_layout.addStretch()
batch_layout.addLayout(ext_layout, 6, 0, 1, 2)
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)
# 多进程设置
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("导出到Excel")
self.process_btn.clicked.connect(self.on_start_processing)
self.excel_to_jpg_btn = QPushButton("Eexcel导出到JPG")
self.excel_to_jpg_btn.clicked.connect(self.on_export_excel_to_jpg)
self.cancel_btn = QPushButton("取消")
# self.cancel_btn.clicked.connect(self.close)
btn_layout.addWidget(self.process_btn)
btn_layout.addWidget(self.excel_to_jpg_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:
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_xzq_features(self):
"""浏览选择乡镇界线要素 (支持 shapefile)"""
file_path, _ = QFileDialog.getOpenFileName(
self, "选择乡镇界线要素 (Shapefile)", "",
"Shapefile 文件 (*.shp);;所有文件 (*)"
)
if not file_path:
return
if file_path.lower().endswith(".shp"):
self.xzq_features_edit.setText(file_path)
self.log_message(f"已选择 Shapefile: {file_path}")
else:
QMessageBox.warning(self, "选择错误", "不支持的文件类型。请选择 .shp 文件。")
self.xzq_features_edit.clear()
def browse_dltb_features(self):
"""浏览选择地类图斑要素 (支持 shapefile)"""
file_path, _ = QFileDialog.getOpenFileName(
self, "选择地类图斑要素 (Shapefile)", "",
"Shapefile 文件 (*.shp);;所有文件 (*)"
)
if not file_path:
return
if file_path.lower().endswith(".shp"):
self.dltb_features_edit.setText(file_path)
self.log_message(f"已选择 Shapefile: {file_path}")
else:
QMessageBox.warning(self, "选择错误", "不支持的文件类型。请选择 .shp 文件。")
self.dltb_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_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
# 获取所有SHP 文件
shp_files = arcpy.ListFeatureClasses("*.shp")
if shp_files:
for raster in shp_files:
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():
QMessageBox.warning(self, "输入错误", "请选择批量处理输出文件夹")
return False
if not self.xzqmc_edit.text():
QMessageBox.warning(self, "输入错误", "请输入县区名称")
return False
# 验证地类图斑文件
dltb_path = self.dltb_features_edit.text()
if not dltb_path and not arcpy.Exists(dltb_path):
QMessageBox.warning(self, "输入错误", "请选择有效的地类图斑文件")
return False
# 3. 验证乡镇界线文件
xzjx_path = self.xzq_features_edit.text()
if not xzjx_path or not arcpy.Exists(xzjx_path):
QMessageBox.warning(self, "输入错误", "请选择有效的乡镇界线文件。")
return False
# Optional: Check if the path points to a Feature Class (SHP or inside GDB)
try:
desc = arcpy.Describe(xzjx_path)
if desc.dataType != 'FeatureClass' and desc.dataType != 'ShapeFile':
QMessageBox.warning(self, "输入错误", f"选择的乡镇界线文件不是一个要素类:\n{xzjx_path}")
return False
except Exception as e:
QMessageBox.warning(self, "输入错误", f"无法描述乡镇界线文件,请检查是否为有效要素类:\n{xzjx_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
# 获取公共参数 (从 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)
# 计算DLTB字段
try:
dltb_polygon = self.dltb_features_edit.text()
if arcpy.Exists(dltb_polygon) and not arcpy.ListFields(dltb_polygon,"YJDLBM"):
arcpy.management.CalculateField(dltb_polygon, "YJDLBM", "!DLBM![:2]", "PYTHON3")
if arcpy.Exists(dltb_polygon) and not arcpy.ListFields(dltb_polygon,"YNDLBM"):
arcpy.management.CalculateField(dltb_polygon, "YNDLBM", "yunnan_dlbm(!DLBM!)", "PYTHON3", yunnan_dlbm)
# else:
# QMessageBox.critical(self, "处理错误", "地类图斑文件不存在")
# return
except Exception as e:
print(f"计算SHFJ字段时发生错误: {e}")
return
# 调用批量处理脚本
for raster_file in selected_files:
params = {
"settings_path": settings_file,
"reclassed_polygon": raster_file,
}
# 统计面积
task_id = self.script_runner.run_area_stat(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 on_export_excel_to_jpg(self):
self.log_message("[GUI] 开始处理")
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
params = {
"settings_path":settings_file
}
task_id = self.script_runner.run_excel_to_jpg(params)
if task_id:
self.log_message(f"[GUI] 批量处理任务 {task_id} 已添加到列队")
except Exception as e:
error_msg = f"处理过程中出错: {str(e)}"
QMessageBox.critical(self, "处理错误", error_msg)
self.log_message(error_msg)
def get_area_stat_settings(self):
"""保存当前配置到JSON文件"""
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(),
"xzq_features": self.xzq_features_edit.text(),
"dltb_features": self.dltb_features_edit.text(),
"xzqmc": self.xzqmc_edit.text(),
"is_by_xzq": self.is_adjust_xzq_and_landuse_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)
self.xzq_features_edit.setText(settings.get("xzq_features", ""))
self.dltb_features_edit.setText(settings.get("dltb_features", ""))
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)
self.excel_to_jpg_btn.setEnabled(enabled)
def update_config_file(self, config_file_path):
"""更新配置文件路径"""
self.config_file_edit.setText(config_file_path)
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):
"""脚本执行完成"""
self.set_buttons_enabled(True)
status = "完成" if success else "失败"
self.log_message(f"[{task_id}] 脚本执行{status}: {message}")
def on_script_error(self,task_id, error_msg):
"""脚本执行出错"""
self.set_buttons_enabled(True)
self.log_message(f"错误:{task_id}-{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 = XlsxToJpgTab()
window.show()
sys.exit(app.exec())