Python分块处理信号高低状态：跨块状态传递实现问询

核心思路 #

要实现跨块的信号状态计算，必须在循环外部维护全局状态变量，记录上一个块结束时的关键信息：

• 上一次上升沿的全局索引
• 当前是否处于高电平状态
• 未完成的高电平长度（块末尾为高电平时，需带到下一块继续计算）

完整实现代码 #

import numpy as np

data = np.array([0,1,1,0,1,1,1,0,0,0,0,0,1,1,1,1,1,1,0,0,0,0,0,0,0,1,1,1,0,0,0,0,1,1,1,1,0,0])
samples_to_process = 8
threshold = 0.5

# 全局状态变量，用于跨块传递
prev_start_global_idx = None     # 上一次上升沿的全局索引
in_high_state = False            # 当前是否处于高电平
unfinished_high_length = 0       # 未完成的高电平长度（块末尾是高电平时）

for block_start in range(0, data.size, samples_to_process):
    block = data[block_start:block_start + samples_to_process]
    block_length = len(block)
    print(f"当前处理块: {block}")

    # 生成块内高低电平掩码，检测上升沿和下降沿
    is_high = block > threshold
    rising_edges = np.where(np.diff(np.concatenate([[False], is_high])))[0]
    falling_edges = np.where(np.diff(np.concatenate([is_high, [False]])))[0]

    # 处理上一块遗留的高电平状态
    if in_high_state:
        if len(falling_edges) > 0:
            # 当前块内有下降沿，完成高电平长度计算
            high_state_length = unfinished_high_length + falling_edges[0]
            print(f"high_state_length = {high_state_length}")
            # 更新状态
            in_high_state = False
            falling_edges = falling_edges[1:]
        else:
            # 当前块全是高电平，继续累积长度
            unfinished_high_length += block_length
            continue

    # 处理当前块内的高低电平对
    edge_idx = 0
    while edge_idx < len(rising_edges):
        curr_start_global = block_start + rising_edges[edge_idx]
        
        # 计算上升沿间隔
        if prev_start_global_idx is not None:
            length_between_high_states = curr_start_global - prev_start_global_idx
            print(f"length_between_high_states = {length_between_high_states}")
        
        # 匹配对应下降沿
        if edge_idx < len(falling_edges):
            curr_end_global = block_start + falling_edges[edge_idx]
            high_state_length = curr_end_global - curr_start_global
            print(f"high_state_length = {high_state_length}")
            edge_idx += 1
        else:
            # 块末尾是高电平，记录未完成状态
            unfinished_high_length = block_length - rising_edges[edge_idx]
            in_high_state = True
            break
        
        prev_start_global_idx = curr_start_global
        edge_idx += 1
    print("---")

代码说明 #

1. 全局状态维护：循环外初始化的变量衔接块与块之间的状态，避免计算断裂。
2. 遗留状态处理：若上一块结束时处于高电平，优先在当前块中寻找下降沿完成计算；若当前块无下降沿，则累积长度到下一块。
3. 边缘检测与全局索引转换：用np.diff检测电平跳变，将块内索引转为全局索引，确保计算的准确性。
4. 结果输出：每次计算出目标值后立即打印，与示例逻辑一致。

测试验证 #

运行代码后，第一次处理块[0 1 1 0 1 1 1 0]时，输出符合示例要求：

high_state_length = 2
length_between_high_states = 3
high_state_length = 3

第二次处理块[0 0 0 0 1 1 1 1]时，输出：

length_between_high_states = 9
high_state_length = 4

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