Procesado por bloques

Normas aplicables: IEC 61260 IEC 61672Referencias: Oppenheim y Schafer 2010

Algunas mediciones nunca caben en memoria: una grabación ambiental de una hora, un monitor en vivo que debe informar niveles mientras el micrófono sigue capturando, o un registrador embebido que solo ve un búfer cada vez. En todos esos casos la señal tiene que procesarse bloque a bloque, y los filtros deben comportarse exactamente como si hubieran visto la señal completa de una vez.

Dos carriles que comparan el procesado por bloques conservando el estado del filtro entre bloques, con una envolvente continua, frente a reiniciarlo en cada bloque, donde la envolvente arranca desde cero en cada unión

Las clases OctaveFilterBank, WeightingFilter (cualquier curva de ponderación: A, C, Z y las especiales B, D, G y AU) y TimeWeighting admiten procesado por bloques (streaming): el estado interno del filtro se conserva entre llamadas, de modo que las salidas concatenadas por bloques coinciden con una única pasada sobre la señal completa. El streaming usa siempre el diseño bilineal clásico de las ponderaciones — la vía high_accuracy, sobremuestreada, remuestrea internamente y no se puede partir en bloques —, así que una ponderación A en streaming arrastra el error en altas frecuencias del diseño simple que describe Ponderación frecuencial, y una ponderación G en streaming a una frecuencia de muestreo baja pierde el sobremuestreo interno que su diseño aplica normalmente (ver Ponderaciones especiales).

Procesado por bloques con estado igual al resultado continuo frente a bloques independientes que reinician el transitorio

Con stateful=True las salidas concatenadas por bloques coinciden exactamente con el resultado continuo; sin estado, cada frontera de bloque reinicia el transitorio del filtro.

Un banco con estado es forzosamente un banco distinto, y el código de la propia figura lo dice dos veces: tanto el banco en streaming como el offline se construyen con design=FilterDesign(resample=False). El diezmado no puede transportar estado entre bloques, así que todas las bandas de un banco en streaming corren a la frecuencia de entrada, y la pasada offline contra la que se compara hay que construirla igual. Compara un resultado en streaming con el octave_filter(x, fs, fraction=3) por defecto y verás un residuo: medido sobre 4 s de ruido rosa a 48 kHz, los dos diseños difieren como mucho en 0,083 dB, con una mediana de 0,007 dB y las mayores diferencias en las bandas graves (0,083 dB a 50 Hz). Eso es lo esperado, no un fallo, y no cambia la clase; pero si estás validando una cadena en streaming, construye los dos lados igual o acabarás persiguiendo la diferencia de diseño en vez del transporte de estado.

Mostrar el código de esta figura

import matplotlib.pyplot as plt
import numpy as np
from phonometry import filters

# Banda de octava de 1 kHz: cuatro bloques con estado frente a una pasada continua
fs, block = 8000, 1000
rng = np.random.default_rng(42)
x = rng.standard_normal(4 * block)
t = np.arange(x.size) / fs

bank = filters.OctaveFilterBank(
    fs, fraction=1, limits=[900, 1100],
    design=filters.FilterDesign(resample=False),
    block_processing=filters.BlockProcessing(stateful=True),
)
streamed = np.concatenate([
    bank.filter(x[i * block:(i + 1) * block], sigbands=True,
                detrend=False, calculate_level=False)[2][0]
    for i in range(4)
])
offline = filters.OctaveFilterBank(
    fs, fraction=1, limits=[900, 1100],
    design=filters.FilterDesign(resample=False),
).filter(x, sigbands=True, detrend=False, calculate_level=False)[2][0]
print(np.max(np.abs(streamed - offline)))     # 0.0 (exacto bit a bit)

fig, ax = plt.subplots(figsize=(9, 4.5))
ax.plot(t, offline, linewidth=2.5, alpha=0.35, label="Continuo (señal completa)")
ax.plot(t, streamed, label="Bloques con estado (estado conservado)")
ax.axvline(block / fs, color="gray", linestyle=":", label="Frontera de bloque")
ax.set(xlim=(0.11, 0.17), xlabel="Tiempo [s]", ylabel="Amplitud")
ax.legend()
plt.show()

Crea un banco con estado usando block_processing=BlockProcessing(stateful=True). El estado interno se inicializa a cero por defecto, pero puede inicializarse al régimen permanente de la respuesta al escalón (como scipy.signal.sosfilt_zi) con BlockProcessing(steady_ic=True). Las opciones que deben desactivarse en modo stateful se resumen en la tabla de restricciones al final de esta guía.

Ejemplo: procesar un WAV bloque a bloque

Este ejemplo procesa un WAV por bloques con soundfile, una dependencia opcional (pip install soundfile). Sirve cualquier fuente de bloques: scipy.io.wavfile con troceado manual, o una función de captura en vivo.

import soundfile as sf
from phonometry import filters

fs = 48000
octave_filter = filters.OctaveFilterBank(
    fs, 1,
    design=filters.FilterDesign(resample=False),
    block_processing=filters.BlockProcessing(stateful=True),
)
afilter = filters.WeightingFilter(fs, "A", stateful=True)

for block in sf.blocks("measurement.wav", blocksize=256, overlap=0):

    # Aplicar el filtro A
    weighted = afilter.filter(block)

    # Dividir en bandas de octava
    block_spl, _, block_output = octave_filter.filter(weighted, sigbands=True, detrend=False)

    # procesado posterior de la señal
    ...

Acumular un nivel entre bloques

La tarea de streaming más común es justamente la que las trampas del streaming dan por fácil: el $L_{eq}$ , o el espectro de bandas, de una grabación que no cabe en memoria. Y es fácil, y además es exacta y no aproximada, porque el $L_{eq}$ no tiene constante de tiempo ni detector: solo una suma de cuadrados y un recuento de muestras.

total_square, n_samples = 0.0, 0

for x in audio_stream(block):        # tu callback de captura
    total_square += float(np.sum(np.square(x)))
    n_samples += x.shape[-1]

leq = 10 * np.log10(total_square / (n_samples * (2e-5) ** 2))

En esas cuatro líneas hay dos trampas. Ponderar todos los bloques por igual (promediar los cuadrados medios de cada bloque) solo es correcto si todos los bloques tienen la misma longitud, y el último casi nunca la tiene; acumula la suma y el recuento por separado, como arriba, y de eso se encarga la aritmética. Y nunca promedies las lecturas en decibelios de cada bloque: ese es el error de la media aritmética de Niveles integrados y estadísticos, es unilateral y siempre subestima.

La versión por bandas es el mismo acumulador con un vector a la izquierda:

band_squares = np.zeros(bank.num_bands)   # de tu OctaveFilterBank con estado

for x in audio_stream(block):
    _, _, bands = bank.filter(x, sigbands=True, detrend=False,
                              calculate_level=False)
    band_squares += np.sum(np.square(bands), axis=-1)

Los percentiles no se acumulan así. Procesa la envolvente en streaming, guárdala y calcula los percentiles una sola vez sobre el resultado conjunto.

Ponderación temporal entre bloques

Usa la clase TimeWeighting (el estado se lleva automáticamente):

from phonometry import filters

tw = filters.TimeWeighting(fs, mode="fast")
# audio_blocks: tramas sucesivas de tu grabación de micrófono (Pa),
#   p. ej. de sf.blocks("measurement.wav", ...) como en el bloque anterior.
for block in audio_blocks:
    envelope = tw.process(block)

O gestiona el estado tú mismo con la API funcional; consulta Ponderación temporal.

Estado multicanal

Todas las clases con estado gestionan entrada multicanal (channels, samples): el estado se reserva de forma perezosa en la primera llamada para ajustarse al número de canales, y se vuelve a reservar si este cambia (p. ej. pasar de estéreo a mono reinicia el estado).

Qué es el estado que se transporta

Todos los filtros de la biblioteca se ejecutan como una cascada de secciones de segundo orden en forma directa II transpuesta. Para una sección con coeficientes $(b_{0}, b_{1}, b_{2}, a_{1}, a_{2})$ :

y [n] = b_{0} x [n] + z_{1} [n - 1], z_{1} [n] = b_{1} x [n] - a_{1} y [n] + z_{2} [n - 1], z_{2} [n] = b_{2} x [n] - a_{2} y [n]

El par $(z_{1}, z_{2})$ por sección es toda la memoria del pasado que tiene el filtro. stateful=True guarda esos valores cuando termina un bloque y los restaura cuando empieza el siguiente, de modo que la recursión no puede saber dónde acabó un búfer y empezó el siguiente: la salida concatenada es igual a la de una sola pasada exactamente, no de forma aproximada. El detector TimeWeighting transporta aún menos, solo su último valor de envolvente $y [n - 1]$ , que es el mismo valor que la API funcional devuelve como initial_state (ver Ponderación temporal).

Trampas del streaming

El preprocesado por bloque crea costuras. Cualquier cosa calculada a partir de un solo bloque que debería ser global, como la eliminación de tendencia (restar la media del propio bloque) o la normalización, da a cada bloque una operación ligeramente distinta: las salidas ya no se concatenan en el resultado continuo. Por eso detrend debe ser False en modo con estado.
No toda métrica admite streaming. Las métricas de energía se acumulan limpiamente (un $L_{eq}$ acumulado es una suma de energía acumulada), pero los estadísticos de rango no: el $L_{90}$ de una grabación no es ninguna combinación de los $L_{90}$ por bloque. Procesa la envolvente en streaming y calcula los percentiles una vez, sobre el resultado conjunto.
El primer bloque sigue llevando el transitorio de arranque. El estado empieza en reposo, así que el filtro se asienta durante los primeros instantes igual que en una sola pasada. Cuánto dura se deduce del ancho de banda: una banda se asienta en unas pocas veces $1/ B$ , y con $B = 0, 231 f_{m}$ para tercios de octava eso es alrededor de un segundo a 12,5 Hz, una décima de segundo a 125 Hz y trece milisegundos a 1 kHz (la regla de duración del registro de Bancos de filtros). Un detector exponencial suma encima sus propios $5 τ$ : 0,6 s en Fast y 5 s en Slow. steady_ic=True arranca las secciones en el régimen permanente de un escalón, lo que elimina el transitorio ligado a la componente continua pero no el arranque resonante de un paso banda, así que para las bandas graves el remedio honesto sigue siendo descartar los primeros segundos del flujo una vez (no una por bloque). En la práctica: arranca el registrador unos segundos antes del intervalo que vayas a informar.
Un flujo por objeto. Una instancia con estado guarda la memoria de una señal; pasarle dos flujos intercalados corrompe ambos. Crea un objeto de filtro por flujo: el coste de diseño se paga una vez al construir, no por bloque.
Un búfer perdido o repetido anula la garantía en silencio. La exactitud bit a bit da por hecho que los bloques eran realmente consecutivos. Cuenta los bloques o compara las marcas de tiempo del dispositivo, y marca el hueco en la salida en lugar de fingir que el flujo fue continuo.
Diseña los filtros a la frecuencia con la que el dispositivo abrió de verdad, no a la que pediste. Lee la frecuencia de muestreo del flujo al arrancar y compruébala una vez; un banco diseñado para 48 kHz al que se le dan muestras a 44,1 kHz informa números plausibles de las bandas equivocadas.
La saturación hay que detectarla bloque a bloque. El bucle sustituye al indicador de sobrecarga del instrumento, así que comprueba en cada bloque si hay rachas a fondo de escala e informa un bloque marcado como no válido en lugar de promediarlo con el resto.
Un registrador desatendido necesita su propia contabilidad: un tono de calibrador al principio y al final del despliegue, y un intervalo fijo de almacenamiento de niveles.

Dos paneles de nivel ponderado A con respuesta Fast frente al tiempo. A la izquierda, ocho bloques de 100 ms de un ruido con escalones de nivel: la traza continua de una sola pasada y la traza en streaming con estado se superponen exactamente, con un error máximo de 0 dB, mientras que la traza sin estado se desploma a 30 dB en cada frontera de bloque y vuelve a subir por la rampa del detector, hasta 64 dB por debajo del nivel real en una unión. A la derecha, esos mismos primeros 0,8 s ampliados con los primeros 0,63 segundos sombreados: la traza con steady_ic False sube desde 30 dB por la rampa de asentamiento mientras que la traza con steady_ic True arranca cerca de 64 dB, y ambas convergen cuando termina la rampa

Las trampas anteriores, en el dominio que un lector en streaming mira de verdad. Conservar el estado no cuesta nada y lo gana todo: la traza azul es la gris. Reinicia en cada bloque y cada unión se convierte en una rampa nueva del detector, de decenas de decibelios, a un ritmo que fija la longitud del bloque y no el sonido. El panel derecho es la tercera trampa: los primeros $5 τ$ son del detector, steady_ic=True arranca en otro sitio en vez de en la verdad, y ninguno de los dos es el nivel hasta que la rampa acaba.

Mostrar el código de esta figura

import matplotlib.pyplot as plt

fs, block = 48000, 4800                      # bloques de 100 ms
rng = np.random.default_rng(17)
x = np.concatenate([a * rng.standard_normal(block)
                    for a in (0.02, 0.02, 0.08, 0.08, 0.02, 0.02, 0.05, 0.05)])
n_blocks = x.size // block


def to_db(env):
    return 10 * np.log10(np.maximum(env, 1e-16) / (2e-5) ** 2)


# La referencia tiene que usar el diseño que puede usar el streaming:
# high_accuracy=False.
reference = to_db(filters.time_weighting(
    filters.weighting_filter(x, fs, curve="A", high_accuracy=False), fs,
    mode="fast"))

aw = filters.WeightingFilter(fs, "A", stateful=True)
tw = filters.TimeWeighting(fs, mode="fast")
streamed = np.concatenate([to_db(tw.process(aw.filter(
    x[i * block:(i + 1) * block]))) for i in range(n_blocks)])
print(f"max |streamed - continuo| = {np.max(np.abs(streamed - reference)):.3g} dB")
# max |streamed - continuo| = 0 dB

plt.plot(np.arange(x.size) / fs, reference, linewidth=3, alpha=0.4)
plt.plot(np.arange(x.size) / fs, streamed)
plt.xlabel("Tiempo [s]")
plt.ylabel("LAF [dB re 20 uPa]")
plt.show()

Patrón de sonómetro en tiempo real

El bucle de streaming canónico pondera, obtiene la envolvente e informa bloque a bloque conservando todo el estado entre llamadas:

import numpy as np
from phonometry import filters

fs, block = 48000, 4800  # bloques de 100 ms
aw = filters.WeightingFilter(fs, "A", stateful=True)
env = filters.TimeWeighting(fs, mode="fast")  # la clase es inherentemente stateful

laf_max = -np.inf

for x in audio_stream(block):            # tu callback de captura
    y = env.process(aw.filter(x))
    spl = 10 * np.log10(y[..., -1] / (2e-5) ** 2)  # LAF instantáneo
    laf_max = max(laf_max, 10 * np.log10(y.max() / (2e-5) ** 2))
    display(spl)

y[..., -1] es la salida del detector en el instante en que acaba el bloque, así que la longitud del bloque es el ritmo de refresco de la indicación y nada de lo que ocurre entre finales de bloque llega a observarse. Con la constante Fast de 125 ms, un bloque de 20 a 100 ms sigue la envolvente fielmente; bloques mucho más largos que $τ$ convierten la indicación en un muestreo disperso de ella. El máximo acumulado de arriba existe por lo mismo: un pico que ocurra a mitad de bloque no aparece nunca en las muestras mostradas, así que el $L_{AFmax}$ hay que acumularlo como el máximo dentro de cada bloque. Y los primeros bloques contienen la propia rampa de ataque del detector ( $5 τ$ : 0,6 s en Fast, 5 s en Slow), que hay que excluir de cualquier máximo o percentil.

Restricciones del modo con estado

El streaming prohíbe todo lo que necesite la señal completa o lo que recalcule algo en cada bloque. Cada fila de la tabla se hace cumplir con una excepción, no se ignora en silencio, salvo high_accuracy, que se resuelve al diseño clásico a menos que pases True explícitamente.

Opción	Comportamiento con estado	Motivo
`detrend`	debe ser `False`	La eliminación de tendencia por bloque crea discontinuidades en las fronteras
`resample`	debe ser `False`	El remuestreador no conserva estado
`zero_phase`	no admitido	El filtrado bidireccional necesita la señal completa
`high_accuracy` (ponderación)	por defecto se resuelve a `False` (el diseño bilineal clásico, consulta Ponderación frecuencial); pasar `True` explícito lanza `ValueError`	El remuestreo polifásico interno es incompatible con bloques
`steady_ic`	opcional	Arranca los filtros en el régimen permanente de la respuesta al escalón

Qué cubre esta guía

Cubierto
La IEC 61260-1:2014 y la IEC 61672-1:2013 exactamente en la medida en que ya las implementan Bancos de filtros, Ponderación frecuencial y Ponderación temporal: stateful=True en OctaveFilterBank, WeightingFilter y TimeWeighting conserva el estado de sección en forma directa II transpuesta ( $z_{1}$ , $z_{2}$ ), o el último valor de la envolvente en TimeWeighting, entre fronteras de bloque, de modo que la salida en streaming concatenada es exactamente igual, bit a bit, a una sola pasada sobre la señal completa. Esta página no añade contenido normativo propio; solo demuestra y documenta esa equivalencia.
No cubierto
El filtrado bidireccional zero_phase necesita la señal completa, así que no se admite en modo con estado; usa la vía offline de Bancos de filtros en su lugar. El diseño high_accuracy de ponderación se resuelve al filtro bilineal clásico en modo con estado (pasar True explícitamente lanza ValueError), porque su etapa de remuestreo polifásico no es compatible con bloques; consulta Ponderación frecuencial para el diseño de alta precisión offline. Los estadísticos de rango como $L_{90}$ tampoco admiten streaming: procesa la envolvente en streaming y calcula los percentiles una vez sobre el resultado conjunto, como describe Niveles integrados y estadísticos.

Véase también

Bancos de filtros: el banco offline que el streaming reproduce bit a bit, y el modo zero_phase que el streaming no puede usar.
Multicanal y rendimiento: un estado por canal, y dónde se va realmente el tiempo de cálculo.
Ponderación temporal: la API funcional del detector, con el estado pasado a mano.
Niveles integrados y estadísticos: qué métricas se acumulan entre bloques y cuáles hay que recalcular sobre la envolvente conjunta.
Referencia de la API: filters.weighting y filters.core.
Teoría: Diseño del banco y estabilidad numérica: por qué un banco se diseña como secciones de segundo orden y se diezma por banda, que es lo que hace posible el estado en streaming.

Referencias

International Electrotechnical Commission. (2013). Electroacoustics — Sound level meters — Part 1: Specifications (IEC 61672-1:2013). Las ponderaciones frecuenciales y temporales en streaming son los mismos diseños que rige esta norma; el estado transportado mantiene válidas en streaming las afirmaciones de clase y tolerancia de las páginas de ponderación frecuencial y temporal.
International Electrotechnical Commission. (2014). Electroacoustics — Octave-band and fractional-octave-band filters — Part 1: Specifications (IEC 61260-1:2014). El procesado por bloques no añade contenido normativo propio: los filtros de octava y de fracción de octava en streaming son los mismos diseños que rige esta norma, y conservar el estado interno del filtro entre bloques es exactamente lo que hace que la salida concatenada sea idéntica a una pasada única, por lo que toda afirmación de clase y tolerancia de la página de bancos de filtros se mantiene sin cambios en uso streaming.
Oppenheim, A. V. y Schafer, R. W. (2010). Discrete-time signal processing (3.ª ed.). Pearson. Las estructuras de filtro en forma directa y la recursión de estado que sustentan la ecuación del estado transportado (ISBN 978-0-13-198842-2).

Creado y mantenido por José Manuel Requena Plens ORCID Todos los proyectos