eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

push-mqttpywkpspw #19

Merged
eric merged 52 commits from push-mqttpywkpspw into main 2025-11-04 21:15:56 +01:00
Conversation 0 Commits 52 Files Changed 679 +485 -407
5 changed files with 485 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit d5aa26edff - Show all commits

327
pmoflac/src/aiff.rs
View File

@@ -153,176 +153,195 @@ where
let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, AiffError>>();
let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), AiffError> {
let channel_reader = ChannelReader::<AiffError>::new(ingest_rx);
let mut aiff_reader = StreamingAiffReader::new(channel_reader);
let mut info_tx = Some(info_tx);
// Parse FORM header
let form_header = aiff_reader.read_exact_vec(12)?;
if &form_header[0..4] != b"FORM" {
return Err(AiffError::Decode("missing FORM header".into()));
}
let form_type = <[u8; 4]>::try_from(&form_header[8..12]).unwrap();
if form_type != *b"AIFF" && form_type != *b"AIFC" {
return Err(AiffError::Decode(
"unsupported FORM type (expected AIFF/AIFC)".into(),
));
}
let result: Result<(), AiffError> = (|| {
let channel_reader = ChannelReader::<AiffError>::new(ingest_rx);
let mut aiff_reader = StreamingAiffReader::new(channel_reader);
let mut comm_chunk: Option<CommChunk> = None;
let mut stream_info_sent = false;
// Parse FORM header
let form_header = aiff_reader.read_exact_vec(12)?;
if &form_header[0..4] != b"FORM" {
return Err(AiffError::Decode("missing FORM header".into()));
}
let form_type = <[u8; 4]>::try_from(&form_header[8..12]).unwrap();
if form_type != *b"AIFF" && form_type != *b"AIFC" {
return Err(AiffError::Decode(
"unsupported FORM type (expected AIFF/AIFC)".into(),
));
}
loop {
let header = match aiff_reader.read_exact_vec(8) {
Ok(bytes) => bytes,
Err(AiffError::Decode(msg)) if msg.contains("unexpected EOF") => break,
Err(err) => return Err(err),
};
let chunk_id = <[u8; 4]>::try_from(&header[..4]).unwrap();
let chunk_size =
u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize;
let mut comm_chunk: Option<CommChunk> = None;
let mut stream_info_sent = false;
let padded_size = if chunk_size % 2 == 0 {
chunk_size
} else {
chunk_size + 1
};
loop {
let header = match aiff_reader.read_exact_vec(8) {
Ok(bytes) => bytes,
Err(AiffError::Decode(msg)) if msg.contains("unexpected EOF") => break,
Err(err) => return Err(err),
};
let chunk_id = <[u8; 4]>::try_from(&header[..4]).unwrap();
let chunk_size =
u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize;
match &chunk_id {
b"COMM" => {
let data = aiff_reader.read_exact_vec(chunk_size)?;
if form_type == *b"AIFF" && data.len() < 18 {
return Err(AiffError::Decode("COMM chunk too small".into()));
}
if data.len() < 18 {
return Err(AiffError::Decode("COMM chunk too small for AIFC".into()));
}
let channels = u16::from_be_bytes([data[0], data[1]]);
let num_frames = u32::from_be_bytes([data[2], data[3], data[4], data[5]]);
let bits_per_sample = u16::from_be_bytes([data[6], data[7]]);
let sample_rate = parse_extended_f80(&data[8..18])?;
let padded_size = if chunk_size % 2 == 0 {
chunk_size
} else {
chunk_size + 1
};
let compression = if form_type == *b"AIFC" {
if data.len() < 22 {
return Err(AiffError::Decode(
"AIFC COMM chunk missing compression type".into(),
));
match &chunk_id {
b"COMM" => {
let data = aiff_reader.read_exact_vec(chunk_size)?;
if form_type == *b"AIFF" && data.len() < 18 {
return Err(AiffError::Decode("COMM chunk too small".into()));
}
match &data[18..22] {
b"NONE" => Compression::BigEndianPcm,
b"sowt" => Compression::LittleEndianPcm,
code => {
return Err(AiffError::Decode(format!(
"unsupported AIFC compression type: {}",
String::from_utf8_lossy(code)
)))
if data.len() < 18 {
return Err(AiffError::Decode("COMM chunk too small for AIFC".into()));
}
let channels = u16::from_be_bytes([data[0], data[1]]);
let num_frames =
u32::from_be_bytes([data[2], data[3], data[4], data[5]]);
let bits_per_sample = u16::from_be_bytes([data[6], data[7]]);
let sample_rate = parse_extended_f80(&data[8..18])?;
let compression = if form_type == *b"AIFC" {
if data.len() < 22 {
return Err(AiffError::Decode(
"AIFC COMM chunk missing compression type".into(),
));
}
}
} else {
Compression::BigEndianPcm
};
let comm = CommChunk {
channels,
num_frames,
bits_per_sample,
sample_rate,
compression,
};
comm.validate()?;
comm_chunk = Some(comm);
if padded_size > chunk_size {
aiff_reader.skip(padded_size - chunk_size)?;
}
}
b"SSND" => {
let comm = comm_chunk.as_ref().ok_or_else(|| {
AiffError::Decode("SSND chunk encountered before COMM".into())
})?;
let header = aiff_reader.read_exact_vec(8)?;
let offset =
u32::from_be_bytes([header[0], header[1], header[2], header[3]]) as usize;
let _block_size =
u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize;
if offset > 0 {
aiff_reader.skip(offset)?;
}
let data_bytes = chunk_size
.checked_sub(8)
.ok_or_else(|| AiffError::Decode("invalid SSND chunk size".into()))?;
let bytes_per_sample = comm.bytes_per_sample();
let info = StreamInfo {
sample_rate: comm.sample_rate,
channels: comm.channels as u8,
bits_per_sample: comm.bits_per_sample as u8,
total_samples: Some(comm.num_frames as u64),
max_block_size: 0,
min_block_size: 0,
};
if !stream_info_sent {
if info_tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
stream_info_sent = true;
}
let mut remaining = data_bytes;
while remaining > 0 {
let mut to_read = remaining.min(8192);
let residue = to_read % bytes_per_sample;
if residue != 0 {
to_read -= residue;
}
if to_read == 0 {
to_read = bytes_per_sample;
}
let mut chunk = aiff_reader.read_exact_vec(to_read)?;
match comm.compression {
Compression::BigEndianPcm => {
chunk = convert_be_pcm(chunk, comm.bits_per_sample)?;
match &data[18..22] {
b"NONE" => Compression::BigEndianPcm,
b"sowt" => Compression::LittleEndianPcm,
code => {
return Err(AiffError::Decode(format!(
"unsupported AIFC compression type: {}",
String::from_utf8_lossy(code)
)))
}
}
Compression::LittleEndianPcm => {
// data already little-endian; no conversion
}
}
if !chunk.is_empty() {
if pcm_tx.blocking_send(Ok(chunk)).is_err() {
return Ok(());
}
}
remaining = remaining
.checked_sub(to_read)
.ok_or_else(|| AiffError::Decode("SSND chunk underflow".into()))?;
}
} else {
Compression::BigEndianPcm
};
if padded_size > chunk_size {
aiff_reader.skip(1)?;
}
let comm = CommChunk {
channels,
num_frames,
bits_per_sample,
sample_rate,
compression,
};
comm.validate()?;
break;
}
_ => {
aiff_reader.skip(chunk_size)?;
if padded_size > chunk_size {
aiff_reader.skip(padded_size - chunk_size)?;
comm_chunk = Some(comm);
if padded_size > chunk_size {
aiff_reader.skip(padded_size - chunk_size)?;
}
}
b"SSND" => {
let comm = comm_chunk.as_ref().ok_or_else(|| {
AiffError::Decode("SSND chunk encountered before COMM".into())
})?;
let header = aiff_reader.read_exact_vec(8)?;
let offset = u32::from_be_bytes([
header[0], header[1], header[2], header[3],
]) as usize;
let _block_size = u32::from_be_bytes([
header[4], header[5], header[6], header[7],
]) as usize;
if offset > 0 {
aiff_reader.skip(offset)?;
}
let data_bytes = chunk_size
.checked_sub(8)
.ok_or_else(|| AiffError::Decode("invalid SSND chunk size".into()))?;
let bytes_per_sample = comm.bytes_per_sample();
let info = StreamInfo {
sample_rate: comm.sample_rate,
channels: comm.channels as u8,
bits_per_sample: comm.bits_per_sample as u8,
total_samples: Some(comm.num_frames as u64),
max_block_size: 0,
min_block_size: 0,
};
if !stream_info_sent {
if let Some(tx) = info_tx.take() {
if tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
}
stream_info_sent = true;
}
let mut remaining = data_bytes;
while remaining > 0 {
let mut to_read = remaining.min(8192);
let residue = to_read % bytes_per_sample;
if residue != 0 {
to_read -= residue;
}
if to_read == 0 {
to_read = bytes_per_sample;
}
let mut chunk = aiff_reader.read_exact_vec(to_read)?;
match comm.compression {
Compression::BigEndianPcm => {
chunk = convert_be_pcm(chunk, comm.bits_per_sample)?;
}
Compression::LittleEndianPcm => {
// data already little-endian; no conversion
}
}
if !chunk.is_empty() {
if pcm_tx.blocking_send(Ok(chunk)).is_err() {
return Ok(());
}
}
remaining = remaining
.checked_sub(to_read)
.ok_or_else(|| AiffError::Decode("SSND chunk underflow".into()))?;
}
if padded_size > chunk_size {
aiff_reader.skip(1)?;
}
break;
}
_ => {
aiff_reader.skip(chunk_size)?;
if padded_size > chunk_size {
aiff_reader.skip(padded_size - chunk_size)?;
}
}
}
}
}
if !stream_info_sent {
return Err(AiffError::Decode(
"no SSND chunk found in AIFF stream".into(),
));
}
if !stream_info_sent {
return Err(AiffError::Decode(
"no SSND chunk found in AIFF stream".into(),
));
}
Ok(())
Ok(())
})();
match result {
Ok(()) => Ok(()),
Err(err) => {
if let Some(tx) = info_tx.take() {
let _ = tx.send(Err(err.clone()));
}
Err(err)
}
}
});
let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "aiff-decode");

137
pmoflac/src/ogg.rs
View File

@@ -184,81 +184,98 @@ where
let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, OggError>>();
let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggError> {
let channel_reader = ChannelReader::<OggContainerError>::new(ingest_rx);
let mut packet_reader = OggPacketReader::new(channel_reader, OggReaderOptions::default());
let mut info_tx = Some(info_tx);
// Read Vorbis headers (3 packets: identification, comment, setup)
let ident_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggError::Decode("missing Vorbis identification header".into()))?;
let ident_hdr = read_header_ident(&ident_packet)?;
let result: Result<(), OggError> = (|| {
let channel_reader = ChannelReader::<OggContainerError>::new(ingest_rx);
let mut packet_reader =
OggPacketReader::new(channel_reader, OggReaderOptions::default());
let comment_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggError::Decode("missing Vorbis comment header".into()))?;
let _comment_hdr: CommentHeader = read_header_comment(&comment_packet)?;
// Read Vorbis headers (3 packets: identification, comment, setup)
let ident_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggError::Decode("missing Vorbis identification header".into()))?;
let ident_hdr = read_header_ident(&ident_packet)?;
let setup_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggError::Decode("missing Vorbis setup header".into()))?;
let setup_hdr = read_header_setup(
&setup_packet,
ident_hdr.audio_channels,
(ident_hdr.blocksize_0, ident_hdr.blocksize_1),
)?;
let comment_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggError::Decode("missing Vorbis comment header".into()))?;
let _comment_hdr: CommentHeader = read_header_comment(&comment_packet)?;
let info = StreamInfo {
sample_rate: ident_hdr.audio_sample_rate,
channels: ident_hdr.audio_channels,
bits_per_sample: 16,
total_samples: None,
max_block_size: 1 << ident_hdr.blocksize_1,
min_block_size: 1 << ident_hdr.blocksize_0,
};
let setup_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggError::Decode("missing Vorbis setup header".into()))?;
let setup_hdr = read_header_setup(
&setup_packet,
ident_hdr.audio_channels,
(ident_hdr.blocksize_0, ident_hdr.blocksize_1),
)?;
if info_tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
let info = StreamInfo {
sample_rate: ident_hdr.audio_sample_rate,
channels: ident_hdr.audio_channels,
bits_per_sample: 16,
total_samples: None,
max_block_size: 1 << ident_hdr.blocksize_1,
min_block_size: 1 << ident_hdr.blocksize_0,
};
// Decode audio packets
let mut pcm_bytes = Vec::new();
let mut produced_audio = false;
let mut pwr = PreviousWindowRight::new();
while let Some(packet) = packet_reader.next_packet()? {
let decoded: InterleavedSamples<i16> =
read_audio_packet_generic(&ident_hdr, &setup_hdr, &packet, &mut pwr)?;
if decoded.samples.is_empty() {
continue;
if let Some(tx) = info_tx.take() {
if tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
}
produced_audio = true;
// Decode audio packets
let mut pcm_bytes = Vec::new();
let mut produced_audio = false;
let mut pwr = PreviousWindowRight::new();
// Reuse buffer capacity from previous iteration
pcm_bytes.clear();
pcm_bytes.reserve(decoded.samples.len() * 2);
for sample in decoded.samples {
pcm_bytes.extend_from_slice(&sample.to_le_bytes());
while let Some(packet) = packet_reader.next_packet()? {
let decoded: InterleavedSamples<i16> =
read_audio_packet_generic(&ident_hdr, &setup_hdr, &packet, &mut pwr)?;
if decoded.samples.is_empty() {
continue;
}
produced_audio = true;
// Reuse buffer capacity from previous iteration
pcm_bytes.clear();
pcm_bytes.reserve(decoded.samples.len() * 2);
for sample in decoded.samples {
pcm_bytes.extend_from_slice(&sample.to_le_bytes());
}
let chunk = std::mem::take(&mut pcm_bytes);
if pcm_tx.blocking_send(Ok(chunk)).is_err() {
break;
}
// Pre-allocate for next iteration
pcm_bytes =
Vec::with_capacity(info.max_block_size as usize * info.channels as usize * 2);
}
let chunk = std::mem::take(&mut pcm_bytes);
if pcm_tx.blocking_send(Ok(chunk)).is_err() {
break;
if !produced_audio {
let err = OggError::Decode("stream contained no decodable Vorbis packets".into());
let _ = pcm_tx.blocking_send(Err(err.clone()));
return Err(err);
}
// Pre-allocate for next iteration
pcm_bytes =
Vec::with_capacity(info.max_block_size as usize * info.channels as usize * 2);
}
Ok(())
})();
if !produced_audio {
let err = OggError::Decode("stream contained no decodable Vorbis packets".into());
let _ = pcm_tx.blocking_send(Err(err.clone()));
return Err(err);
match result {
Ok(()) => Ok(()),
Err(err) => {
if let Some(tx) = info_tx.take() {
let _ = tx.send(Err(err.clone()));
}
Err(err)
}
}
Ok(())
});
let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "ogg-decode");

18
pmoflac/src/ogg_common.rs
View File

@@ -82,6 +82,7 @@ pub struct OggPacketReader {
finished: bool,
stream_serial: Option<u32>,
sync_buffer: Vec<u8>,
sync_bytes_read: usize,
synced: bool,
options: OggReaderOptions,
}
@@ -96,6 +97,7 @@ impl OggPacketReader {
finished: false,
stream_serial: None,
sync_buffer: Vec::new(),
sync_bytes_read: 0,
synced: !options.find_sync, // If we don't need to find sync, we're already synced
options,
}
@@ -183,14 +185,21 @@ impl OggPacketReader {
return Ok(());
}
while self.sync_buffer.len() < self.options.max_sync_search {
while self.sync_bytes_read < self.options.max_sync_search {
let mut chunk = [0u8; 1024];
let n = Read::read(&mut self.reader, &mut chunk).map_err(OggContainerError::from)?;
let remaining = self.options.max_sync_search - self.sync_bytes_read;
let to_read = remaining.min(chunk.len());
if to_read == 0 {
break;
}
let n = Read::read(&mut self.reader, &mut chunk[..to_read])
.map_err(OggContainerError::from)?;
if n == 0 {
return Err(OggContainerError::Decode(
"EOF reached while searching for Ogg sync pattern".into(),
));
}
self.sync_bytes_read += n;
self.sync_buffer.extend_from_slice(&chunk[..n]);
// Search for "OggS" pattern
@@ -205,9 +214,8 @@ impl OggPacketReader {
return Ok(());
}
// If buffer is getting large and still no sync, keep only last 3 bytes
// (in case "OggS" is split across chunk boundary)
if self.sync_buffer.len() >= self.options.max_sync_search {
// Keep the buffer compact to avoid unbounded growth.
if self.sync_buffer.len() > 3 {
let keep_len = 3.min(self.sync_buffer.len());
self.sync_buffer.drain(..self.sync_buffer.len() - keep_len);
}

193
pmoflac/src/opus.rs
View File

@@ -48,112 +48,129 @@ where
let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, OggOpusError>>();
let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggOpusError> {
let channel_reader = ChannelReader::<OggContainerError>::new(ingest_rx);
let mut packet_reader = OggPacketReader::new(
channel_reader,
OggReaderOptions {
validate_crc: false,
find_sync: false,
..OggReaderOptions::default()
},
);
let mut info_tx = Some(info_tx);
let header_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggOpusError::Decode("missing OpusHead packet".into()))?;
let header = OpusHead::parse(&header_packet)?;
let result: Result<(), OggOpusError> = (|| {
let channel_reader = ChannelReader::<OggContainerError>::new(ingest_rx);
let mut packet_reader = OggPacketReader::new(
channel_reader,
OggReaderOptions {
validate_crc: false,
find_sync: false,
..OggReaderOptions::default()
},
);
let tags_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggOpusError::Decode("missing OpusTags packet".into()))?;
let _tags = OpusTags::parse(&tags_packet)?;
let header_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggOpusError::Decode("missing OpusHead packet".into()))?;
let header = OpusHead::parse(&header_packet)?;
let channels_enum = match header.channels {
1 => Channels::Mono,
2 => Channels::Stereo,
other => {
return Err(OggOpusError::Decode(format!(
"unsupported channel count: {}",
other
)))
}
};
let tags_packet = packet_reader
.next_packet()?
.ok_or_else(|| OggOpusError::Decode("missing OpusTags packet".into()))?;
let _tags = OpusTags::parse(&tags_packet)?;
let mut decoder = OpusDecoder::new(48_000, channels_enum)?;
if header.output_gain != 0 {
decoder.set_gain(i32::from(header.output_gain))?;
}
let channels_enum = match header.channels {
1 => Channels::Mono,
2 => Channels::Stereo,
other => {
return Err(OggOpusError::Decode(format!(
"unsupported channel count: {}",
other
)))
}
};
let info = StreamInfo {
sample_rate: 48_000,
channels: header.channels,
bits_per_sample: 16,
total_samples: None,
max_block_size: MAX_FRAME_SAMPLES as u16,
min_block_size: 0,
};
if info_tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
let channels = header.channels as usize;
let mut pcm_buffer = vec![0i16; MAX_FRAME_SAMPLES * channels];
let mut pcm_bytes = Vec::new();
let mut pre_skip = header.pre_skip as usize;
let mut produced_audio = false;
while let Some(packet) = packet_reader.next_packet()? {
if pcm_buffer.len() < MAX_FRAME_SAMPLES * channels {
pcm_buffer.resize(MAX_FRAME_SAMPLES * channels, 0);
let mut decoder = OpusDecoder::new(48_000, channels_enum)?;
if header.output_gain != 0 {
decoder.set_gain(i32::from(header.output_gain))?;
}
let decoded_frames = decoder.decode(
&packet,
&mut pcm_buffer[..MAX_FRAME_SAMPLES * channels],
false,
)?;
if decoded_frames == 0 {
continue;
}
let info = StreamInfo {
sample_rate: 48_000,
channels: header.channels,
bits_per_sample: 16,
total_samples: None,
max_block_size: MAX_FRAME_SAMPLES as u16,
min_block_size: 0,
};
let mut start_frame = 0;
if pre_skip > 0 {
let drop = pre_skip.min(decoded_frames);
pre_skip -= drop;
start_frame = drop;
if start_frame == decoded_frames {
continue;
if let Some(tx) = info_tx.take() {
if tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
}
let start_index = start_frame * channels;
let end_index = decoded_frames * channels;
let channels = header.channels as usize;
let mut pcm_buffer = vec![0i16; MAX_FRAME_SAMPLES * channels];
let mut pcm_bytes = Vec::new();
let mut pre_skip = header.pre_skip as usize;
let mut produced_audio = false;
pcm_bytes.clear();
pcm_bytes.reserve((end_index - start_index) * 2);
for sample in &pcm_buffer[start_index..end_index] {
pcm_bytes.extend_from_slice(&sample.to_le_bytes());
while let Some(packet) = packet_reader.next_packet()? {
if pcm_buffer.len() < MAX_FRAME_SAMPLES * channels {
pcm_buffer.resize(MAX_FRAME_SAMPLES * channels, 0);
}
let decoded_frames = decoder.decode(
&packet,
&mut pcm_buffer[..MAX_FRAME_SAMPLES * channels],
false,
)?;
if decoded_frames == 0 {
continue;
}
let mut start_frame = 0;
if pre_skip > 0 {
let drop = pre_skip.min(decoded_frames);
pre_skip -= drop;
start_frame = drop;
if start_frame == decoded_frames {
continue;
}
}
let start_index = start_frame * channels;
let end_index = decoded_frames * channels;
pcm_bytes.clear();
pcm_bytes.reserve((end_index - start_index) * 2);
for sample in &pcm_buffer[start_index..end_index] {
pcm_bytes.extend_from_slice(&sample.to_le_bytes());
}
if pcm_bytes.is_empty() {
continue;
}
produced_audio = true;
let chunk = std::mem::take(&mut pcm_bytes);
if pcm_tx.blocking_send(Ok(chunk)).is_err() {
break;
}
pcm_bytes = Vec::with_capacity(MAX_FRAME_SAMPLES * channels * 2);
}
if pcm_bytes.is_empty() {
continue;
if !produced_audio {
return Err(OggOpusError::Decode(
"stream contained no decodable Opus packets".into(),
));
}
produced_audio = true;
let chunk = pcm_bytes.clone();
if pcm_tx.blocking_send(Ok(chunk)).is_err() {
break;
Ok(())
})();
match result {
Ok(()) => Ok(()),
Err(err) => {
if let Some(tx) = info_tx.take() {
let _ = tx.send(Err(err.clone()));
}
Err(err)
}
}
if !produced_audio {
return Err(OggOpusError::Decode(
"stream contained no decodable Opus packets".into(),
));
}
Ok(())
});
let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "ogg-opus");

217
pmoflac/src/wav.rs
View File

@@ -157,126 +157,143 @@ where
let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, WavError>>();
let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), WavError> {
let channel_reader = ChannelReader::<WavError>::new(ingest_rx);
let mut wav_reader = StreamingWavReader::new(channel_reader);
let mut info_tx = Some(info_tx);
let riff = wav_reader.read_exact(12)?;
if &riff[0..4] != b"RIFF" {
return Err(WavError::Decode("missing RIFF header".into()));
}
if &riff[8..12] != b"WAVE" {
return Err(WavError::Decode("missing WAVE signature".into()));
}
let result: Result<(), WavError> = (|| {
let channel_reader = ChannelReader::<WavError>::new(ingest_rx);
let mut wav_reader = StreamingWavReader::new(channel_reader);
let mut fmt_chunk: Option<FmtChunk> = None;
let mut data_found = false;
loop {
let mut chunk_header = [0u8; 8];
match wav_reader.read_exact(8) {
Ok(bytes) => chunk_header.copy_from_slice(bytes),
Err(WavError::Decode(msg)) if msg.contains("unexpected EOF") => break,
Err(err) => return Err(err),
let riff = wav_reader.read_exact(12)?;
if &riff[0..4] != b"RIFF" {
return Err(WavError::Decode("missing RIFF header".into()));
}
if &riff[8..12] != b"WAVE" {
return Err(WavError::Decode("missing WAVE signature".into()));
}
let chunk_id = &chunk_header[..4];
let chunk_size = u32::from_le_bytes([
chunk_header[4],
chunk_header[5],
chunk_header[6],
chunk_header[7],
]) as usize;
let padded_size = (chunk_size + 1) & !1; // align to even bytes
let mut fmt_chunk: Option<FmtChunk> = None;
let mut data_found = false;
match chunk_id {
b"fmt " => {
let bytes = wav_reader.read_exact(chunk_size)?;
if chunk_size < 16 {
return Err(WavError::Decode("fmt chunk too small".into()));
}
let audio_format = u16::from_le_bytes([bytes[0], bytes[1]]);
let channels = u16::from_le_bytes([bytes[2], bytes[3]]);
let sample_rate = u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
let bits_per_sample = u16::from_le_bytes([bytes[14], bytes[15]]);
let fmt = FmtChunk {
audio_format,
channels,
sample_rate,
bits_per_sample,
};
fmt.validate()?;
fmt_chunk = Some(fmt);
if padded_size > chunk_size {
wav_reader.skip(padded_size - chunk_size)?;
}
loop {
let mut chunk_header = [0u8; 8];
match wav_reader.read_exact(8) {
Ok(bytes) => chunk_header.copy_from_slice(bytes),
Err(WavError::Decode(msg)) if msg.contains("unexpected EOF") => break,
Err(err) => return Err(err),
}
b"data" => {
let fmt = fmt_chunk
.as_ref()
.ok_or_else(|| WavError::Decode("data chunk before fmt chunk".into()))?;
let chunk_id = &chunk_header[..4];
let chunk_size = u32::from_le_bytes([
chunk_header[4],
chunk_header[5],
chunk_header[6],
chunk_header[7],
]) as usize;
let info = StreamInfo {
sample_rate: fmt.sample_rate,
channels: fmt.channels as u8,
bits_per_sample: fmt.bits_per_sample as u8,
total_samples: None,
max_block_size: 0,
min_block_size: 0,
};
let padded_size = (chunk_size + 1) & !1; // align to even bytes
if info_tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
let mut remaining = chunk_size;
let bytes_per_frame = fmt.bytes_per_sample() * fmt.channels as usize;
let mut buffer = vec![0u8; 4096];
while remaining > 0 {
let to_read = remaining.min(buffer.len());
let read = wav_reader.reader.read(&mut buffer[..to_read])?;
if read == 0 {
break;
match chunk_id {
b"fmt " => {
let bytes = wav_reader.read_exact(chunk_size)?;
if chunk_size < 16 {
return Err(WavError::Decode("fmt chunk too small".into()));
}
remaining -= read;
let aligned = read - (read % bytes_per_frame);
if aligned > 0 {
if pcm_tx
.blocking_send(Ok(buffer[..aligned].to_vec()))
.is_err()
{
let audio_format = u16::from_le_bytes([bytes[0], bytes[1]]);
let channels = u16::from_le_bytes([bytes[2], bytes[3]]);
let sample_rate =
u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
let bits_per_sample = u16::from_le_bytes([bytes[14], bytes[15]]);
let fmt = FmtChunk {
audio_format,
channels,
sample_rate,
bits_per_sample,
};
fmt.validate()?;
fmt_chunk = Some(fmt);
if padded_size > chunk_size {
wav_reader.skip(padded_size - chunk_size)?;
}
}
b"data" => {
let fmt = fmt_chunk.as_ref().ok_or_else(|| {
WavError::Decode("data chunk before fmt chunk".into())
})?;
let info = StreamInfo {
sample_rate: fmt.sample_rate,
channels: fmt.channels as u8,
bits_per_sample: fmt.bits_per_sample as u8,
total_samples: None,
max_block_size: 0,
min_block_size: 0,
};
if let Some(tx) = info_tx.take() {
if tx.send(Ok(info.clone())).is_err() {
return Ok(());
}
}
if aligned < read {
return Err(WavError::Decode(
"incomplete frame at end of chunk".into(),
));
let mut remaining = chunk_size;
let bytes_per_frame = fmt.bytes_per_sample() * fmt.channels as usize;
let mut buffer = vec![0u8; 4096];
while remaining > 0 {
let to_read = remaining.min(buffer.len());
let read = wav_reader.reader.read(&mut buffer[..to_read])?;
if read == 0 {
break;
}
remaining -= read;
let aligned = read - (read % bytes_per_frame);
if aligned > 0 {
if pcm_tx
.blocking_send(Ok(buffer[..aligned].to_vec()))
.is_err()
{
return Ok(());
}
}
if aligned < read {
return Err(WavError::Decode(
"incomplete frame at end of chunk".into(),
));
}
}
}
if padded_size > chunk_size {
let mut pad = [0u8; 1];
wav_reader.reader.read_exact(&mut pad)?;
}
if padded_size > chunk_size {
let mut pad = [0u8; 1];
wav_reader.reader.read_exact(&mut pad)?;
}
data_found = true;
break;
}
_ => {
wav_reader.skip(chunk_size)?;
if padded_size > chunk_size {
wav_reader.skip(padded_size - chunk_size)?;
data_found = true;
break;
}
_ => {
wav_reader.skip(chunk_size)?;
if padded_size > chunk_size {
wav_reader.skip(padded_size - chunk_size)?;
}
}
}
}
}
if !data_found {
return Err(WavError::Decode("no data chunk found in WAV stream".into()));
}
if !data_found {
return Err(WavError::Decode("no data chunk found in WAV stream".into()));
}
Ok(())
Ok(())
})();
match result {
Ok(()) => Ok(()),
Err(err) => {
if let Some(tx) = info_tx.take() {
let _ = tx.send(Err(err.clone()));
}
Err(err)
}
}
});
let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "wav-decode");