POO（面向对象编程）
本练习涉及面向对象编程、property属性和魔术方法的使用。目标是实现一个 Signal1D 类，用于表示一个离散时间信号，使其可以像对象一样进行各种操作。
信号由以下两个核心属性定义：一个numpy数据向量，存储在属性 .raw 中；一个采样频率，存储在属性 .fs 中。采样频率的物理意义是每秒采集的样本数，单位是赫兹（Hz）。
在开发过程中还需要注意以下实践要求：在git和uv仓库中对工作进行版本控制；在模块文件上执行 ruff、black 和 mypy 进行代码质量检查；在项目根目录创建测试脚本来验证模块功能。
基础类的构建
创建类及其构造函数
首先定义 Signal1D 类的基本结构。在Python中，__init__ 方法是类的构造函数，在创建对象实例时自动调用。self 参数代表实例本身，通过它可以访问和设置实例的属性。
import numpy as np
class Signal1D:
def __init__(self, data, fs: float):
self.raw = np.asarray(data)
self.fs = fs
np.asarray(data) 的作用是将输入数据转换为numpy数组。与 np.array() 不同的是，如果输入本身已经是ndarray，asarray 不会创建副本，而是直接返回原数组的引用，这样更高效。fs: float 是类型注解，表明参数 fs 应该是浮点数类型。
验证类的基本功能
在编写代码的过程中应该逐步验证其有效性，这是良好的开发习惯。
# 测试代码
s = Signal1D([1, 2, 3, 4, 5], fs=100.0)
print(s.raw) # [1 2 3 4 5]
print(s.fs) # 100.0
执行结果：
[1 2 3 4 5]
100.0
创建了一个包含5个采样点的信号，采样频率为100Hz。s.raw 返回存储的数据数组，s.fs 返回采样频率。
确保数据类型的一致性
构造函数应该能够接受 np.array 能接受的任何输入类型（列表、元组、numpy数组等），并统一转换为 np.ndarray 类型。这种设计提高了类的灵活性和易用性。
import numpy as np
class Signal1D:
def __init__(self, data, fs: float):
self.raw = np.asarray(data) # 自动将任何输入转换为 np.ndarray
self.fs = fs
测试不同输入类型的转换效果：
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0) # 列表
s2 = Signal1D((4, 5, 6), fs=100.0) # 元组
s3 = Signal1D(np.array([7, 8, 9]), fs=100.0) # numpy数组
print(type(s1.raw)) # <class 'numpy.ndarray'>
print(type(s2.raw)) # <class 'numpy.ndarray'>
print(type(s3.raw)) # <class 'numpy.ndarray'>
执行结果：
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
无论输入是列表、元组还是numpy数组，self.raw 最终都是 numpy.ndarray 类型，保证了后续处理的一致性。
参数有效性验证
采样频率 fs 从物理意义上必须严格为正数，因为负数或零的采样频率没有实际意义。需要在构造函数中添加验证逻辑。
import numpy as np
class Signal1D:
def __init__(self, data, fs: float):
if fs <= 0:
raise ValueError("采样频率 fs 必须严格为正")
self.raw = np.asarray(data)
self.fs = fs
raise 语句用于主动抛出异常。ValueError 是Python内置的异常类型，适用于参数值不合法的情况。将验证逻辑放在赋值操作之前，可以确保对象在创建失败时不会处于不一致的状态。
测试正常情况：
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0) # 正常
print(s1.fs)
执行结果：
100.0
异常处理机制
如果传入的采样频率不满足条件（小于等于零），构造函数应该抛出异常，阻止创建无效的对象。
s2 = Signal1D([1, 2, 3], fs=-10.0) # 如果 fs < 0, 抛出异常
执行结果：
---------------------------------------------------------------------------
ValueError Traceback (most recent call last)
...
----> 1 s2 = Signal1D([1, 2, 3], fs=-10.0) # 抛出 ValueError
4 def __init__(self, data, fs: float):
5 if fs <= 0:
----> 6 raise ValueError("采样频率 fs 必须严格为正")
7 self.raw = np.asarray(data)
8 self.fs = fs
ValueError: 采样频率 fs 必须严格为正
异常信息清晰地指出了问题所在，便于调试和错误处理。在实际应用中，调用者可以使用 try-except 语句捕获这个异常并进行适当处理。
添加duration属性
使用 @property 装饰器可以将方法转换为只读属性。duration 表示信号的持续时间，以秒为单位，计算方式是采样点数除以采样频率。
import numpy as np
class Signal1D:
def __init__(self, data, fs: float):
if fs <= 0:
raise ValueError("采样频率 fs 必须严格为正")
self.raw = np.asarray(data)
self.fs = fs
@property
def duration(self) -> float:
"""返回信号持续时间 (秒)"""
return len(self.raw) / self.fs
@property 装饰器的作用是让方法可以像属性一样被访问，即使用 s.duration 而不是 s.duration()。这种设计使得接口更加简洁直观。-> float 是返回值类型注解，表明该方法返回浮点数。
持续时间的计算公式为：
其中 N 是采样点数，f_s 是采样频率。例如，100个采样点、采样频率100Hz的信号，持续时间为1秒。由于这是一个计算属性而非存储属性，每次访问时都会根据当前的 raw 和 fs 值重新计算，确保结果始终正确。将其设置为只读（只有getter没有setter）是因为持续时间是由数据长度和采样频率决定的派生量，不应该被直接修改。
将raw改为property并添加数据验证
之前 raw 是一个普通的实例属性，可以被任意赋值。为了增加数据验证功能（例如拒绝空数组），需要将其改为property。实际的数据存储在私有属性 ._data 中（Python约定以单下划线开头的属性为内部使用），而 raw 作为访问接口。
@property
def raw(self) -> np.ndarray:
"""返回信号数据"""
return self._data
@raw.setter
def raw(self, value):
"""设置信号数据，拒绝空数组"""
arr = np.asarray(value)
if arr.size == 0:
raise ValueError("数据不能为空")
self._data = arr
@property 装饰器定义了getter方法，当访问 s.raw 时被调用。@raw.setter 装饰器定义了setter方法，当执行 s.raw = value 时被调用。在setter中，先将输入转换为numpy数组，然后检查数组大小，如果为空则抛出异常，否则才赋值给私有属性 _data。
构造函数也需要相应修改，使用私有属性并添加空数组检查：
def __init__(self, data, fs: float):
if fs <= 0:
raise ValueError("采样频率 fs 必须严格为正")
self._data = np.asarray(data)
if self._data.size == 0:
raise ValueError("数据不能为空")
self.fs = fs
同时，duration 属性的实现也要改为使用 self._data：
@property
def duration(self) -> float:
"""返回信号持续时间 (秒)"""
return len(self._data) / self.fs
测试property的读写功能和验证逻辑：
# 测试代码
s = Signal1D([1, 2, 3], fs=100.0)
print(s.raw) # [1 2 3]
s.raw = [4, 5, 6, 7] # 正常赋值
print(s.raw) # [4 5 6 7]
try:
s.raw = [] # 空数组，抛出异常
except ValueError as e:
print(f"捕获异常: {e}")
执行结果：
[1 2 3]
[4 5 6 7]
捕获异常: 数据不能为空
添加norm属性
norm 是一个只读属性，用于计算向量的范数（均方根值，RMS）。计算公式为：
其中 N 是采样点数，x_i 是第 i 个采样值。这个值表示信号的平均能量水平。
@property
def norm(self) -> float:
"""返回向量的范数: sqrt(1/N * sum(x_i^2))"""
return np.sqrt(np.mean(self._data ** 2))
self._data ** 2 对数组中每个元素求平方，np.mean() 计算平均值，np.sqrt() 开平方根。这是一个只读属性，没有定义setter。
测试范数计算：
# 测试代码
s = Signal1D([1, 2, 3, 4, 5], fs=100.0)
print(s.norm) # sqrt((1+4+9+16+25)/5) = sqrt(11) = 3.3166
执行结果：
3.3166247903554
手动验证：(1^2 + 2^2 + 3^2 + 4^2 + 5^2) / 5 = (1+4+9+16+25)/5 = 55/5 = 11，\sqrt{11} \approx 3.3166。
魔术方法
魔术方法（Magic Methods）是Python中以双下划线开头和结尾的特殊方法，它们定义了对象在特定操作下的行为。通过实现这些方法，可以让自定义类的对象像内置类型一样自然地使用。
实现__repr__方法
__repr__ 方法定义了对象的字符串表示形式，当使用 print() 或在交互式环境中直接输入对象名时被调用。目标是显示信号的有用信息，包括点数、采样频率、持续时间等。
def __repr__(self) -> str:
return f"Signal1D(points={len(self._data)}, fs={self.fs} Hz, duration={self.duration} s)"
f-string 格式化字符串中，{len(self._data)} 插入采样点数，{self.fs} 插入采样频率，{self.duration} 插入持续时间。返回的字符串清晰地展示了信号的关键参数。
测试：
# 测试代码
s = Signal1D([1, 2, 3, 4, 5], fs=100.0)
print(s) # Signal1D(points=5, fs=100.0 Hz, duration=0.05 s)
执行结果：
Signal1D(points=5, fs=100.0 Hz, duration=0.05 s)
5个采样点、100Hz采样频率，持续时间为 5/100 = 0.05 秒。
实现__len__方法
__len__ 方法使对象可以被 len() 函数调用，返回信号的长度（采样点数量）。
def __len__(self) -> int:
return len(self._data)
实现后，可以直接对 Signal1D 对象使用 len() 函数，而不必访问 .raw 属性再取长度。
测试：
# 测试代码
s = Signal1D([1, 2, 3, 4, 5], fs=100.0)
print(len(s)) # 5
执行结果：
5
实现__getitem__方法
如果想用类似numpy数组的方式 s[i] 访问元素，必须先访问 .raw 属性，即 s.raw[i]，这不太方便。实现 __getitem__ 方法可以让对象直接支持索引和切片操作。
def __getitem__(self, index):
return self._data[index]
当使用 s[index] 语法时，Python会自动调用 s.__getitem__(index)。这里直接将索引操作委托给内部的numpy数组 _data，因此支持numpy的所有索引方式，包括正索引、负索引和切片。
测试各种索引方式：
# 测试代码
s = Signal1D([10, 20, 30, 40, 50], fs=100.0)
print(s[0]) # 10
print(s[2]) # 30
print(s[-1]) # 50
print(s[1:4]) # [20 30 40] 切片也支持
执行结果：
10
30
50
[20 30 40]
s[0] 访问第一个元素，s[-1] 访问最后一个元素（Python支持负索引从末尾计数），s[1:4] 返回索引1到3的切片（不包含索引4）。由于底层是numpy数组，所有numpy的高级索引功能都可以使用。
实现__add__方法
__add__ 方法定义了加法运算符 + 的行为，使两个 Signal1D 对象可以直接相加。基本用法如下：
s1 = Signal1D(...)
s2 = Signal1D(...)
s3 = s1 + s2
print(s3)
初始版本的实现直接将两个信号的数据相加，并返回一个新的 Signal1D 对象：
def __add__(self, other: "Signal1D") -> "Signal1D":
return Signal1D(self._data + other._data, self.fs)
类型注解 other: "Signal1D" 使用字符串形式是因为在类定义内部引用自身类型时，类尚未完全定义，需要用字符串延迟求值。self._data + other._data 利用numpy数组的逐元素加法，结果仍是numpy数组。返回值是一个新的 Signal1D 实例，采样频率继承自 self。
测试基本加法：
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0)
s2 = Signal1D([4, 5, 6], fs=100.0)
s3 = s1 + s2
print(s3) # Signal1D(points=3, fs=100.0 Hz, duration=0.03 s)
print(s3.raw) # [5 7 9]
执行结果：
Signal1D(points=3, fs=100.0 Hz, duration=0.03 s)
[5 7 9]
逐元素相加：1+4=5，2+5=7，3+6=9。
添加加法运算的验证逻辑
两个信号相加在物理上需要满足条件：采样频率必须相同（否则时间尺度不一致），信号长度必须相同（否则无法逐点对应）。需要在 __add__ 方法中添加验证，不满足条件时抛出 ValueError。
def __add__(self, other: "Signal1D") -> "Signal1D":
if self.fs != other.fs:
raise ValueError("采样频率必须相同")
if len(self) != len(other):
raise ValueError("信号长度必须相同")
return Signal1D(self._data + other._data, self.fs)
测试正常情况和异常情况：
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0)
s2 = Signal1D([4, 5, 6], fs=100.0)
s3 = s1 + s2
print(s3.raw) # [5 7 9]
# 测试不同采样频率
try:
s4 = Signal1D([1, 2, 3], fs=200.0)
s5 = s1 + s4
except ValueError as e:
print(f"捕获异常: {e}") # 采样频率必须相同
# 测试不同长度
try:
s6 = Signal1D([1, 2, 3, 4], fs=100.0)
s7 = s1 + s6
except ValueError as e:
print(f"捕获异常: {e}") # 信号长度必须相同
执行结果：
[5 7 9]
捕获异常: 采样频率必须相同
捕获异常: 信号长度必须相同
确保返回值类型正确
加法运算的结果必须是一个 Signal1D 对象，而不是普通的numpy数组。可以通过 type() 和 isinstance() 来验证。
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0)
s2 = Signal1D([4, 5, 6], fs=100.0)
s3 = s1 + s2
print(type(s3)) # <class '__main__.Signal1D'>
print(isinstance(s3, Signal1D)) # True
print(s3) # Signal1D(points=3, fs=100.0 Hz, duration=0.03 s)
执行结果：
<class '__main__.Signal1D'>
True
Signal1D(points=3, fs=100.0 Hz, duration=0.03 s)
返回的 s3 是完整的 Signal1D 对象，拥有所有的属性和方法。
实现__mul__方法
__mul__ 方法定义了乘法运算符 * 的行为。这里实现的是信号与标量（数字）的乘法，即将信号的每个采样值乘以一个系数，常用于信号的放大或衰减。
def __mul__(self, scalar: float) -> "Signal1D":
if not isinstance(scalar, (int, float)):
raise TypeError("必须与标量相乘")
if scalar <= 0:
raise ValueError("标量必须为正")
return Signal1D(self._data * scalar, self.fs)
首先用 isinstance() 检查 scalar 是否为数字类型（整数或浮点数），如果不是则抛出 TypeError。然后检查标量是否为正数，这里的设计选择是只允许正数乘法（根据具体需求，也可以允许负数或零）。最后返回新的 Signal1D 对象，数据是原数据乘以标量。
测试标量乘法：
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0)
s2 = s1 * 2.5
print(s2.raw) # [2.5 5. 7.5]
print(s2) # Signal1D(points=3, fs=100.0 Hz, duration=0.03 s)
# 测试负标量
try:
s3 = s1 * (-1)
except ValueError as e:
print(f"捕获异常: {e}") # 标量必须为正
执行结果：
[2.5 5. 7.5]
Signal1D(points=3, fs=100.0 Hz, duration=0.03 s)
捕获异常: 标量必须为正
信号 [1, 2, 3] 乘以 2.5 得到 [2.5, 5.0, 7.5]。采样频率和持续时间保持不变，因为标量乘法只改变幅度，不改变时间特性。
方法
normalize方法
normalize 方法用于对信号进行归一化处理，使信号的幅值最大值变为1。这在信号处理中是常见操作，可以消除不同信号之间的幅度差异，便于比较和后续处理。
def normalize(self) -> "Signal1D":
"""归一化信号，使得幅值最大为1"""
max_val = np.max(np.abs(self._data))
return Signal1D(self._data / max_val, self.fs)
归一化的计算过程分为两步：首先用 np.abs(self._data) 取所有数据的绝对值（因为信号可能有正有负），然后用 np.max() 找到绝对值的最大值；接着将整个数据数组除以这个最大值。这样处理后，所有数据都会落在 [-1, 1] 范围内，且绝对值最大的点变为 \pm 1。方法返回一个新的 Signal1D 对象，原信号不被修改（不可变设计）。
测试归一化：
# 测试代码
s = Signal1D([2, 4, -8, 6], fs=100.0)
s_norm = s.normalize()
print(s_norm.raw) # [ 0.25 0.5 -1. 0.75]
print(np.max(np.abs(s_norm.raw))) # 1.0
执行结果：
[ 0.25 0.5 -1. 0.75]
1.0
原信号中绝对值最大的是 -8，因此所有值除以 8：2/8=0.25，4/8=0.5，-8/8=-1，6/8=0.75。归一化后最大绝对值确实为 1。
concatenate方法
concatenate 方法用于将两个信号首尾相连，生成一个更长的信号。这在拼接音频片段、组合不同信号段时非常有用。
def concatenate(self, other: "Signal1D") -> "Signal1D":
"""连接两个信号，采样频率必须相同"""
if self.fs != other.fs:
raise ValueError("采样频率必须相同")
return Signal1D(np.concatenate([self._data, other._data]), self.fs)
连接操作在物理上要求两个信号具有相同的采样频率，否则时间尺度不一致，连接后的信号没有意义。np.concatenate() 函数接受一个数组列表，将它们沿指定轴（默认为第一个轴）连接起来。返回的新信号长度等于两个原信号长度之和，持续时间也相应增加。
测试信号连接：
# 测试代码
s1 = Signal1D([1, 2, 3], fs=100.0)
s2 = Signal1D([4, 5, 6, 7], fs=100.0)
s3 = s1.concatenate(s2)
print(s3.raw) # [1 2 3 4 5 6 7]
print(s3) # Signal1D(points=7, fs=100.0 Hz, duration=0.07 s)
# 测试不同采样频率
try:
s4 = Signal1D([8, 9], fs=200.0)
s5 = s1.concatenate(s4)
except ValueError as e:
print(f"捕获异常: {e}") # 采样频率必须相同
执行结果：
[1 2 3 4 5 6 7]
Signal1D(points=7, fs=100.0 Hz, duration=0.07 s)
捕获异常: 采样频率必须相同
s1 有3个点，s2 有4个点，连接后 s3 有7个点。持续时间为 7/100 = 0.07 秒。尝试连接采样频率不同的信号会触发异常。
play方法
play 方法将信号作为音频播放。在Jupyter/Colab环境中，可以使用 IPython.display.Audio 类来实现音频播放功能。首先需要安装 sounddevice 库：
!pip install sounddevice
实现代码：
from IPython.display import Audio
def play(self):
"""播放信号（返回 Audio 对象）"""
return Audio(self._data, rate=int(self.fs))
Audio 类接受两个主要参数：self._data 是音频数据数组，rate 是采样率（必须是整数）。返回的 Audio 对象在Jupyter环境中会自动渲染为一个可交互的音频播放器。
测试播放功能，生成一个标准A4音符（440Hz）的正弦波：
# 测试代码 - 生成一个 440Hz 的正弦波（A4音符）并播放
fs = 44100 # 采样频率
duration = 1.0 # 持续1秒
t = np.linspace(0, duration, int(fs * duration), endpoint=False)
data = 0.5 * np.sin(2 * np.pi * 440 * t) # 440Hz 正弦波
s = Signal1D(data, fs=fs)
print(s)
s.play() # 会显示一个播放器，点击播放
采样频率 44100 Hz 是CD音质的标准采样率。np.linspace(0, duration, int(fs * duration), endpoint=False) 生成从0到1秒的时间轴，共44100个采样点。正弦波公式 0.5 \times \sin(2\pi \times 440 \times t) 生成频率为440Hz、幅度为0.5的正弦信号。440Hz对应音乐中的标准A4音符。执行 s.play() 后，在Notebook环境中会显示一个音频播放器控件，点击即可听到这个音符。
迭代器和闭包
本节将编写一种迷你处理流水线，涉及的数值计算任务包括：将信号分割成块、应用滤波器、计算时间计时、参数管理。
数据类ParametresFiltre
dataclass 是Python 3.7引入的装饰器，用于简化数据容器类的定义。它自动生成 __init__、__repr__、__eq__ 等方法，减少样板代码。创建一个 ParametresFiltre 数据类来管理滤波器参数，包含增益（浮点数）、块大小（整数）和可选的名称。
from dataclasses import dataclass
from typing import Optional
@dataclass
class ParametresFiltre:
gain: float
block_size: int
name: Optional[str] = None
@dataclass 装饰器自动根据类属性生成构造函数。gain: float 和 block_size: int 是类型注解，同时也定义了必需的参数。name: Optional[str] = None 表示这是一个可选参数，默认值为 None。Optional[str] 类型注解表示该值可以是字符串或 None。
测试数据类：
# 测试代码
p1 = ParametresFiltre(gain=0.8, block_size=512)
print(p1) # ParametresFiltre(gain=0.8, block_size=512, name=None)
p2 = ParametresFiltre(gain=1.2, block_size=256, name="低通滤波器")
print(p2) # ParametresFiltre(gain=1.2, block_size=256, name='低通滤波器')
执行结果：
ParametresFiltre(gain=0.8, block_size=512, name=None)
ParametresFiltre(gain=1.2, block_size=256, name='低通滤波器')
@dataclass 自动生成的 __repr__ 方法提供了清晰的字符串表示。
迭代器的基本结构
迭代器（iterator）是Python中实现迭代协议的对象。一个类要成为迭代器，必须实现两个方法：__iter__(self) 返回迭代器对象本身，__next__(self) 返回下一个元素或在没有更多元素时抛出 StopIteration 异常。迭代器使得对象可以在 for 循环中使用。
定义 BlockIterator 类的基本结构：
import numpy as np
class BlockIterator:
def __init__(self, signal: Signal1D, block_size: int):
self.signal = signal
self.block_size = block_size
self.index = 0
def __iter__(self):
return self
def __next__(self) -> np.ndarray:
raise StopIteration
__init__ 方法接收两个参数：要切割的信号（Signal1D 对象）和块大小（整数）。self.index 用于跟踪当前迭代位置，初始为0。
__iter__方法
__iter__ 方法是迭代协议的一部分，当对象被用于 for 循环或 iter() 函数时调用。最常见的实现是返回 self，表示对象本身就是迭代器。这是 for 循环的初始化步骤。
def __iter__(self):
"""返回 self，对象本身就是迭代器"""
return self
验证 __iter__ 返回自身：
# 测试代码
s = Signal1D([1, 2, 3, 4, 5], fs=100.0)
it = BlockIterator(s, 2)
print(iter(it) is it) # True
执行结果：
True
iter(it) 调用 it.__iter__()，返回 self，因此 iter(it) is it 为 True。
__next__方法
__next__ 方法负责返回序列中的下一个元素。在信号分块的场景中，它返回信号的下一个块。当所有块都已返回后，抛出 StopIteration 异常来通知 for 循环结束。
def __next__(self) -> np.ndarray:
"""返回信号的下一个块"""
if self.index >= len(self.signal):
raise StopIteration
block = self.signal.raw[self.index:self.index + self.block_size]
self.index += self.block_size
return block
首先检查当前索引是否已超过信号长度，如果是则抛出 StopIteration。否则，使用切片 self.signal.raw[self.index:self.index + self.block_size] 提取当前块，然后将索引前移一个块大小，最后返回提取的块。
测试迭代功能：
# 测试代码
s = Signal1D([1, 2, 3, 4, 5, 6], fs=100.0)
for bloc in BlockIterator(s, 2):
print(bloc)
执行结果：
[1 2]
[3 4]
[5 6]
信号被分成3个大小为2的块，for 循环依次获取每个块直到 StopIteration 被抛出。
处理不完整的最后一个块
当信号长度不能被块大小整除时，最后一个块的大小会小于指定的块大小。Python的切片操作自动处理这种情况：如果切片的结束索引超过数组长度，切片会返回到数组末尾为止的所有元素，不会报错。
def __next__(self) -> np.ndarray:
"""返回下一个块，最后一个块可能小于 block_size"""
if self.index >= len(self.signal):
raise StopIteration
# 切片自动处理最后一个块不足的情况
block = self.signal.raw[self.index:self.index + self.block_size]
self.index += self.block_size
return block
测试信号长度为7、块大小为3的情况：
# 测试代码 - 信号长度为7，块大小为3
s = Signal1D([1, 2, 3, 4, 5, 6, 7], fs=100.0)
for bloc in BlockIterator(s, 3):
print(f"块: {bloc}, 大小: {len(bloc)}")
执行结果：
块: [1 2 3], 大小: 3
块: [4 5 6], 大小: 3
块: [7], 大小: 1
前两个块大小为3，最后一个块只有1个元素，因为 7 = 3 \times 2 + 1。
迭代器的实际使用
完整的 BlockIterator 类可以直接在 for 循环中使用，遍历信号的所有块：
for bloc in BlockIterator(signal, 512):
print(bloc.shape)
测试更大规模的信号：
# 测试代码 - 按题目要求的方式使用
s = Signal1D(np.random.randn(2000), fs=44100.0)
for bloc in BlockIterator(s, 512):
print(bloc.shape)
执行结果：
(512,)
(512,)
(512,)
(464,)
信号共2000个采样点，块大小512。2000 = 512 \times 3 + 464，因此产生3个完整块和1个大小为464的最后块。bloc.shape 返回numpy数组的形状，一维数组显示为 (n,) 形式。
滤波器工厂函数
make_filtre 是一个工厂函数（factory function），它返回另一个函数。这种模式称为闭包（closure）：内部函数 filtre 可以访问外部函数 make_filtre 的参数 gain，即使外部函数已经返回。闭包使得我们可以创建具有预设参数的函数。
def make_filtre(gain: float):
"""返回一个对信号应用增益的函数"""
def filtre(signal):
return signal * gain
return filtre
make_filtre(gain) 不直接处理信号，而是返回一个新函数 filtre。这个返回的函数接受信号作为参数，将其乘以预设的增益值。gain 变量被"捕获"在闭包中，即使 make_filtre 执行完毕，filtre 仍然可以访问它。
测试工厂函数：
# 测试代码
filtre = make_filtre(0.5)
data = np.array([2, 4, 6, 8, 10])
result = filtre(data)
print(result) # [1. 2. 3. 4. 5.]
执行结果：
[1. 2. 3. 4. 5.]
所有值乘以0.5：2 \times 0.5 = 1，4 \times 0.5 = 2，以此类推。
滤波器作为单参数函数
通过 make_filtre 创建的 filtre 是一个单参数函数，只需要传入信号即可，增益已经预设好了。这种设计使得滤波器可以方便地与其他组件（如迭代器）配合使用。
# 测试代码
filtre = make_filtre(0.8)
# filtre 是一个单参数函数
print(callable(filtre)) # True
# 使用 filtre
data = np.array([10, 20, 30])
result = filtre(data)
print(result) # [ 8. 16. 24.]
执行结果：
True
[ 8. 16. 24.]
callable(filtre) 检查对象是否可调用，函数当然是可调用的。
将滤波器与 BlockIterator 配合使用，对每个块应用滤波：
# 也可以与 BlockIterator 配合使用
s = Signal1D([1, 2, 3, 4, 5, 6], fs=100.0)
for bloc in BlockIterator(s, 2):
filtered_bloc = filtre(bloc)
print(f"原始: {bloc}, 滤波后: {filtered_bloc}")
执行结果：
原始: [1 2], 滤波后: [0.8 1.6]
原始: [3 4], 滤波后: [2.4 3.2]
原始: [5 6], 滤波后: [4. 4.8]
每个块被迭代器提取后，立即通过滤波器处理。
扩展为低通滤波器
可以扩展 make_filtre 函数，增加低通滤波功能。低通滤波器允许低频信号通过，抑制高频成分（如噪声）。
from scipy import signal as sp_signal
def make_filtre(gain: float, lowpass: bool = False, cutoff: float = 0.5):
"""
返回一个滤波器函数
:param gain: 增益
:param lowpass: 是否启用低通滤波
:param cutoff: 截止频率（归一化，0到1之间，1表示奈奎斯特频率）
"""
def filtre(signal_data):
result = signal_data * gain
if lowpass:
# 设计低通滤波器 (Butterworth, 4阶)
b, a = sp_signal.butter(4, cutoff, btype='low')
result = sp_signal.filtfilt(b, a, result)
return result
return filtre
sp_signal.butter(4, cutoff, btype='low') 设计一个4阶Butterworth低通滤波器，返回滤波器系数 b 和 a。cutoff 是归一化截止频率，范围0到1，其中1对应奈奎斯特频率（采样频率的一半）。sp_signal.filtfilt 进行零相位滤波，即正向和反向各滤波一次，消除相位延迟。
测试低通滤波效果：
# 低通滤波器
filtre_lowpass = make_filtre(gain=1.0, lowpass=True, cutoff=0.3)
# 生成带噪声的信号
fs = 1000
t = np.linspace(0, 1, fs, endpoint=False)
clean_signal = np.sin(2 * np.pi * 5 * t) # 5Hz 正弦波
noise = 0.5 * np.sin(2 * np.pi * 100 * t) # 100Hz 噪声
noisy_signal = clean_signal + noise
# 应用低通滤波
filtered_signal = filtre_lowpass(noisy_signal)
print(f"原始信号方差: {np.var(noisy_signal):.4f}")
print(f"滤波后信号方差: {np.var(filtered_signal):.4f}")
原始信号包含5Hz的有用信号和100Hz的噪声。低通滤波器截止频率设为0.3（对应150Hz），可以让5Hz信号通过，同时抑制100Hz噪声。滤波后方差降低表明高频噪声被有效去除。
装饰器
计时装饰器
装饰器（decorator）是一种修改函数行为的技术。创建一个装饰器来计算函数执行时间，并将时间存储在列表中，便于后续分析。
import time
def timing_decorator(func):
"""装饰器：计算函数执行时间并存储在列表中"""
def wrapper(*args, **kwargs):
start = time.process_time()
result = func(*args, **kwargs)
end = time.process_time()
wrapper.times.append(end - start)
return result
wrapper.times = [] # 存储时间的列表
return wrapper
timing_decorator 接受一个函数 func 作为参数，返回一个新函数 wrapper。wrapper 在调用原函数前后记录时间，计算执行时长并存入 wrapper.times 列表。*args, **kwargs 使得 wrapper 可以接受任意参数并传递给原函数。
测试计时装饰器：
# 测试代码
@timing_decorator
def my_filtre(signal):
return signal * 0.8
data = np.random.randn(10000)
for _ in range(5):
my_filtre(data)
print(f"执行时间列表: {my_filtre.times}")
@timing_decorator 语法等价于 my_filtre = timing_decorator(my_filtre)。每次调用 my_filtre 时，实际执行的是 wrapper，执行时间被记录。
函数作为对象
Python中函数是一等对象（first-class object），可以像其他对象一样被赋值、传递、添加属性。wrapper.times = [] 给函数对象添加了一个 times 属性，用于存储时间数据。
# 测试代码 - 验证函数属性
@timing_decorator
def my_filtre(signal):
return signal * 0.8
# 函数是对象，可以有属性
print(type(my_filtre)) # <class 'function'>
print(hasattr(my_filtre, 'times')) # True
print(my_filtre.times) # []
# 执行几次
my_filtre(np.array([1, 2, 3]))
my_filtre(np.array([4, 5, 6]))
# 查看时间列表
print(my_filtre.times) # [时间1, 时间2]
执行结果：
<class 'function'>
True
[]
[时间1, 时间2]
使用process_time
time.process_time() 返回当前进程的CPU时间，不包括睡眠时间，比 time.time() 更适合测量计算密集型任务的执行时间。time.time() 返回的是墙钟时间（wall-clock time），会受到系统负载、I/O等待等因素影响。
import time
def timing_decorator(func):
"""
装饰器：计算函数执行时间并存储在列表中
使用 time.process_time() 测量CPU时间（更精确）
"""
def wrapper(*args, **kwargs):
start = time.process_time() # CPU时间，比time.time()更精确
result = func(*args, **kwargs)
end = time.process_time()
elapsed = end - start
wrapper.times.append(elapsed)
return result
wrapper.times = []
return wrapper
测试并统计执行时间：
# 测试代码
@timing_decorator
def my_filtre(signal):
# 模拟一些计算
return signal * 0.8 + np.sin(signal)
# 执行多次
data = np.random.randn(100000)
for i in range(10):
my_filtre(data)
print(f"执行次数: {len(my_filtre.times)}")
print(f"时间列表: {my_filtre.times}")
print(f"平均时间: {np.mean(my_filtre.times):.6f} 秒")
print(f"总时间: {np.sum(my_filtre.times):.6f} 秒")
完整的信号处理流水线
将所有组件整合在一起：Signal1D 类用于表示信号，ParametresFiltre 数据类管理参数，BlockIterator 迭代器分块处理，timing_decorator 装饰器计时，make_filtre 工厂函数创建滤波器。
import numpy as np
import time
from dataclasses import dataclass
from typing import Optional
from IPython.display import Audio
# ============== Signal1D 类 ==============
class Signal1D:
# ... 完整实现如前所述 ...
# ============== 数据类 ==============
@dataclass
class ParametresFiltre:
gain: float
block_size: int
name: Optional[str] = None
# ============== 迭代器 ==============
class BlockIterator:
def __init__(self, signal: Signal1D, block_size: int):
self.signal = signal
self.block_size = block_size
self.index = 0
def __iter__(self):
return self
def __next__(self) -> np.ndarray:
if self.index >= len(self.signal):
raise StopIteration
block = self.signal.raw[self.index:self.index + self.block_size]
self.index += self.block_size
return block
# ============== 装饰器 ==============
def timing_decorator(func):
def wrapper(*args, **kwargs):
start = time.process_time()
result = func(*args, **kwargs)
end = time.process_time()
wrapper.times.append(end - start)
return result
wrapper.times = []
return wrapper
# ============== 滤波器工厂 ==============
def make_filtre(gain: float):
@timing_decorator
def filtre(signal):
return signal * gain
return filtre
使用这套流水线处理信号时，首先用参数类配置滤波器参数，然后用工厂函数创建滤波器，接着用迭代器将信号分块，对每个块应用滤波器，最后将滤波后的块重新组装。装饰器自动记录每个块的处理时间，便于性能分析。
完整流水线测试
将所有组件组装起来，实现一个完整的信号处理流水线：创建参数、生成信号、创建滤波器、分块处理、合并结果。
# 测试代码 - 完整流水线
# 1. 创建参数
params = ParametresFiltre(gain=0.8, block_size=512, name="增益滤波器")
print(f"参数: {params}")
# 2. 创建信号
fs = 44100
duration = 1.0
t = np.linspace(0, duration, int(fs * duration), endpoint=False)
data = np.sin(2 * np.pi * 440 * t) # 440Hz 正弦波
signal = Signal1D(data, fs=fs)
print(f"信号: {signal}")
# 3. 创建滤波器
filtre = make_filtre(params.gain)
# 4. 处理流水线：遍历块并应用滤波器
filtered_blocks = []
for bloc in BlockIterator(signal, params.block_size):
filtered_bloc = filtre(bloc)
filtered_blocks.append(filtered_bloc)
# 5. 合并结果
filtered_data = np.concatenate(filtered_blocks)
filtered_signal = Signal1D(filtered_data, fs=fs)
# 6. 显示结果
print(f"滤波后信号: {filtered_signal}")
print(f"处理块数: {len(filtre.times)}")
print(f"总处理时间: {sum(filtre.times):.6f} 秒")
print(f"平均每块时间: {np.mean(filtre.times):.6f} 秒")
# 7. 播放滤波后的信号
filtered_signal.play()
流水线的执行流程如下：
第一步使用 ParametresFiltre 数据类创建参数对象，包含增益0.8、块大小512和滤波器名称。数据类自动提供了清晰的字符串表示。
第二步生成测试信号。采样频率44100Hz是CD音质标准，持续时间1秒，共44100个采样点。使用 np.linspace 生成时间轴，然后生成440Hz的正弦波（标准A4音符）。
第三步通过工厂函数 make_filtre 创建滤波器。由于在 make_filtre 内部使用了 @timing_decorator 装饰器，返回的滤波器函数会自动记录每次调用的执行时间。
第四步是核心处理循环。BlockIterator 将信号分成512个采样点的块，for 循环遍历每个块，对每个块应用滤波器，将滤波后的块添加到列表中。每次调用 filtre(bloc) 时，装饰器自动记录执行时间到 filtre.times 列表。
第五步使用 np.concatenate 将所有滤波后的块重新合并成完整的数组，然后创建新的 Signal1D 对象。
第六步显示处理结果。len(filtre.times) 给出处理的块数，44100 / 512 \approx 86 个块（最后一个块可能不完整）。sum(filtre.times) 给出总处理时间，np.mean(filtre.times) 给出平均每块的处理时间。这些性能数据对于优化流水线和评估实时处理能力非常有用。
第七步调用 play() 方法播放滤波后的信号。由于增益为0.8，滤波后的信号幅度是原信号的80%，听起来会稍微安静一些。
这个流水线展示了面向对象编程、迭代器模式、闭包、装饰器等Python高级特性的综合应用，构成了一个模块化、可扩展、可测量的信号处理框架。