372 lines
13 KiB
Python
372 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
高光谱分析工具包统一入口
|
|
提供命令行接口调用所有功能模块
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import traceback
|
|
|
|
from registry import (
|
|
REGISTRY, create_module_parser, execute_task,
|
|
get_module_info, get_call_sequence
|
|
)
|
|
from validators import validate_task_inputs
|
|
from output_handler import create_unified_output
|
|
|
|
|
|
def create_parser() -> argparse.ArgumentParser:
|
|
"""创建主参数解析器"""
|
|
parser = argparse.ArgumentParser(
|
|
prog='hyperspectral-toolkit',
|
|
description='高光谱图像分析工具包 - 统一入口',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
使用示例:
|
|
# 降维分析
|
|
python main.py dim-reduction --input data.hdr --method pca --n-components 5
|
|
|
|
# 图像分割
|
|
python main.py segmentation --input data.hdr --method otsu --band-index 100
|
|
|
|
# 分类分析
|
|
python main.py classification --input data.hdr --roi-file roi.xml --method svm
|
|
|
|
# 光谱指数计算
|
|
python main.py spectral-index --input data.hdr --batch --png
|
|
|
|
# 色差计算
|
|
python main.py delta-e --mode image --input image.lab --standards colors.csv --method CIEDE2000
|
|
|
|
# 光谱到颜色转换
|
|
python main.py spectral-to-color --input spectrum.csv --color-space Lab --illuminant D65
|
|
|
|
# XYZ到RGB转换
|
|
python main.py xyz-to-rgb --input xyz_image.dat --rgb-space sRGB --gamma-method sRGB
|
|
|
|
# 查看特定任务帮助
|
|
python main.py <task_name> --help
|
|
"""
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--version', '-v', action='version',
|
|
version='高光谱分析工具包 v1.0.0'
|
|
)
|
|
|
|
# 创建子解析器
|
|
subparsers = parser.add_subparsers(
|
|
dest='task',
|
|
help='选择要执行的任务',
|
|
metavar='TASK'
|
|
)
|
|
|
|
# 为每个任务创建子解析器
|
|
for task_name in REGISTRY.keys():
|
|
create_module_parser(subparsers, task_name)
|
|
|
|
return parser
|
|
|
|
|
|
def validate_input_file(file_path: str, task_name: str) -> bool:
|
|
"""验证输入文件"""
|
|
if not os.path.exists(file_path):
|
|
print(f"错误: 输入文件不存在: {file_path}")
|
|
return False
|
|
|
|
# 检查文件扩展名
|
|
valid_extensions = ['.hdr', '.dat', '.csv', '.xml', '.bil', '.bsq', '.bip']
|
|
file_ext = Path(file_path).suffix.lower()
|
|
|
|
if file_ext not in valid_extensions:
|
|
print(f"警告: 文件扩展名 {file_ext} 可能不受支持")
|
|
print(f"支持的格式: {', '.join(valid_extensions)}")
|
|
|
|
return True
|
|
|
|
|
|
def validate_roi_file(roi_path: str) -> bool:
|
|
"""验证ROI文件"""
|
|
if not os.path.exists(roi_path):
|
|
print(f"错误: ROI文件不存在: {roi_path}")
|
|
return False
|
|
|
|
if not roi_path.endswith('.xml'):
|
|
print(f"错误: ROI文件必须是XML格式: {roi_path}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def validate_output_dir(output_dir: str) -> bool:
|
|
"""验证输出目录"""
|
|
try:
|
|
Path(output_dir).mkdir(parents=True, exist_ok=True)
|
|
return True
|
|
except Exception as e:
|
|
print(f"错误: 无法创建输出目录 {output_dir}: {e}")
|
|
return False
|
|
|
|
|
|
def validate_args(args) -> bool:
|
|
"""验证命令行参数"""
|
|
# 获取任务信息
|
|
task_info = REGISTRY.get(args.task)
|
|
if not task_info:
|
|
print(f"错误: 未知任务 '{args.task}'")
|
|
return False
|
|
|
|
info = task_info["info"]
|
|
|
|
# 验证输入文件
|
|
if info.requires_data_file:
|
|
# 根据任务类型确定输入文件参数名
|
|
input_param = None
|
|
if args.task == 'delta-e':
|
|
input_param = getattr(args, 'input', None)
|
|
elif args.task == 'spectral-to-color':
|
|
input_param = getattr(args, 'input', None)
|
|
elif args.task == 'xyz-to-rgb':
|
|
input_param = getattr(args, 'input', None)
|
|
else:
|
|
input_param = getattr(args, 'input', None)
|
|
|
|
if not input_param:
|
|
print(f"错误: 任务 '{args.task}' 需要指定输入文件")
|
|
return False
|
|
|
|
if not validate_input_file(input_param, args.task):
|
|
return False
|
|
|
|
# 验证ROI文件
|
|
if info.requires_roi_file:
|
|
roi_attr = getattr(args, 'roi_file', None)
|
|
if not roi_attr:
|
|
print(f"错误: 任务 '{args.task}' 需要指定ROI文件 (--roi-file)")
|
|
return False
|
|
if not validate_roi_file(roi_attr):
|
|
return False
|
|
|
|
# 验证输出目录或文件
|
|
if args.task in ['delta-e', 'spectral-to-color', 'xyz-to-rgb', 'prosail-gui']:
|
|
# 颜色分析任务和GUI应用使用自己的输出参数验证
|
|
if args.task == 'prosail-gui':
|
|
# PROSAIL GUI 不需要输出目录验证
|
|
pass
|
|
elif args.task == 'delta-e':
|
|
output_dir = getattr(args, 'output_dir', './results')
|
|
if not validate_output_dir(output_dir):
|
|
return False
|
|
elif args.task == 'spectral-to-color':
|
|
# spectral-to-color 使用 -o 参数作为输出路径，需要提取目录部分
|
|
output_param = getattr(args, 'output', None)
|
|
if output_param:
|
|
output_dir = os.path.dirname(output_param) or '.'
|
|
else:
|
|
output_dir = './results'
|
|
if not validate_output_dir(output_dir):
|
|
return False
|
|
elif args.task == 'xyz-to-rgb':
|
|
# xyz-to-rgb 使用 --output 参数作为输出路径，需要提取目录部分
|
|
output_param = getattr(args, 'output', None)
|
|
if output_param:
|
|
output_dir = os.path.dirname(output_param) or '.'
|
|
else:
|
|
output_dir = './results'
|
|
if not validate_output_dir(output_dir):
|
|
return False
|
|
else:
|
|
# 标准任务的输出验证
|
|
if hasattr(args, 'output_file') and args.output_file:
|
|
# 如果指定了输出文件，验证其目录
|
|
output_dir = os.path.dirname(args.output_file) or '.'
|
|
if not validate_output_dir(output_dir):
|
|
return False
|
|
else:
|
|
# 否则验证输出目录
|
|
if not validate_output_dir(args.output_dir):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def print_task_summary(task_name: str, args, success: bool = True):
|
|
"""打印任务执行摘要"""
|
|
status = "✓ 成功" if success else "✗ 失败"
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"任务执行摘要")
|
|
print(f"{'='*60}")
|
|
print(f"任务名称: {task_name}")
|
|
print(f"执行状态: {status}")
|
|
|
|
# 根据任务类型显示不同的参数信息
|
|
if task_name == 'prosail-gui':
|
|
print("PROSAIL植被光谱模拟器GUI")
|
|
print(f"GUI模式: {'否' if getattr(args, 'no_gui', False) else '是'}")
|
|
elif task_name == 'delta-e':
|
|
print(f"模式: {args.mode}")
|
|
print(f"输入文件: {getattr(args, 'input', 'N/A')}")
|
|
print(f"标准色文件: {getattr(args, 'standards', 'N/A')}")
|
|
print(f"色差方法: {getattr(args, 'method', 'N/A')}")
|
|
print(f"输出目录: {getattr(args, 'output_dir', './results')}")
|
|
elif task_name == 'spectral-to-color':
|
|
print(f"输入文件: {getattr(args, 'input_file', 'N/A')}")
|
|
print(f"颜色空间: {getattr(args, 'color_space', 'N/A')}")
|
|
print(f"光源: {getattr(args, 'illuminant', 'N/A')}")
|
|
print(f"观察者: {getattr(args, 'observer', 'N/A')}")
|
|
print(f"输出文件: {getattr(args, 'output', 'N/A')}")
|
|
elif task_name == 'xyz-to-rgb':
|
|
print(f"输入文件: {getattr(args, 'input', 'N/A')}")
|
|
print(f"RGB空间: {getattr(args, 'rgb_space', 'N/A')}")
|
|
print(f"Gamma方法: {getattr(args, 'gamma', 'N/A')}")
|
|
print(f"输出文件: {getattr(args, 'output', 'N/A')}")
|
|
elif task_name == 'regression-prediction':
|
|
print("使用训练好的回归模型进行高光谱图像预测")
|
|
print(f"输入图像: {getattr(args, 'input', 'N/A')}")
|
|
print(f"模型路径: {getattr(args, 'model_path', 'N/A')}")
|
|
if hasattr(args, 'mask_path') and getattr(args, 'mask_path', None):
|
|
print(f"遮罩文件: {args.mask_path}")
|
|
print(f"输出目录: {getattr(args, 'output_dir', './results')}")
|
|
print(f"使用遮罩: {'是' if getattr(args, 'use_mask', True) else '否'}")
|
|
elif task_name == 'glcm':
|
|
print("GLCM纹理特征提取")
|
|
print(f"输入图像: {getattr(args, 'input', 'N/A')}")
|
|
print(f"波段索引: {getattr(args, 'band_index', 25)}")
|
|
print(f"灰度级数: {getattr(args, 'nbit', 64)}")
|
|
print(f"滑动窗口: {getattr(args, 'slide_window', 7)}")
|
|
print(f"步长距离: {getattr(args, 'step', [2])}")
|
|
print(f"角度: {[f'{a:.2f}rad' for a in getattr(args, 'angle', [0])]}")
|
|
print(f"输出目录: {getattr(args, 'output_dir', './results')}")
|
|
print(f"保存DAT: {'是' if getattr(args, 'save_dat', True) else '否'}")
|
|
else:
|
|
# 标准任务的参数显示
|
|
print(f"输入文件: {getattr(args, 'input', 'N/A')}")
|
|
if hasattr(args, 'roi_file') and args.roi_file:
|
|
print(f"ROI文件: {args.roi_file}")
|
|
print(f"输出目录: {getattr(args, 'output_dir', './results')}")
|
|
print(f"输出前缀: {getattr(args, 'output_prefix', 'result')}")
|
|
|
|
print(f"{'='*60}\n")
|
|
|
|
|
|
def list_available_tasks():
|
|
"""列出所有可用任务"""
|
|
print("可用任务列表:")
|
|
print("-" * 60)
|
|
|
|
categories = {}
|
|
for task_name, task_info in REGISTRY.items():
|
|
category = task_info["info"].category
|
|
if category not in categories:
|
|
categories[category] = []
|
|
categories[category].append((task_name, task_info["info"].description))
|
|
|
|
for category, tasks in categories.items():
|
|
print(f"\n{category}:")
|
|
for task_name, description in tasks:
|
|
print(f" {task_name:20} - {description}")
|
|
|
|
print(f"\n{'='*60}")
|
|
print("使用方法: python main.py <task_name> --help 查看具体任务参数")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
parser = create_parser()
|
|
args = parser.parse_args()
|
|
|
|
# 如果没有指定任务，显示帮助信息
|
|
if not args.task:
|
|
parser.print_help()
|
|
print("\n" + "="*60)
|
|
list_available_tasks()
|
|
return 0
|
|
|
|
# 如果是特殊任务，处理后退出
|
|
if args.task == 'list':
|
|
list_available_tasks()
|
|
return 0
|
|
|
|
try:
|
|
# 验证参数
|
|
if not validate_args(args):
|
|
return 1
|
|
|
|
# 高级验证
|
|
validation_result = validate_task_inputs(args.task, **vars(args))
|
|
if not validation_result['valid']:
|
|
print("验证失败:")
|
|
for error in validation_result['errors']:
|
|
print(f" ✗ {error}")
|
|
return 1
|
|
|
|
if validation_result['warnings']:
|
|
print("验证警告:")
|
|
for warning in validation_result['warnings']:
|
|
print(f" ⚠ {warning}")
|
|
|
|
print(f"开始执行任务: {args.task}")
|
|
if args.task != 'prosail-gui':
|
|
print(f"输入文件: {getattr(args, 'input', 'N/A')}")
|
|
print(f"输出目录: {getattr(args, 'output_dir', './results')}")
|
|
else:
|
|
print("PROSAIL植被光谱模拟器GUI")
|
|
print("-" * 40)
|
|
|
|
# 记录开始时间
|
|
start_time = time.time()
|
|
|
|
# 执行任务
|
|
result = execute_task(args.task, args)
|
|
|
|
# 计算执行时间
|
|
execution_time = time.time() - start_time
|
|
|
|
# 创建统一输出
|
|
if args.task != 'prosail-gui':
|
|
output_files = create_unified_output(
|
|
args.task, args, result, getattr(args, 'output_dir', './results'),
|
|
execution_time, success=True
|
|
)
|
|
else:
|
|
output_files = None
|
|
|
|
# 显示输出文件信息
|
|
if output_files:
|
|
print("\n生成的文件:")
|
|
for file_type, file_path in output_files.items():
|
|
print(f" ✓ {file_type}: {file_path}")
|
|
|
|
return 0
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n用户中断执行")
|
|
return 1
|
|
except Exception as e:
|
|
print(f"\n执行失败: {e}")
|
|
if '--debug' in sys.argv:
|
|
traceback.print_exc()
|
|
|
|
# 即使失败也尝试创建输出摘要
|
|
try:
|
|
execution_time = time.time() - start_time if 'start_time' in locals() else None
|
|
if args.task != 'prosail-gui':
|
|
create_unified_output(
|
|
args.task, args, None, getattr(args, 'output_dir', './results'),
|
|
execution_time, success=False
|
|
)
|
|
except:
|
|
pass # 如果输出失败，忽略
|
|
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|