lib/audio_proxy/peaks.ex
Rama main · réplica de lectura del repositorio original.
defmodule AudioProxy.Peaks do
@moduledoc """
Raw `s16le` PCM in, waveform peaks out (API doc §3.3).
ffmpeg does the decoding, the trim and the downmix — see
`AudioProxy.Ffmpeg.Command`'s peaks note — and hands over interleaved 16-bit
little-endian samples. This module does the arithmetic: reduce those samples
to `pts` min/max pairs and serialize them in one of audiowaveform's two
formats, which is the schema decision recorded in CLAUDE.md. There is
nothing to invent here. peaks.js reads only 8-bit data, so it needs
`pk_bits:8`. The default width is 16 bits.
## Streaming, not buffering
`feed/2` consumes chunks as they arrive and keeps only the reduced result:
four integers per pixel for stereo, two for mono, plus at most one partial
frame of carry-over. A ten-minute source is tens of megabytes of PCM and a
few kilobytes of peaks, and only the second number is ever resident.
That is why the reducer is built from a *sample count* rather than from the
stream (`new/2`): bucket boundaries have to be known before the first sample
arrives, and the count comes from `AudioProxy.Ffprobe` running ahead of
the decode. `samples_per_pixel` is `ceil(frames / count)`, so the last
bucket may be short.
## When the probe and the decode disagree
They can, by a frame or two — a duration in the container header is not a
promise about how many samples a decoder will produce. Both directions are
absorbed rather than reported:
* **more** samples than expected fold into the final bucket, which is why
that bucket alone never closes on width;
* **fewer** leave trailing buckets empty, and those serialize as `0, 0`.
So `length` is always exactly the `pts` that was asked for, whatever the
decoder did, which is what makes the output shape a function of the URL.
## Chunk boundaries are not sample boundaries
A port hands over whatever the OS had, so a chunk can end mid-sample or
mid-frame. The leftover bytes are carried into the next `feed/2`, and the
reduction is therefore independent of how the stream was cut — the property
test that random-chunks the same PCM is the statement of it.
iex> pcm = <<100::little-signed-16, -200::little-signed-16, 50::little-signed-16>>
iex> AudioProxy.Peaks.new(3, count: 3) |> AudioProxy.Peaks.feed(pcm)
...> |> AudioProxy.Peaks.finish() |> Map.fetch!(:data)
[100, 100, -200, -200, 50, 50]
"""
alias AudioProxy.Options
# audiowaveform's data format version. v2 is v1 plus the channel count, and
# emitting it unconditionally keeps mono and stereo one code path.
@dat_version 2
# Sentinels for an untouched bucket, deliberately at the far end of each
# bound so the first sample replaces both. A bucket is only ever closed
# after at least one sample, so these cannot reach the output; the padding
# in `finish/1` emits an honest zero pair instead.
@empty_mono {32_767, -32_768}
@empty_stereo {32_767, -32_768, 32_767, -32_768}
@typedoc """
A reduction in progress. Opaque — build with `new/2`, drive with `feed/2`.
"""
@opaque t :: %__MODULE__{}
@typedoc """
The finished reduction, ready to serialize.
`data` is interleaved min/max per channel per pixel, so it holds
`length * 2 * channels` integers, exactly as audiowaveform's JSON does.
"""
@type result :: %{
version: 2,
channels: 1 | 2,
sample_rate: non_neg_integer(),
samples_per_pixel: pos_integer(),
bits: 8 | 16,
length: pos_integer(),
data: [integer()]
}
defstruct [
:count,
:channels,
:sample_rate,
:samples_per_pixel,
:bits,
:acc,
:remaining,
closed: [],
n_closed: 0,
leftover: <<>>
]
@doc """
A reducer for `frames` frames of PCM.
A *frame* is one sample per channel, so `frames` is the sample count of one
channel and not the number of integers on the wire.
Options:
* `:count` — how many min/max pairs to produce (`pts`); required.
* `:channels` — 1 or 2, matching the `-ac` the decode was given.
Defaults to 1.
* `:sample_rate` — carried into the serialized output, where consumers use
it to turn a pixel index into a time. Defaults to 0, meaning unknown.
* `:bits` — 8 or 16, the width of each serialized value. Defaults to 16.
"""
@spec new(non_neg_integer(), keyword()) :: t()
def new(frames, opts) when is_integer(frames) and frames >= 0 do
count = Keyword.fetch!(opts, :count)
channels = Keyword.get(opts, :channels, 1)
bits = Keyword.get(opts, :bits, 16)
# Fail here, not in finish/1, where the cause is far from the call.
unless bits in [8, 16] do
raise ArgumentError, "bits must be 8 or 16, got: #{inspect(bits)}"
end
# At least one: a source shorter than `pts` frames still gets one frame per
# pixel, and the pixels past its end are the padding described above.
samples_per_pixel = max(ceil_div(frames, count), 1)
%__MODULE__{
count: count,
channels: channels,
sample_rate: Keyword.get(opts, :sample_rate, 0),
samples_per_pixel: samples_per_pixel,
bits: bits,
acc: empty(channels),
remaining: samples_per_pixel
}
end
@doc """
A reducer configured from `options` — the `pts`, `ch` and `pk_bits` half of
the URL — plus what only the probe knows.
"""
@spec new(non_neg_integer(), Options.t(), pos_integer()) :: t()
def new(frames, %Options{} = options, sample_rate) do
new(frames,
count: Options.peak_count(options),
channels: Options.peak_channels(options),
sample_rate: sample_rate,
bits: Options.peak_bits(options)
)
end
@doc """
Folds a chunk of interleaved `s16le` PCM into the reduction.
Any trailing bytes that do not complete a frame are carried to the next
call, so chunk boundaries do not have to fall anywhere in particular.
"""
@spec feed(t(), binary()) :: t()
def feed(%__MODULE__{} = state, <<>>), do: state
def feed(%__MODULE__{channels: 1} = state, data) do
{acc, remaining, closed, n_closed, leftover} =
scan_mono(
state.leftover <> data,
state.acc,
state.remaining,
state.closed,
state.n_closed,
state.count - 1,
state.samples_per_pixel
)
%{
state
| acc: acc,
remaining: remaining,
closed: closed,
n_closed: n_closed,
leftover: leftover
}
end
def feed(%__MODULE__{channels: 2} = state, data) do
{acc, remaining, closed, n_closed, leftover} =
scan_stereo(
state.leftover <> data,
state.acc,
state.remaining,
state.closed,
state.n_closed,
state.count - 1,
state.samples_per_pixel
)
%{
state
| acc: acc,
remaining: remaining,
closed: closed,
n_closed: n_closed,
leftover: leftover
}
end
@doc """
Closes the reduction and returns the audiowaveform-shaped result.
The bucket in progress is closed if anything reached it, and the tail is
padded to `count` pairs. A partial final frame — the decoder stopped
mid-sample — is discarded rather than being read as a sample it is not.
"""
@spec finish(t()) :: result()
def finish(%__MODULE__{} = state) do
closed =
if state.acc == empty(state.channels) do
state.closed
else
[state.acc | state.closed]
end
data =
closed
|> Enum.reverse()
|> Enum.flat_map(&Tuple.to_list/1)
|> pad(state.count * 2 * state.channels)
|> narrow(state.bits)
%{
version: @dat_version,
channels: state.channels,
sample_rate: state.sample_rate,
samples_per_pixel: state.samples_per_pixel,
bits: state.bits,
length: state.count,
data: data
}
end
@doc """
audiowaveform's JSON serialization: the keys of `t:result/0`, verbatim.
iex> pcm = <<-3000::little-signed-16, 9000::little-signed-16>>
iex> AudioProxy.Peaks.new(2, count: 1, sample_rate: 8000)
...> |> AudioProxy.Peaks.feed(pcm) |> AudioProxy.Peaks.finish()
...> |> AudioProxy.Peaks.to_json() |> JSON.decode!()
%{"version" => 2, "channels" => 1, "sample_rate" => 8000,
"samples_per_pixel" => 2, "bits" => 16, "length" => 1,
"data" => [-3000, 9000]}
"""
@spec to_json(result()) :: binary()
def to_json(result), do: JSON.encode!(result)
@doc """
audiowaveform's binary serialization: a little-endian v2 header followed by
the same integers as `int16`, or as `int8` when `bits` is 8.
The header is version, flags, sample rate, samples per pixel, length and
channel count, in that order — six 32-bit fields, and then the data. A
consumer decoding this and the JSON above gets identical pairs, which is the
round-trip the spec asks for.
"""
@spec to_dat(result()) :: binary()
def to_dat(result) do
header = <<
@dat_version::little-signed-32,
dat_flags(result.bits)::little-unsigned-32,
result.sample_rate::little-signed-32,
result.samples_per_pixel::little-signed-32,
result.length::little-unsigned-32,
result.channels::little-signed-32
>>
body = for value <- result.data, do: <<value::little-signed-size(result.bits)>>
IO.iodata_to_binary([header | body])
end
@doc """
Serializes for `format` — the `pk_fmt` half of the URL.
"""
@spec serialize(result(), Options.peak_format()) :: binary()
def serialize(result, :json), do: to_json(result)
def serialize(result, :dat), do: to_dat(result)
## Reduction
# Two hand-written loops rather than one generic one, because this is the
# only hot path in the project: a five-minute mono source is thirteen
# million iterations, and a per-frame list or map would dominate the render.
# `ch` is 1 or 2 and nothing else, so there is no third case to miss.
#
# `limit` is the index of the final bucket. Reaching it switches the loop
# into its accumulate-forever clause, which is how extra samples from a
# decode that outran its probe fold in instead of overflowing `count`.
defp scan_mono(
<<sample::little-signed-16, rest::binary>>,
{lo, hi},
remaining,
closed,
n,
limit,
spp
)
when remaining > 1 or n >= limit do
scan_mono(rest, {min(lo, sample), max(hi, sample)}, remaining - 1, closed, n, limit, spp)
end
defp scan_mono(
<<sample::little-signed-16, rest::binary>>,
{lo, hi},
_remaining,
closed,
n,
limit,
spp
) do
bucket = {min(lo, sample), max(hi, sample)}
scan_mono(rest, @empty_mono, spp, [bucket | closed], n + 1, limit, spp)
end
defp scan_mono(leftover, acc, remaining, closed, n, _limit, _spp) do
{acc, remaining, closed, n, leftover}
end
defp scan_stereo(
<<left::little-signed-16, right::little-signed-16, rest::binary>>,
{llo, lhi, rlo, rhi},
remaining,
closed,
n,
limit,
spp
)
when remaining > 1 or n >= limit do
acc = {min(llo, left), max(lhi, left), min(rlo, right), max(rhi, right)}
scan_stereo(rest, acc, remaining - 1, closed, n, limit, spp)
end
defp scan_stereo(
<<left::little-signed-16, right::little-signed-16, rest::binary>>,
{llo, lhi, rlo, rhi},
_remaining,
closed,
n,
limit,
spp
) do
bucket = {min(llo, left), max(lhi, left), min(rlo, right), max(rhi, right)}
scan_stereo(rest, @empty_stereo, spp, [bucket | closed], n + 1, limit, spp)
end
defp scan_stereo(leftover, acc, remaining, closed, n, _limit, _spp) do
{acc, remaining, closed, n, leftover}
end
defp empty(1), do: @empty_mono
defp empty(2), do: @empty_stereo
# Trailing pixels the decode never reached. A flat zero line is the honest
# picture of "no samples here" and is what audiowaveform itself writes.
defp pad(data, width) do
case width - length(data) do
missing when missing > 0 -> data ++ List.duplicate(0, missing)
_complete -> data
end
end
# The reduction always runs at 16 bits, and 8-bit output is that result
# narrowed. Thus the two widths are one waveform by construction.
# `div/2` truncates toward zero, which is audiowaveform's own rule for
# `-b 8`. An int16 divided by 256 is always inside -128..127.
defp narrow(data, 16), do: data
defp narrow(data, 8), do: Enum.map(data, &div(&1, 256))
# audiowaveform's flags field. Bit 0 set means 8-bit values. Bits 1-31 are
# unused.
defp dat_flags(16), do: 0
defp dat_flags(8), do: 1
defp ceil_div(_numerator, 0), do: 0
defp ceil_div(numerator, denominator), do: div(numerator + denominator - 1, denominator)
end