use std::borrow::Cow;
use std::io::{Read, Write};
use std::sync::Arc;
use ad_core_rs::codec::{Codec, CodecName, CodecStatus};
use ad_core_rs::ndarray::{NDArray, NDDataBuffer, NDDataType, NDDimension};
use ad_core_rs::ndarray_pool::NDArrayPool;
use ad_core_rs::plugin::runtime::{NDPluginProcess, ParamUpdate, ProcessResult};
use flate2::Compression;
use flate2::read::ZlibDecoder;
use flate2::write::ZlibEncoder;
use lz4_flex::block::{compress, decompress};
use rust_hdf5::format::messages::filter::{
FILTER_BLOSC, Filter, FilterPipeline, apply_filters, reverse_filters,
};
pub fn original_data_type(array: &NDArray) -> NDDataType {
match &array.codec {
Some(c) => c.original_data_type,
None => array.data.data_type(),
}
}
pub(crate) fn buffer_from_bytes(bytes: &[u8], data_type: NDDataType) -> Option<NDDataBuffer> {
let elem_size = data_type.element_size();
if bytes.len() % elem_size != 0 {
return None;
}
let count = bytes.len() / elem_size;
Some(match data_type {
NDDataType::Int8 => {
let mut v = vec![0i8; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::I8(v)
}
NDDataType::UInt8 => NDDataBuffer::U8(bytes.to_vec()),
NDDataType::Int16 => {
let mut v = vec![0i16; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::I16(v)
}
NDDataType::UInt16 => {
let mut v = vec![0u16; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::U16(v)
}
NDDataType::Int32 => {
let mut v = vec![0i32; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::I32(v)
}
NDDataType::UInt32 => {
let mut v = vec![0u32; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::U32(v)
}
NDDataType::Int64 => {
let mut v = vec![0i64; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::I64(v)
}
NDDataType::UInt64 => {
let mut v = vec![0u64; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::U64(v)
}
NDDataType::Float32 => {
let mut v = vec![0f32; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::F32(v)
}
NDDataType::Float64 => {
let mut v = vec![0f64; count];
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
v.as_mut_ptr() as *mut u8,
bytes.len(),
);
}
NDDataBuffer::F64(v)
}
})
}
pub fn compress_lz4(src: &NDArray) -> NDArray {
let raw = src.data.as_u8_slice();
let original_data_type = src.data.data_type();
let original_size = raw.len();
let compressed = compress(raw);
let compressed_size = compressed.len();
let mut arr = src.clone();
arr.data = NDDataBuffer::U8(compressed);
arr.codec = Some(Codec {
name: CodecName::LZ4,
compressed_size,
level: 0,
shuffle: 0,
compressor: 0,
original_data_type,
});
tracing::debug!(
original_size,
compressed_size,
ratio = original_size as f64 / compressed_size.max(1) as f64,
"LZ4 compress"
);
arr
}
pub fn decompress_lz4(src: &NDArray) -> Option<NDArray> {
if src.codec.as_ref().map(|c| c.name) != Some(CodecName::LZ4) {
return None;
}
let compressed = src.data.as_u8_slice();
let original_type = original_data_type(src);
let num_elements: usize = src.dims.iter().map(|d| d.size).product();
let uncompressed_size = num_elements * original_type.element_size();
let decompressed = decompress(compressed, uncompressed_size).ok()?;
let buffer = buffer_from_bytes(&decompressed, original_type)?;
let mut arr = src.clone();
arr.data = buffer;
arr.codec = None;
Some(arr)
}
const ZLIB_DEFAULT_LEVEL: u32 = 6;
pub fn compress_zlib(src: &NDArray) -> NDArray {
let raw = src.data.as_u8_slice();
let original_data_type = src.data.data_type();
let original_size = raw.len();
let mut encoder = ZlibEncoder::new(Vec::<u8>::new(), Compression::new(ZLIB_DEFAULT_LEVEL));
if encoder.write_all(raw).is_err() {
return src.clone();
}
let compressed = match encoder.finish() {
Ok(buf) => buf,
Err(_) => return src.clone(),
};
let compressed_size = compressed.len();
let mut arr = src.clone();
arr.data = NDDataBuffer::U8(compressed);
arr.codec = Some(Codec {
name: CodecName::Zlib,
compressed_size,
level: ZLIB_DEFAULT_LEVEL as i32,
shuffle: 0,
compressor: 0,
original_data_type,
});
tracing::debug!(
original_size,
compressed_size,
ratio = original_size as f64 / compressed_size.max(1) as f64,
"Zlib compress"
);
arr
}
pub fn decompress_zlib(src: &NDArray) -> Option<NDArray> {
if src.codec.as_ref().map(|c| c.name) != Some(CodecName::Zlib) {
return None;
}
let compressed = src.data.as_u8_slice();
let original_type = original_data_type(src);
let num_elements: usize = src.dims.iter().map(|d| d.size).product();
let uncompressed_size = num_elements * original_type.element_size();
let mut decoder = ZlibDecoder::new(compressed);
let mut decompressed = Vec::with_capacity(uncompressed_size);
decoder.read_to_end(&mut decompressed).ok()?;
let buffer = buffer_from_bytes(&decompressed, original_type)?;
let mut arr = src.clone();
arr.data = buffer;
arr.codec = None;
Some(arr)
}
const LZ4HDF5_DEFAULT_BLOCK_SIZE: usize = 1 << 20;
pub fn compress_lz4hdf5(src: &NDArray) -> NDArray {
let raw = src.data.as_u8_slice();
let data_type = src.data.data_type();
let original_size = raw.len();
let block_size = LZ4HDF5_DEFAULT_BLOCK_SIZE;
let mut out: Vec<u8> = Vec::with_capacity(original_size / 2 + 12);
out.extend_from_slice(&(original_size as u64).to_be_bytes());
out.extend_from_slice(&(block_size as u32).to_be_bytes());
let mut pos = 0usize;
while pos < raw.len() {
let n = block_size.min(raw.len() - pos);
let block = &raw[pos..pos + n];
let comp = compress(block);
if comp.len() < n {
out.extend_from_slice(&(comp.len() as u32).to_be_bytes());
out.extend_from_slice(&comp);
} else {
out.extend_from_slice(&(n as u32).to_be_bytes());
out.extend_from_slice(block);
}
pos += n;
}
let compressed_size = out.len();
let mut arr = src.clone();
arr.data = NDDataBuffer::U8(out);
arr.codec = Some(Codec {
name: CodecName::LZ4HDF5,
compressed_size,
level: 0,
shuffle: 0,
compressor: 0,
original_data_type: data_type,
});
tracing::debug!(
original_size,
compressed_size,
ratio = original_size as f64 / compressed_size.max(1) as f64,
"LZ4HDF5 compress"
);
arr
}
pub fn decompress_lz4hdf5(src: &NDArray) -> Option<NDArray> {
if src.codec.as_ref().map(|c| c.name) != Some(CodecName::LZ4HDF5) {
return None;
}
let buf = src.data.as_u8_slice();
if buf.len() < 12 {
return None;
}
let total_bytes = u64::from_be_bytes(buf[0..8].try_into().ok()?) as usize;
let block_size = u32::from_be_bytes(buf[8..12].try_into().ok()?) as usize;
if block_size == 0 {
return None;
}
let original_type = original_data_type(src);
let mut out: Vec<u8> = Vec::with_capacity(total_bytes);
let mut pos = 12usize;
while out.len() < total_bytes {
let n = block_size.min(total_bytes - out.len());
if pos + 4 > buf.len() {
return None;
}
let clen = u32::from_be_bytes(buf[pos..pos + 4].try_into().ok()?) as usize;
pos += 4;
if pos + clen > buf.len() {
return None;
}
let block_payload = &buf[pos..pos + clen];
if clen == n {
out.extend_from_slice(block_payload);
} else {
let block = decompress(block_payload, n).ok()?;
if block.len() != n {
return None;
}
out.extend_from_slice(&block);
}
pos += clen;
}
if out.len() != total_bytes {
return None;
}
let buffer = buffer_from_bytes(&out, original_type)?;
let mut arr = src.clone();
arr.data = buffer;
arr.codec = None;
Some(arr)
}
const BSHUF_TARGET_BLOCK_SIZE_B: usize = 8192;
const BSHUF_BLOCKED_MULT: usize = 8;
const BSHUF_MIN_RECOMMEND_BLOCK: usize = 128;
pub(crate) fn bshuf_default_block_size(elem_size: usize) -> usize {
let bs = BSHUF_TARGET_BLOCK_SIZE_B / elem_size.max(1);
let bs = (bs / BSHUF_BLOCKED_MULT) * BSHUF_BLOCKED_MULT;
bs.max(BSHUF_MIN_RECOMMEND_BLOCK)
}
#[inline]
fn trans_bit_8x8(mut x: u64) -> u64 {
let t = (x ^ (x >> 7)) & 0x00AA_00AA_00AA_00AA;
x = x ^ t ^ (t << 7);
let t = (x ^ (x >> 14)) & 0x0000_CCCC_0000_CCCC;
x = x ^ t ^ (t << 14);
let t = (x ^ (x >> 28)) & 0x0000_0000_F0F0_F0F0;
x = x ^ t ^ (t << 28);
x
}
#[inline]
fn read_u64_le(b: &[u8], off: usize) -> u64 {
u64::from_le_bytes(b[off..off + 8].try_into().unwrap())
}
fn bshuf_trans_byte_elem(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
let mut ii = 0;
while ii + 7 < size {
for jj in 0..elem_size {
for kk in 0..8 {
out[jj * size + ii + kk] = input[ii * elem_size + kk * elem_size + jj];
}
}
ii += 8;
}
let mut ii = size - size % 8;
while ii < size {
for jj in 0..elem_size {
out[jj * size + ii] = input[ii * elem_size + jj];
}
ii += 1;
}
}
fn bshuf_trans_bit_byte(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
let nbyte = elem_size * size;
let nbyte_bitrow = nbyte / 8;
for ii in 0..nbyte_bitrow {
let mut x = trans_bit_8x8(read_u64_le(input, ii * 8));
for kk in 0..8 {
out[kk * nbyte_bitrow + ii] = x as u8;
x >>= 8;
}
}
}
fn bshuf_trans_bitrow_eight(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
let nbyte_bitrow = size / 8;
for ii in 0..8 {
for jj in 0..elem_size {
let src = (ii * elem_size + jj) * nbyte_bitrow;
let dst = (jj * 8 + ii) * nbyte_bitrow;
out[dst..dst + nbyte_bitrow].copy_from_slice(&input[src..src + nbyte_bitrow]);
}
}
}
fn bshuf_trans_bit_elem(input: &[u8], size: usize, elem_size: usize) -> Vec<u8> {
debug_assert_eq!(size % 8, 0);
let nbyte = size * elem_size;
let mut a = vec![0u8; nbyte];
bshuf_trans_byte_elem(input, &mut a, size, elem_size);
let mut b = vec![0u8; nbyte];
bshuf_trans_bit_byte(&a, &mut b, size, elem_size);
let mut out = vec![0u8; nbyte];
bshuf_trans_bitrow_eight(&b, &mut out, size, elem_size);
out
}
fn bshuf_trans_byte_bitrow(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
let nbyte_row = size / 8;
for jj in 0..elem_size {
for ii in 0..nbyte_row {
for kk in 0..8 {
out[ii * 8 * elem_size + jj * 8 + kk] = input[(jj * 8 + kk) * nbyte_row + ii];
}
}
}
}
fn bshuf_shuffle_bit_eightelem(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
let nbyte = elem_size * size;
let mut jj = 0;
while jj < 8 * elem_size {
let mut ii = 0;
while ii + 8 * elem_size - 1 < nbyte {
let mut x = trans_bit_8x8(read_u64_le(input, ii + jj));
for kk in 0..8 {
out[ii + jj / 8 + kk * elem_size] = x as u8;
x >>= 8;
}
ii += 8 * elem_size;
}
jj += 8;
}
}
fn bshuf_untrans_bit_elem(input: &[u8], size: usize, elem_size: usize) -> Vec<u8> {
debug_assert_eq!(size % 8, 0);
let nbyte = size * elem_size;
let mut tmp = vec![0u8; nbyte];
bshuf_trans_byte_bitrow(input, &mut tmp, size, elem_size);
let mut out = vec![0u8; nbyte];
bshuf_shuffle_bit_eightelem(&tmp, &mut out, size, elem_size);
out
}
fn bshuf_compress_lz4_block(
out: &mut Vec<u8>,
raw: &[u8],
elem_start: usize,
size: usize,
elem_size: usize,
) {
let off = elem_start * elem_size;
let shuffled = bshuf_trans_bit_elem(&raw[off..off + size * elem_size], size, elem_size);
let comp = compress(&shuffled);
out.extend_from_slice(&(comp.len() as u32).to_be_bytes());
out.extend_from_slice(&comp);
}
fn bshuf_decompress_lz4_block(
buf: &[u8],
pos: usize,
size: usize,
elem_size: usize,
) -> Option<(Vec<u8>, usize)> {
if pos + 4 > buf.len() {
return None;
}
let clen = u32::from_be_bytes(buf[pos..pos + 4].try_into().ok()?) as usize;
let dstart = pos + 4;
if dstart + clen > buf.len() {
return None;
}
let shuffled = decompress(&buf[dstart..dstart + clen], size * elem_size).ok()?;
if shuffled.len() != size * elem_size {
return None;
}
Some((
bshuf_untrans_bit_elem(&shuffled, size, elem_size),
dstart + clen,
))
}
pub fn compress_bslz4(src: &NDArray) -> NDArray {
let raw = src.data.as_u8_slice();
let data_type = src.data.data_type();
let elem_size = data_type.element_size();
let total_elems = if elem_size > 0 {
raw.len() / elem_size
} else {
0
};
let block_size = bshuf_default_block_size(elem_size);
let mut out: Vec<u8> = Vec::with_capacity(raw.len() / 2 + 16);
let n_full = total_elems / block_size;
let mut elem = 0usize;
for _ in 0..n_full {
bshuf_compress_lz4_block(&mut out, raw, elem, block_size, elem_size);
elem += block_size;
}
let mut last_block = total_elems % block_size;
last_block -= last_block % BSHUF_BLOCKED_MULT;
if last_block > 0 {
bshuf_compress_lz4_block(&mut out, raw, elem, last_block, elem_size);
elem += last_block;
}
if elem < total_elems {
out.extend_from_slice(&raw[elem * elem_size..total_elems * elem_size]);
}
let compressed_size = out.len();
let mut arr = src.clone();
arr.data = NDDataBuffer::U8(out);
arr.codec = Some(Codec {
name: CodecName::BSLZ4,
compressed_size,
level: 0,
shuffle: 0,
compressor: 0,
original_data_type: data_type,
});
tracing::debug!(
original_size = raw.len(),
compressed_size,
ratio = raw.len() as f64 / compressed_size.max(1) as f64,
"BSLZ4 compress"
);
arr
}
pub fn decompress_bslz4(src: &NDArray) -> Option<NDArray> {
let codec = src.codec.as_ref()?;
if codec.name != CodecName::BSLZ4 {
return None;
}
let buf = src.data.as_u8_slice();
let original_type = original_data_type(src);
let elem_size = original_type.element_size();
if elem_size == 0 {
return None;
}
let total_elems: usize = src.dims.iter().map(|d| d.size).product();
let total_bytes = total_elems * elem_size;
let block_size = bshuf_default_block_size(elem_size);
let mut out: Vec<u8> = Vec::with_capacity(total_bytes);
let mut pos = 0usize;
let n_full = total_elems / block_size;
for _ in 0..n_full {
let (block, next) = bshuf_decompress_lz4_block(buf, pos, block_size, elem_size)?;
out.extend_from_slice(&block);
pos = next;
}
let mut last_block = total_elems % block_size;
last_block -= last_block % BSHUF_BLOCKED_MULT;
if last_block > 0 {
let (block, next) = bshuf_decompress_lz4_block(buf, pos, last_block, elem_size)?;
out.extend_from_slice(&block);
pos = next;
}
let leftover_bytes = (total_elems % BSHUF_BLOCKED_MULT) * elem_size;
if leftover_bytes > 0 {
if pos + leftover_bytes > buf.len() {
return None;
}
out.extend_from_slice(&buf[pos..pos + leftover_bytes]);
}
if out.len() != total_bytes {
return None;
}
let buffer = buffer_from_bytes(&out, original_type)?;
let mut arr = src.clone();
arr.data = buffer;
arr.codec = None;
Some(arr)
}
pub fn compress_jpeg(src: &NDArray, quality: u8) -> Result<NDArray, JpegCompressError> {
use ad_core_rs::color::{NDColorMode, convert_rgb_layout};
match src.data.data_type() {
NDDataType::UInt8 | NDDataType::Int8 => {}
_ => return Err(JpegCompressError::NotEightBit),
}
let info = src.info();
if !matches!(src.dims.len(), 2 | 3) {
return Err(JpegCompressError::UnsupportedArrayStructure);
}
match info.color_mode {
NDColorMode::Mono | NDColorMode::RGB1 | NDColorMode::RGB2 | NDColorMode::RGB3 => {}
mode => return Err(JpegCompressError::UnknownColorMode(mode as i32)),
}
if info.x_size > u16::MAX as usize || info.y_size > u16::MAX as usize {
return Err(JpegCompressError::EncodeFailed);
}
let (color_type, interleaved) = match (src.dims.len(), info.color_mode) {
(2, NDColorMode::Mono | NDColorMode::RGB1) => (jpeg_encoder::ColorType::Luma, None),
(3, NDColorMode::RGB1) if info.color_size == 3 => (jpeg_encoder::ColorType::Rgb, None),
(3, mode @ (NDColorMode::RGB2 | NDColorMode::RGB3)) if info.color_size == 3 => (
jpeg_encoder::ColorType::Rgb,
Some(
convert_rgb_layout(src, mode, NDColorMode::RGB1)
.map_err(|_| JpegCompressError::EncodeFailed)?,
),
),
_ => return Err(JpegCompressError::EncodeFailed),
};
let width = info.x_size as u16;
let height = info.y_size as u16;
let pixels = interleaved
.as_ref()
.map_or_else(|| src.data.as_u8_slice(), |a| a.data.as_u8_slice());
let mut jpeg_buf = Vec::new();
let encoder = jpeg_encoder::Encoder::new(&mut jpeg_buf, quality);
if encoder.encode(pixels, width, height, color_type).is_err() {
return Err(JpegCompressError::EncodeFailed);
}
let compressed_size = jpeg_buf.len();
let original_size = src.data.as_u8_slice().len();
let mut arr = src.clone();
arr.data = NDDataBuffer::U8(jpeg_buf);
arr.codec = Some(Codec {
name: CodecName::JPEG,
compressed_size,
level: 0,
shuffle: 0,
compressor: 0,
original_data_type: src.data.data_type(),
});
tracing::debug!(
original_size,
compressed_size,
ratio = original_size as f64 / compressed_size.max(1) as f64,
"JPEG compress (quality={})",
quality,
);
Ok(arr)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JpegCompressError {
NotEightBit,
UnsupportedArrayStructure,
UnknownColorMode(i32),
EncodeFailed,
}
impl JpegCompressError {
pub fn message(&self) -> Cow<'static, str> {
match self {
Self::NotEightBit => "JPEG only supports 8-bit data".into(),
Self::UnsupportedArrayStructure => "Unsupported array structure".into(),
Self::UnknownColorMode(mode) => format!("Unknown color mode {}", mode).into(),
Self::EncodeFailed => "Error writing JPEG data".into(),
}
}
}
pub fn decompress_jpeg(src: &NDArray) -> Option<NDArray> {
use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
use ad_core_rs::color::NDColorMode;
if src.codec.as_ref().map(|c| c.name) != Some(CodecName::JPEG) {
return None;
}
let compressed = src.data.as_u8_slice();
let mut decoder = jpeg_decoder::Decoder::new(compressed);
let pixels = decoder.decode().ok()?;
let metadata = decoder.info()?;
let width = metadata.width as usize;
let height = metadata.height as usize;
let (dims, color_mode) = match metadata.pixel_format {
jpeg_decoder::PixelFormat::L8 => (
vec![NDDimension::new(width), NDDimension::new(height)],
NDColorMode::Mono,
),
jpeg_decoder::PixelFormat::RGB24 => (
vec![
NDDimension::new(3),
NDDimension::new(width),
NDDimension::new(height),
],
NDColorMode::RGB1,
),
_ => return None,
};
let mut arr = src.clone();
arr.dims = dims;
arr.data = NDDataBuffer::U8(pixels);
arr.codec = None;
arr.attributes.add(NDAttribute::new_static(
"ColorMode",
"Color Mode",
NDAttrSource::Driver,
NDAttrValue::Int32(color_mode as i32),
));
Some(arr)
}
#[derive(Debug, Clone, Copy)]
pub struct BloscConfig {
pub compressor: u32,
pub clevel: u32,
pub shuffle: u32,
}
impl Default for BloscConfig {
fn default() -> Self {
Self {
compressor: 0,
clevel: 5,
shuffle: 0,
}
}
}
pub fn compress_blosc(src: &NDArray, config: &BloscConfig) -> NDArray {
let raw = src.data.as_u8_slice();
let element_size = src.data.data_type().element_size();
let pipeline = FilterPipeline {
filters: vec![Filter {
id: FILTER_BLOSC,
flags: 0,
cd_values: vec![
2, 2, element_size as u32, raw.len() as u32, config.clevel, config.shuffle, config.compressor, ],
}],
};
let compressed = match apply_filters(&pipeline, raw) {
Ok(data) => data,
Err(_) => return src.clone(),
};
let compressed_size = compressed.len();
let original_data_type = src.data.data_type();
let mut arr = src.clone();
arr.data = NDDataBuffer::U8(compressed);
arr.codec = Some(Codec {
name: CodecName::Blosc,
compressed_size,
level: config.clevel as i32,
shuffle: config.shuffle as i32,
compressor: config.compressor as i32,
original_data_type,
});
arr
}
pub fn decompress_blosc(src: &NDArray) -> Option<NDArray> {
let codec = src.codec.as_ref()?;
if codec.name != CodecName::Blosc {
return None;
}
let compressed = src.data.as_u8_slice();
let original_type = original_data_type(src);
let element_size = original_type.element_size();
let pipeline = FilterPipeline {
filters: vec![Filter {
id: FILTER_BLOSC,
flags: 0,
cd_values: vec![
2,
2,
element_size as u32,
0,
codec.level as u32,
codec.shuffle as u32,
codec.compressor as u32,
],
}],
};
let decompressed = reverse_filters(&pipeline, compressed).ok()?;
let buffer = buffer_from_bytes(&decompressed, original_type)?;
let mut arr = src.clone();
arr.data = buffer;
arr.codec = None;
Some(arr)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CodecMode {
Compress { codec: CodecName, quality: u8 },
Decompress,
}
#[derive(Default)]
struct CodecParamIndices {
mode: Option<usize>,
compressor: Option<usize>,
comp_factor: Option<usize>,
jpeg_quality: Option<usize>,
blosc_compressor: Option<usize>,
blosc_clevel: Option<usize>,
blosc_shuffle: Option<usize>,
blosc_numthreads: Option<usize>,
codec_status: Option<usize>,
codec_error: Option<usize>,
}
pub struct CodecProcessor {
mode: CodecMode,
compression_ratio: f64,
jpeg_quality: u8,
blosc_config: BloscConfig,
params: CodecParamIndices,
}
impl CodecProcessor {
pub fn new(mode: CodecMode) -> Self {
let quality = match mode {
CodecMode::Compress { quality, .. } => quality,
_ => 85,
};
Self {
mode,
compression_ratio: 1.0,
jpeg_quality: quality,
blosc_config: BloscConfig::default(),
params: CodecParamIndices::default(),
}
}
pub fn compression_ratio(&self) -> f64 {
self.compression_ratio
}
}
enum CodecOutcome {
PassThrough,
Skipped(&'static str),
Converted(NDArray),
Failed(Cow<'static, str>),
}
impl CodecOutcome {
fn status(&self) -> CodecStatus {
match self {
Self::PassThrough | Self::Converted(_) => CodecStatus::Success,
Self::Skipped(_) => CodecStatus::Warning,
Self::Failed(_) => CodecStatus::Error,
}
}
fn error_message(&self) -> &str {
match self {
Self::PassThrough | Self::Converted(_) => "",
Self::Skipped(message) => message,
Self::Failed(message) => message,
}
}
}
impl NDPluginProcess for CodecProcessor {
fn process_array(&mut self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
let original_bytes = array.data.as_u8_slice().len();
let mut compressor: Option<i32> = None;
let outcome = match self.mode {
CodecMode::Compress {
codec: CodecName::None,
..
} => CodecOutcome::PassThrough,
CodecMode::Compress { .. } if array.codec.is_some() => {
CodecOutcome::Skipped("Array already compressed")
}
CodecMode::Compress { codec, .. } => match codec {
CodecName::LZ4 => CodecOutcome::Converted(compress_lz4(array)),
CodecName::JPEG => match compress_jpeg(array, self.jpeg_quality) {
Ok(out) => CodecOutcome::Converted(out),
Err(e) => CodecOutcome::Failed(e.message()),
},
CodecName::Zlib => CodecOutcome::Converted(compress_zlib(array)),
CodecName::Blosc => {
CodecOutcome::Converted(compress_blosc(array, &self.blosc_config))
}
CodecName::LZ4HDF5 => CodecOutcome::Converted(compress_lz4hdf5(array)),
CodecName::BSLZ4 => CodecOutcome::Converted(compress_bslz4(array)),
CodecName::None => CodecOutcome::PassThrough,
},
CodecMode::Decompress => {
let name = array
.codec
.as_ref()
.map(|c| c.name)
.unwrap_or(CodecName::None);
compressor = Some(name.ordinal());
match name {
CodecName::None => CodecOutcome::PassThrough,
CodecName::LZ4 => match decompress_lz4(array) {
Some(out) => CodecOutcome::Converted(out),
None => CodecOutcome::Failed("Failed to LZ4 decompress".into()),
},
CodecName::JPEG => match decompress_jpeg(array) {
Some(out) => CodecOutcome::Converted(out),
None => CodecOutcome::Failed("Error decoding JPEG".into()),
},
CodecName::Zlib => match decompress_zlib(array) {
Some(out) => CodecOutcome::Converted(out),
None => CodecOutcome::Failed("Failed to Zlib decompress".into()),
},
CodecName::Blosc => match decompress_blosc(array) {
Some(out) => CodecOutcome::Converted(out),
None => CodecOutcome::Failed("Failed to Blosc decompress".into()),
},
CodecName::LZ4HDF5 => match decompress_lz4hdf5(array) {
Some(out) => CodecOutcome::Converted(out),
None => CodecOutcome::Failed("Failed to LZ4 decompress".into()),
},
CodecName::BSLZ4 => match decompress_bslz4(array) {
Some(out) => CodecOutcome::Converted(out),
None => CodecOutcome::Failed("Failed to Blosc decompress".into()),
},
}
}
};
let status = outcome.status();
let error = outcome.error_message().to_string();
let output = match outcome {
CodecOutcome::Converted(out) => {
let output_bytes = out.data.as_u8_slice().len();
self.compression_ratio = match self.mode {
CodecMode::Compress { .. } => {
original_bytes as f64 / output_bytes.max(1) as f64
}
CodecMode::Decompress => output_bytes as f64 / original_bytes.max(1) as f64,
};
out
}
CodecOutcome::PassThrough | CodecOutcome::Skipped(_) | CodecOutcome::Failed(_) => {
self.compression_ratio = 1.0;
array.clone()
}
};
let mut updates = Vec::new();
if let Some(idx) = self.params.comp_factor {
updates.push(ParamUpdate::float64(idx, self.compression_ratio));
}
if let (Some(idx), Some(value)) = (self.params.compressor, compressor) {
updates.push(ParamUpdate::int32(idx, value));
}
if let Some(idx) = self.params.codec_status {
updates.push(ParamUpdate::int32(idx, status.as_i32()));
}
if let Some(idx) = self.params.codec_error {
updates.push(ParamUpdate::Octet {
reason: idx,
addr: 0,
value: error,
});
}
let mut r = ProcessResult::arrays(vec![Arc::new(output)]);
r.param_updates = updates;
r
}
fn plugin_type(&self) -> &str {
"NDPluginCodec"
}
fn compression_aware(&self) -> bool {
true
}
fn register_params(
&mut self,
base: &mut asyn_rs::port::PortDriverBase,
) -> asyn_rs::error::AsynResult<()> {
use asyn_rs::param::ParamType;
base.create_param("MODE", ParamType::Int32)?;
base.create_param("COMPRESSOR", ParamType::Int32)?;
base.create_param("COMP_FACTOR", ParamType::Float64)?;
base.create_param("JPEG_QUALITY", ParamType::Int32)?;
base.create_param("BLOSC_COMPRESSOR", ParamType::Int32)?;
base.create_param("BLOSC_CLEVEL", ParamType::Int32)?;
base.create_param("BLOSC_SHUFFLE", ParamType::Int32)?;
base.create_param("BLOSC_NUMTHREADS", ParamType::Int32)?;
base.create_param("CODEC_STATUS", ParamType::Int32)?;
base.create_param("CODEC_ERROR", ParamType::Octet)?;
self.params.mode = base.find_param("MODE");
self.params.compressor = base.find_param("COMPRESSOR");
self.params.comp_factor = base.find_param("COMP_FACTOR");
self.params.jpeg_quality = base.find_param("JPEG_QUALITY");
self.params.blosc_compressor = base.find_param("BLOSC_COMPRESSOR");
self.params.blosc_clevel = base.find_param("BLOSC_CLEVEL");
self.params.blosc_shuffle = base.find_param("BLOSC_SHUFFLE");
self.params.blosc_numthreads = base.find_param("BLOSC_NUMTHREADS");
self.params.codec_status = base.find_param("CODEC_STATUS");
self.params.codec_error = base.find_param("CODEC_ERROR");
Ok(())
}
fn on_param_change(
&mut self,
reason: usize,
params: &ad_core_rs::plugin::runtime::PluginParamSnapshot,
) -> ad_core_rs::plugin::runtime::ParamChangeResult {
if Some(reason) == self.params.mode {
let v = params.value.as_i32();
if v == 0 {
let codec = match self.mode {
CodecMode::Compress { codec, .. } => codec,
_ => CodecName::LZ4,
};
self.mode = CodecMode::Compress {
codec,
quality: self.jpeg_quality,
};
} else {
self.mode = CodecMode::Decompress;
}
} else if Some(reason) == self.params.compressor {
let codec = CodecName::from_ordinal(params.value.as_i32());
if let CodecMode::Compress { .. } = self.mode {
self.mode = CodecMode::Compress {
codec,
quality: self.jpeg_quality,
};
}
} else if Some(reason) == self.params.jpeg_quality {
self.jpeg_quality = params.value.as_i32().clamp(1, 100) as u8;
if let CodecMode::Compress { codec, .. } = self.mode {
self.mode = CodecMode::Compress {
codec,
quality: self.jpeg_quality,
};
}
} else if Some(reason) == self.params.blosc_compressor {
self.blosc_config.compressor = params.value.as_i32().max(0) as u32;
} else if Some(reason) == self.params.blosc_clevel {
self.blosc_config.clevel = params.value.as_i32().clamp(0, 9) as u32;
} else if Some(reason) == self.params.blosc_shuffle {
self.blosc_config.shuffle = params.value.as_i32().max(0) as u32;
}
ad_core_rs::plugin::runtime::ParamChangeResult::updates(vec![])
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_u8_array(width: usize, height: usize) -> NDArray {
let mut arr = NDArray::new(
vec![NDDimension::new(width), NDDimension::new(height)],
NDDataType::UInt8,
);
if let NDDataBuffer::U8(ref mut v) = arr.data {
for i in 0..v.len() {
v[i] = (i % 256) as u8;
}
}
arr
}
fn make_rgb_array(width: usize, height: usize) -> NDArray {
use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
let mut arr = NDArray::new(
vec![
NDDimension::new(3),
NDDimension::new(width),
NDDimension::new(height),
],
NDDataType::UInt8,
);
arr.attributes.add(NDAttribute::new_static(
"ColorMode",
"Color Mode",
NDAttrSource::Driver,
NDAttrValue::Int32(2), ));
if let NDDataBuffer::U8(ref mut v) = arr.data {
for i in 0..v.len() {
v[i] = (i % 256) as u8;
}
}
arr
}
#[test]
fn compressors_record_type_in_codec_not_an_attribute() {
let mut arr = NDArray::new(vec![NDDimension::new(8)], NDDataType::UInt16);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 7) as u16;
}
}
for compressed in [
compress_lz4(&arr),
compress_zlib(&arr),
compress_lz4hdf5(&arr),
compress_bslz4(&arr),
compress_blosc(&arr, &BloscConfig::default()),
] {
assert_eq!(
compressed.codec.as_ref().unwrap().original_data_type,
NDDataType::UInt16,
"the original element type must travel in the codec"
);
assert!(
compressed
.attributes
.get("CODEC_ORIGINAL_DATA_TYPE")
.is_none(),
"no codec carrier attribute may be attached to a compressed frame"
);
}
}
#[test]
fn test_adp29_blosc_default_clevel_and_codec_params() {
assert_eq!(
BloscConfig::default().clevel,
5,
"default Blosc clevel must be 5 (C parity)"
);
let mut arr = NDArray::new(vec![NDDimension::new(8)], NDDataType::UInt16);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 7) as u16;
}
}
let out = compress_blosc(&arr, &BloscConfig::default());
let codec = out.codec.as_ref().expect("blosc codec metadata");
assert_eq!(codec.level, 5, "codec.level records the default clevel 5");
assert_eq!(codec.shuffle, 0, "codec.shuffle records shuffle");
assert_eq!(codec.compressor, 0, "codec.compressor records compressor");
}
#[test]
fn test_blosc_roundtrip_u16_default_compressor() {
let mut arr = NDArray::new(
vec![NDDimension::new(100), NDDimension::new(20)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 37 % 65521) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_blosc(&arr, &BloscConfig::default());
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::Blosc);
assert_ne!(
compressed.data.as_u8_slice(),
original.as_slice(),
"blosc must actually compress (not fall back to the raw clone)"
);
let decompressed = decompress_blosc(&compressed).expect("blosc round-trip");
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_blosc_roundtrip_u16_lz4_subcompressor() {
let cfg = BloscConfig {
compressor: 1,
clevel: 5,
shuffle: 1,
};
let mut arr = NDArray::new(vec![NDDimension::new(256)], NDDataType::UInt16);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 13 % 65521) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_blosc(&arr, &cfg);
let codec = compressed.codec.as_ref().unwrap();
assert_eq!(codec.compressor, 1, "records the LZ4 sub-compressor");
assert_eq!(codec.shuffle, 1, "records byte shuffle");
let decompressed = decompress_blosc(&compressed).expect("blosc lz4 round-trip");
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_lz4_roundtrip_u8() {
let arr = make_u8_array(4, 4);
let original_data = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4(&arr);
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4);
assert_ne!(compressed.data.as_u8_slice(), original_data.as_slice());
let decompressed = decompress_lz4(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
assert_eq!(decompressed.data.as_u8_slice(), original_data.as_slice());
}
#[test]
fn test_decompress_runtime_does_not_drop_compressed_input() {
use ad_core_rs::plugin::channel::{NDArrayOutput, ndarray_channel};
use ad_core_rs::plugin::runtime::create_plugin_runtime_with_output;
use ad_core_rs::plugin::wiring::WiringRegistry;
use std::sync::atomic::Ordering;
let mut raw = make_u8_array(4, 4);
raw.unique_id = 1;
let original_data = raw.data.as_u8_slice().to_vec();
let compressed = compress_lz4(&raw);
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4);
assert_eq!(compressed.unique_id, 1);
let mut sentinel = make_u8_array(4, 4);
sentinel.unique_id = 2;
let pool = Arc::new(NDArrayPool::new(1_000_000));
let (ds_sender, mut ds_rx) = ndarray_channel("DS", 10);
let mut output = NDArrayOutput::new();
output.add(ds_sender);
let (handle, _jh) = create_plugin_runtime_with_output(
"CODEC_DECOMP",
CodecProcessor::new(CodecMode::Decompress),
pool,
10,
output,
"",
Arc::new(WiringRegistry::new()),
);
let dropped = handle.array_sender().dropped_arrays_counter().clone();
handle
.port_runtime()
.port_handle()
.write_int32_blocking(handle.plugin_params.enable_callbacks, 0, 1)
.unwrap();
assert!(
handle.wait_params_applied(std::time::Duration::from_secs(10)),
"data thread did not apply EnableCallbacks"
);
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(handle.array_sender().publish(Arc::new(compressed)));
rt.block_on(handle.array_sender().publish(Arc::new(sentinel)));
let first = ds_rx.blocking_recv().expect("downstream array");
assert_eq!(
first.unique_id, 1,
"compressed input must be decompressed and delivered, not dropped"
);
assert!(
first.codec.is_none(),
"delivered array must be decompressed (codec cleared)"
);
assert_eq!(first.data.as_u8_slice(), original_data.as_slice());
assert_eq!(
dropped.load(Ordering::Acquire),
0,
"compression-aware Codec must not count its compressed input as dropped"
);
}
#[test]
fn test_lz4_roundtrip_u16() {
let mut arr = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for i in 0..v.len() {
v[i] = (i * 100) as u16;
}
}
let original_bytes = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4(&arr);
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4);
assert_eq!(
compressed.codec.as_ref().unwrap().original_data_type,
NDDataType::UInt16
);
assert!(
compressed
.attributes
.get("CODEC_ORIGINAL_DATA_TYPE")
.is_none()
);
let decompressed = decompress_lz4(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
assert_eq!(decompressed.data.as_u8_slice(), original_bytes.as_slice());
}
#[test]
fn test_lz4_roundtrip_f64() {
let mut arr = NDArray::new(vec![NDDimension::new(16)], NDDataType::Float64);
if let NDDataBuffer::F64(ref mut v) = arr.data {
for i in 0..v.len() {
v[i] = i as f64 * 1.5;
}
}
let original_bytes = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4(&arr);
let decompressed = decompress_lz4(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
assert_eq!(decompressed.data.as_u8_slice(), original_bytes.as_slice());
}
#[test]
fn test_lz4_compresses_repetitive_data() {
let mut arr = NDArray::new(
vec![NDDimension::new(256), NDDimension::new(256)],
NDDataType::UInt8,
);
if let NDDataBuffer::U8(ref mut v) = arr.data {
for x in v.iter_mut() {
*x = 0;
}
}
let original_size = arr.data.as_u8_slice().len();
let compressed = compress_lz4(&arr);
let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
assert!(
compressed_size < original_size,
"compressed ({}) should be smaller than original ({})",
compressed_size,
original_size,
);
}
#[test]
fn test_lz4_preserves_metadata() {
let mut arr = make_u8_array(4, 4);
arr.unique_id = 42;
let compressed = compress_lz4(&arr);
assert_eq!(compressed.unique_id, 42);
assert_eq!(compressed.dims.len(), 2);
assert_eq!(compressed.dims[0].size, 4);
assert_eq!(compressed.dims[1].size, 4);
}
#[test]
fn test_bitshuffle_block_transpose_roundtrip() {
for &(n, elem_size) in &[(16usize, 4usize), (8, 2), (256, 8), (128, 1)] {
let input: Vec<u8> = (0..n * elem_size).map(|i| (i * 7 + 3) as u8).collect();
let shuffled = bshuf_trans_bit_elem(&input, n, elem_size);
assert_eq!(shuffled.len(), input.len());
let restored = bshuf_untrans_bit_elem(&shuffled, n, elem_size);
assert_eq!(restored, input, "elem_size {elem_size}, n {n}");
}
}
#[test]
fn test_bitshuffle_matches_c_reference_vector() {
let input: Vec<u8> = (0..16u16).flat_map(|v| v.to_le_bytes()).collect();
let shuffled = bshuf_trans_bit_elem(&input, 16, 2);
let mut expected = vec![0u8; 32];
expected[..8].copy_from_slice(&[170, 170, 204, 204, 240, 240, 0, 255]);
assert_eq!(
shuffled, expected,
"canonical bitshuffle transpose must match the C library bytes"
);
}
#[test]
fn test_bslz4_roundtrip_u8() {
let mut arr = NDArray::new(
vec![NDDimension::new(64), NDDimension::new(64)],
NDDataType::UInt8,
);
if let NDDataBuffer::U8(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i % 251) as u8;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_bslz4(&arr);
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::BSLZ4);
assert_ne!(compressed.data.as_u8_slice(), original.as_slice());
let decompressed = decompress_bslz4(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_bslz4_roundtrip_u16() {
let mut arr = NDArray::new(
vec![NDDimension::new(100), NDDimension::new(20)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 37 % 65521) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_bslz4(&arr);
assert_eq!(
compressed.codec.as_ref().unwrap().original_data_type,
NDDataType::UInt16
);
let decompressed = decompress_bslz4(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_bslz4_roundtrip_f64_with_negatives() {
let mut arr = NDArray::new(vec![NDDimension::new(73)], NDDataType::Float64);
if let NDDataBuffer::F64(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i as f64 - 36.0) * 2.5;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_bslz4(&arr);
let decompressed = decompress_bslz4(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_bslz4_roundtrip_multi_block() {
let elem_size = 4usize;
let block = bshuf_default_block_size(elem_size);
let count = block * 2 + block / 2 + 3;
let mut arr = NDArray::new(vec![NDDimension::new(count)], NDDataType::Int32);
if let NDDataBuffer::I32(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i as i32).wrapping_mul(2_654_435_761u32 as i32);
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_bslz4(&arr);
let decompressed = decompress_bslz4(&compressed).unwrap();
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_bslz4_compresses_repetitive_data() {
let arr = NDArray::new(
vec![NDDimension::new(256), NDDimension::new(256)],
NDDataType::UInt16,
);
let original_size = arr.data.as_u8_slice().len();
let compressed = compress_bslz4(&arr);
let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
assert!(
compressed_size < original_size,
"bslz4 compressed ({compressed_size}) should be < original ({original_size})"
);
}
#[test]
fn test_r9_71_corrupt_bslz4_reports_cs_blosc_text() {
use ad_core_rs::plugin::runtime::ParamUpdate;
let mut arr = NDArray::new(
vec![NDDimension::new(32), NDDimension::new(32)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 11) as u16;
}
}
let pool = NDArrayPool::new(10_000_000);
let mut compressed = compress_bslz4(&arr);
if let NDDataBuffer::U8(ref mut v) = compressed.data {
for b in v.iter_mut() {
*b = 0xFF;
}
}
assert!(
decompress_bslz4(&compressed).is_none(),
"the corrupted frame must fail to decompress"
);
let mut decomp = CodecProcessor::new(CodecMode::Decompress);
decomp.params.codec_error = Some(13);
let result = decomp.process_array(&compressed, &pool);
let text = result
.param_updates
.iter()
.find_map(|u| match u {
ParamUpdate::Octet {
reason: 13, value, ..
} => Some(value.clone()),
_ => None,
})
.expect("CodecError posted");
assert_eq!(text, "Failed to Blosc decompress");
}
#[test]
fn test_bslz4_via_processor() {
let mut arr = NDArray::new(
vec![NDDimension::new(32), NDDimension::new(32)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 11) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let pool = NDArrayPool::new(10_000_000);
let mut comp = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::BSLZ4,
quality: 0,
});
let compressed = comp.process_array(&arr, &pool);
let compressed_arr = &compressed.output_arrays[0];
assert_eq!(
compressed_arr.codec.as_ref().unwrap().name,
CodecName::BSLZ4
);
let mut decomp = CodecProcessor::new(CodecMode::Decompress);
let result = decomp.process_array(compressed_arr, &pool);
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
original.as_slice()
);
}
#[test]
fn test_jpeg_compress_mono() {
let arr = make_u8_array(16, 16);
let compressed = compress_jpeg(&arr, 90).unwrap();
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::JPEG);
let data = compressed.data.as_u8_slice();
assert_eq!(&data[0..2], &[0xFF, 0xD8]);
}
#[test]
fn test_jpeg_compress_rgb() {
let arr = make_rgb_array(16, 16);
let compressed = compress_jpeg(&arr, 90).unwrap();
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::JPEG);
let data = compressed.data.as_u8_slice();
assert_eq!(&data[0..2], &[0xFF, 0xD8]);
}
#[test]
fn test_jpeg_roundtrip_mono() {
let arr = make_u8_array(16, 16);
let compressed = compress_jpeg(&arr, 100).unwrap();
let decompressed = decompress_jpeg(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.dims.len(), 2);
assert_eq!(decompressed.dims[0].size, 16); assert_eq!(decompressed.dims[1].size, 16); assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
assert_eq!(decompressed.data.len(), 16 * 16);
}
#[test]
fn test_jpeg_roundtrip_rgb() {
let arr = make_rgb_array(16, 16);
let compressed = compress_jpeg(&arr, 100).unwrap();
let decompressed = decompress_jpeg(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.dims.len(), 3);
assert_eq!(decompressed.dims[0].size, 3); assert_eq!(decompressed.dims[1].size, 16); assert_eq!(decompressed.dims[2].size, 16); assert_eq!(decompressed.data.len(), 3 * 16 * 16);
}
fn make_rgb_layout(mode: ad_core_rs::color::NDColorMode, w: usize, h: usize) -> NDArray {
use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
use ad_core_rs::color::NDColorMode;
let pixel = |x: usize, y: usize, c: usize| ((x * 7 + y * 13 + c * 61) % 256) as u8;
let dims = match mode {
NDColorMode::RGB1 => vec![3, w, h],
NDColorMode::RGB2 => vec![w, 3, h],
NDColorMode::RGB3 => vec![w, h, 3],
other => panic!("not an RGB layout: {other:?}"),
};
let mut arr = NDArray::new(
dims.into_iter().map(NDDimension::new).collect(),
NDDataType::UInt8,
);
arr.attributes.add(NDAttribute::new_static(
"ColorMode",
"Color Mode",
NDAttrSource::Driver,
NDAttrValue::Int32(mode as i32),
));
if let NDDataBuffer::U8(ref mut v) = arr.data {
for y in 0..h {
for x in 0..w {
for c in 0..3 {
let idx = match mode {
NDColorMode::RGB1 => c + x * 3 + y * w * 3,
NDColorMode::RGB2 => x + c * w + y * w * 3,
NDColorMode::RGB3 => x + y * w + c * w * h,
_ => unreachable!(),
};
v[idx] = pixel(x, y, c);
}
}
}
}
arr
}
#[test]
fn test_r8_62_jpeg_compresses_rgb2_and_rgb3_as_reinterleaved_rgb() {
use ad_core_rs::color::NDColorMode;
let rgb1 = make_rgb_layout(NDColorMode::RGB1, 16, 8);
let reference = compress_jpeg(&rgb1, 90).expect("RGB1 must compress");
for mode in [NDColorMode::RGB2, NDColorMode::RGB3] {
let src = make_rgb_layout(mode, 16, 8);
let out = compress_jpeg(&src, 90)
.unwrap_or_else(|e| panic!("{mode:?} must compress, C encodes it: {e:?}"));
assert_eq!(out.codec.as_ref().unwrap().name, CodecName::JPEG);
assert_eq!(&out.data.as_u8_slice()[0..2], &[0xFF, 0xD8], "SOI marker");
assert_eq!(
out.data.as_u8_slice(),
reference.data.as_u8_slice(),
"{mode:?} must encode the same pixels as the RGB1 image — a wrong \
(or missing) scanline re-interleave changes the JPEG bytes"
);
assert_eq!(out.dims.len(), 3);
}
}
#[test]
fn test_r8_62_decompressed_jpeg_reports_rgb1_colormode() {
use ad_core_rs::color::NDColorMode;
let src = make_rgb_layout(NDColorMode::RGB2, 16, 8);
let compressed = compress_jpeg(&src, 90).expect("rgb2 jpeg");
assert_eq!(
compressed
.attributes
.get("ColorMode")
.unwrap()
.value
.as_i64(),
Some(NDColorMode::RGB2 as i64),
"the compressed frame keeps the source ColorMode"
);
let out = decompress_jpeg(&compressed).expect("jpeg decode");
assert_eq!(
out.attributes.get("ColorMode").unwrap().value.as_i64(),
Some(NDColorMode::RGB1 as i64),
"decompressed JPEG must be reported as RGB1"
);
assert_eq!(out.dims[0].size, 3);
assert_eq!(out.dims[1].size, 16);
assert_eq!(out.dims[2].size, 8);
assert_eq!(out.info().color_mode, NDColorMode::RGB1);
let mono = decompress_jpeg(&compress_jpeg(&make_u8_array(16, 16), 90).unwrap()).unwrap();
assert_eq!(
mono.attributes.get("ColorMode").unwrap().value.as_i64(),
Some(NDColorMode::Mono as i64)
);
}
#[test]
fn test_r8_62_jpeg_accepts_int8_like_c() {
let mut arr = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::Int8,
);
if let NDDataBuffer::I8(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i as i32 - 32) as i8;
}
}
let out = compress_jpeg(&arr, 90).expect("Int8 must compress");
assert_eq!(&out.data.as_u8_slice()[0..2], &[0xFF, 0xD8]);
assert_eq!(
out.codec.as_ref().unwrap().original_data_type,
NDDataType::Int8
);
}
#[test]
fn test_r8_62_jpeg_rejects_3d_without_an_rgb_colormode() {
let arr = NDArray::new(
vec![
NDDimension::new(3),
NDDimension::new(8),
NDDimension::new(8),
],
NDDataType::UInt8,
);
assert_eq!(
compress_jpeg(&arr, 90).unwrap_err(),
JpegCompressError::EncodeFailed,
"3-D Mono (no ColorMode attribute) is not a JPEG-encodable layout in C"
);
}
#[test]
fn test_jpeg_rejects_non_u8() {
let arr = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::UInt16,
);
let err = compress_jpeg(&arr, 90).unwrap_err();
assert_eq!(err, JpegCompressError::NotEightBit);
assert_eq!(err.message(), "JPEG only supports 8-bit data");
}
#[test]
fn test_jpeg_rejects_1d() {
let arr = NDArray::new(vec![NDDimension::new(64)], NDDataType::UInt8);
let err = compress_jpeg(&arr, 90).unwrap_err();
assert_eq!(err, JpegCompressError::UnsupportedArrayStructure);
assert_eq!(err.message(), "Unsupported array structure");
}
#[test]
fn test_r8_74_jpeg_compress_failures_carry_the_c_error_texts() {
use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
use ad_core_rs::color::NDColorMode;
let arr = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::Float32,
);
assert_eq!(
compress_jpeg(&arr, 90).unwrap_err().message(),
"JPEG only supports 8-bit data"
);
let arr = NDArray::new(
vec![
NDDimension::new(2),
NDDimension::new(2),
NDDimension::new(2),
NDDimension::new(2),
],
NDDataType::UInt8,
);
assert_eq!(
compress_jpeg(&arr, 90).unwrap_err().message(),
"Unsupported array structure"
);
for (mode, text) in [
(NDColorMode::Bayer, "Unknown color mode 1"),
(NDColorMode::YUV444, "Unknown color mode 5"),
(NDColorMode::YUV411, "Unknown color mode 7"),
] {
let mut arr = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::UInt8,
);
arr.attributes.add(NDAttribute::new_static(
"ColorMode",
"",
NDAttrSource::Driver,
NDAttrValue::Int32(mode as i32),
));
assert_eq!(compress_jpeg(&arr, 90).unwrap_err().message(), text);
}
}
#[test]
fn test_r8_74_codec_error_pv_carries_the_jpeg_text() {
use ad_core_rs::plugin::runtime::ParamUpdate;
let mut proc = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::JPEG,
quality: 90,
});
proc.params.codec_error = Some(13);
let arr = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::UInt16,
);
let result = proc.process_array(&arr, &NDArrayPool::new(0));
let text = result
.param_updates
.iter()
.find_map(|u| match u {
ParamUpdate::Octet {
reason: 13, value, ..
} => Some(value.clone()),
_ => None,
})
.expect("CodecError posted");
assert_eq!(text, "JPEG only supports 8-bit data");
}
#[test]
fn test_jpeg_quality_affects_size() {
let arr = make_u8_array(64, 64);
let high = compress_jpeg(&arr, 95).unwrap();
let low = compress_jpeg(&arr, 10).unwrap();
let high_size = high.codec.as_ref().unwrap().compressed_size;
let low_size = low.codec.as_ref().unwrap().compressed_size;
assert!(
high_size > low_size,
"high quality ({}) should produce larger output than low quality ({})",
high_size,
low_size,
);
}
#[test]
fn test_zlib_roundtrip_u8() {
let arr = make_u8_array(8, 8);
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_zlib(&arr);
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::Zlib);
assert_ne!(compressed.data.as_u8_slice(), original.as_slice());
let decompressed = decompress_zlib(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_zlib_roundtrip_u16() {
let mut arr = NDArray::new(
vec![NDDimension::new(16), NDDimension::new(16)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 257 % 65521) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_zlib(&arr);
assert_eq!(
compressed.codec.as_ref().unwrap().original_data_type,
NDDataType::UInt16
);
let decompressed = decompress_zlib(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_zlib_roundtrip_f64_with_negatives() {
let mut arr = NDArray::new(vec![NDDimension::new(64)], NDDataType::Float64);
if let NDDataBuffer::F64(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i as f64 - 32.0) * 3.25;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_zlib(&arr);
let decompressed = decompress_zlib(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_zlib_compresses_repetitive_data() {
let arr = NDArray::new(
vec![NDDimension::new(256), NDDimension::new(256)],
NDDataType::UInt8,
);
let original_size = arr.data.as_u8_slice().len();
let compressed = compress_zlib(&arr);
let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
assert!(
compressed_size < original_size,
"zlib compressed ({compressed_size}) should be < original ({original_size})"
);
}
#[test]
fn test_zlib_via_processor() {
let mut arr = NDArray::new(
vec![NDDimension::new(32), NDDimension::new(32)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 13) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let pool = NDArrayPool::new(10_000_000);
let mut comp = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::Zlib,
quality: 0,
});
let compressed = comp.process_array(&arr, &pool);
let compressed_arr = &compressed.output_arrays[0];
assert_eq!(compressed_arr.codec.as_ref().unwrap().name, CodecName::Zlib);
let mut decomp = CodecProcessor::new(CodecMode::Decompress);
let result = decomp.process_array(compressed_arr, &pool);
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
original.as_slice()
);
}
#[test]
fn test_lz4hdf5_roundtrip_u8() {
let mut arr = NDArray::new(
vec![NDDimension::new(64), NDDimension::new(64)],
NDDataType::UInt8,
);
if let NDDataBuffer::U8(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i % 251) as u8;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4hdf5(&arr);
assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4HDF5);
assert_ne!(compressed.data.as_u8_slice(), original.as_slice());
let decompressed = decompress_lz4hdf5(&compressed).unwrap();
assert!(decompressed.codec.is_none());
assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_lz4hdf5_roundtrip_u16() {
let mut arr = NDArray::new(
vec![NDDimension::new(80), NDDimension::new(40)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 37 % 65521) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4hdf5(&arr);
assert_eq!(
compressed.codec.as_ref().unwrap().original_data_type,
NDDataType::UInt16
);
let decompressed = decompress_lz4hdf5(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_lz4hdf5_roundtrip_f64_with_negatives() {
let mut arr = NDArray::new(vec![NDDimension::new(97)], NDDataType::Float64);
if let NDDataBuffer::F64(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i as f64 - 48.0) * 1.75;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4hdf5(&arr);
let decompressed = decompress_lz4hdf5(&compressed).unwrap();
assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_lz4hdf5_multi_block_roundtrip() {
let block = LZ4HDF5_DEFAULT_BLOCK_SIZE;
let count = block * 2 + block / 3 + 7; let mut arr = NDArray::new(vec![NDDimension::new(count)], NDDataType::UInt8);
if let NDDataBuffer::U8(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i.wrapping_mul(2_654_435_761) % 251) as u8;
}
}
let original = arr.data.as_u8_slice().to_vec();
let compressed = compress_lz4hdf5(&arr);
let decompressed = decompress_lz4hdf5(&compressed).unwrap();
assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
}
#[test]
fn test_lz4hdf5_compresses_repetitive_data() {
let arr = NDArray::new(
vec![NDDimension::new(256), NDDimension::new(256)],
NDDataType::UInt16,
);
let original_size = arr.data.as_u8_slice().len();
let compressed = compress_lz4hdf5(&arr);
let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
assert!(
compressed_size < original_size,
"lz4hdf5 compressed ({compressed_size}) should be < original ({original_size})"
);
}
#[test]
fn test_lz4hdf5_via_processor() {
let mut arr = NDArray::new(
vec![NDDimension::new(48), NDDimension::new(48)],
NDDataType::UInt16,
);
if let NDDataBuffer::U16(ref mut v) = arr.data {
for (i, x) in v.iter_mut().enumerate() {
*x = (i * 7) as u16;
}
}
let original = arr.data.as_u8_slice().to_vec();
let pool = NDArrayPool::new(10_000_000);
let mut comp = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::LZ4HDF5,
quality: 0,
});
let compressed = comp.process_array(&arr, &pool);
let compressed_arr = &compressed.output_arrays[0];
assert_eq!(
compressed_arr.codec.as_ref().unwrap().name,
CodecName::LZ4HDF5
);
let mut decomp = CodecProcessor::new(CodecMode::Decompress);
let result = decomp.process_array(compressed_arr, &pool);
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
original.as_slice()
);
}
#[test]
fn test_compressor_ordinal_mapping() {
use ad_core_rs::plugin::runtime::{ParamChangeValue, PluginParamSnapshot};
let cases = [
(0i32, CodecName::None),
(1, CodecName::JPEG),
(2, CodecName::Blosc),
(3, CodecName::LZ4),
(4, CodecName::BSLZ4),
(5, CodecName::Zlib),
(6, CodecName::LZ4HDF5),
];
for (ordinal, expected) in cases {
let mut proc = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::LZ4,
quality: 85,
});
proc.params.compressor = Some(0);
let snapshot = PluginParamSnapshot {
enable_callbacks: true,
reason: 0,
addr: 0,
value: ParamChangeValue::Int32(ordinal),
};
proc.on_param_change(0, &snapshot);
match proc.mode {
CodecMode::Compress { codec, .. } => assert_eq!(
codec, expected,
"ordinal {ordinal} should select {expected:?}"
),
other => panic!("expected Compress mode, got {other:?}"),
}
}
}
#[test]
fn test_decompress_wrong_codec() {
let arr = make_u8_array(4, 4);
assert!(decompress_lz4(&arr).is_none());
assert!(decompress_jpeg(&arr).is_none());
assert!(decompress_zlib(&arr).is_none());
assert!(decompress_lz4hdf5(&arr).is_none());
}
#[test]
fn test_processor_lz4_compress() {
let pool = NDArrayPool::new(1_000_000);
let mut proc = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::LZ4,
quality: 0,
});
let arr = make_u8_array(32, 32);
let result = proc.process_array(&arr, &pool);
assert_eq!(result.output_arrays.len(), 1);
assert_eq!(
result.output_arrays[0].codec.as_ref().unwrap().name,
CodecName::LZ4
);
assert!(proc.compression_ratio() >= 1.0);
}
#[test]
fn test_processor_jpeg_compress() {
let pool = NDArrayPool::new(1_000_000);
let mut proc = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::JPEG,
quality: 80,
});
let arr = make_u8_array(16, 16);
let result = proc.process_array(&arr, &pool);
assert_eq!(result.output_arrays.len(), 1);
assert_eq!(
result.output_arrays[0].codec.as_ref().unwrap().name,
CodecName::JPEG
);
}
#[test]
fn test_processor_decompress_auto_lz4() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(16, 16);
let compressed = compress_lz4(&arr);
let mut proc = CodecProcessor::new(CodecMode::Decompress);
let result = proc.process_array(&compressed, &pool);
assert_eq!(result.output_arrays.len(), 1);
assert!(result.output_arrays[0].codec.is_none());
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
arr.data.as_u8_slice()
);
assert!(proc.compression_ratio() > 0.0);
}
#[test]
fn test_processor_decompress_auto_jpeg() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(16, 16);
let compressed = compress_jpeg(&arr, 90).unwrap();
let mut proc = CodecProcessor::new(CodecMode::Decompress);
let result = proc.process_array(&compressed, &pool);
assert_eq!(result.output_arrays.len(), 1);
assert!(result.output_arrays[0].codec.is_none());
}
#[test]
fn test_processor_decompress_no_codec() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(8, 8);
let mut proc = CodecProcessor::new(CodecMode::Decompress);
let result = proc.process_array(&arr, &pool);
assert_eq!(result.output_arrays.len(), 1);
assert_eq!(proc.compression_ratio(), 1.0);
}
fn processor_with_params(mode: CodecMode) -> CodecProcessor {
let mut proc = CodecProcessor::new(mode);
proc.params.comp_factor = Some(10);
proc.params.compressor = Some(11);
proc.params.codec_status = Some(12);
proc.params.codec_error = Some(13);
proc
}
fn int32_update(updates: &[ParamUpdate], reason: usize) -> Option<i32> {
updates.iter().find_map(|u| match u {
ParamUpdate::Int32 {
reason: r, value, ..
} if *r == reason => Some(*value),
_ => None,
})
}
fn octet_update(updates: &[ParamUpdate], reason: usize) -> Option<String> {
updates.iter().find_map(|u| match u {
ParamUpdate::Octet {
reason: r, value, ..
} if *r == reason => Some(value.clone()),
_ => None,
})
}
#[test]
fn test_r8_61_decompress_uncompressed_input_is_success_passthrough() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(8, 8);
let mut proc = processor_with_params(CodecMode::Decompress);
let result = proc.process_array(&arr, &pool);
assert_eq!(
int32_update(&result.param_updates, 12),
Some(0),
"CodecStatus must stay SUCCESS on an uncompressed input"
);
assert_eq!(
octet_update(&result.param_updates, 13),
Some(String::new()),
"no error string on a pass-through"
);
assert_eq!(
int32_update(&result.param_updates, 11),
Some(0),
"COMPRESSOR must be set to NDCODEC_NONE"
);
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
arr.data.as_u8_slice(),
"the input array is passed through unchanged"
);
assert_eq!(proc.compression_ratio(), 1.0);
}
#[test]
fn test_r8_61_decompress_reports_compressor_of_the_input_codec() {
let pool = NDArrayPool::new(1_000_000);
let src = make_u8_array(16, 16);
for (codec, ordinal) in [
(compress_lz4(&src), 3),
(compress_blosc(&src, &BloscConfig::default()), 2),
(compress_bslz4(&src), 4),
(compress_jpeg(&src, 90).expect("jpeg"), 1),
] {
let mut proc = processor_with_params(CodecMode::Decompress);
let result = proc.process_array(&codec, &pool);
assert_eq!(
int32_update(&result.param_updates, 11),
Some(ordinal),
"COMPRESSOR must report the input codec's C ordinal"
);
assert_eq!(
int32_update(&result.param_updates, 12),
Some(0),
"a successful decompress is SUCCESS"
);
}
}
#[test]
fn test_r8_61_compress_with_compressor_none_is_success_passthrough() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(8, 8);
let mut proc = processor_with_params(CodecMode::Compress {
codec: CodecName::None,
quality: 85,
});
let result = proc.process_array(&arr, &pool);
assert_eq!(int32_update(&result.param_updates, 12), Some(0));
assert_eq!(octet_update(&result.param_updates, 13), Some(String::new()));
assert_eq!(
int32_update(&result.param_updates, 11),
None,
"compress mode must not overwrite the operator's COMPRESSOR selection"
);
assert!(result.output_arrays[0].codec.is_none());
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
arr.data.as_u8_slice()
);
}
#[test]
fn test_r8_61_genuine_decompress_failure_still_reports_an_error() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(16, 16);
let mut corrupted = compress_lz4(&arr);
if let NDDataBuffer::U8(ref mut v) = corrupted.data {
v.truncate(3);
}
let mut proc = processor_with_params(CodecMode::Decompress);
let result = proc.process_array(&corrupted, &pool);
assert_ne!(
int32_update(&result.param_updates, 12),
Some(0),
"a failed decompress must not report SUCCESS"
);
assert_eq!(
octet_update(&result.param_updates, 13),
Some("Failed to LZ4 decompress".to_string())
);
assert_eq!(int32_update(&result.param_updates, 11), Some(3));
assert_eq!(
result.output_arrays[0].data.as_u8_slice(),
corrupted.data.as_u8_slice(),
"the input array is republished on failure"
);
}
#[test]
fn test_r8_63_status_levels_match_c() {
assert_eq!(CodecStatus::Success.as_i32(), 0);
assert_eq!(CodecStatus::Warning.as_i32(), 1);
assert_eq!(CodecStatus::Error.as_i32(), 2);
}
#[test]
fn test_r8_63_already_compressed_is_a_warning_not_success() {
let pool = NDArrayPool::new(1_000_000);
let compressed = compress_lz4(&make_u8_array(16, 16));
let mut proc = processor_with_params(CodecMode::Compress {
codec: CodecName::Zlib,
quality: 85,
});
let result = proc.process_array(&compressed, &pool);
assert_eq!(
int32_update(&result.param_updates, 12),
Some(CodecStatus::Warning.as_i32()),
"already-compressed input must report WARNING(1)"
);
assert_eq!(
octet_update(&result.param_updates, 13),
Some("Array already compressed".to_string())
);
assert_eq!(
result.output_arrays[0].codec.as_ref().unwrap().name,
CodecName::LZ4
);
}
#[test]
fn test_r8_63_genuine_failures_are_error_not_warning() {
let pool = NDArrayPool::new(1_000_000);
let wide = NDArray::new(
vec![NDDimension::new(8), NDDimension::new(8)],
NDDataType::UInt16,
);
let mut proc = processor_with_params(CodecMode::Compress {
codec: CodecName::JPEG,
quality: 85,
});
let result = proc.process_array(&wide, &pool);
assert_eq!(
int32_update(&result.param_updates, 12),
Some(CodecStatus::Error.as_i32()),
"an unsupported JPEG input is an ERROR"
);
let mut corrupted = compress_lz4(&make_u8_array(16, 16));
if let NDDataBuffer::U8(ref mut v) = corrupted.data {
v.truncate(3);
}
let mut proc = processor_with_params(CodecMode::Decompress);
let result = proc.process_array(&corrupted, &pool);
assert_eq!(
int32_update(&result.param_updates, 12),
Some(CodecStatus::Error.as_i32()),
"a failed decompress is an ERROR"
);
}
#[test]
fn test_r8_63_successful_and_passthrough_paths_report_success() {
let pool = NDArrayPool::new(1_000_000);
let arr = make_u8_array(16, 16);
let mut proc = processor_with_params(CodecMode::Compress {
codec: CodecName::LZ4,
quality: 85,
});
let compressed = proc.process_array(&arr, &pool);
assert_eq!(
int32_update(&compressed.param_updates, 12),
Some(CodecStatus::Success.as_i32())
);
let mut proc = processor_with_params(CodecMode::Decompress);
let passthrough = proc.process_array(&arr, &pool);
assert_eq!(
int32_update(&passthrough.param_updates, 12),
Some(CodecStatus::Success.as_i32())
);
}
#[test]
fn test_processor_compression_ratio() {
let pool = NDArrayPool::new(1_000_000);
let mut arr = NDArray::new(
vec![NDDimension::new(128), NDDimension::new(128)],
NDDataType::UInt8,
);
if let NDDataBuffer::U8(ref mut v) = arr.data {
for x in v.iter_mut() {
*x = 0;
}
}
let mut proc = CodecProcessor::new(CodecMode::Compress {
codec: CodecName::LZ4,
quality: 0,
});
let _ = proc.process_array(&arr, &pool);
let ratio = proc.compression_ratio();
assert!(
ratio > 2.0,
"all-zeros 128x128 should compress at least 2x, got {}",
ratio,
);
}
#[test]
fn test_processor_plugin_type() {
let proc = CodecProcessor::new(CodecMode::Decompress);
assert_eq!(proc.plugin_type(), "NDPluginCodec");
}
#[test]
fn test_buffer_from_bytes_u8() {
let data = vec![1u8, 2, 3, 4];
let buf = buffer_from_bytes(&data, NDDataType::UInt8).unwrap();
assert_eq!(buf.data_type(), NDDataType::UInt8);
assert_eq!(buf.len(), 4);
assert_eq!(buf.as_u8_slice(), &[1, 2, 3, 4]);
}
#[test]
fn test_buffer_from_bytes_u16() {
let original = vec![1000u16, 2000, 3000];
let bytes: Vec<u8> = original.iter().flat_map(|v| v.to_ne_bytes()).collect();
let buf = buffer_from_bytes(&bytes, NDDataType::UInt16).unwrap();
assert_eq!(buf.data_type(), NDDataType::UInt16);
assert_eq!(buf.len(), 3);
if let NDDataBuffer::U16(v) = buf {
assert_eq!(v, original);
} else {
panic!("wrong buffer type");
}
}
#[test]
fn test_buffer_from_bytes_bad_alignment() {
let data = vec![0u8; 3];
assert!(buffer_from_bytes(&data, NDDataType::UInt16).is_none());
}
#[test]
fn test_buffer_from_bytes_f64_roundtrip() {
let original = vec![1.5f64, -2.7, 3.14159];
let bytes: Vec<u8> = original.iter().flat_map(|v| v.to_ne_bytes()).collect();
let buf = buffer_from_bytes(&bytes, NDDataType::Float64).unwrap();
if let NDDataBuffer::F64(v) = buf {
assert_eq!(v, original);
} else {
panic!("wrong buffer type");
}
}
}