37 KiB
错误处理与异常管理是构建健壮 SunvStation 脚本的核心能力。通过系统化的错误处理机制,您的脚本能够优雅地处理异常情况、记录关键错误信息、确保数据完整性,并在用户界面中提供清晰的反馈。本页面将深入解析 sunvpy 库的错误处理架构、实用模式和最佳实践。
错误处理架构概览
sunvpy 库采用分层设计,通过 LogMixin 混入类提供统一的错误记录能力,结合多种防御性编程模式形成完整的错误处理体系。这种架构将错误记录、异常捕获和状态验证分离到不同层次,确保代码的可维护性和扩展性。
classDiagram
class Logger {
<<Python标准库>>
+debug(msg)
+info(msg)
+warning(msg)
+error(msg)
+critical(msg)
}
class LogAware {
<<抽象基类>>
+set_logger(Logger)*
+log_error_msg(str)*
}
class LogMixin {
+Logger logger
+set_logger(Logger)
+log_error_msg(str)
}
class SSProcessManager {
+Logger logger
+log_error_msg(str)
}
class ErrorPatterns {
<<错误处理模式>>
+null_check_return
+type_conversion_try
+silent_exception_pass
+early_return_validation
}
LogAware <|-- LogMixin
LogMixin <|-- SSProcessManager
SSProcessManager ..> Logger : 使用
SSProcessManager ..> ErrorPatterns : 应用
这种架构设计的核心优势在于关注点分离:业务逻辑代码通过统一的 log_error_msg() 接口记录错误,而日志的具体输出配置由您完全掌控。错误处理模式则分布在各个 Mixin 类中,针对不同场景提供灵活的应对策略。
Sources: log_mixin.py, PySSProcess.py
错误处理机制层次
sunvpy 的错误处理分为四个层次,每个层次应对不同类型的错误场景。理解这个层次结构对于编写健壮的脚本至关重要。
flowchart LR
A[错误输入] --> B[第一层:参数验证<br/>空值检查/范围检查]
B --> C{是否通过?}
C -->|否| D[返回默认值/空值<br/>记录警告日志]
C -->|是| E[第二层:类型转换<br/>try-except 捕获]
E --> F{转换成功?}
F -->|否| G[使用容错值/返回 False<br/>记录错误日志]
F -->|是| H[第三层:业务操作<br/>空指针检查]
H --> I{操作成功?}
I -->|否| J[返回空结果/失败标志<br/>记录错误日志]
I -->|是| K[第四层:资源清理<br/>try-finally 确保]
K --> L[正常返回结果]
style D fill:#fff4e1
style G fill:#ffe1e1
style J fill:#ffe1e1
style L fill:#e1ffe1
第一层:参数验证与空值检查
这是错误处理的第一道防线,通过前置条件检查防止无效输入进入核心逻辑。sunvpy 广泛使用边界检查和空值检查模式。
空值检查返回默认值模式:
def getSysSelGeoList(self):
"""获取当前选中的几何对象列表。"""
map = self.getCurrentMap()
if map is None:
return None # 空值检查,返回 None 表示失败
return GeoBaseList(map.m_selGeoList)
Sources: selection_mixin.py
边界检查返回空字符串模式:
def getSelGeoValue(self, index: int, fieldName: str) -> str:
"""获取当前选择集中特定几何对象的属性值。"""
if index < 0 or index >= len(self.selGeoList):
return "" # 边界检查,返回空字符串
geo = self.selGeoList[index]
if geo is None:
return "" # 空值检查
# ... 后续逻辑
Sources: selection_mixin.py
第二层:类型转换异常捕获
当涉及字符串到数值的转换时,sunvpy 使用 try-except 块捕获 ValueError 异常,并提供容错机制或失败标记。
类型转换容错模式:
def setSelGeoValue(self, index: int, fieldName: str, value: str):
"""设置当前选择集中特定几何对象的属性值。"""
# ... 前置检查代码
if fieldName.lower() == "ssobj_code":
# 类型转换容错:若不能转为 int 则直接用原值
try:
code_value = int(value)
except Exception:
code_value = value # 容错处理
for geo in geoList:
geo.setCode(code_value)
Sources: selection_mixin.py
类型转换失败返回 False 模式:
case "ssobj_angle":
try:
angle = float(value)
geo.setAngle(angle)
except ValueError:
return False # 转换失败,返回 False 标记
Sources: geo_edit_mixin.py
| 错误类型 | 异常类 | 处理策略 | 使用场景 |
|---|---|---|---|
| 数值转换失败 | ValueError |
返回 False 或使用容错值 | 角度、比例、时间等数值属性设置 |
| 参数类型错误 | TypeError |
返回 False 或跳过操作 | 几何对象属性赋值 |
| 通用异常 | Exception |
使用容错值或静默处理 | 关键操作的容错处理 |
第三层:业务操作错误处理
在执行地理空间操作时,可能遇到数据源不可用、数据集不存在等运行时错误。sunvpy 通过错误日志记录和提前返回模式处理这类错误。
错误日志记录模式:
def createNewObjByCode(self, code: int) -> tuple:
"""根据指定的代码创建一个新的几何对象。"""
self.curNewObj = (None, None) # 重置当前新建对象
ds_eps = self.map.getCurrentDataSourceEPS()
if ds_eps is None:
self.log_error_msg("无法获取当前数据源!") # 记录错误
return self.curNewObj # 返回空元组
fea = ds_eps.getFeature(code)
if fea is None:
fea = ds_eps.getFeature(0)
dataset = getDatasetByFeaCode(ds_eps, fea)
if dataset is None:
self.log_error_msg(f"无法获取数据集, code:{code}") # 记录错误
return self.curNewObj # 返回空元组
# ... 后续逻辑
Sources: geo_edit_mixin.py
错误日志与循环继续模式:
def persistTempGeoList(self, geoVec: GeoBaseVector):
"""将临时几何对象列表持久化到地图中。"""
for i in range(geoVec.size()):
geo: GeoBase = geoVec.at(i)
code = geo.getCode()
fea = ds_eps.getFeature(code)
if fea is None:
fea = ds_eps.getFeature(0)
dataset = getDatasetByFeaCode(ds_eps, fea)
if dataset is None:
self.log_error_msg(f"无法获取数据集, code:{code}")
continue # 跳过当前对象,继续处理下一个
obj = GeoObject(dataset, generateUniqueId())
obj.setGeoBase(geo)
persistVec.push_back(geo, obj)
return persistVec
Sources: geo_edit_mixin.py
第四层:静默异常处理
对于非关键操作(如进度条关闭、UI 清理),sunvpy 使用静默异常处理模式,确保即使操作失败也不会影响主流程。
静默异常处理模式:
def disable_progress(self):
"""关闭进度条显示并立即关闭正在显示的进度条。"""
self.enable_progress = False
if self.progress is not None:
try:
closeProgress() # 尝试关闭进度条
except Exception:
pass # 静默处理,不中断流程
self.progress = None
Sources: progress_mixin.py
| 处理模式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 异常抛出 | 关键错误、不可恢复状态 | 强制调用方处理,问题可见 | 需要额外 try-except 代码 |
| 日志记录 | 可恢复错误、业务逻辑错误 | 不中断流程,便于追踪 | 需要配置日志系统 |
| 返回失败值 | 可选操作、容错场景 | 调用方灵活处理 | 需要检查返回值 |
| 静默处理 | 非关键清理、UI 操作 | 简洁,不影响主流程 | 错误信息丢失 |
错误日志记录机制
sunvpy 通过 LogMixin 提供统一的错误日志记录接口,支持将自定义配置的 Logger 注入到 SSProcess 中。这种设计允许您灵活控制日志的输出方式、格式和存储位置。
日志记录接口
LogMixin 提供两个核心方法用于日志记录:
| 方法 | 签名 | 功能 | 使用场景 |
|---|---|---|---|
set_logger() |
set_logger(logger: Logger) |
注入自定义日志记录器 | 脚本初始化时配置 |
log_error_msg() |
log_error_msg(error_msg: str) |
记录错误消息 | 发生错误时调用 |
LogMixin 实现细节:
class LogMixin(LogAware):
"""Mixin class for logging operations"""
logger : Optional[Logger] = None
def set_logger(self, logger: Logger):
self.logger = logger
def log_error_msg(self, error_msg: str):
print("error log in!")
if self.logger is not None:
self.logger.error(error_msg) # 调用标准库 Logger 的 error 方法
Sources: log_mixin.py
日志配置流程
配置错误日志记录需要创建 Logger 实例、设置处理器和格式化器,最后注入到 SSProcess 中。
flowchart TD
A[开始配置日志] --> B[创建 Logger 实例<br/>logging.getLogger'script_name']
B --> C[创建 FileHandler<br/>输出到文件]
C --> D[创建 StreamHandler<br/>输出到控制台]
D --> E[设置 Formatter<br/>定义日志格式]
E --> F[配置日志级别<br/>DEBUG/INFO/WARNING/ERROR]
F --> G[添加 Handler 到 Logger]
G --> H[调用 SSProcess.set_logger'logger']
H --> I[日志配置完成]
I --> J{调用 log_error_msg}
J -->|logger 未配置| K[仅打印到控制台]
J -->|logger 已配置| L[按配置输出到文件/控制台]
style H fill:#e1f5ff
style L fill:#e1ffe1
完整日志配置示例:
import logging
from sunvpy import SSProcess
# 步骤 1:创建 Logger 实例
logger = logging.getLogger('my_script')
logger.setLevel(logging.DEBUG)
# 步骤 2:创建文件处理器
file_handler = logging.FileHandler('script_log.log', mode='a', encoding='utf-8')
file_handler.setLevel(logging.ERROR) # 只记录 ERROR 及以上级别
# 步骤 3:创建控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.WARNING) # 只显示 WARNING 及以上级别
# 步骤 4:设置格式化器
formatter = logging.Formatter(
fmt='%(asctime)s [%(filename)s:%(lineno)d] %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# 步骤 5:将处理器添加到 Logger
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# 步骤 6:注入到 SSProcess
SSProcess.set_logger(logger)
# 步骤 7:使用示例
SSProcess.createNewObjByCode(3103013) # 如果出错,会记录到日志
Sources: log_mixin.py
日志级别选择指南
根据错误的严重程度和影响范围,合理选择日志级别对于问题诊断至关重要。
| 日志级别 | 错误类型 | 示例场景 | 处理策略 |
|---|---|---|---|
| DEBUG | 调试信息 | 对象创建过程、参数传递细节 | 仅开发阶段启用 |
| INFO | 重要操作 | 脚本开始/结束、批量处理完成 | 记录关键执行节点 |
| WARNING | 潜在问题 | 使用默认值、对象不存在但可继续 | 不影响脚本运行 |
| ERROR | 错误但可恢复 | 数据源不可用、数据集获取失败 | 记录详细错误信息 |
| CRITICAL | 严重错误 | 无法恢复的异常、致命失败 | 建议终止脚本执行 |
最佳实践:为关键操作添加错误日志记录
def batch_export_data(dataset_name: str):
"""批量导出数据示例"""
# 关键操作前记录日志
SSProcess.logger.info(f"开始批量导出数据集: {dataset_name}")
try:
dataset = SSProcess.map.getDataset(dataset_name)
if dataset is None:
SSProcess.logger.error(f"数据集不存在: {dataset_name}")
return False
SSProcess.startProgress("批量导出", 1000)
for i in range(1000):
# ... 导出逻辑
SSProcess.stepProgress(f"处理第 {i} 条记录")
SSProcess.logger.info(f"批量导出完成,共处理 1000 条记录")
SSProcess.closeProgress()
return True
except Exception as e:
SSProcess.logger.critical(f"批量导出发生严重错误: {str(e)}")
SSProcess.closeProgress()
return False
Sources: geo_edit_mixin.py
常见错误场景与处理策略
了解 sunvpy 中的常见错误场景及其标准处理模式,能够帮助您编写更加健壮的脚本。以下总结了几类典型错误及其推荐处理策略。
场景一:数据源和数据集访问错误
访问地理空间数据时,经常遇到数据源不可用或数据集不存在的情况。
| 错误场景 | 检测方法 | 推荐处理策略 |
|---|---|---|
| 数据源不可用 | getCurrentDataSourceEPS() 返回 None |
记录错误日志,返回空结果 |
| 数据集不存在 | getDataset() 返回 None |
记录错误日志,跳过当前对象 |
| 特征代码无效 | getFeature(code) 返回 None |
使用默认特征代码 0 |
标准处理模式:
def createGeoObjectByCode(self, code: int):
"""根据编码创建地理对象的标准错误处理模式"""
# 获取数据源
ds_eps = self.map.getCurrentDataSourceEPS()
if ds_eps is None:
self.log_error_msg("无法获取当前数据源!")
return (None, None) # 返回空元组
# 获取特征
fea = ds_eps.getFeature(code)
if fea is None:
fea = ds_eps.getFeature(0) # 使用默认特征
if fea is None:
self.log_error_msg(f"无法获取特征,code: {code}")
return (None, None)
# 获取数据集
dataset = getDatasetByFeaCode(ds_eps, fea)
if dataset is None:
self.log_error_msg(f"无法获取数据集,code: {code}")
return (None, None)
# 创建对象
obj_type = fea.nObjectType
obj, geo = self.createDefaultGeoBase(obj_type, dataset)
if obj is None or geo is None:
self.log_error_msg(f"创建地物失败,code: {code}")
return (None, None)
return (obj, geo)
Sources: geo_edit_mixin.py
场景二:属性类型转换错误
设置对象属性时,需要将字符串值转换为正确的数据类型(int、float、日期等),可能遇到格式错误。
| 属性类型 | 目标类型 | 转换错误处理 |
|---|---|---|
| 角度、比例 | float | 捕获 ValueError,返回 False |
| 编码、线型 | int | 捕获 Exception,使用原值容错 |
| 创建/修改时间 | DateTime | 捕获 ValueError,返回 False |
| 坐标值 | float | 捕获 ValueError,返回 False |
类型转换错误处理模式:
def setPropertyWithValidation(self, geo: GeoBase, field_name: str, value: str) -> bool:
"""带类型转换验证的属性设置方法"""
field_name = field_name.lower().strip()
match field_name:
case "ssobj_angle" | "ssobj_scalex" | "ssobj_scaley":
try:
float_value = float(value)
geo.setAngle(float_value) if field_name == "ssobj_angle" else None
# ... 其他赋值逻辑
return True
except ValueError:
self.log_error_msg(f"数值转换失败: {field_name} = {value}")
return False
case "ssobj_createtime" | "ssobj_modifytime":
try:
datetime_value = parseDateTime(value)
geo.setCreateTime(datetime_value) if field_name == "ssobj_createtime" else None
return True
except ValueError:
self.log_error_msg(f"日期转换失败: {field_name} = {value}")
return False
case "ssobj_code":
try:
int_value = int(value)
geo.setCode(int_value)
return True
except Exception:
# 容错处理:保持原值
self.log_error_msg(f"编码转换失败,保持原值: {value}")
return True
case _:
# 其他属性直接设置
geo.setObjName(value)
return True
Sources: geo_edit_mixin.py
场景三:索引越界与空对象访问
在遍历选择集或访问集合元素时,可能遇到索引越界或空对象引用错误。
| 错误场景 | 检测条件 | 推荐处理 |
|---|---|---|
| 索引越界 | index < 0 或 index >= len(list) |
返回空值,记录警告 |
| 空对象引用 | obj is None |
跳过处理,返回 False |
| 列表为空 | list.empty() is True |
直接返回,不执行操作 |
集合访问安全模式:
def safeGetGeoValue(self, index: int, field_name: str) -> str:
"""安全的属性值获取方法"""
# 第一层检查:索引边界
if index < 0 or index >= len(self.selGeoList):
self.log_error_msg(f"索引越界: {index}, 选择集大小: {len(self.selGeoList)}")
return ""
# 第二层检查:对象是否为 None
geo = self.selGeoList[index]
if geo is None:
self.log_error_msg(f"索引 {index} 处的对象为空")
return ""
# 第三层检查:属性是否存在
if field_name.startswith('['):
# 扩展属性
fields, values = StringArray(), StringArray()
self.map.getExtentAttr(geo, fields, values)
idx = self.findIndexInStringArray(field_name[1:-1], fields)
if idx == -1:
return "" # 属性不存在,返回空字符串
return values[idx]
else:
# 基本属性
return self.objBaseAttr.getGeoValue(self.map, geo, field_name)
Sources: selection_mixin.py
场景四:资源清理错误
在进度条关闭、图形释放等资源清理操作中,可能遇到异常,但不应中断主流程。
| 操作类型 | 异常原因 | 推荐处理 |
|---|---|---|
| 关闭进度条 | UI 已关闭、句柄无效 | try-except 静默处理 |
| 释放图形 | 图形已被释放 | 检查对象状态后操作 |
| 清除选择集 | 选择集为空 | 检查状态后再清理 |
资源清理模式:
def cleanupResources(self):
"""安全的资源清理方法"""
# 安全关闭进度条
if self.progress is not None:
try:
closeProgress()
except Exception:
pass # 静默处理,不中断流程
finally:
self.progress = None
# 安全释放图形
for geo in self.selGeoList:
if geo is not None:
try:
geo.freeGraphics()
except Exception:
pass # 静默处理
# 安全清除选择集
try:
self.selGeoList.clear()
self.selNoteList.clear()
except Exception:
pass # 清理失败不影响其他操作
Sources: progress_mixin.py
错误处理最佳实践
基于 sunvpy 库的设计理念和实际应用经验,以下错误处理最佳实践将帮助您构建更加健壮和可维护的脚本。
实践一:前置条件检查
在任何操作开始前进行参数和状态验证,遵循"先检查,后执行"的原则。
def updateGeoAttributes(self, index: int, attributes: dict) -> bool:
"""更新地物属性的最佳实践示例"""
# 前置条件 1:检查选择集不为空
if len(self.selGeoList) == 0:
self.log_error_msg("选择集为空,无法更新属性")
return False
# 前置条件 2:检查索引有效性
if index < 0 or index >= len(self.selGeoList):
self.log_error_msg(f"索引无效: {index}")
return False
# 前置条件 3:检查参数不为空
if not attributes:
self.log_error_msg("属性字典为空")
return False
# 前置条件 4:检查地图可用
if self.map is None:
self.log_error_msg("地图实例不可用")
return False
# 通过所有验证后,执行核心逻辑
geo = self.selGeoList[index]
success_count = 0
for field, value in attributes.items():
if self.setSelGeoValue(index, field, str(value)):
success_count += 1
else:
self.log_error_msg(f"设置属性失败: {field} = {value}")
return success_count > 0
实践二:精确的异常捕获
避免使用过于宽泛的 except Exception,优先捕获具体的异常类型,提高错误定位的准确性。
| 推荐做法 | 不推荐做法 | 原因 |
|---|---|---|
except ValueError |
except Exception |
捕获精确的异常类型 |
except (ValueError, TypeError) |
except |
明确捕获多种预期异常 |
| 先捕获具体异常,最后捕获 Exception | 只捕获 Exception | 分层处理不同错误 |
精确异常捕获示例:
def convertAngleValue(self, value: str) -> Optional[float]:
"""角度值转换的精确异常处理"""
try:
return float(value)
except ValueError:
self.log_error_msg(f"角度值格式错误: {value}")
return None
except TypeError:
self.log_error_msg(f"角度值类型错误,应为字符串: {type(value)}")
return None
except Exception as e:
self.log_error_msg(f"角度值转换发生未预期错误: {str(e)}")
return None
实践三:错误恢复与回滚
对于涉及多个步骤的操作,实现错误恢复机制,确保失败时能够回滚到一致状态。
def batchUpdateWithRollback(self, indices: list, field_name: str, value: str) -> bool:
"""带回滚机制的批量更新"""
# 保存原始值
original_values = []
valid_indices = []
for idx in indices:
if 0 <= idx < len(self.selGeoList):
original_value = self.getSelGeoValue(idx, field_name)
original_values.append(original_value)
valid_indices.append(idx)
if not valid_indices:
self.log_error_msg("没有有效的索引需要更新")
return False
# 执行更新
updated_count = 0
for i, idx in enumerate(valid_indices):
if self.setSelGeoValue(idx, field_name, value):
updated_count += 1
else:
# 更新失败,执行回滚
self.log_error_msg(f"索引 {idx} 更新失败,开始回滚...")
for j, rollback_idx in enumerate(valid_indices[:i]):
self.setSelGeoValue(rollback_idx, field_name, original_values[j])
return False
self.logger.info(f"批量更新成功: {updated_count} 个对象")
return True
实践四:日志与用户反馈结合
将技术性的错误日志与面向用户的友好反馈结合,提升用户体验。
def importDataWithFeedback(self, file_path: str) -> bool:
"""带用户反馈的数据导入"""
try:
# 技术日志
self.logger.info(f"开始导入文件: {file_path}")
# 用户反馈
SSProcess.startProgress("数据导入", 100)
SSProcess.stepProgress("正在读取文件...")
# 导入逻辑
# ... 执行导入操作
SSProcess.stepProgress("导入完成")
SSProcess.closeProgress()
# 成功日志
self.logger.info(f"文件导入成功: {file_path}")
return True
except FileNotFoundError:
# 技术日志
self.logger.error(f"文件不存在: {file_path}")
# 用户反馈(通过进度条标题)
SSProcess.startProgress("导入失败", 1)
SSProcess.stepProgress(f"错误:文件不存在 - {file_path}")
SSProcess.closeProgress()
return False
except PermissionError:
self.logger.error(f"文件访问权限不足: {file_path}")
SSProcess.startProgress("导入失败", 1)
SSProcess.stepProgress("错误:无文件访问权限")
SSProcess.closeProgress()
return False
except Exception as e:
self.logger.critical(f"导入发生严重错误: {str(e)}")
SSProcess.startProgress("导入失败", 1)
SSProcess.stepProgress("错误:未知错误,请查看日志")
SSProcess.closeProgress()
return False
实践五:错误码与状态管理
为不同的错误场景定义清晰的错误码,便于调用方根据错误类型采取不同的恢复策略。
class ScriptError:
"""脚本错误码定义"""
SUCCESS = 0
INVALID_PARAM = 1
DATA_SOURCE_ERROR = 2
DATASET_ERROR = 3
CONVERSION_ERROR = 4
INDEX_OUT_OF_RANGE = 5
OBJECT_NULL = 6
def createObjectWithErrorCode(self, code: int) -> tuple[int, tuple]:
"""带错误码的对象创建方法"""
# 检查数据源
ds_eps = self.map.getCurrentDataSourceEPS()
if ds_eps is None:
self.log_error_msg("数据源不可用")
return (ScriptError.DATA_SOURCE_ERROR, (None, None))
# 检查特征
fea = ds_eps.getFeature(code)
if fea is None:
fea = ds_eps.getFeature(0)
if fea is None:
self.log_error_msg(f"特征代码无效: {code}")
return (ScriptError.DATASET_ERROR, (None, None))
# 检查数据集
dataset = getDatasetByFeaCode(ds_eps, fea)
if dataset is None:
self.log_error_msg(f"数据集不存在,code: {code}")
return (ScriptError.DATASET_ERROR, (None, None))
# 创建对象
obj_type = fea.nObjectType
obj, geo = self.createDefaultGeoBase(obj_type, dataset)
if obj is None or geo is None:
self.log_error_msg(f"对象创建失败,code: {code}")
return (ScriptError.DATASET_ERROR, (None, None))
return (ScriptError.SUCCESS, (obj, geo))
# 使用示例
error_code, (obj, geo) = SSProcess.createObjectWithErrorCode(3103013)
if error_code == ScriptError.SUCCESS:
print("对象创建成功")
elif error_code == ScriptError.DATA_SOURCE_ERROR:
print("请检查数据源配置")
elif error_code == ScriptError.DATASET_ERROR:
print("请检查数据集和特征代码")
错误调试与问题排查
当脚本遇到错误时,系统化的调试流程能够快速定位问题根源。本节提供一套完整的错误调试方法论。
错误调试流程
flowchart TD
A[脚本发生错误] --> B{是否已配置日志?}
B -->|否| C[配置日志记录器<br/>参考[配置日志记录器] 37-pei-zhi-ri-zhi-ji-lu-qi]
B -->|是| D[检查日志文件<br/>查找 ERROR/CRITICAL 级别消息]
C --> D
D --> E{错误类型判断}
E -->|数据源/数据集错误| F[检查数据源连接<br/>验证数据集名称]
E -->|类型转换错误| G[检查输入值格式<br/>验证数据类型]
E -->|索引越界错误| H[检查选择集大小<br/>验证索引范围]
E -->|空对象引用错误| I[检查对象初始化<br/>验证对象状态]
E -->|未知错误| J[启用 DEBUG 级别日志<br/>添加详细打印信息]
F --> K[修复问题]
G --> K
H --> K
I --> K
J --> L[分析堆栈跟踪<br/>定位代码位置]
L --> K
K --> M[重新运行脚本<br/>验证修复效果]
M --> N{问题解决?}
N -->|是| O[调试完成]
N -->|否| P[重复调试流程]
style C fill:#fff4e1
style K fill:#e1ffe1
style O fill:#e1ffe1
常见错误排查清单
| 错误现象 | 可能原因 | 排查步骤 | 相关页面 |
|---|---|---|---|
| "无法获取当前数据源" | 工作空间未初始化或数据源关闭 | 检查 工作空间与地图概念 | - |
| "无法获取数据集" | 图层名称错误或图层不存在 | 验证图层名称,使用 获取工作空间信息 | - |
| "索引越界" | 选择集为空或索引超出范围 | 使用 获取选择集对象数量 检查 | - |
| "角度/比例转换失败" | 输入值格式不正确 | 验证输入值,确保为数字字符串 | - |
| "对象创建失败" | 特征代码无效或对象类型不支持 | 参考 地理对象类型定义 | - |
调试辅助代码
以下辅助函数可以帮助您在调试时获取详细的系统状态信息。
def printSystemStatus():
"""打印系统状态信息,用于调试"""
print("=" * 60)
print("SunvStation 系统状态")
print("=" * 60)
# 工作空间状态
workspace = SSProcess.getWorkspace()
if workspace:
print(f"✓ 工作空间已初始化")
map = workspace.getCurrentScaleMap()
if map:
print(f"✓ 当前地图已加载")
print(f" - 图层数量: {map.getLayerSize()}")
print(f" - 总对象数: {map.getObjectSize(e_Null_Obj)}")
else:
print(f"✗ 当前地图未加载")
else:
print(f"✗ 工作空间未初始化")
# 数据源状态
if SSProcess.map:
ds_eps = SSProcess.map.getCurrentDataSourceEPS()
if ds_eps:
print(f"✓ 数据源已连接")
else:
print(f"✗ 数据源未连接")
# 选择集状态
geo_count = SSProcess.getSelGeoCount()
note_count = SSProcess.getSelNoteCount()
print(f"✓ 选择集状态: {geo_count} 个地物, {note_count} 个注记")
# 日志状态
if SSProcess.logger:
print(f"✓ 日志记录器已配置: {SSProcess.logger.name}")
print(f" - 日志级别: {logging.getLevelName(SSProcess.logger.level)}")
print(f" - 处理器数量: {len(SSProcess.logger.handlers)}")
else:
print(f"✗ 日志记录器未配置")
# 进度条状态
if SSProcess.progress:
print(f"✓ 进度条已启动")
else:
print(f" 进度条未启动")
print("=" * 60)
# 使用示例
printSystemStatus()
错误处理与相关功能的集成
错误处理机制与 sunvpy 的其他核心功能紧密协作,形成完整的脚本开发生态。理解这些集成关系有助于您构建更加全面的解决方案。
错误处理与进度条管理
进度条操作与错误处理密切相关,特别是在长时间运行的任务中。当发生错误时,需要正确关闭进度条,避免 UI 卡顿。
def batchProcessWithErrorHandling(self, total_count: int) -> bool:
"""批量操作中的错误处理与进度条协作"""
try:
SSProcess.startProgress("批量处理", total_count)
for i in range(total_count):
try:
# 执行单次处理
result = self.processSingleItem(i)
if result:
SSProcess.stepProgress(f"处理完成: {i+1}/{total_count}")
else:
self.log_error_msg(f"项目处理失败: {i}")
# 继续处理下一个项目,不中断整个流程
except Exception as e:
self.log_error_msg(f"项目处理发生异常: {i}, 错误: {str(e)}")
# 继续处理下一个项目
SSProcess.closeProgress()
return True
except Exception as e:
# 外层异常,确保进度条关闭
self.logger.critical(f"批量操作严重错误: {str(e)}")
try:
SSProcess.closeProgress()
except Exception:
pass # 静默处理关闭错误
return False
Sources: progress_mixin.py
错误处理与对象缓存机制
对象缓存操作中的错误需要特别小心,避免导致数据不一致。当缓存操作失败时,需要清理相关状态。
def saveBufferWithValidation(self) -> bool:
"""带验证的对象缓存保存"""
try:
# 验证缓存不为空
if (self.bufferObjList.empty() and
self.bufferNoteList.empty() and
len(self.newBufferObjList) == 0 and
len(self.newBufferNoteList) == 0):
self.logger.info("缓存为空,无需保存")
return True
# 保存新建对象
all_new_objs = self.newBufferObjList + self.newBufferNoteList
if all_new_objs:
for obj, geo in all_new_objs:
try:
if geo is None:
self.log_error_msg("新建对象为空,跳过保存")
continue
dataset = geo.getDatasetName()
if not dataset:
self.log_error_msg("对象数据集名称为空,跳过保存")
continue
# ... 保存逻辑
except Exception as e:
self.log_error_msg(f"保存对象失败: {str(e)}")
continue
# 保存修改对象
if not self.bufferObjList.empty():
self.transMemoDataToExtendAttr(self.bufferObjList)
self.map.saveDatabase(self.bufferObjList)
if not self.bufferNoteList.empty():
self.transMemoDataToExtendAttr(self.bufferNoteList)
self.map.saveDatabase(self.bufferNoteList)
# 清理缓存(无论成功或失败)
self.newBufferObjList.clear()
self.newBufferNoteList.clear()
self.bufferObjList.clear()
self.bufferNoteList.clear()
self.logger.info("缓存保存成功")
return True
except Exception as e:
self.logger.critical(f"缓存保存严重错误: {str(e)}")
# 清理缓存以避免不一致状态
self.newBufferObjList.clear()
self.newBufferNoteList.clear()
return False
Sources: geo_edit_mixin.py
错误处理与选择集管理
选择集操作中的错误处理需要特别注意,避免破坏选择集的状态一致性。
def safeSelectAndProcess(self, condition_field: str, condition_op: str, condition_value: str) -> bool:
"""安全的选择和操作流程"""
try:
# 清理现有选择集
SSProcess.clearSelection()
SSProcess.clearSelectCondition()
# 设置选择条件
SSProcess.setSelectCondition(condition_field, condition_op, condition_value)
# 执行过滤
SSProcess.selectFilter()
# 检查选择结果
geo_count = SSProcess.getSelGeoCount()
note_count = SSProcess.getSelNoteCount()
total_count = geo_count + note_count
if total_count == 0:
self.logger.warning(f"没有找到符合条件的对象: {condition_field} {condition_op} {condition_value}")
return True # 没有对象也是正常情况
self.logger.info(f"选择集创建成功: {geo_count} 个地物, {note_count} 个注记")
# 处理选择集
success_count = 0
for i in range(total_count):
try:
# 处理单个对象
if i < geo_count:
result = self.processGeo(i)
else:
result = self.processNote(i - geo_count)
if result:
success_count += 1
except Exception as e:
self.log_error_msg(f"处理对象 {i} 失败: {str(e)}")
continue
self.logger.info(f"选择集处理完成: {success_count}/{total_count} 成功")
return success_count > 0
except Exception as e:
self.logger.critical(f"选择集操作严重错误: {str(e)}")
# 清理选择集
try:
SSProcess.clearSelection()
SSProcess.clearSelectCondition()
except Exception:
pass
return False
Sources: selection_mixin.py
下一步学习建议
掌握错误处理与异常管理后,建议按照以下路径深入学习相关主题,构建完整的 SunvStation 脚本开发知识体系: