Python中按时间轴统计3D数组内连续3个及以上1的事件数

需求回顾 #

给定3D xarray.DataArray（维度：time(155) × rlat(6) × rlon(4)），数组值仅为0和1。需要对每个(rlat, rlon)空间点，统计连续3个及以上1的事件次数（连续3个或更长的1序列算作1次事件），最终输出2D统计结果数组。

解决方案代码 #

核心思路是对每个空间点的时间序列，逐元素遍历识别连续1的块，统计长度≥3的块数量。利用xarray.apply_ufunc实现批量处理：

import xarray as xr
import numpy as np
import pandas as pd

# 模拟用户的3D DataArray（可替换为实际数据）
np.random.seed(42)
da = xr.DataArray(
    np.random.randint(0, 2, size=(155, 6, 4)),
    dims=['time', 'rlat', 'rlon'],
    coords={'time': pd.date_range('2020-01-01', periods=155),
            'rlat': np.arange(6),
            'rlon': np.arange(4)}
)

# 定义统计连续事件的函数
def count_valid_runs(arr):
    """统计1D数组中连续3个及以上1的事件数"""
    current_run_length = 0
    event_count = 0
    for val in arr:
        if val == 1:
            current_run_length += 1
        else:
            # 遇到0时，检查之前的连续序列长度是否达标
            if current_run_length >= 3:
                event_count += 1
            current_run_length = 0
    # 处理序列末尾未被0截断的连续1
    if current_run_length >= 3:
        event_count += 1
    return event_count

# 批量应用函数到每个空间点的时间序列
event_stats = xr.apply_ufunc(
    count_valid_runs,
    da,
    input_core_dims=[['time']],  # 指定输入的核心维度为time
    vectorize=True,  # 自动向量化处理每个空间点
    output_core_dims=[[]],  # 输出为标量（每个空间点对应一个值）
    dask='allowed'  # 支持dask懒加载数组（可选）
)

# 查看2D统计结果
print(event_stats)

代码说明 #

1. count_valid_runs函数：逐元素遍历时间序列，维护当前连续1的长度。每次遇到0时判断之前的连续长度是否≥3，达标则事件数+1；遍历结束后检查末尾的连续序列。
2. xr.apply_ufunc：将1D统计函数批量映射到每个(rlat, rlon)对应的时间序列，自动处理维度对齐，避免手动循环空间维度。
3. 结果维度：输出的event_stats为2D数组，维度与输入的rlat、rlon完全一致，每个值对应该空间点的有效事件数。

避免常见误区 #

如果直接用rolling(time=3).sum() >=3标记连续窗口，会导致同一个连续事件被多次计数（比如连续5个1会生成3个符合条件的窗口），无法得到正确的独立事件数量。上述遍历方法则精准统计每个独立的连续事件。

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