# -*- 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 class XlsxToJpgTab(QWidget): """栅格处理窗口部件类""" def __init__(self, parent=None): super(XlsxToJpgTab, 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() 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) # 选择乡镇界线 batch_layout.addWidget(QLabel("乡镇界线:"), 3, 0) self.xzq_features_edit = QLineEdit() batch_layout.addWidget(self.xzq_features_edit, 3, 1, 1, 2) self.xzq_features_btn = QPushButton("浏览") self.xzq_features_btn.clicked.connect(self.browse_xzq_features) batch_layout.addWidget(self.xzq_features_btn, 3, 3) # 选择地类图斑 batch_layout.addWidget(QLabel("地类图斑:"), 4, 0) self.dltb_features_edit = QLineEdit() batch_layout.addWidget(self.dltb_features_edit, 4, 1, 1, 2) self.dltb_features_btn = QPushButton("浏览") self.dltb_features_btn.clicked.connect(self.browse_dltb_features) batch_layout.addWidget(self.dltb_features_btn, 4, 3) # 输入行政区名称 batch_layout.addWidget(QLabel("行政区名称:"), 5, 0) self.xzqmc_edit = QLineEdit() batch_layout.addWidget(self.xzqmc_edit, 5, 1, 1, 2) # 单选框 batch_layout.addWidget(QLabel("是否同时平差行政区和地类:"), 6, 0) self.is_adjust_xzq_and_landuse_checkbox = QCheckBox() batch_layout.addWidget(self.is_adjust_xzq_and_landuse_checkbox, 6, 1) 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("开始处理") 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: 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 clip_path = self.xzq_features_edit.text() if not clip_path or not arcpy.Exists(clip_path): QMessageBox.warning(self, "输入错误", "已启用裁剪，但裁剪要素路径无效或不存在") return False # 3. 验证乡镇界限 clip_path = self.xzq_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 != 'FeatureClass' and 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, "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 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(), } 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) 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) self.log_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())