UAV/post_pro/merge_laz.py

import os
import logging
import numpy as np
from typing import Dict, Tuple
import pandas as pd
import subprocess


class MergePly:
    def __init__(self, output_dir: str):
        self.output_dir = output_dir
        self.logger = logging.getLogger('UAV_Preprocess.MergePly')

    def merge_grid_laz(self, grid_points: Dict[tuple, list]):
        """合并所有网格的点云"""
        self.logger.info("开始合并所有网格的laz点云")

        if len(grid_points) < 2:
            self.logger.info("只有一个网格，无需合并")
            return

        try:
            laz_lt = []
            for grid_id, points in grid_points.items():
                grid_laz = os.path.join(
                    self.output_dir,
                    f"grid_{grid_id[0]}_{grid_id[1]}",
                    "project",
                    "odm_georeferencing",
                    "odm_georeferenced_model.laz"
                )

                if not os.path.exists(grid_laz):
                    self.logger.warning(f"网格 ({grid_id[0]},{grid_id[1]}) 的laz文件不存在: {grid_laz}")
                    continue
                laz_lt.append(grid_laz)

            kwargs = {
                'all_inputs': " ".join(laz_lt),
                'output': os.path.join(self.output_dir, 'merged_pointcloud.laz')
            }

            subprocess.run('D:\\software\\LAStools\\bin\\lasmerge64.exe -i {all_inputs} -o "{output}"'.format(**kwargs))
   
        except Exception as e:
            self.logger.error(f"PLY点云合并过程中发生错误: {str(e)}", exc_info=True)
            raise


if __name__ == "__main__":
    from utils.logger import setup_logger

    # 设置输出目录和日志
    output_dir = r"G:\ODM_output\1009"
    setup_logger(output_dir)

    # 构造测试用的grid_points字典
    grid_points = {
        (0, 0): [],  # 不再需要GPS点信息
        (0, 1): []
    }

    # 创建MergePly实例并执行合并
    merge_ply = MergePly(output_dir)
    merge_ply.merge_grid_laz(grid_points)