优化监控进程
This commit is contained in:
龙澳
parent
a808e020a4
commit
451fdbdf32
@ -307,8 +307,8 @@ class ImagePreprocessor:
|
||||
if __name__ == "__main__":
|
||||
# 创建配置
|
||||
config = PreprocessConfig(
|
||||
image_dir=r"F:\error_data\20241024100834\code\images",
|
||||
output_dir=r"G:\ODM_output\20241024100834\output",
|
||||
image_dir=r"F:\error_data\20241024100834\project\images",
|
||||
output_dir=r"G:\ODM_output\20241024100834",
|
||||
|
||||
cluster_eps=0.01,
|
||||
cluster_min_samples=5,
|
||||
|
||||
@ -25,27 +25,23 @@ i
|
||||
self.mode = mode
|
||||
|
||||
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}')
|
||||
grid_dir = grid_dir[0].lower() + grid_dir[1:].replace('\\', '/')
|
||||
|
||||
# 简化命令构建
|
||||
base_command = (
|
||||
f"docker run --rm " # 移除 -i 参数
|
||||
f"-v {grid_dir}:/datasets "
|
||||
f"opendronemap/odm "
|
||||
f"--project-path /datasets project "
|
||||
f"--max-concurrency 10 "
|
||||
f"--force-gps "
|
||||
)
|
||||
|
||||
if self.mode == "快拼模式":
|
||||
command = (
|
||||
f"docker run -i --rm "
|
||||
f"-v {grid_dir}:/datasets "
|
||||
f"opendronemap/odm "
|
||||
f"--project-path /datasets project "
|
||||
f"--max-concurrency 10 "
|
||||
f"--force-gps "
|
||||
command = base_command + (
|
||||
f"--feature-quality lowest "
|
||||
f"--orthophoto-resolution 8 "
|
||||
f"--fast-orthophoto "
|
||||
@ -53,17 +49,8 @@ i
|
||||
f"--rerun-all"
|
||||
)
|
||||
else:
|
||||
command = (
|
||||
f"docker run -i --rm "
|
||||
f"-v {grid_dir}:/datasets "
|
||||
f"opendronemap/odm "
|
||||
f"--project-path /datasets project "
|
||||
f"--max-concurrency 10 "
|
||||
f"--force-gps "
|
||||
f"--rerun-all"
|
||||
)
|
||||
command = base_command + "--rerun-all"
|
||||
|
||||
self.logger.info(f"开始执行命令: {command}")
|
||||
success, error_msg = self.monitor.run_odm_with_monitor(
|
||||
command, grid_dir, grid_idx)
|
||||
|
||||
|
||||
@ -4,6 +4,7 @@ import psutil
|
||||
import logging
|
||||
import subprocess
|
||||
from typing import Optional, Tuple
|
||||
from collections import deque
|
||||
|
||||
|
||||
class ODMProcessMonitor:
|
||||
@ -22,28 +23,14 @@ class ODMProcessMonitor:
|
||||
self.check_interval = check_interval
|
||||
self.logger = logging.getLogger('UAV_Preprocess.ODMMonitor')
|
||||
self.mode = mode
|
||||
|
||||
def _check_docker_container(self, process_name: str = "opendronemap/odm") -> bool:
|
||||
"""检查是否有指定的Docker容器在运行"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["docker", "ps", "--filter",
|
||||
f"ancestor={process_name}", "--format", "{{.ID}}"],
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
return bool(result.stdout.strip())
|
||||
except Exception as e:
|
||||
self.logger.error(f"检查Docker容器状态时发生错误: {str(e)}")
|
||||
return False
|
||||
# 用于存储最后N行输出
|
||||
self.last_outputs = deque(maxlen=50)
|
||||
|
||||
def _check_success(self, grid_dir: str) -> bool:
|
||||
"""检查ODM是否执行成功"""
|
||||
if self.mode == "快拼模式":
|
||||
success_markers = ['odm_orthophoto', 'odm_georeferencing']
|
||||
else:
|
||||
success_markers = ['odm_orthophoto',
|
||||
'odm_georeferencing', 'odm_texturing']
|
||||
success_markers = ['odm_orthophoto', 'odm_georeferencing']
|
||||
if self.mode != "快拼模式":
|
||||
success_markers.append('odm_texturing')
|
||||
return all(os.path.exists(os.path.join(grid_dir, 'project', marker)) for marker in success_markers)
|
||||
|
||||
def run_odm_with_monitor(self, command: str, grid_dir: str, grid_idx: int) -> Tuple[bool, str]:
|
||||
@ -52,61 +39,46 @@ class ODMProcessMonitor:
|
||||
while attempt < self.max_retries:
|
||||
try:
|
||||
self.logger.info(f"网格 {grid_idx + 1} 第 {attempt + 1} 次尝试执行ODM")
|
||||
|
||||
# 创建日志文件
|
||||
log_file = os.path.join(grid_dir, f'odm_attempt_{attempt + 1}.log')
|
||||
|
||||
# 使用 subprocess.Popen 启动进程，并设置适当的参数
|
||||
|
||||
# 使用 subprocess.Popen 启动进程
|
||||
process = subprocess.Popen(
|
||||
command,
|
||||
shell=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
bufsize=1, # 使用行缓冲
|
||||
text=True, # 使用文本模式
|
||||
universal_newlines=True # 统一换行符处理
|
||||
bufsize=-1, # 使用系统默认缓冲
|
||||
text=True
|
||||
)
|
||||
|
||||
# 使用非阻塞方式读取输出
|
||||
while True:
|
||||
# 读取输出
|
||||
output = process.stdout.readline()
|
||||
if output:
|
||||
with open(log_file, 'a', encoding='utf-8') as f:
|
||||
f.write(output)
|
||||
self.logger.info(f"网格 {grid_idx + 1}: {output.strip()}")
|
||||
|
||||
# 检查进程是否结束
|
||||
if process.poll() is not None:
|
||||
break
|
||||
|
||||
# 检查Docker容器状态
|
||||
if not self._check_docker_container():
|
||||
break
|
||||
|
||||
time.sleep(0.1) # 短暂休眠，避免CPU过度使用
|
||||
|
||||
# 获取最终返回码
|
||||
return_code = process.wait()
|
||||
|
||||
|
||||
# 等待进程完成并获取所有输出
|
||||
stdout, stderr = process.communicate()
|
||||
return_code = process.returncode
|
||||
|
||||
# 只保存最后N行输出
|
||||
last_lines = stdout.strip().split('\n')[-50:]
|
||||
if last_lines:
|
||||
self.logger.info(f"网格 {grid_idx + 1} 最后的输出:")
|
||||
for line in last_lines:
|
||||
self.logger.info(line)
|
||||
|
||||
# 检查是否成功完成
|
||||
if return_code == 0 and self._check_success(grid_dir):
|
||||
self.logger.info(f"网格 {grid_idx + 1} ODM处理成功")
|
||||
return True, ""
|
||||
|
||||
|
||||
self.logger.warning(f"网格 {grid_idx + 1} 第 {attempt + 1} 次尝试失败")
|
||||
|
||||
if stderr:
|
||||
self.logger.error(f"错误输出: {stderr}")
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"监控进程发生异常: {str(e)}"
|
||||
self.logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
|
||||
attempt += 1
|
||||
if attempt < self.max_retries:
|
||||
wait_time = (attempt + 1) * 30
|
||||
self.logger.info(f"等待 {wait_time} 秒后重试...")
|
||||
time.sleep(wait_time)
|
||||
|
||||
time.sleep(30) # 固定等待时间
|
||||
|
||||
error_msg = f"网格 {grid_idx + 1} 在 {self.max_retries} 次尝试后仍然失败"
|
||||
self.logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
Loading…
Reference in New Issue
Block a user