import os import logging import subprocess from typing import Dict import pandas as pd class CommandRunner: """执行网格处理命令的类""" def __init__(self, output_dir: str): """ 初始化命令执行器 Args: output_dir: 输出目录路径 """ self.output_dir = output_dir self.logger = logging.getLogger('UAV_Preprocess.CommandRunner') def run_grid_commands(self, grid_points: Dict[int, pd.DataFrame], enable_grid_division: bool = True): """ 为每个网格顺序运行指定命令 Args: grid_points: 网格点数据字典,键为网格索引,值为该网格的点数据 enable_grid_division: 是否启用网格划分 """ if not enable_grid_division: self._run_command(0) return self.logger.info("开始执行网格处理命令") # 顺序执行每个网格的命令 for grid_idx in grid_points.keys(): try: self._run_command(grid_idx) except Exception as e: self.logger.error(f"网格 {grid_idx + 1} 处理命令执行失败: {str(e)}") raise # 如果一个网格失败,停止后续执行 def _run_command(self, grid_idx: int): """ 执行单个网格的命令 Args: grid_idx: 网格索引 Raises: Exception: 当命令执行失败时抛出异常 """ try: # 确定网格目录和命令 grid_dir = os.path.join(self.output_dir, f'grid_{grid_idx + 1}') command = f"docker run -ti --rm -v {grid_dir}:/datasets opendronemap/odm --project-path /datasets project --feature-quality lowest --force-gps --use-3dmesh" self.logger.info(f"执行命令: {command} 在目录: {grid_dir}") # 在指定目录下执行命令 process = subprocess.Popen( command, shell=True, cwd=grid_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 获取命令输出 stdout, stderr = process.communicate() # 检查命令执行结果 if process.returncode == 0: self.logger.info(f"网格 {grid_idx + 1} 命令执行成功") self.logger.debug(f"命令输出至日志文件") with open(os.path.join(grid_dir, 'odm_success.log'), 'a', encoding='utf-8') as f: f.write(f"{stdout}") else: self.logger.error(f"网格 {grid_idx + 1} 命令执行失败") self.logger.error(f"错误信息输出至日志文件") with open(os.path.join(grid_dir, 'odm_error.log'), 'a', encoding='utf-8') as f: f.write(f"{stdout}") f.write(f"\n错误日志:\n") f.write(f"{stderr}") raise Exception(f"命令执行失败: {stderr}") except Exception as e: self.logger.error(f"网格 {grid_idx + 1} 命令执行出错: {str(e)}") raise