推荐订阅源

自从有了AI，处理数据再也不烧脑了~喂进去一篇说明书啥都有了

以下代码从XSE格式多波束测深数据提取坐标点和时间，用于画航迹图和查找作业时间

##读XSE文件提取时间和坐标（高性能版）
import struct
import math
import mmap
import multiprocessing
from datetime import datetime, timedelta
from pathlib import Path
import csv

# 全局常量（提前定义，避免重复计算）
BLOCK_START = b'\x24\x48\x53\x46'  # "$HSF" 大端序字节值（直接写死，避免重复pack）
BLOCK_END = b'\x23\x48\x53\x46'    # "#HSF"
GROUP_START = b'\x24\x48\x53\x47'  # "$HSG"
GROUP_END = b'\x23\x48\x53\x47'    # "#HSG"
TIME_BASE_UTC = datetime(1901, 1, 1, 0, 0, 0)
BEIJING_OFFSET = timedelta(hours=8)
RTD = 180.0 / math.pi  # 弧度转度
CHUNK_SIZE = 1024 * 1024 * 8  # 8MB块大小（平衡IO和内存）

def parse_beijing_time(seconds, micro):
    """快速转换北京时间（抽离为独立函数，减少重复代码）"""
    total_seconds = seconds + micro / 1e6
    utc_time = TIME_BASE_UTC + timedelta(seconds=total_seconds)
    return utc_time + BEIJING_OFFSET

def process_single_xse(xse_file, output_dir):
    """处理单个XSE文件（核心解析逻辑，内存映射+边读边写）"""
    xse_path = Path(xse_file)
    output_csv = Path(output_dir) / f"{xse_path.stem}_blh.csv"
    
    # 提前创建CSV并写入表头（边读边写）
    with open(output_csv, 'w', encoding='utf-8-sig', newline='') as f_csv:
        csv_writer = csv.writer(f_csv)
        csv_writer.writerow(['latitude', 'longitude', 'timestamp'])
        
        # 内存映射打开文件（大文件读取效率提升10倍以上）
        with open(xse_file, 'rb') as f:
            with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
                file_size = len(mm)
                ptr = 0  # 内存映射指针
                
                while ptr < file_size - 24:  # 保留最小头部空间
                    # 快速定位数据块起始$HSF（替代逐字节查找）
                    hsf_pos = mm.find(BLOCK_START, ptr, min(ptr + CHUNK_SIZE, file_size))
                    if hsf_pos == -1:
                        break  # 无更多数据块
                    ptr = hsf_pos + 4  # 跳过$HSF
                    
                    # 解析数据块头部（仅保留必要字段）
                    try:
                        # Byte Count（其后字节数）+ Block ID
                        block_content_size = struct.unpack('>I', mm[ptr:ptr+4])[0]
                        ptr += 4
                        block_id = struct.unpack('>I', mm[ptr:ptr+4])[0]
                        ptr += 4
                        
                        if block_id != 1:  # 仅处理blockid=1
                            ptr += block_content_size + 4  # 跳过内容+结束标识
                            continue
                        
                        # 跳过source，读取时间戳（seconds+micro）
                        ptr += 4
                        block_seconds = struct.unpack('>I', mm[ptr:ptr+4])[0]
                        ptr += 4
                        block_micro = struct.unpack('>I', mm[ptr:ptr+4])[0]
                        ptr += 4
                        
                        # 读取数据块内容（仅处理必要部分）
                        content_end = ptr + block_content_size
                        content_ptr = ptr
                        
                        # 在数据块内容中查找位置组$HSG
                        while content_ptr < content_end - 12:
                            hsg_pos = mm.find(GROUP_START, content_ptr, min(content_ptr + CHUNK_SIZE, content_end))
                            if hsg_pos == -1:
                                break
                            content_ptr = hsg_pos + 4
                            
                            # 解析组头部
                            group_content_size = struct.unpack('>I', mm[content_ptr:content_ptr+4])[0]
                            content_ptr += 4
                            group_id = struct.unpack('>I', mm[content_ptr:content_ptr+4])[0]
                            content_ptr += 4
                            
                            if group_id != 2:  # 仅处理groupid=2
                                content_ptr += group_content_size
                                continue
                            
                            # 跳过坐标系描述（N+Description）
                            n = struct.unpack('>I', mm[content_ptr:content_ptr+4])[0]
                            content_ptr += 4 + n
                            
                            # 提取弧度坐标并转度
                            lon_rad = struct.unpack('>d', mm[content_ptr:content_ptr+8])[0]
                            content_ptr += 8
                            lat_rad = struct.unpack('>d', mm[content_ptr:content_ptr+8])[0]
                            content_ptr += 8
                            height = struct.unpack('>d', mm[content_ptr:content_ptr+8])[0]
                            content_ptr += 8
                            
                            # 弧度转度
                            longitude = lon_rad * RTD
                            latitude = lat_rad * RTD
                            
                            # 转换北京时间并格式化
                            beijing_time = parse_beijing_time(block_seconds, block_micro)
                            time_str = beijing_time.strftime('%Y-%m-%d %H:%M:%S.%f')
                            
                            # 立即写入CSV，不缓存
                            csv_writer.writerow([latitude, longitude, time_str])
                        
                        # 跳过当前块剩余内容+结束标识
                        ptr = content_end + 4
                    
                    except Exception as e:
                        # 跳过损坏数据，不中断整体处理
                        ptr += CHUNK_SIZE
                        continue
    
    print(f"✅ 处理完成：{xse_file} → {output_csv}")
    return str(output_csv)

def batch_process_xse(xse_dir, output_dir, num_workers=None):
    """批量处理目录下所有XSE文件（多进程）"""
    # 创建输出目录
    Path(output_dir).mkdir(exist_ok=True, parents=True)
    
    # 获取所有XSE文件
    xse_files = list(Path(xse_dir).glob("*.xse"))
    if not xse_files:
        print("❌ 未找到XSE文件")
        return
    
    # 多进程处理（默认使用CPU核心数）
    num_workers = num_workers or multiprocessing.cpu_count()
    with multiprocessing.Pool(num_workers) as pool:
        # 构建参数（每个文件对应一个任务）
        tasks = [(str(f), output_dir) for f in xse_files]
        # 异步执行
        pool.starmap(process_single_xse, tasks)
    
    print(f"\n🎉 全部处理完成！共处理 {len(xse_files)} 个文件，输出目录：{output_dir}")

if __name__ == "__main__":
    # 配置参数（替换为你的路径）
    XSE_DIR = r"c:\\Temp"  # 存放所有XSE文件的目录
    OUTPUT_DIR = r"D:\\Temp"  # CSV输出目录
    NUM_WORKERS = 12  # 进程数（根据CPU核心数调整，如8核设为8）
    
    # 启动批量处理
    batch_process_xse(XSE_DIR, OUTPUT_DIR, NUM_WORKERS)