Astropy时间转换性能瓶颈问题及优化方案咨询

我写了一个从二进制“包”流读取数据的脚本，每个包包含多个参数，读取的参数存入对应包的字典，再把字典加入代表包流的数组，最终将这个字典数组写入CSV文件。

数据里包含CUC7格式的时间，以GPS时间的coarse/fine整数部分存储，需要转成UTC ISO字符串。最初用的代码是：

from astropy.time import Time

def cuc2gps_time(coarse, fine):
    return Time(coarse + fine / (2**24), format='gps')


def gps2utc_time(gps):
    return Time(gps, format='isot', scale='utc')

问题是这两个时间转换占了脚本总运行时间的90%，剩下的读取二进制文件、解码15个其他参数、写入CSV只占10%。我改成批量处理Numpy数组后，总耗时降了一些，但还是很慢：读取几个GB文件里的大量包并写入CSV要几十秒，去掉时间转换后只需要几秒。

现在问两个问题：

1. Astropy的时间转换这么慢正常吗？我的用法有没有问题？有没有更合适的替代库？
2. 能不能优化现有实现？我怀疑剩下的for循环开销很大，但找不到替代方案。

优化后的批量处理代码如下：

import numpy as np

while end_not_reached:

    # 读取1个包
    # (...)

    nb_packets_discover
    end_not_reached = ... # 布尔值

    # 批量处理提升性能
    if not nb_packets % 1000 or not end_not_reached:
        # 将CUC7时间转换为UTC时间
        all_coarse = np.array([packet['lobt_coarse'] for packet in packets])
        all_fine = np.array([packet['lobt_fine'] for packet in packets])
        all_gps = cuc2gps_time(all_coarse, all_fine)
        all_utc = gps2utc_time(all_gps)

        # 将时间加入包字典
        for packet, gps_time, utc_time in zip(packets, all_gps, all_utc):
            packet.update({'gps_time': gps_time, 'utc_time': utc_time})

问题1：Astropy时间转换的性能问题及替代方案 #

Astropy的Time类偏向天文领域的高精度时间处理，内部做了大量校验、时间尺度转换和格式兼容工作，所以单条或小批量转换时性能确实不算高，你的情况属于正常现象——尤其是处理百万级以上时间点时，Astropy的开销会被放大。

你的用法里有个明显问题：gps2utc_time函数重复创建了Time对象，其实可以直接在cuc2gps_time的结果上做尺度转换，不用再初始化一次Time。合并后的代码如下：

def cuc2utc_time(coarse, fine):
    gps_time = Time(coarse + fine / (2**24), format='gps')
    return gps_time.utc.isot

这能减少一次Time对象的创建开销。

如果追求极致性能，推荐用numpy+datetime组合，或直接用pytz处理时区转换，避开Astropy的重型封装：

• 直接基于GPS起始时间和闰秒差计算UTC时间（截至2024年，GPS比UTC快18秒，需根据数据时间跨度调整闰秒数）：

import numpy as np
from datetime import datetime, timedelta

# 预定义GPS起始时间和闰秒差
GPS_EPOCH = datetime(1980, 1, 6, 0, 0, 0)
LEAP_SECONDS = 18

def cuc2utc_fast(coarse, fine):
    # 计算总GPS秒数
    gps_seconds = coarse + fine / (2**24)
    # 转UTC：GPS时间 = UTC时间 + 闰秒
    utc_seconds = gps_seconds - LEAP_SECONDS
    # 转datetime对象并生成ISO字符串
    utc_datetime = GPS_EPOCH + timedelta(seconds=utc_seconds)
    return utc_datetime.isoformat()

# 批量处理示例
all_gps_seconds = all_coarse + all_fine / (2**24)
all_utc_seconds = all_gps_seconds - LEAP_SECONDS
all_utc_datetime = GPS_EPOCH + np.array(all_utc_seconds, dtype='timedelta64[s]')
all_utc_isot = np.array([dt.isoformat() for dt in all_utc_datetime])

这种方法完全避开Astropy，性能能提升一个数量级，但需注意维护闰秒表以适配不同时间范围的数据。

另外，pandas的to_datetime也支持批量转换GPS时间，性能表现也不错：

import pandas as pd

def cuc2utc_pandas(coarse, fine):
    gps_seconds = coarse + fine / (2**24)
    utc_dt = pd.to_datetime(gps_seconds, unit='s', origin='1980-01-06') - pd.Timedelta(seconds=LEAP_SECONDS)
    return utc_dt.dt.isoformat().to_numpy()

问题2：优化现有实现的循环开销 #

你当前批量处理后仍用for循环逐个更新字典，包数量大时开销确实可观，可尝试以下优化：

1. 直接索引赋值替代字典update
   不要用update方法，直接通过索引给字典字段赋值，能降低开销：

   # 批量计算时间
   all_coarse = np.array([p['lobt_coarse'] for p in packets])
   all_fine = np.array([p['lobt_fine'] for p in packets])
   all_gps = all_coarse + all_fine/(2**24)
   all_utc_isot = cuc2utc_fast(all_coarse, all_fine)
   
   # 批量赋值
   for i in range(len(packets)):
       packets[i]['gps_time'] = all_gps[i]
       packets[i]['utc_time'] = all_utc_isot[i]
   

2. 用结构化数组替代字典列表
   放弃字典列表格式，改用numpy结构化数组或pandas DataFrame存储数据，彻底消除Python层面的循环开销：

   # 定义结构化数组的 dtype
   dtype = [
       ('lobt_coarse', 'int64'),
       ('lobt_fine', 'int64'),
       # 其他参数的 dtype...
       ('gps_time', 'float64'),
       ('utc_time', 'U20')
   ]
   packet_array = np.array([tuple(p.values()) for p in packets], dtype=dtype)
   
   # 批量计算并赋值时间
   packet_array['gps_time'] = packet_array['lobt_coarse'] + packet_array['lobt_fine']/(2**24)
   packet_array['utc_time'] = cuc2utc_fast(packet_array['lobt_coarse'], packet_array['lobt_fine'])
   
   # 直接写入CSV
   np.savetxt('output.csv', packet_array, fmt='%s', delimiter=',', header=','.join(packet_array.dtype.names))
   

3. 边读边处理，减少内存占用
   若文件过大，无需将所有包存入内存，每读一批（如10000个）就处理一批并写入CSV：

   import csv
   
   with open('output.csv', 'w', newline='') as f:
       writer = csv.DictWriter(f, fieldnames=['lobt_coarse', 'lobt_fine', ..., 'gps_time', 'utc_time'])
       writer.writeheader()
   
       batch = []
       while end_not_reached:
           # 读取单个包并转为字典
           packet = read_one_packet()
           batch.append(packet)
   
           if len(batch) >= 10000 or not end_not_reached:
               # 批量处理时间
               all_coarse = np.array([p['lobt_coarse'] for p in batch])
               all_fine = np.array([p['lobt_fine'] for p in batch])
               all_gps = all_coarse + all_fine/(2**24)
               all_utc = cuc2utc_fast(all_coarse, all_fine)
   
               # 赋值后写入CSV
               for i in range(len(batch)):
                   batch[i]['gps_time'] = all_gps[i]
                   batch[i]['utc_time'] = all_utc[i]
               writer.writerows(batch)
   
               # 清空批次
               batch = []
   

内容的提问来源于stack exchange，提问作者Guiux