1use std::borrow::Cow;
2use std::io::{Read, Write};
3use std::sync::Arc;
4
5use ad_core_rs::codec::{Codec, CodecName, CodecStatus};
6use ad_core_rs::ndarray::{NDArray, NDDataBuffer, NDDataType, NDDimension};
7use ad_core_rs::ndarray_pool::NDArrayPool;
8use ad_core_rs::plugin::runtime::{NDPluginProcess, ParamUpdate, ProcessResult};
9
10use flate2::Compression;
11use flate2::read::ZlibDecoder;
12use flate2::write::ZlibEncoder;
13use lz4_flex::block::{compress, decompress};
14use rust_hdf5::format::messages::filter::{
15 FILTER_BLOSC, Filter, FilterPipeline, apply_filters, reverse_filters,
16};
17
18pub fn original_data_type(array: &NDArray) -> NDDataType {
30 match &array.codec {
31 Some(c) => c.original_data_type,
32 None => array.data.data_type(),
33 }
34}
35
36pub(crate) fn buffer_from_bytes(bytes: &[u8], data_type: NDDataType) -> Option<NDDataBuffer> {
41 let elem_size = data_type.element_size();
42 if bytes.len() % elem_size != 0 {
43 return None;
44 }
45 let count = bytes.len() / elem_size;
46
47 Some(match data_type {
48 NDDataType::Int8 => {
49 let mut v = vec![0i8; count];
50 unsafe {
52 std::ptr::copy_nonoverlapping(
53 bytes.as_ptr(),
54 v.as_mut_ptr() as *mut u8,
55 bytes.len(),
56 );
57 }
58 NDDataBuffer::I8(v)
59 }
60 NDDataType::UInt8 => NDDataBuffer::U8(bytes.to_vec()),
61 NDDataType::Int16 => {
62 let mut v = vec![0i16; count];
63 unsafe {
64 std::ptr::copy_nonoverlapping(
65 bytes.as_ptr(),
66 v.as_mut_ptr() as *mut u8,
67 bytes.len(),
68 );
69 }
70 NDDataBuffer::I16(v)
71 }
72 NDDataType::UInt16 => {
73 let mut v = vec![0u16; count];
74 unsafe {
75 std::ptr::copy_nonoverlapping(
76 bytes.as_ptr(),
77 v.as_mut_ptr() as *mut u8,
78 bytes.len(),
79 );
80 }
81 NDDataBuffer::U16(v)
82 }
83 NDDataType::Int32 => {
84 let mut v = vec![0i32; count];
85 unsafe {
86 std::ptr::copy_nonoverlapping(
87 bytes.as_ptr(),
88 v.as_mut_ptr() as *mut u8,
89 bytes.len(),
90 );
91 }
92 NDDataBuffer::I32(v)
93 }
94 NDDataType::UInt32 => {
95 let mut v = vec![0u32; count];
96 unsafe {
97 std::ptr::copy_nonoverlapping(
98 bytes.as_ptr(),
99 v.as_mut_ptr() as *mut u8,
100 bytes.len(),
101 );
102 }
103 NDDataBuffer::U32(v)
104 }
105 NDDataType::Int64 => {
106 let mut v = vec![0i64; count];
107 unsafe {
108 std::ptr::copy_nonoverlapping(
109 bytes.as_ptr(),
110 v.as_mut_ptr() as *mut u8,
111 bytes.len(),
112 );
113 }
114 NDDataBuffer::I64(v)
115 }
116 NDDataType::UInt64 => {
117 let mut v = vec![0u64; count];
118 unsafe {
119 std::ptr::copy_nonoverlapping(
120 bytes.as_ptr(),
121 v.as_mut_ptr() as *mut u8,
122 bytes.len(),
123 );
124 }
125 NDDataBuffer::U64(v)
126 }
127 NDDataType::Float32 => {
128 let mut v = vec![0f32; count];
129 unsafe {
130 std::ptr::copy_nonoverlapping(
131 bytes.as_ptr(),
132 v.as_mut_ptr() as *mut u8,
133 bytes.len(),
134 );
135 }
136 NDDataBuffer::F32(v)
137 }
138 NDDataType::Float64 => {
139 let mut v = vec![0f64; count];
140 unsafe {
141 std::ptr::copy_nonoverlapping(
142 bytes.as_ptr(),
143 v.as_mut_ptr() as *mut u8,
144 bytes.len(),
145 );
146 }
147 NDDataBuffer::F64(v)
148 }
149 })
150}
151
152pub fn compress_lz4(src: &NDArray) -> NDArray {
158 let raw = src.data.as_u8_slice();
159 let original_data_type = src.data.data_type();
160 let original_size = raw.len();
161 let compressed = compress(raw);
163 let compressed_size = compressed.len();
164
165 let mut arr = src.clone();
166 arr.data = NDDataBuffer::U8(compressed);
167 arr.codec = Some(Codec {
168 name: CodecName::LZ4,
169 compressed_size,
170 level: 0,
171 shuffle: 0,
172 compressor: 0,
173 original_data_type,
176 });
177
178 tracing::debug!(
179 original_size,
180 compressed_size,
181 ratio = original_size as f64 / compressed_size.max(1) as f64,
182 "LZ4 compress"
183 );
184
185 arr
186}
187
188pub fn decompress_lz4(src: &NDArray) -> Option<NDArray> {
193 if src.codec.as_ref().map(|c| c.name) != Some(CodecName::LZ4) {
194 return None;
195 }
196 let compressed = src.data.as_u8_slice();
197 let original_type = original_data_type(src);
200 let num_elements: usize = src.dims.iter().map(|d| d.size).product();
201 let uncompressed_size = num_elements * original_type.element_size();
202 let decompressed = decompress(compressed, uncompressed_size).ok()?;
203
204 let buffer = buffer_from_bytes(&decompressed, original_type)?;
205
206 let mut arr = src.clone();
207 arr.data = buffer;
208 arr.codec = None;
209
210 Some(arr)
211}
212
213const ZLIB_DEFAULT_LEVEL: u32 = 6;
224
225pub fn compress_zlib(src: &NDArray) -> NDArray {
231 let raw = src.data.as_u8_slice();
232 let original_data_type = src.data.data_type();
233 let original_size = raw.len();
234
235 let mut encoder = ZlibEncoder::new(Vec::<u8>::new(), Compression::new(ZLIB_DEFAULT_LEVEL));
236 if encoder.write_all(raw).is_err() {
238 return src.clone();
239 }
240 let compressed = match encoder.finish() {
241 Ok(buf) => buf,
242 Err(_) => return src.clone(),
243 };
244 let compressed_size = compressed.len();
245
246 let mut arr = src.clone();
247 arr.data = NDDataBuffer::U8(compressed);
248 arr.codec = Some(Codec {
249 name: CodecName::Zlib,
250 compressed_size,
251 level: ZLIB_DEFAULT_LEVEL as i32,
252 shuffle: 0,
253 compressor: 0,
254 original_data_type,
255 });
256
257 tracing::debug!(
258 original_size,
259 compressed_size,
260 ratio = original_size as f64 / compressed_size.max(1) as f64,
261 "Zlib compress"
262 );
263 arr
264}
265
266pub fn decompress_zlib(src: &NDArray) -> Option<NDArray> {
271 if src.codec.as_ref().map(|c| c.name) != Some(CodecName::Zlib) {
272 return None;
273 }
274 let compressed = src.data.as_u8_slice();
275
276 let original_type = original_data_type(src);
277 let num_elements: usize = src.dims.iter().map(|d| d.size).product();
278 let uncompressed_size = num_elements * original_type.element_size();
279
280 let mut decoder = ZlibDecoder::new(compressed);
281 let mut decompressed = Vec::with_capacity(uncompressed_size);
282 decoder.read_to_end(&mut decompressed).ok()?;
283
284 let buffer = buffer_from_bytes(&decompressed, original_type)?;
285
286 let mut arr = src.clone();
287 arr.data = buffer;
288 arr.codec = None;
289 Some(arr)
290}
291
292const LZ4HDF5_DEFAULT_BLOCK_SIZE: usize = 1 << 20;
312
313pub fn compress_lz4hdf5(src: &NDArray) -> NDArray {
320 let raw = src.data.as_u8_slice();
321 let data_type = src.data.data_type();
322 let original_size = raw.len();
323 let block_size = LZ4HDF5_DEFAULT_BLOCK_SIZE;
324
325 let mut out: Vec<u8> = Vec::with_capacity(original_size / 2 + 12);
327 out.extend_from_slice(&(original_size as u64).to_be_bytes());
328 out.extend_from_slice(&(block_size as u32).to_be_bytes());
329
330 let mut pos = 0usize;
331 while pos < raw.len() {
332 let n = block_size.min(raw.len() - pos);
333 let block = &raw[pos..pos + n];
334 let comp = compress(block);
335 if comp.len() < n {
338 out.extend_from_slice(&(comp.len() as u32).to_be_bytes());
339 out.extend_from_slice(&comp);
340 } else {
341 out.extend_from_slice(&(n as u32).to_be_bytes());
342 out.extend_from_slice(block);
343 }
344 pos += n;
345 }
346
347 let compressed_size = out.len();
348 let mut arr = src.clone();
349 arr.data = NDDataBuffer::U8(out);
350 arr.codec = Some(Codec {
351 name: CodecName::LZ4HDF5,
352 compressed_size,
353 level: 0,
354 shuffle: 0,
355 compressor: 0,
356 original_data_type: data_type,
357 });
358
359 tracing::debug!(
360 original_size,
361 compressed_size,
362 ratio = original_size as f64 / compressed_size.max(1) as f64,
363 "LZ4HDF5 compress"
364 );
365 arr
366}
367
368pub fn decompress_lz4hdf5(src: &NDArray) -> Option<NDArray> {
372 if src.codec.as_ref().map(|c| c.name) != Some(CodecName::LZ4HDF5) {
373 return None;
374 }
375 let buf = src.data.as_u8_slice();
376 if buf.len() < 12 {
377 return None;
378 }
379 let total_bytes = u64::from_be_bytes(buf[0..8].try_into().ok()?) as usize;
380 let block_size = u32::from_be_bytes(buf[8..12].try_into().ok()?) as usize;
381 if block_size == 0 {
382 return None;
383 }
384
385 let original_type = original_data_type(src);
386
387 let mut out: Vec<u8> = Vec::with_capacity(total_bytes);
388 let mut pos = 12usize;
389 while out.len() < total_bytes {
390 let n = block_size.min(total_bytes - out.len());
391 if pos + 4 > buf.len() {
392 return None;
393 }
394 let clen = u32::from_be_bytes(buf[pos..pos + 4].try_into().ok()?) as usize;
395 pos += 4;
396 if pos + clen > buf.len() {
397 return None;
398 }
399 let block_payload = &buf[pos..pos + clen];
400 if clen == n {
401 out.extend_from_slice(block_payload);
403 } else {
404 let block = decompress(block_payload, n).ok()?;
405 if block.len() != n {
406 return None;
407 }
408 out.extend_from_slice(&block);
409 }
410 pos += clen;
411 }
412 if out.len() != total_bytes {
413 return None;
414 }
415
416 let buffer = buffer_from_bytes(&out, original_type)?;
417 let mut arr = src.clone();
418 arr.data = buffer;
419 arr.codec = None;
420 Some(arr)
421}
422
423const BSHUF_TARGET_BLOCK_SIZE_B: usize = 8192;
446const BSHUF_BLOCKED_MULT: usize = 8;
448const BSHUF_MIN_RECOMMEND_BLOCK: usize = 128;
450
451pub(crate) fn bshuf_default_block_size(elem_size: usize) -> usize {
458 let bs = BSHUF_TARGET_BLOCK_SIZE_B / elem_size.max(1);
459 let bs = (bs / BSHUF_BLOCKED_MULT) * BSHUF_BLOCKED_MULT;
460 bs.max(BSHUF_MIN_RECOMMEND_BLOCK)
461}
462
463#[inline]
466fn trans_bit_8x8(mut x: u64) -> u64 {
467 let t = (x ^ (x >> 7)) & 0x00AA_00AA_00AA_00AA;
468 x = x ^ t ^ (t << 7);
469 let t = (x ^ (x >> 14)) & 0x0000_CCCC_0000_CCCC;
470 x = x ^ t ^ (t << 14);
471 let t = (x ^ (x >> 28)) & 0x0000_0000_F0F0_F0F0;
472 x = x ^ t ^ (t << 28);
473 x
474}
475
476#[inline]
478fn read_u64_le(b: &[u8], off: usize) -> u64 {
479 u64::from_le_bytes(b[off..off + 8].try_into().unwrap())
480}
481
482fn bshuf_trans_byte_elem(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
485 let mut ii = 0;
486 while ii + 7 < size {
487 for jj in 0..elem_size {
488 for kk in 0..8 {
489 out[jj * size + ii + kk] = input[ii * elem_size + kk * elem_size + jj];
490 }
491 }
492 ii += 8;
493 }
494 let mut ii = size - size % 8;
496 while ii < size {
497 for jj in 0..elem_size {
498 out[jj * size + ii] = input[ii * elem_size + jj];
499 }
500 ii += 1;
501 }
502}
503
504fn bshuf_trans_bit_byte(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
507 let nbyte = elem_size * size;
508 let nbyte_bitrow = nbyte / 8;
509 for ii in 0..nbyte_bitrow {
510 let mut x = trans_bit_8x8(read_u64_le(input, ii * 8));
511 for kk in 0..8 {
512 out[kk * nbyte_bitrow + ii] = x as u8;
513 x >>= 8;
514 }
515 }
516}
517
518fn bshuf_trans_bitrow_eight(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
521 let nbyte_bitrow = size / 8;
522 for ii in 0..8 {
523 for jj in 0..elem_size {
524 let src = (ii * elem_size + jj) * nbyte_bitrow;
525 let dst = (jj * 8 + ii) * nbyte_bitrow;
526 out[dst..dst + nbyte_bitrow].copy_from_slice(&input[src..src + nbyte_bitrow]);
527 }
528 }
529}
530
531fn bshuf_trans_bit_elem(input: &[u8], size: usize, elem_size: usize) -> Vec<u8> {
535 debug_assert_eq!(size % 8, 0);
536 let nbyte = size * elem_size;
537 let mut a = vec![0u8; nbyte];
538 bshuf_trans_byte_elem(input, &mut a, size, elem_size);
539 let mut b = vec![0u8; nbyte];
540 bshuf_trans_bit_byte(&a, &mut b, size, elem_size);
541 let mut out = vec![0u8; nbyte];
542 bshuf_trans_bitrow_eight(&b, &mut out, size, elem_size);
543 out
544}
545
546fn bshuf_trans_byte_bitrow(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
549 let nbyte_row = size / 8;
550 for jj in 0..elem_size {
551 for ii in 0..nbyte_row {
552 for kk in 0..8 {
553 out[ii * 8 * elem_size + jj * 8 + kk] = input[(jj * 8 + kk) * nbyte_row + ii];
554 }
555 }
556 }
557}
558
559fn bshuf_shuffle_bit_eightelem(input: &[u8], out: &mut [u8], size: usize, elem_size: usize) {
562 let nbyte = elem_size * size;
563 let mut jj = 0;
564 while jj < 8 * elem_size {
565 let mut ii = 0;
566 while ii + 8 * elem_size - 1 < nbyte {
567 let mut x = trans_bit_8x8(read_u64_le(input, ii + jj));
568 for kk in 0..8 {
569 out[ii + jj / 8 + kk * elem_size] = x as u8;
570 x >>= 8;
571 }
572 ii += 8 * elem_size;
573 }
574 jj += 8;
575 }
576}
577
578fn bshuf_untrans_bit_elem(input: &[u8], size: usize, elem_size: usize) -> Vec<u8> {
581 debug_assert_eq!(size % 8, 0);
582 let nbyte = size * elem_size;
583 let mut tmp = vec![0u8; nbyte];
584 bshuf_trans_byte_bitrow(input, &mut tmp, size, elem_size);
585 let mut out = vec![0u8; nbyte];
586 bshuf_shuffle_bit_eightelem(&tmp, &mut out, size, elem_size);
587 out
588}
589
590fn bshuf_compress_lz4_block(
594 out: &mut Vec<u8>,
595 raw: &[u8],
596 elem_start: usize,
597 size: usize,
598 elem_size: usize,
599) {
600 let off = elem_start * elem_size;
601 let shuffled = bshuf_trans_bit_elem(&raw[off..off + size * elem_size], size, elem_size);
602 let comp = compress(&shuffled);
603 out.extend_from_slice(&(comp.len() as u32).to_be_bytes());
604 out.extend_from_slice(&comp);
605}
606
607fn bshuf_decompress_lz4_block(
611 buf: &[u8],
612 pos: usize,
613 size: usize,
614 elem_size: usize,
615) -> Option<(Vec<u8>, usize)> {
616 if pos + 4 > buf.len() {
617 return None;
618 }
619 let clen = u32::from_be_bytes(buf[pos..pos + 4].try_into().ok()?) as usize;
620 let dstart = pos + 4;
621 if dstart + clen > buf.len() {
622 return None;
623 }
624 let shuffled = decompress(&buf[dstart..dstart + clen], size * elem_size).ok()?;
625 if shuffled.len() != size * elem_size {
626 return None;
627 }
628 Some((
629 bshuf_untrans_bit_elem(&shuffled, size, elem_size),
630 dstart + clen,
631 ))
632}
633
634pub fn compress_bslz4(src: &NDArray) -> NDArray {
647 let raw = src.data.as_u8_slice();
648 let data_type = src.data.data_type();
649 let elem_size = data_type.element_size();
650 let total_elems = if elem_size > 0 {
651 raw.len() / elem_size
652 } else {
653 0
654 };
655 let block_size = bshuf_default_block_size(elem_size);
656
657 let mut out: Vec<u8> = Vec::with_capacity(raw.len() / 2 + 16);
658
659 let n_full = total_elems / block_size;
660 let mut elem = 0usize;
661 for _ in 0..n_full {
662 bshuf_compress_lz4_block(&mut out, raw, elem, block_size, elem_size);
663 elem += block_size;
664 }
665 let mut last_block = total_elems % block_size;
667 last_block -= last_block % BSHUF_BLOCKED_MULT;
668 if last_block > 0 {
669 bshuf_compress_lz4_block(&mut out, raw, elem, last_block, elem_size);
670 elem += last_block;
671 }
672 if elem < total_elems {
674 out.extend_from_slice(&raw[elem * elem_size..total_elems * elem_size]);
675 }
676
677 let compressed_size = out.len();
678 let mut arr = src.clone();
679 arr.data = NDDataBuffer::U8(out);
680 arr.codec = Some(Codec {
681 name: CodecName::BSLZ4,
682 compressed_size,
683 level: 0,
684 shuffle: 0,
685 compressor: 0,
686 original_data_type: data_type,
687 });
688
689 tracing::debug!(
690 original_size = raw.len(),
691 compressed_size,
692 ratio = raw.len() as f64 / compressed_size.max(1) as f64,
693 "BSLZ4 compress"
694 );
695 arr
696}
697
698pub fn decompress_bslz4(src: &NDArray) -> Option<NDArray> {
706 let codec = src.codec.as_ref()?;
707 if codec.name != CodecName::BSLZ4 {
708 return None;
709 }
710 let buf = src.data.as_u8_slice();
711 let original_type = original_data_type(src);
712 let elem_size = original_type.element_size();
713 if elem_size == 0 {
714 return None;
715 }
716 let total_elems: usize = src.dims.iter().map(|d| d.size).product();
717 let total_bytes = total_elems * elem_size;
718 let block_size = bshuf_default_block_size(elem_size);
719
720 let mut out: Vec<u8> = Vec::with_capacity(total_bytes);
721 let mut pos = 0usize;
722
723 let n_full = total_elems / block_size;
724 for _ in 0..n_full {
725 let (block, next) = bshuf_decompress_lz4_block(buf, pos, block_size, elem_size)?;
726 out.extend_from_slice(&block);
727 pos = next;
728 }
729 let mut last_block = total_elems % block_size;
731 last_block -= last_block % BSHUF_BLOCKED_MULT;
732 if last_block > 0 {
733 let (block, next) = bshuf_decompress_lz4_block(buf, pos, last_block, elem_size)?;
734 out.extend_from_slice(&block);
735 pos = next;
736 }
737 let leftover_bytes = (total_elems % BSHUF_BLOCKED_MULT) * elem_size;
739 if leftover_bytes > 0 {
740 if pos + leftover_bytes > buf.len() {
741 return None;
742 }
743 out.extend_from_slice(&buf[pos..pos + leftover_bytes]);
744 }
745 if out.len() != total_bytes {
746 return None;
747 }
748
749 let buffer = buffer_from_bytes(&out, original_type)?;
750 let mut arr = src.clone();
751 arr.data = buffer;
752 arr.codec = None;
753 Some(arr)
754}
755
756pub fn compress_jpeg(src: &NDArray, quality: u8) -> Result<NDArray, JpegCompressError> {
775 use ad_core_rs::color::{NDColorMode, convert_rgb_layout};
776
777 match src.data.data_type() {
779 NDDataType::UInt8 | NDDataType::Int8 => {}
780 _ => return Err(JpegCompressError::NotEightBit),
781 }
782
783 let info = src.info();
784
785 if !matches!(src.dims.len(), 2 | 3) {
788 return Err(JpegCompressError::UnsupportedArrayStructure);
789 }
790
791 match info.color_mode {
796 NDColorMode::Mono | NDColorMode::RGB1 | NDColorMode::RGB2 | NDColorMode::RGB3 => {}
797 mode => return Err(JpegCompressError::UnknownColorMode(mode as i32)),
798 }
799
800 if info.x_size > u16::MAX as usize || info.y_size > u16::MAX as usize {
802 return Err(JpegCompressError::EncodeFailed);
803 }
804
805 let (color_type, interleaved) = match (src.dims.len(), info.color_mode) {
808 (2, NDColorMode::Mono | NDColorMode::RGB1) => (jpeg_encoder::ColorType::Luma, None),
809 (3, NDColorMode::RGB1) if info.color_size == 3 => (jpeg_encoder::ColorType::Rgb, None),
810 (3, mode @ (NDColorMode::RGB2 | NDColorMode::RGB3)) if info.color_size == 3 => (
811 jpeg_encoder::ColorType::Rgb,
812 Some(
813 convert_rgb_layout(src, mode, NDColorMode::RGB1)
814 .map_err(|_| JpegCompressError::EncodeFailed)?,
815 ),
816 ),
817 _ => return Err(JpegCompressError::EncodeFailed),
820 };
821
822 let width = info.x_size as u16;
823 let height = info.y_size as u16;
824 let pixels = interleaved
825 .as_ref()
826 .map_or_else(|| src.data.as_u8_slice(), |a| a.data.as_u8_slice());
827
828 let mut jpeg_buf = Vec::new();
829 let encoder = jpeg_encoder::Encoder::new(&mut jpeg_buf, quality);
830 if encoder.encode(pixels, width, height, color_type).is_err() {
831 return Err(JpegCompressError::EncodeFailed);
832 }
833
834 let compressed_size = jpeg_buf.len();
835 let original_size = src.data.as_u8_slice().len();
836
837 let mut arr = src.clone();
838 arr.data = NDDataBuffer::U8(jpeg_buf);
839 arr.codec = Some(Codec {
840 name: CodecName::JPEG,
841 compressed_size,
842 level: 0,
843 shuffle: 0,
844 compressor: 0,
845 original_data_type: src.data.data_type(),
848 });
849
850 tracing::debug!(
851 original_size,
852 compressed_size,
853 ratio = original_size as f64 / compressed_size.max(1) as f64,
854 "JPEG compress (quality={})",
855 quality,
856 );
857
858 Ok(arr)
859}
860
861#[derive(Debug, Clone, Copy, PartialEq, Eq)]
870pub enum JpegCompressError {
871 NotEightBit,
873 UnsupportedArrayStructure,
875 UnknownColorMode(i32),
878 EncodeFailed,
888}
889
890impl JpegCompressError {
891 pub fn message(&self) -> Cow<'static, str> {
893 match self {
894 Self::NotEightBit => "JPEG only supports 8-bit data".into(),
895 Self::UnsupportedArrayStructure => "Unsupported array structure".into(),
896 Self::UnknownColorMode(mode) => format!("Unknown color mode {}", mode).into(),
899 Self::EncodeFailed => "Error writing JPEG data".into(),
900 }
901 }
902}
903
904pub fn decompress_jpeg(src: &NDArray) -> Option<NDArray> {
918 use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
919 use ad_core_rs::color::NDColorMode;
920
921 if src.codec.as_ref().map(|c| c.name) != Some(CodecName::JPEG) {
922 return None;
923 }
924
925 let compressed = src.data.as_u8_slice();
926 let mut decoder = jpeg_decoder::Decoder::new(compressed);
927 let pixels = decoder.decode().ok()?;
928 let metadata = decoder.info()?;
929
930 let width = metadata.width as usize;
931 let height = metadata.height as usize;
932
933 let (dims, color_mode) = match metadata.pixel_format {
934 jpeg_decoder::PixelFormat::L8 => (
935 vec![NDDimension::new(width), NDDimension::new(height)],
936 NDColorMode::Mono,
937 ),
938 jpeg_decoder::PixelFormat::RGB24 => (
939 vec![
940 NDDimension::new(3),
941 NDDimension::new(width),
942 NDDimension::new(height),
943 ],
944 NDColorMode::RGB1,
945 ),
946 _ => return None,
947 };
948
949 let mut arr = src.clone();
950 arr.dims = dims;
951 arr.data = NDDataBuffer::U8(pixels);
952 arr.codec = None;
953 arr.attributes.add(NDAttribute::new_static(
954 "ColorMode",
955 "Color Mode",
956 NDAttrSource::Driver,
957 NDAttrValue::Int32(color_mode as i32),
958 ));
959
960 Some(arr)
961}
962
963#[derive(Debug, Clone, Copy)]
965pub struct BloscConfig {
966 pub compressor: u32,
968 pub clevel: u32,
970 pub shuffle: u32,
972}
973
974impl Default for BloscConfig {
975 fn default() -> Self {
976 Self {
977 compressor: 0,
978 clevel: 5,
983 shuffle: 0,
984 }
985 }
986}
987
988pub fn compress_blosc(src: &NDArray, config: &BloscConfig) -> NDArray {
990 let raw = src.data.as_u8_slice();
991 let element_size = src.data.data_type().element_size();
992
993 let pipeline = FilterPipeline {
1000 filters: vec![Filter {
1001 id: FILTER_BLOSC,
1002 flags: 0,
1003 cd_values: vec![
1004 2, 2, element_size as u32, raw.len() as u32, config.clevel, config.shuffle, config.compressor, ],
1012 }],
1013 };
1014
1015 let compressed = match apply_filters(&pipeline, raw) {
1016 Ok(data) => data,
1017 Err(_) => return src.clone(),
1018 };
1019
1020 let compressed_size = compressed.len();
1021 let original_data_type = src.data.data_type();
1022 let mut arr = src.clone();
1023 arr.data = NDDataBuffer::U8(compressed);
1024 arr.codec = Some(Codec {
1025 name: CodecName::Blosc,
1026 compressed_size,
1027 level: config.clevel as i32,
1030 shuffle: config.shuffle as i32,
1031 compressor: config.compressor as i32,
1032 original_data_type,
1033 });
1034 arr
1035}
1036
1037pub fn decompress_blosc(src: &NDArray) -> Option<NDArray> {
1039 let codec = src.codec.as_ref()?;
1040 if codec.name != CodecName::Blosc {
1041 return None;
1042 }
1043
1044 let compressed = src.data.as_u8_slice();
1045 let original_type = original_data_type(src);
1046 let element_size = original_type.element_size();
1047
1048 let pipeline = FilterPipeline {
1053 filters: vec![Filter {
1054 id: FILTER_BLOSC,
1055 flags: 0,
1056 cd_values: vec![
1057 2,
1058 2,
1059 element_size as u32,
1060 0,
1061 codec.level as u32,
1062 codec.shuffle as u32,
1063 codec.compressor as u32,
1064 ],
1065 }],
1066 };
1067
1068 let decompressed = reverse_filters(&pipeline, compressed).ok()?;
1069
1070 let buffer = buffer_from_bytes(&decompressed, original_type)?;
1071
1072 let mut arr = src.clone();
1073 arr.data = buffer;
1074 arr.codec = None;
1075 Some(arr)
1076}
1077
1078#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1080pub enum CodecMode {
1081 Compress { codec: CodecName, quality: u8 },
1083 Decompress,
1085}
1086
1087#[derive(Default)]
1091struct CodecParamIndices {
1092 mode: Option<usize>,
1093 compressor: Option<usize>,
1094 comp_factor: Option<usize>,
1095 jpeg_quality: Option<usize>,
1096 blosc_compressor: Option<usize>,
1097 blosc_clevel: Option<usize>,
1098 blosc_shuffle: Option<usize>,
1099 blosc_numthreads: Option<usize>,
1100 codec_status: Option<usize>,
1101 codec_error: Option<usize>,
1102}
1103
1104pub struct CodecProcessor {
1105 mode: CodecMode,
1106 compression_ratio: f64,
1107 jpeg_quality: u8,
1108 blosc_config: BloscConfig,
1109 params: CodecParamIndices,
1110}
1111
1112impl CodecProcessor {
1113 pub fn new(mode: CodecMode) -> Self {
1114 let quality = match mode {
1115 CodecMode::Compress { quality, .. } => quality,
1116 _ => 85,
1117 };
1118 Self {
1119 mode,
1120 compression_ratio: 1.0,
1121 jpeg_quality: quality,
1122 blosc_config: BloscConfig::default(),
1123 params: CodecParamIndices::default(),
1124 }
1125 }
1126
1127 pub fn compression_ratio(&self) -> f64 {
1130 self.compression_ratio
1131 }
1132}
1133
1134enum CodecOutcome {
1150 PassThrough,
1153 Skipped(&'static str),
1157 Converted(NDArray),
1160 Failed(Cow<'static, str>),
1167}
1168
1169impl CodecOutcome {
1170 fn status(&self) -> CodecStatus {
1172 match self {
1173 Self::PassThrough | Self::Converted(_) => CodecStatus::Success,
1174 Self::Skipped(_) => CodecStatus::Warning,
1175 Self::Failed(_) => CodecStatus::Error,
1176 }
1177 }
1178
1179 fn error_message(&self) -> &str {
1182 match self {
1183 Self::PassThrough | Self::Converted(_) => "",
1184 Self::Skipped(message) => message,
1185 Self::Failed(message) => message,
1186 }
1187 }
1188}
1189
1190impl NDPluginProcess for CodecProcessor {
1191 fn process_array(&mut self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
1192 let original_bytes = array.data.as_u8_slice().len();
1193
1194 let mut compressor: Option<i32> = None;
1199
1200 let outcome = match self.mode {
1201 CodecMode::Compress {
1206 codec: CodecName::None,
1207 ..
1208 } => CodecOutcome::PassThrough,
1209 CodecMode::Compress { .. } if array.codec.is_some() => {
1210 CodecOutcome::Skipped("Array already compressed")
1213 }
1214 CodecMode::Compress { codec, .. } => match codec {
1215 CodecName::LZ4 => CodecOutcome::Converted(compress_lz4(array)),
1216 CodecName::JPEG => match compress_jpeg(array, self.jpeg_quality) {
1220 Ok(out) => CodecOutcome::Converted(out),
1221 Err(e) => CodecOutcome::Failed(e.message()),
1222 },
1223 CodecName::Zlib => CodecOutcome::Converted(compress_zlib(array)),
1224 CodecName::Blosc => {
1225 CodecOutcome::Converted(compress_blosc(array, &self.blosc_config))
1226 }
1227 CodecName::LZ4HDF5 => CodecOutcome::Converted(compress_lz4hdf5(array)),
1228 CodecName::BSLZ4 => CodecOutcome::Converted(compress_bslz4(array)),
1229 CodecName::None => CodecOutcome::PassThrough,
1231 },
1232 CodecMode::Decompress => {
1233 let name = array
1238 .codec
1239 .as_ref()
1240 .map(|c| c.name)
1241 .unwrap_or(CodecName::None);
1242 compressor = Some(name.ordinal());
1243 match name {
1244 CodecName::None => CodecOutcome::PassThrough,
1247 CodecName::LZ4 => match decompress_lz4(array) {
1248 Some(out) => CodecOutcome::Converted(out),
1249 None => CodecOutcome::Failed("Failed to LZ4 decompress".into()),
1250 },
1251 CodecName::JPEG => match decompress_jpeg(array) {
1252 Some(out) => CodecOutcome::Converted(out),
1253 None => CodecOutcome::Failed("Error decoding JPEG".into()),
1254 },
1255 CodecName::Zlib => match decompress_zlib(array) {
1256 Some(out) => CodecOutcome::Converted(out),
1257 None => CodecOutcome::Failed("Failed to Zlib decompress".into()),
1258 },
1259 CodecName::Blosc => match decompress_blosc(array) {
1260 Some(out) => CodecOutcome::Converted(out),
1261 None => CodecOutcome::Failed("Failed to Blosc decompress".into()),
1262 },
1263 CodecName::LZ4HDF5 => match decompress_lz4hdf5(array) {
1264 Some(out) => CodecOutcome::Converted(out),
1265 None => CodecOutcome::Failed("Failed to LZ4 decompress".into()),
1266 },
1267 CodecName::BSLZ4 => match decompress_bslz4(array) {
1272 Some(out) => CodecOutcome::Converted(out),
1273 None => CodecOutcome::Failed("Failed to Blosc decompress".into()),
1274 },
1275 }
1276 }
1277 };
1278
1279 let status = outcome.status();
1280 let error = outcome.error_message().to_string();
1281
1282 let output = match outcome {
1286 CodecOutcome::Converted(out) => {
1287 let output_bytes = out.data.as_u8_slice().len();
1288 self.compression_ratio = match self.mode {
1289 CodecMode::Compress { .. } => {
1290 original_bytes as f64 / output_bytes.max(1) as f64
1291 }
1292 CodecMode::Decompress => output_bytes as f64 / original_bytes.max(1) as f64,
1293 };
1294 out
1295 }
1296 CodecOutcome::PassThrough | CodecOutcome::Skipped(_) | CodecOutcome::Failed(_) => {
1297 self.compression_ratio = 1.0;
1298 array.clone()
1299 }
1300 };
1301
1302 let mut updates = Vec::new();
1303 if let Some(idx) = self.params.comp_factor {
1304 updates.push(ParamUpdate::float64(idx, self.compression_ratio));
1305 }
1306 if let (Some(idx), Some(value)) = (self.params.compressor, compressor) {
1307 updates.push(ParamUpdate::int32(idx, value));
1308 }
1309 if let Some(idx) = self.params.codec_status {
1310 updates.push(ParamUpdate::int32(idx, status.as_i32()));
1311 }
1312 if let Some(idx) = self.params.codec_error {
1313 updates.push(ParamUpdate::Octet {
1314 reason: idx,
1315 addr: 0,
1316 value: error,
1317 });
1318 }
1319
1320 let mut r = ProcessResult::arrays(vec![Arc::new(output)]);
1321 r.param_updates = updates;
1322 r
1323 }
1324
1325 fn plugin_type(&self) -> &str {
1326 "NDPluginCodec"
1327 }
1328
1329 fn compression_aware(&self) -> bool {
1338 true
1339 }
1340
1341 fn register_params(
1342 &mut self,
1343 base: &mut asyn_rs::port::PortDriverBase,
1344 ) -> asyn_rs::error::AsynResult<()> {
1345 use asyn_rs::param::ParamType;
1346 base.create_param("MODE", ParamType::Int32)?;
1347 base.create_param("COMPRESSOR", ParamType::Int32)?;
1348 base.create_param("COMP_FACTOR", ParamType::Float64)?;
1349 base.create_param("JPEG_QUALITY", ParamType::Int32)?;
1350 base.create_param("BLOSC_COMPRESSOR", ParamType::Int32)?;
1351 base.create_param("BLOSC_CLEVEL", ParamType::Int32)?;
1352 base.create_param("BLOSC_SHUFFLE", ParamType::Int32)?;
1353 base.create_param("BLOSC_NUMTHREADS", ParamType::Int32)?;
1354 base.create_param("CODEC_STATUS", ParamType::Int32)?;
1355 base.create_param("CODEC_ERROR", ParamType::Octet)?;
1356
1357 self.params.mode = base.find_param("MODE");
1358 self.params.compressor = base.find_param("COMPRESSOR");
1359 self.params.comp_factor = base.find_param("COMP_FACTOR");
1360 self.params.jpeg_quality = base.find_param("JPEG_QUALITY");
1361 self.params.blosc_compressor = base.find_param("BLOSC_COMPRESSOR");
1362 self.params.blosc_clevel = base.find_param("BLOSC_CLEVEL");
1363 self.params.blosc_shuffle = base.find_param("BLOSC_SHUFFLE");
1364 self.params.blosc_numthreads = base.find_param("BLOSC_NUMTHREADS");
1365 self.params.codec_status = base.find_param("CODEC_STATUS");
1366 self.params.codec_error = base.find_param("CODEC_ERROR");
1367 Ok(())
1368 }
1369
1370 fn on_param_change(
1371 &mut self,
1372 reason: usize,
1373 params: &ad_core_rs::plugin::runtime::PluginParamSnapshot,
1374 ) -> ad_core_rs::plugin::runtime::ParamChangeResult {
1375 if Some(reason) == self.params.mode {
1376 let v = params.value.as_i32();
1377 if v == 0 {
1378 let codec = match self.mode {
1380 CodecMode::Compress { codec, .. } => codec,
1381 _ => CodecName::LZ4,
1382 };
1383 self.mode = CodecMode::Compress {
1384 codec,
1385 quality: self.jpeg_quality,
1386 };
1387 } else {
1388 self.mode = CodecMode::Decompress;
1389 }
1390 } else if Some(reason) == self.params.compressor {
1391 let codec = CodecName::from_ordinal(params.value.as_i32());
1395 if let CodecMode::Compress { .. } = self.mode {
1396 self.mode = CodecMode::Compress {
1397 codec,
1398 quality: self.jpeg_quality,
1399 };
1400 }
1401 } else if Some(reason) == self.params.jpeg_quality {
1402 self.jpeg_quality = params.value.as_i32().clamp(1, 100) as u8;
1403 if let CodecMode::Compress { codec, .. } = self.mode {
1404 self.mode = CodecMode::Compress {
1405 codec,
1406 quality: self.jpeg_quality,
1407 };
1408 }
1409 } else if Some(reason) == self.params.blosc_compressor {
1410 self.blosc_config.compressor = params.value.as_i32().max(0) as u32;
1411 } else if Some(reason) == self.params.blosc_clevel {
1412 self.blosc_config.clevel = params.value.as_i32().clamp(0, 9) as u32;
1413 } else if Some(reason) == self.params.blosc_shuffle {
1414 self.blosc_config.shuffle = params.value.as_i32().max(0) as u32;
1415 }
1416
1417 ad_core_rs::plugin::runtime::ParamChangeResult::updates(vec![])
1418 }
1419}
1420
1421#[cfg(test)]
1422mod tests {
1423 use super::*;
1424
1425 fn make_u8_array(width: usize, height: usize) -> NDArray {
1426 let mut arr = NDArray::new(
1427 vec![NDDimension::new(width), NDDimension::new(height)],
1428 NDDataType::UInt8,
1429 );
1430 if let NDDataBuffer::U8(ref mut v) = arr.data {
1431 for i in 0..v.len() {
1432 v[i] = (i % 256) as u8;
1433 }
1434 }
1435 arr
1436 }
1437
1438 fn make_rgb_array(width: usize, height: usize) -> NDArray {
1439 use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
1440 let mut arr = NDArray::new(
1441 vec![
1442 NDDimension::new(3),
1443 NDDimension::new(width),
1444 NDDimension::new(height),
1445 ],
1446 NDDataType::UInt8,
1447 );
1448 arr.attributes.add(NDAttribute::new_static(
1450 "ColorMode",
1451 "Color Mode",
1452 NDAttrSource::Driver,
1453 NDAttrValue::Int32(2), ));
1455 if let NDDataBuffer::U8(ref mut v) = arr.data {
1456 for i in 0..v.len() {
1457 v[i] = (i % 256) as u8;
1458 }
1459 }
1460 arr
1461 }
1462
1463 #[test]
1469 fn compressors_record_type_in_codec_not_an_attribute() {
1470 let mut arr = NDArray::new(vec![NDDimension::new(8)], NDDataType::UInt16);
1471 if let NDDataBuffer::U16(ref mut v) = arr.data {
1472 for (i, x) in v.iter_mut().enumerate() {
1473 *x = (i * 7) as u16;
1474 }
1475 }
1476 for compressed in [
1477 compress_lz4(&arr),
1478 compress_zlib(&arr),
1479 compress_lz4hdf5(&arr),
1480 compress_bslz4(&arr),
1481 compress_blosc(&arr, &BloscConfig::default()),
1482 ] {
1483 assert_eq!(
1484 compressed.codec.as_ref().unwrap().original_data_type,
1485 NDDataType::UInt16,
1486 "the original element type must travel in the codec"
1487 );
1488 assert!(
1489 compressed
1490 .attributes
1491 .get("CODEC_ORIGINAL_DATA_TYPE")
1492 .is_none(),
1493 "no codec carrier attribute may be attached to a compressed frame"
1494 );
1495 }
1496 }
1497
1498 #[test]
1499 fn test_adp29_blosc_default_clevel_and_codec_params() {
1500 assert_eq!(
1503 BloscConfig::default().clevel,
1504 5,
1505 "default Blosc clevel must be 5 (C parity)"
1506 );
1507
1508 let mut arr = NDArray::new(vec![NDDimension::new(8)], NDDataType::UInt16);
1511 if let NDDataBuffer::U16(ref mut v) = arr.data {
1512 for (i, x) in v.iter_mut().enumerate() {
1513 *x = (i * 7) as u16;
1514 }
1515 }
1516 let out = compress_blosc(&arr, &BloscConfig::default());
1517 let codec = out.codec.as_ref().expect("blosc codec metadata");
1518 assert_eq!(codec.level, 5, "codec.level records the default clevel 5");
1521 assert_eq!(codec.shuffle, 0, "codec.shuffle records shuffle");
1522 assert_eq!(codec.compressor, 0, "codec.compressor records compressor");
1523 }
1524
1525 #[test]
1526 fn test_blosc_roundtrip_u16_default_compressor() {
1527 let mut arr = NDArray::new(
1533 vec![NDDimension::new(100), NDDimension::new(20)],
1534 NDDataType::UInt16,
1535 );
1536 if let NDDataBuffer::U16(ref mut v) = arr.data {
1537 for (i, x) in v.iter_mut().enumerate() {
1538 *x = (i * 37 % 65521) as u16;
1539 }
1540 }
1541 let original = arr.data.as_u8_slice().to_vec();
1542
1543 let compressed = compress_blosc(&arr, &BloscConfig::default());
1544 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::Blosc);
1545 assert_ne!(
1546 compressed.data.as_u8_slice(),
1547 original.as_slice(),
1548 "blosc must actually compress (not fall back to the raw clone)"
1549 );
1550
1551 let decompressed = decompress_blosc(&compressed).expect("blosc round-trip");
1552 assert!(decompressed.codec.is_none());
1553 assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
1554 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
1555 }
1556
1557 #[test]
1558 fn test_blosc_roundtrip_u16_lz4_subcompressor() {
1559 let cfg = BloscConfig {
1562 compressor: 1,
1563 clevel: 5,
1564 shuffle: 1,
1565 };
1566 let mut arr = NDArray::new(vec![NDDimension::new(256)], NDDataType::UInt16);
1567 if let NDDataBuffer::U16(ref mut v) = arr.data {
1568 for (i, x) in v.iter_mut().enumerate() {
1569 *x = (i * 13 % 65521) as u16;
1570 }
1571 }
1572 let original = arr.data.as_u8_slice().to_vec();
1573
1574 let compressed = compress_blosc(&arr, &cfg);
1575 let codec = compressed.codec.as_ref().unwrap();
1576 assert_eq!(codec.compressor, 1, "records the LZ4 sub-compressor");
1577 assert_eq!(codec.shuffle, 1, "records byte shuffle");
1578
1579 let decompressed = decompress_blosc(&compressed).expect("blosc lz4 round-trip");
1580 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
1581 }
1582
1583 #[test]
1586 fn test_lz4_roundtrip_u8() {
1587 let arr = make_u8_array(4, 4);
1588 let original_data = arr.data.as_u8_slice().to_vec();
1589
1590 let compressed = compress_lz4(&arr);
1591 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4);
1592 assert_ne!(compressed.data.as_u8_slice(), original_data.as_slice());
1594
1595 let decompressed = decompress_lz4(&compressed).unwrap();
1596 assert!(decompressed.codec.is_none());
1597 assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
1598 assert_eq!(decompressed.data.as_u8_slice(), original_data.as_slice());
1599 }
1600
1601 #[test]
1602 fn test_decompress_runtime_does_not_drop_compressed_input() {
1603 use ad_core_rs::plugin::channel::{NDArrayOutput, ndarray_channel};
1610 use ad_core_rs::plugin::runtime::create_plugin_runtime_with_output;
1611 use ad_core_rs::plugin::wiring::WiringRegistry;
1612 use std::sync::atomic::Ordering;
1613
1614 let mut raw = make_u8_array(4, 4);
1616 raw.unique_id = 1;
1617 let original_data = raw.data.as_u8_slice().to_vec();
1618 let compressed = compress_lz4(&raw);
1619 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4);
1620 assert_eq!(compressed.unique_id, 1);
1621
1622 let mut sentinel = make_u8_array(4, 4);
1626 sentinel.unique_id = 2;
1627
1628 let pool = Arc::new(NDArrayPool::new(1_000_000));
1629 let (ds_sender, mut ds_rx) = ndarray_channel("DS", 10);
1630 let mut output = NDArrayOutput::new();
1631 output.add(ds_sender);
1632 let (handle, _jh) = create_plugin_runtime_with_output(
1633 "CODEC_DECOMP",
1634 CodecProcessor::new(CodecMode::Decompress),
1635 pool,
1636 10,
1637 output,
1638 "",
1639 Arc::new(WiringRegistry::new()),
1640 );
1641 let dropped = handle.array_sender().dropped_arrays_counter().clone();
1642 handle
1643 .port_runtime()
1644 .port_handle()
1645 .write_int32_blocking(handle.plugin_params.enable_callbacks, 0, 1)
1646 .unwrap();
1647 assert!(
1649 handle.wait_params_applied(std::time::Duration::from_secs(10)),
1650 "data thread did not apply EnableCallbacks"
1651 );
1652
1653 let rt = tokio::runtime::Builder::new_current_thread()
1654 .enable_all()
1655 .build()
1656 .unwrap();
1657 rt.block_on(handle.array_sender().publish(Arc::new(compressed)));
1658 rt.block_on(handle.array_sender().publish(Arc::new(sentinel)));
1659
1660 let first = ds_rx.blocking_recv().expect("downstream array");
1661 assert_eq!(
1662 first.unique_id, 1,
1663 "compressed input must be decompressed and delivered, not dropped"
1664 );
1665 assert!(
1666 first.codec.is_none(),
1667 "delivered array must be decompressed (codec cleared)"
1668 );
1669 assert_eq!(first.data.as_u8_slice(), original_data.as_slice());
1670 assert_eq!(
1671 dropped.load(Ordering::Acquire),
1672 0,
1673 "compression-aware Codec must not count its compressed input as dropped"
1674 );
1675 }
1676
1677 #[test]
1678 fn test_lz4_roundtrip_u16() {
1679 let mut arr = NDArray::new(
1680 vec![NDDimension::new(8), NDDimension::new(8)],
1681 NDDataType::UInt16,
1682 );
1683 if let NDDataBuffer::U16(ref mut v) = arr.data {
1684 for i in 0..v.len() {
1685 v[i] = (i * 100) as u16;
1686 }
1687 }
1688 let original_bytes = arr.data.as_u8_slice().to_vec();
1689
1690 let compressed = compress_lz4(&arr);
1691 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4);
1692 assert_eq!(
1694 compressed.codec.as_ref().unwrap().original_data_type,
1695 NDDataType::UInt16
1696 );
1697 assert!(
1699 compressed
1700 .attributes
1701 .get("CODEC_ORIGINAL_DATA_TYPE")
1702 .is_none()
1703 );
1704
1705 let decompressed = decompress_lz4(&compressed).unwrap();
1706 assert!(decompressed.codec.is_none());
1707 assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
1708 assert_eq!(decompressed.data.as_u8_slice(), original_bytes.as_slice());
1709 }
1710
1711 #[test]
1712 fn test_lz4_roundtrip_f64() {
1713 let mut arr = NDArray::new(vec![NDDimension::new(16)], NDDataType::Float64);
1714 if let NDDataBuffer::F64(ref mut v) = arr.data {
1715 for i in 0..v.len() {
1716 v[i] = i as f64 * 1.5;
1717 }
1718 }
1719 let original_bytes = arr.data.as_u8_slice().to_vec();
1720
1721 let compressed = compress_lz4(&arr);
1722 let decompressed = decompress_lz4(&compressed).unwrap();
1723 assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
1724 assert_eq!(decompressed.data.as_u8_slice(), original_bytes.as_slice());
1725 }
1726
1727 #[test]
1728 fn test_lz4_compresses_repetitive_data() {
1729 let mut arr = NDArray::new(
1731 vec![NDDimension::new(256), NDDimension::new(256)],
1732 NDDataType::UInt8,
1733 );
1734 if let NDDataBuffer::U8(ref mut v) = arr.data {
1736 for x in v.iter_mut() {
1737 *x = 0;
1738 }
1739 }
1740 let original_size = arr.data.as_u8_slice().len();
1741
1742 let compressed = compress_lz4(&arr);
1743 let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
1744 assert!(
1745 compressed_size < original_size,
1746 "compressed ({}) should be smaller than original ({})",
1747 compressed_size,
1748 original_size,
1749 );
1750 }
1751
1752 #[test]
1753 fn test_lz4_preserves_metadata() {
1754 let mut arr = make_u8_array(4, 4);
1755 arr.unique_id = 42;
1756
1757 let compressed = compress_lz4(&arr);
1758 assert_eq!(compressed.unique_id, 42);
1759 assert_eq!(compressed.dims.len(), 2);
1760 assert_eq!(compressed.dims[0].size, 4);
1761 assert_eq!(compressed.dims[1].size, 4);
1762 }
1763
1764 #[test]
1767 fn test_bitshuffle_block_transpose_roundtrip() {
1768 for &(n, elem_size) in &[(16usize, 4usize), (8, 2), (256, 8), (128, 1)] {
1771 let input: Vec<u8> = (0..n * elem_size).map(|i| (i * 7 + 3) as u8).collect();
1772 let shuffled = bshuf_trans_bit_elem(&input, n, elem_size);
1773 assert_eq!(shuffled.len(), input.len());
1774 let restored = bshuf_untrans_bit_elem(&shuffled, n, elem_size);
1775 assert_eq!(restored, input, "elem_size {elem_size}, n {n}");
1776 }
1777 }
1778
1779 #[test]
1780 fn test_bitshuffle_matches_c_reference_vector() {
1781 let input: Vec<u8> = (0..16u16).flat_map(|v| v.to_le_bytes()).collect();
1789 let shuffled = bshuf_trans_bit_elem(&input, 16, 2);
1790 let mut expected = vec![0u8; 32];
1791 expected[..8].copy_from_slice(&[170, 170, 204, 204, 240, 240, 0, 255]);
1792 assert_eq!(
1793 shuffled, expected,
1794 "canonical bitshuffle transpose must match the C library bytes"
1795 );
1796 }
1797
1798 #[test]
1799 fn test_bslz4_roundtrip_u8() {
1800 let mut arr = NDArray::new(
1801 vec![NDDimension::new(64), NDDimension::new(64)],
1802 NDDataType::UInt8,
1803 );
1804 if let NDDataBuffer::U8(ref mut v) = arr.data {
1805 for (i, x) in v.iter_mut().enumerate() {
1806 *x = (i % 251) as u8;
1807 }
1808 }
1809 let original = arr.data.as_u8_slice().to_vec();
1810
1811 let compressed = compress_bslz4(&arr);
1812 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::BSLZ4);
1813 assert_ne!(compressed.data.as_u8_slice(), original.as_slice());
1814
1815 let decompressed = decompress_bslz4(&compressed).unwrap();
1816 assert!(decompressed.codec.is_none());
1817 assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
1818 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
1819 }
1820
1821 #[test]
1822 fn test_bslz4_roundtrip_u16() {
1823 let mut arr = NDArray::new(
1824 vec![NDDimension::new(100), NDDimension::new(20)],
1825 NDDataType::UInt16,
1826 );
1827 if let NDDataBuffer::U16(ref mut v) = arr.data {
1828 for (i, x) in v.iter_mut().enumerate() {
1829 *x = (i * 37 % 65521) as u16;
1830 }
1831 }
1832 let original = arr.data.as_u8_slice().to_vec();
1833
1834 let compressed = compress_bslz4(&arr);
1835 assert_eq!(
1836 compressed.codec.as_ref().unwrap().original_data_type,
1837 NDDataType::UInt16
1838 );
1839 let decompressed = decompress_bslz4(&compressed).unwrap();
1840 assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
1841 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
1842 }
1843
1844 #[test]
1845 fn test_bslz4_roundtrip_f64_with_negatives() {
1846 let mut arr = NDArray::new(vec![NDDimension::new(73)], NDDataType::Float64);
1847 if let NDDataBuffer::F64(ref mut v) = arr.data {
1848 for (i, x) in v.iter_mut().enumerate() {
1849 *x = (i as f64 - 36.0) * 2.5;
1850 }
1851 }
1852 let original = arr.data.as_u8_slice().to_vec();
1853
1854 let compressed = compress_bslz4(&arr);
1855 let decompressed = decompress_bslz4(&compressed).unwrap();
1856 assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
1857 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
1858 }
1859
1860 #[test]
1861 fn test_bslz4_roundtrip_multi_block() {
1862 let elem_size = 4usize;
1865 let block = bshuf_default_block_size(elem_size);
1866 let count = block * 2 + block / 2 + 3;
1868 let mut arr = NDArray::new(vec![NDDimension::new(count)], NDDataType::Int32);
1869 if let NDDataBuffer::I32(ref mut v) = arr.data {
1870 for (i, x) in v.iter_mut().enumerate() {
1871 *x = (i as i32).wrapping_mul(2_654_435_761u32 as i32);
1872 }
1873 }
1874 let original = arr.data.as_u8_slice().to_vec();
1875
1876 let compressed = compress_bslz4(&arr);
1877 let decompressed = decompress_bslz4(&compressed).unwrap();
1878 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
1879 }
1880
1881 #[test]
1882 fn test_bslz4_compresses_repetitive_data() {
1883 let arr = NDArray::new(
1885 vec![NDDimension::new(256), NDDimension::new(256)],
1886 NDDataType::UInt16,
1887 );
1888 let original_size = arr.data.as_u8_slice().len();
1889 let compressed = compress_bslz4(&arr);
1890 let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
1891 assert!(
1892 compressed_size < original_size,
1893 "bslz4 compressed ({compressed_size}) should be < original ({original_size})"
1894 );
1895 }
1896
1897 #[test]
1898 fn test_r9_71_corrupt_bslz4_reports_cs_blosc_text() {
1899 use ad_core_rs::plugin::runtime::ParamUpdate;
1904
1905 let mut arr = NDArray::new(
1906 vec![NDDimension::new(32), NDDimension::new(32)],
1907 NDDataType::UInt16,
1908 );
1909 if let NDDataBuffer::U16(ref mut v) = arr.data {
1910 for (i, x) in v.iter_mut().enumerate() {
1911 *x = (i * 11) as u16;
1912 }
1913 }
1914 let pool = NDArrayPool::new(10_000_000);
1915
1916 let mut compressed = compress_bslz4(&arr);
1918 if let NDDataBuffer::U8(ref mut v) = compressed.data {
1919 for b in v.iter_mut() {
1920 *b = 0xFF;
1921 }
1922 }
1923 assert!(
1924 decompress_bslz4(&compressed).is_none(),
1925 "the corrupted frame must fail to decompress"
1926 );
1927
1928 let mut decomp = CodecProcessor::new(CodecMode::Decompress);
1929 decomp.params.codec_error = Some(13);
1930 let result = decomp.process_array(&compressed, &pool);
1931 let text = result
1932 .param_updates
1933 .iter()
1934 .find_map(|u| match u {
1935 ParamUpdate::Octet {
1936 reason: 13, value, ..
1937 } => Some(value.clone()),
1938 _ => None,
1939 })
1940 .expect("CodecError posted");
1941 assert_eq!(text, "Failed to Blosc decompress");
1942 }
1943
1944 #[test]
1945 fn test_bslz4_via_processor() {
1946 let mut arr = NDArray::new(
1948 vec![NDDimension::new(32), NDDimension::new(32)],
1949 NDDataType::UInt16,
1950 );
1951 if let NDDataBuffer::U16(ref mut v) = arr.data {
1952 for (i, x) in v.iter_mut().enumerate() {
1953 *x = (i * 11) as u16;
1954 }
1955 }
1956 let original = arr.data.as_u8_slice().to_vec();
1957 let pool = NDArrayPool::new(10_000_000);
1958
1959 let mut comp = CodecProcessor::new(CodecMode::Compress {
1960 codec: CodecName::BSLZ4,
1961 quality: 0,
1962 });
1963 let compressed = comp.process_array(&arr, &pool);
1964 let compressed_arr = &compressed.output_arrays[0];
1965 assert_eq!(
1966 compressed_arr.codec.as_ref().unwrap().name,
1967 CodecName::BSLZ4
1968 );
1969
1970 let mut decomp = CodecProcessor::new(CodecMode::Decompress);
1971 let result = decomp.process_array(compressed_arr, &pool);
1972 assert_eq!(
1973 result.output_arrays[0].data.as_u8_slice(),
1974 original.as_slice()
1975 );
1976 }
1977
1978 #[test]
1981 fn test_jpeg_compress_mono() {
1982 let arr = make_u8_array(16, 16);
1983 let compressed = compress_jpeg(&arr, 90).unwrap();
1984 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::JPEG);
1985 let data = compressed.data.as_u8_slice();
1987 assert_eq!(&data[0..2], &[0xFF, 0xD8]);
1988 }
1989
1990 #[test]
1991 fn test_jpeg_compress_rgb() {
1992 let arr = make_rgb_array(16, 16);
1993 let compressed = compress_jpeg(&arr, 90).unwrap();
1994 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::JPEG);
1995 let data = compressed.data.as_u8_slice();
1996 assert_eq!(&data[0..2], &[0xFF, 0xD8]);
1997 }
1998
1999 #[test]
2000 fn test_jpeg_roundtrip_mono() {
2001 let arr = make_u8_array(16, 16);
2002 let compressed = compress_jpeg(&arr, 100).unwrap();
2003 let decompressed = decompress_jpeg(&compressed).unwrap();
2004 assert!(decompressed.codec.is_none());
2005 assert_eq!(decompressed.dims.len(), 2);
2006 assert_eq!(decompressed.dims[0].size, 16); assert_eq!(decompressed.dims[1].size, 16); assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
2009 assert_eq!(decompressed.data.len(), 16 * 16);
2011 }
2012
2013 #[test]
2014 fn test_jpeg_roundtrip_rgb() {
2015 let arr = make_rgb_array(16, 16);
2016 let compressed = compress_jpeg(&arr, 100).unwrap();
2017 let decompressed = decompress_jpeg(&compressed).unwrap();
2018 assert!(decompressed.codec.is_none());
2019 assert_eq!(decompressed.dims.len(), 3);
2020 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);
2024 }
2025
2026 fn make_rgb_layout(mode: ad_core_rs::color::NDColorMode, w: usize, h: usize) -> NDArray {
2032 use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
2033 use ad_core_rs::color::NDColorMode;
2034
2035 let pixel = |x: usize, y: usize, c: usize| ((x * 7 + y * 13 + c * 61) % 256) as u8;
2036 let dims = match mode {
2037 NDColorMode::RGB1 => vec![3, w, h],
2038 NDColorMode::RGB2 => vec![w, 3, h],
2039 NDColorMode::RGB3 => vec![w, h, 3],
2040 other => panic!("not an RGB layout: {other:?}"),
2041 };
2042 let mut arr = NDArray::new(
2043 dims.into_iter().map(NDDimension::new).collect(),
2044 NDDataType::UInt8,
2045 );
2046 arr.attributes.add(NDAttribute::new_static(
2047 "ColorMode",
2048 "Color Mode",
2049 NDAttrSource::Driver,
2050 NDAttrValue::Int32(mode as i32),
2051 ));
2052 if let NDDataBuffer::U8(ref mut v) = arr.data {
2053 for y in 0..h {
2054 for x in 0..w {
2055 for c in 0..3 {
2056 let idx = match mode {
2057 NDColorMode::RGB1 => c + x * 3 + y * w * 3,
2058 NDColorMode::RGB2 => x + c * w + y * w * 3,
2059 NDColorMode::RGB3 => x + y * w + c * w * h,
2060 _ => unreachable!(),
2061 };
2062 v[idx] = pixel(x, y, c);
2063 }
2064 }
2065 }
2066 }
2067 arr
2068 }
2069
2070 #[test]
2071 fn test_r8_62_jpeg_compresses_rgb2_and_rgb3_as_reinterleaved_rgb() {
2072 use ad_core_rs::color::NDColorMode;
2077
2078 let rgb1 = make_rgb_layout(NDColorMode::RGB1, 16, 8);
2079 let reference = compress_jpeg(&rgb1, 90).expect("RGB1 must compress");
2080
2081 for mode in [NDColorMode::RGB2, NDColorMode::RGB3] {
2082 let src = make_rgb_layout(mode, 16, 8);
2083 let out = compress_jpeg(&src, 90)
2084 .unwrap_or_else(|e| panic!("{mode:?} must compress, C encodes it: {e:?}"));
2085 assert_eq!(out.codec.as_ref().unwrap().name, CodecName::JPEG);
2086 assert_eq!(&out.data.as_u8_slice()[0..2], &[0xFF, 0xD8], "SOI marker");
2087 assert_eq!(
2088 out.data.as_u8_slice(),
2089 reference.data.as_u8_slice(),
2090 "{mode:?} must encode the same pixels as the RGB1 image — a wrong \
2091 (or missing) scanline re-interleave changes the JPEG bytes"
2092 );
2093 assert_eq!(out.dims.len(), 3);
2095 }
2096 }
2097
2098 #[test]
2099 fn test_r8_62_decompressed_jpeg_reports_rgb1_colormode() {
2100 use ad_core_rs::color::NDColorMode;
2105
2106 let src = make_rgb_layout(NDColorMode::RGB2, 16, 8);
2107 let compressed = compress_jpeg(&src, 90).expect("rgb2 jpeg");
2108 assert_eq!(
2109 compressed
2110 .attributes
2111 .get("ColorMode")
2112 .unwrap()
2113 .value
2114 .as_i64(),
2115 Some(NDColorMode::RGB2 as i64),
2116 "the compressed frame keeps the source ColorMode"
2117 );
2118
2119 let out = decompress_jpeg(&compressed).expect("jpeg decode");
2120 assert_eq!(
2121 out.attributes.get("ColorMode").unwrap().value.as_i64(),
2122 Some(NDColorMode::RGB1 as i64),
2123 "decompressed JPEG must be reported as RGB1"
2124 );
2125 assert_eq!(out.dims[0].size, 3);
2126 assert_eq!(out.dims[1].size, 16);
2127 assert_eq!(out.dims[2].size, 8);
2128 assert_eq!(out.info().color_mode, NDColorMode::RGB1);
2129
2130 let mono = decompress_jpeg(&compress_jpeg(&make_u8_array(16, 16), 90).unwrap()).unwrap();
2132 assert_eq!(
2133 mono.attributes.get("ColorMode").unwrap().value.as_i64(),
2134 Some(NDColorMode::Mono as i64)
2135 );
2136 }
2137
2138 #[test]
2139 fn test_r8_62_jpeg_accepts_int8_like_c() {
2140 let mut arr = NDArray::new(
2144 vec![NDDimension::new(8), NDDimension::new(8)],
2145 NDDataType::Int8,
2146 );
2147 if let NDDataBuffer::I8(ref mut v) = arr.data {
2148 for (i, x) in v.iter_mut().enumerate() {
2149 *x = (i as i32 - 32) as i8;
2150 }
2151 }
2152 let out = compress_jpeg(&arr, 90).expect("Int8 must compress");
2153 assert_eq!(&out.data.as_u8_slice()[0..2], &[0xFF, 0xD8]);
2154 assert_eq!(
2155 out.codec.as_ref().unwrap().original_data_type,
2156 NDDataType::Int8
2157 );
2158 }
2159
2160 #[test]
2161 fn test_r8_62_jpeg_rejects_3d_without_an_rgb_colormode() {
2162 let arr = NDArray::new(
2171 vec![
2172 NDDimension::new(3),
2173 NDDimension::new(8),
2174 NDDimension::new(8),
2175 ],
2176 NDDataType::UInt8,
2177 );
2178 assert_eq!(
2179 compress_jpeg(&arr, 90).unwrap_err(),
2180 JpegCompressError::EncodeFailed,
2181 "3-D Mono (no ColorMode attribute) is not a JPEG-encodable layout in C"
2182 );
2183 }
2184
2185 #[test]
2186 fn test_jpeg_rejects_non_u8() {
2187 let arr = NDArray::new(
2189 vec![NDDimension::new(8), NDDimension::new(8)],
2190 NDDataType::UInt16,
2191 );
2192 let err = compress_jpeg(&arr, 90).unwrap_err();
2193 assert_eq!(err, JpegCompressError::NotEightBit);
2194 assert_eq!(err.message(), "JPEG only supports 8-bit data");
2195 }
2196
2197 #[test]
2198 fn test_jpeg_rejects_1d() {
2199 let arr = NDArray::new(vec![NDDimension::new(64)], NDDataType::UInt8);
2201 let err = compress_jpeg(&arr, 90).unwrap_err();
2202 assert_eq!(err, JpegCompressError::UnsupportedArrayStructure);
2203 assert_eq!(err.message(), "Unsupported array structure");
2204 }
2205
2206 #[test]
2207 fn test_r8_74_jpeg_compress_failures_carry_the_c_error_texts() {
2208 use ad_core_rs::attributes::{NDAttrSource, NDAttrValue, NDAttribute};
2214 use ad_core_rs::color::NDColorMode;
2215
2216 let arr = NDArray::new(
2218 vec![NDDimension::new(8), NDDimension::new(8)],
2219 NDDataType::Float32,
2220 );
2221 assert_eq!(
2222 compress_jpeg(&arr, 90).unwrap_err().message(),
2223 "JPEG only supports 8-bit data"
2224 );
2225
2226 let arr = NDArray::new(
2228 vec![
2229 NDDimension::new(2),
2230 NDDimension::new(2),
2231 NDDimension::new(2),
2232 NDDimension::new(2),
2233 ],
2234 NDDataType::UInt8,
2235 );
2236 assert_eq!(
2237 compress_jpeg(&arr, 90).unwrap_err().message(),
2238 "Unsupported array structure"
2239 );
2240
2241 for (mode, text) in [
2245 (NDColorMode::Bayer, "Unknown color mode 1"),
2246 (NDColorMode::YUV444, "Unknown color mode 5"),
2247 (NDColorMode::YUV411, "Unknown color mode 7"),
2248 ] {
2249 let mut arr = NDArray::new(
2250 vec![NDDimension::new(8), NDDimension::new(8)],
2251 NDDataType::UInt8,
2252 );
2253 arr.attributes.add(NDAttribute::new_static(
2254 "ColorMode",
2255 "",
2256 NDAttrSource::Driver,
2257 NDAttrValue::Int32(mode as i32),
2258 ));
2259 assert_eq!(compress_jpeg(&arr, 90).unwrap_err().message(), text);
2260 }
2261 }
2262
2263 #[test]
2264 fn test_r8_74_codec_error_pv_carries_the_jpeg_text() {
2265 use ad_core_rs::plugin::runtime::ParamUpdate;
2268
2269 let mut proc = CodecProcessor::new(CodecMode::Compress {
2270 codec: CodecName::JPEG,
2271 quality: 90,
2272 });
2273 proc.params.codec_error = Some(13);
2274 let arr = NDArray::new(
2275 vec![NDDimension::new(8), NDDimension::new(8)],
2276 NDDataType::UInt16,
2277 );
2278 let result = proc.process_array(&arr, &NDArrayPool::new(0));
2279 let text = result
2280 .param_updates
2281 .iter()
2282 .find_map(|u| match u {
2283 ParamUpdate::Octet {
2284 reason: 13, value, ..
2285 } => Some(value.clone()),
2286 _ => None,
2287 })
2288 .expect("CodecError posted");
2289 assert_eq!(text, "JPEG only supports 8-bit data");
2290 }
2291
2292 #[test]
2293 fn test_jpeg_quality_affects_size() {
2294 let arr = make_u8_array(64, 64);
2295 let high = compress_jpeg(&arr, 95).unwrap();
2296 let low = compress_jpeg(&arr, 10).unwrap();
2297 let high_size = high.codec.as_ref().unwrap().compressed_size;
2298 let low_size = low.codec.as_ref().unwrap().compressed_size;
2299 assert!(
2300 high_size > low_size,
2301 "high quality ({}) should produce larger output than low quality ({})",
2302 high_size,
2303 low_size,
2304 );
2305 }
2306
2307 #[test]
2310 fn test_zlib_roundtrip_u8() {
2311 let arr = make_u8_array(8, 8);
2312 let original = arr.data.as_u8_slice().to_vec();
2313
2314 let compressed = compress_zlib(&arr);
2315 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::Zlib);
2316 assert_ne!(compressed.data.as_u8_slice(), original.as_slice());
2317
2318 let decompressed = decompress_zlib(&compressed).unwrap();
2319 assert!(decompressed.codec.is_none());
2320 assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
2321 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2322 }
2323
2324 #[test]
2325 fn test_zlib_roundtrip_u16() {
2326 let mut arr = NDArray::new(
2327 vec![NDDimension::new(16), NDDimension::new(16)],
2328 NDDataType::UInt16,
2329 );
2330 if let NDDataBuffer::U16(ref mut v) = arr.data {
2331 for (i, x) in v.iter_mut().enumerate() {
2332 *x = (i * 257 % 65521) as u16;
2333 }
2334 }
2335 let original = arr.data.as_u8_slice().to_vec();
2336
2337 let compressed = compress_zlib(&arr);
2338 assert_eq!(
2339 compressed.codec.as_ref().unwrap().original_data_type,
2340 NDDataType::UInt16
2341 );
2342
2343 let decompressed = decompress_zlib(&compressed).unwrap();
2344 assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
2345 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2346 }
2347
2348 #[test]
2349 fn test_zlib_roundtrip_f64_with_negatives() {
2350 let mut arr = NDArray::new(vec![NDDimension::new(64)], NDDataType::Float64);
2351 if let NDDataBuffer::F64(ref mut v) = arr.data {
2352 for (i, x) in v.iter_mut().enumerate() {
2353 *x = (i as f64 - 32.0) * 3.25;
2354 }
2355 }
2356 let original = arr.data.as_u8_slice().to_vec();
2357
2358 let compressed = compress_zlib(&arr);
2359 let decompressed = decompress_zlib(&compressed).unwrap();
2360 assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
2361 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2362 }
2363
2364 #[test]
2365 fn test_zlib_compresses_repetitive_data() {
2366 let arr = NDArray::new(
2367 vec![NDDimension::new(256), NDDimension::new(256)],
2368 NDDataType::UInt8,
2369 );
2370 let original_size = arr.data.as_u8_slice().len();
2371 let compressed = compress_zlib(&arr);
2372 let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
2373 assert!(
2374 compressed_size < original_size,
2375 "zlib compressed ({compressed_size}) should be < original ({original_size})"
2376 );
2377 }
2378
2379 #[test]
2380 fn test_zlib_via_processor() {
2381 let mut arr = NDArray::new(
2382 vec![NDDimension::new(32), NDDimension::new(32)],
2383 NDDataType::UInt16,
2384 );
2385 if let NDDataBuffer::U16(ref mut v) = arr.data {
2386 for (i, x) in v.iter_mut().enumerate() {
2387 *x = (i * 13) as u16;
2388 }
2389 }
2390 let original = arr.data.as_u8_slice().to_vec();
2391 let pool = NDArrayPool::new(10_000_000);
2392
2393 let mut comp = CodecProcessor::new(CodecMode::Compress {
2394 codec: CodecName::Zlib,
2395 quality: 0,
2396 });
2397 let compressed = comp.process_array(&arr, &pool);
2398 let compressed_arr = &compressed.output_arrays[0];
2399 assert_eq!(compressed_arr.codec.as_ref().unwrap().name, CodecName::Zlib);
2400
2401 let mut decomp = CodecProcessor::new(CodecMode::Decompress);
2402 let result = decomp.process_array(compressed_arr, &pool);
2403 assert_eq!(
2404 result.output_arrays[0].data.as_u8_slice(),
2405 original.as_slice()
2406 );
2407 }
2408
2409 #[test]
2412 fn test_lz4hdf5_roundtrip_u8() {
2413 let mut arr = NDArray::new(
2414 vec![NDDimension::new(64), NDDimension::new(64)],
2415 NDDataType::UInt8,
2416 );
2417 if let NDDataBuffer::U8(ref mut v) = arr.data {
2418 for (i, x) in v.iter_mut().enumerate() {
2419 *x = (i % 251) as u8;
2420 }
2421 }
2422 let original = arr.data.as_u8_slice().to_vec();
2423
2424 let compressed = compress_lz4hdf5(&arr);
2425 assert_eq!(compressed.codec.as_ref().unwrap().name, CodecName::LZ4HDF5);
2426 assert_ne!(compressed.data.as_u8_slice(), original.as_slice());
2427
2428 let decompressed = decompress_lz4hdf5(&compressed).unwrap();
2429 assert!(decompressed.codec.is_none());
2430 assert_eq!(decompressed.data.data_type(), NDDataType::UInt8);
2431 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2432 }
2433
2434 #[test]
2435 fn test_lz4hdf5_roundtrip_u16() {
2436 let mut arr = NDArray::new(
2437 vec![NDDimension::new(80), NDDimension::new(40)],
2438 NDDataType::UInt16,
2439 );
2440 if let NDDataBuffer::U16(ref mut v) = arr.data {
2441 for (i, x) in v.iter_mut().enumerate() {
2442 *x = (i * 37 % 65521) as u16;
2443 }
2444 }
2445 let original = arr.data.as_u8_slice().to_vec();
2446
2447 let compressed = compress_lz4hdf5(&arr);
2448 assert_eq!(
2449 compressed.codec.as_ref().unwrap().original_data_type,
2450 NDDataType::UInt16
2451 );
2452
2453 let decompressed = decompress_lz4hdf5(&compressed).unwrap();
2454 assert_eq!(decompressed.data.data_type(), NDDataType::UInt16);
2455 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2456 }
2457
2458 #[test]
2459 fn test_lz4hdf5_roundtrip_f64_with_negatives() {
2460 let mut arr = NDArray::new(vec![NDDimension::new(97)], NDDataType::Float64);
2461 if let NDDataBuffer::F64(ref mut v) = arr.data {
2462 for (i, x) in v.iter_mut().enumerate() {
2463 *x = (i as f64 - 48.0) * 1.75;
2464 }
2465 }
2466 let original = arr.data.as_u8_slice().to_vec();
2467
2468 let compressed = compress_lz4hdf5(&arr);
2469 let decompressed = decompress_lz4hdf5(&compressed).unwrap();
2470 assert_eq!(decompressed.data.data_type(), NDDataType::Float64);
2471 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2472 }
2473
2474 #[test]
2475 fn test_lz4hdf5_multi_block_roundtrip() {
2476 let block = LZ4HDF5_DEFAULT_BLOCK_SIZE;
2479 let count = block * 2 + block / 3 + 7; let mut arr = NDArray::new(vec![NDDimension::new(count)], NDDataType::UInt8);
2481 if let NDDataBuffer::U8(ref mut v) = arr.data {
2482 for (i, x) in v.iter_mut().enumerate() {
2483 *x = (i.wrapping_mul(2_654_435_761) % 251) as u8;
2484 }
2485 }
2486 let original = arr.data.as_u8_slice().to_vec();
2487
2488 let compressed = compress_lz4hdf5(&arr);
2489 let decompressed = decompress_lz4hdf5(&compressed).unwrap();
2490 assert_eq!(decompressed.data.as_u8_slice(), original.as_slice());
2491 }
2492
2493 #[test]
2494 fn test_lz4hdf5_compresses_repetitive_data() {
2495 let arr = NDArray::new(
2496 vec![NDDimension::new(256), NDDimension::new(256)],
2497 NDDataType::UInt16,
2498 );
2499 let original_size = arr.data.as_u8_slice().len();
2500 let compressed = compress_lz4hdf5(&arr);
2501 let compressed_size = compressed.codec.as_ref().unwrap().compressed_size;
2502 assert!(
2503 compressed_size < original_size,
2504 "lz4hdf5 compressed ({compressed_size}) should be < original ({original_size})"
2505 );
2506 }
2507
2508 #[test]
2509 fn test_lz4hdf5_via_processor() {
2510 let mut arr = NDArray::new(
2511 vec![NDDimension::new(48), NDDimension::new(48)],
2512 NDDataType::UInt16,
2513 );
2514 if let NDDataBuffer::U16(ref mut v) = arr.data {
2515 for (i, x) in v.iter_mut().enumerate() {
2516 *x = (i * 7) as u16;
2517 }
2518 }
2519 let original = arr.data.as_u8_slice().to_vec();
2520 let pool = NDArrayPool::new(10_000_000);
2521
2522 let mut comp = CodecProcessor::new(CodecMode::Compress {
2523 codec: CodecName::LZ4HDF5,
2524 quality: 0,
2525 });
2526 let compressed = comp.process_array(&arr, &pool);
2527 let compressed_arr = &compressed.output_arrays[0];
2528 assert_eq!(
2529 compressed_arr.codec.as_ref().unwrap().name,
2530 CodecName::LZ4HDF5
2531 );
2532
2533 let mut decomp = CodecProcessor::new(CodecMode::Decompress);
2534 let result = decomp.process_array(compressed_arr, &pool);
2535 assert_eq!(
2536 result.output_arrays[0].data.as_u8_slice(),
2537 original.as_slice()
2538 );
2539 }
2540
2541 #[test]
2544 fn test_compressor_ordinal_mapping() {
2545 use ad_core_rs::plugin::runtime::{ParamChangeValue, PluginParamSnapshot};
2549
2550 let cases = [
2551 (0i32, CodecName::None),
2552 (1, CodecName::JPEG),
2553 (2, CodecName::Blosc),
2554 (3, CodecName::LZ4),
2555 (4, CodecName::BSLZ4),
2556 (5, CodecName::Zlib),
2557 (6, CodecName::LZ4HDF5),
2558 ];
2559
2560 for (ordinal, expected) in cases {
2561 let mut proc = CodecProcessor::new(CodecMode::Compress {
2562 codec: CodecName::LZ4,
2563 quality: 85,
2564 });
2565 proc.params.compressor = Some(0);
2568 let snapshot = PluginParamSnapshot {
2569 enable_callbacks: true,
2570 reason: 0,
2571 addr: 0,
2572 value: ParamChangeValue::Int32(ordinal),
2573 };
2574 proc.on_param_change(0, &snapshot);
2575 match proc.mode {
2576 CodecMode::Compress { codec, .. } => assert_eq!(
2577 codec, expected,
2578 "ordinal {ordinal} should select {expected:?}"
2579 ),
2580 other => panic!("expected Compress mode, got {other:?}"),
2581 }
2582 }
2583 }
2584
2585 #[test]
2588 fn test_decompress_wrong_codec() {
2589 let arr = make_u8_array(4, 4);
2590 assert!(decompress_lz4(&arr).is_none());
2591 assert!(decompress_jpeg(&arr).is_none());
2592 assert!(decompress_zlib(&arr).is_none());
2593 assert!(decompress_lz4hdf5(&arr).is_none());
2594 }
2595
2596 #[test]
2599 fn test_processor_lz4_compress() {
2600 let pool = NDArrayPool::new(1_000_000);
2601 let mut proc = CodecProcessor::new(CodecMode::Compress {
2602 codec: CodecName::LZ4,
2603 quality: 0,
2604 });
2605 let arr = make_u8_array(32, 32);
2606 let result = proc.process_array(&arr, &pool);
2607 assert_eq!(result.output_arrays.len(), 1);
2608 assert_eq!(
2609 result.output_arrays[0].codec.as_ref().unwrap().name,
2610 CodecName::LZ4
2611 );
2612 assert!(proc.compression_ratio() >= 1.0);
2613 }
2614
2615 #[test]
2616 fn test_processor_jpeg_compress() {
2617 let pool = NDArrayPool::new(1_000_000);
2618 let mut proc = CodecProcessor::new(CodecMode::Compress {
2619 codec: CodecName::JPEG,
2620 quality: 80,
2621 });
2622 let arr = make_u8_array(16, 16);
2623 let result = proc.process_array(&arr, &pool);
2624 assert_eq!(result.output_arrays.len(), 1);
2625 assert_eq!(
2626 result.output_arrays[0].codec.as_ref().unwrap().name,
2627 CodecName::JPEG
2628 );
2629 }
2630
2631 #[test]
2632 fn test_processor_decompress_auto_lz4() {
2633 let pool = NDArrayPool::new(1_000_000);
2634 let arr = make_u8_array(16, 16);
2635 let compressed = compress_lz4(&arr);
2636
2637 let mut proc = CodecProcessor::new(CodecMode::Decompress);
2638 let result = proc.process_array(&compressed, &pool);
2639 assert_eq!(result.output_arrays.len(), 1);
2640 assert!(result.output_arrays[0].codec.is_none());
2641 assert_eq!(
2642 result.output_arrays[0].data.as_u8_slice(),
2643 arr.data.as_u8_slice()
2644 );
2645 assert!(proc.compression_ratio() > 0.0);
2646 }
2647
2648 #[test]
2649 fn test_processor_decompress_auto_jpeg() {
2650 let pool = NDArrayPool::new(1_000_000);
2651 let arr = make_u8_array(16, 16);
2652 let compressed = compress_jpeg(&arr, 90).unwrap();
2653
2654 let mut proc = CodecProcessor::new(CodecMode::Decompress);
2655 let result = proc.process_array(&compressed, &pool);
2656 assert_eq!(result.output_arrays.len(), 1);
2657 assert!(result.output_arrays[0].codec.is_none());
2658 }
2659
2660 #[test]
2661 fn test_processor_decompress_no_codec() {
2662 let pool = NDArrayPool::new(1_000_000);
2663 let arr = make_u8_array(8, 8);
2664 let mut proc = CodecProcessor::new(CodecMode::Decompress);
2665 let result = proc.process_array(&arr, &pool);
2666 assert_eq!(result.output_arrays.len(), 1);
2668 assert_eq!(proc.compression_ratio(), 1.0);
2669 }
2670
2671 fn processor_with_params(mode: CodecMode) -> CodecProcessor {
2676 let mut proc = CodecProcessor::new(mode);
2677 proc.params.comp_factor = Some(10);
2678 proc.params.compressor = Some(11);
2679 proc.params.codec_status = Some(12);
2680 proc.params.codec_error = Some(13);
2681 proc
2682 }
2683
2684 fn int32_update(updates: &[ParamUpdate], reason: usize) -> Option<i32> {
2685 updates.iter().find_map(|u| match u {
2686 ParamUpdate::Int32 {
2687 reason: r, value, ..
2688 } if *r == reason => Some(*value),
2689 _ => None,
2690 })
2691 }
2692
2693 fn octet_update(updates: &[ParamUpdate], reason: usize) -> Option<String> {
2694 updates.iter().find_map(|u| match u {
2695 ParamUpdate::Octet {
2696 reason: r, value, ..
2697 } if *r == reason => Some(value.clone()),
2698 _ => None,
2699 })
2700 }
2701
2702 #[test]
2703 fn test_r8_61_decompress_uncompressed_input_is_success_passthrough() {
2704 let pool = NDArrayPool::new(1_000_000);
2709 let arr = make_u8_array(8, 8);
2710 let mut proc = processor_with_params(CodecMode::Decompress);
2711 let result = proc.process_array(&arr, &pool);
2712
2713 assert_eq!(
2714 int32_update(&result.param_updates, 12),
2715 Some(0),
2716 "CodecStatus must stay SUCCESS on an uncompressed input"
2717 );
2718 assert_eq!(
2719 octet_update(&result.param_updates, 13),
2720 Some(String::new()),
2721 "no error string on a pass-through"
2722 );
2723 assert_eq!(
2724 int32_update(&result.param_updates, 11),
2725 Some(0),
2726 "COMPRESSOR must be set to NDCODEC_NONE"
2727 );
2728 assert_eq!(
2729 result.output_arrays[0].data.as_u8_slice(),
2730 arr.data.as_u8_slice(),
2731 "the input array is passed through unchanged"
2732 );
2733 assert_eq!(proc.compression_ratio(), 1.0);
2734 }
2735
2736 #[test]
2737 fn test_r8_61_decompress_reports_compressor_of_the_input_codec() {
2738 let pool = NDArrayPool::new(1_000_000);
2741 let src = make_u8_array(16, 16);
2742 for (codec, ordinal) in [
2743 (compress_lz4(&src), 3),
2744 (compress_blosc(&src, &BloscConfig::default()), 2),
2745 (compress_bslz4(&src), 4),
2746 (compress_jpeg(&src, 90).expect("jpeg"), 1),
2747 ] {
2748 let mut proc = processor_with_params(CodecMode::Decompress);
2749 let result = proc.process_array(&codec, &pool);
2750 assert_eq!(
2751 int32_update(&result.param_updates, 11),
2752 Some(ordinal),
2753 "COMPRESSOR must report the input codec's C ordinal"
2754 );
2755 assert_eq!(
2756 int32_update(&result.param_updates, 12),
2757 Some(0),
2758 "a successful decompress is SUCCESS"
2759 );
2760 }
2761 }
2762
2763 #[test]
2764 fn test_r8_61_compress_with_compressor_none_is_success_passthrough() {
2765 let pool = NDArrayPool::new(1_000_000);
2770 let arr = make_u8_array(8, 8);
2771 let mut proc = processor_with_params(CodecMode::Compress {
2772 codec: CodecName::None,
2773 quality: 85,
2774 });
2775 let result = proc.process_array(&arr, &pool);
2776
2777 assert_eq!(int32_update(&result.param_updates, 12), Some(0));
2778 assert_eq!(octet_update(&result.param_updates, 13), Some(String::new()));
2779 assert_eq!(
2780 int32_update(&result.param_updates, 11),
2781 None,
2782 "compress mode must not overwrite the operator's COMPRESSOR selection"
2783 );
2784 assert!(result.output_arrays[0].codec.is_none());
2785 assert_eq!(
2786 result.output_arrays[0].data.as_u8_slice(),
2787 arr.data.as_u8_slice()
2788 );
2789 }
2790
2791 #[test]
2792 fn test_r8_61_genuine_decompress_failure_still_reports_an_error() {
2793 let pool = NDArrayPool::new(1_000_000);
2797 let arr = make_u8_array(16, 16);
2798 let mut corrupted = compress_lz4(&arr);
2799 if let NDDataBuffer::U8(ref mut v) = corrupted.data {
2800 v.truncate(3);
2801 }
2802 let mut proc = processor_with_params(CodecMode::Decompress);
2803 let result = proc.process_array(&corrupted, &pool);
2804
2805 assert_ne!(
2806 int32_update(&result.param_updates, 12),
2807 Some(0),
2808 "a failed decompress must not report SUCCESS"
2809 );
2810 assert_eq!(
2811 octet_update(&result.param_updates, 13),
2812 Some("Failed to LZ4 decompress".to_string())
2813 );
2814 assert_eq!(int32_update(&result.param_updates, 11), Some(3));
2815 assert_eq!(
2816 result.output_arrays[0].data.as_u8_slice(),
2817 corrupted.data.as_u8_slice(),
2818 "the input array is republished on failure"
2819 );
2820 }
2821
2822 #[test]
2825 fn test_r8_63_status_levels_match_c() {
2826 assert_eq!(CodecStatus::Success.as_i32(), 0);
2829 assert_eq!(CodecStatus::Warning.as_i32(), 1);
2830 assert_eq!(CodecStatus::Error.as_i32(), 2);
2831 }
2832
2833 #[test]
2834 fn test_r8_63_already_compressed_is_a_warning_not_success() {
2835 let pool = NDArrayPool::new(1_000_000);
2839 let compressed = compress_lz4(&make_u8_array(16, 16));
2840 let mut proc = processor_with_params(CodecMode::Compress {
2841 codec: CodecName::Zlib,
2842 quality: 85,
2843 });
2844 let result = proc.process_array(&compressed, &pool);
2845
2846 assert_eq!(
2847 int32_update(&result.param_updates, 12),
2848 Some(CodecStatus::Warning.as_i32()),
2849 "already-compressed input must report WARNING(1)"
2850 );
2851 assert_eq!(
2852 octet_update(&result.param_updates, 13),
2853 Some("Array already compressed".to_string())
2854 );
2855 assert_eq!(
2857 result.output_arrays[0].codec.as_ref().unwrap().name,
2858 CodecName::LZ4
2859 );
2860 }
2861
2862 #[test]
2863 fn test_r8_63_genuine_failures_are_error_not_warning() {
2864 let pool = NDArrayPool::new(1_000_000);
2869
2870 let wide = NDArray::new(
2872 vec![NDDimension::new(8), NDDimension::new(8)],
2873 NDDataType::UInt16,
2874 );
2875 let mut proc = processor_with_params(CodecMode::Compress {
2876 codec: CodecName::JPEG,
2877 quality: 85,
2878 });
2879 let result = proc.process_array(&wide, &pool);
2880 assert_eq!(
2881 int32_update(&result.param_updates, 12),
2882 Some(CodecStatus::Error.as_i32()),
2883 "an unsupported JPEG input is an ERROR"
2884 );
2885
2886 let mut corrupted = compress_lz4(&make_u8_array(16, 16));
2888 if let NDDataBuffer::U8(ref mut v) = corrupted.data {
2889 v.truncate(3);
2890 }
2891 let mut proc = processor_with_params(CodecMode::Decompress);
2892 let result = proc.process_array(&corrupted, &pool);
2893 assert_eq!(
2894 int32_update(&result.param_updates, 12),
2895 Some(CodecStatus::Error.as_i32()),
2896 "a failed decompress is an ERROR"
2897 );
2898 }
2899
2900 #[test]
2901 fn test_r8_63_successful_and_passthrough_paths_report_success() {
2902 let pool = NDArrayPool::new(1_000_000);
2905 let arr = make_u8_array(16, 16);
2906
2907 let mut proc = processor_with_params(CodecMode::Compress {
2908 codec: CodecName::LZ4,
2909 quality: 85,
2910 });
2911 let compressed = proc.process_array(&arr, &pool);
2912 assert_eq!(
2913 int32_update(&compressed.param_updates, 12),
2914 Some(CodecStatus::Success.as_i32())
2915 );
2916
2917 let mut proc = processor_with_params(CodecMode::Decompress);
2918 let passthrough = proc.process_array(&arr, &pool);
2919 assert_eq!(
2920 int32_update(&passthrough.param_updates, 12),
2921 Some(CodecStatus::Success.as_i32())
2922 );
2923 }
2924
2925 #[test]
2926 fn test_processor_compression_ratio() {
2927 let pool = NDArrayPool::new(1_000_000);
2928 let mut arr = NDArray::new(
2930 vec![NDDimension::new(128), NDDimension::new(128)],
2931 NDDataType::UInt8,
2932 );
2933 if let NDDataBuffer::U8(ref mut v) = arr.data {
2934 for x in v.iter_mut() {
2935 *x = 0;
2936 }
2937 }
2938
2939 let mut proc = CodecProcessor::new(CodecMode::Compress {
2940 codec: CodecName::LZ4,
2941 quality: 0,
2942 });
2943 let _ = proc.process_array(&arr, &pool);
2944 let ratio = proc.compression_ratio();
2945 assert!(
2946 ratio > 2.0,
2947 "all-zeros 128x128 should compress at least 2x, got {}",
2948 ratio,
2949 );
2950 }
2951
2952 #[test]
2953 fn test_processor_plugin_type() {
2954 let proc = CodecProcessor::new(CodecMode::Decompress);
2955 assert_eq!(proc.plugin_type(), "NDPluginCodec");
2956 }
2957
2958 #[test]
2961 fn test_buffer_from_bytes_u8() {
2962 let data = vec![1u8, 2, 3, 4];
2963 let buf = buffer_from_bytes(&data, NDDataType::UInt8).unwrap();
2964 assert_eq!(buf.data_type(), NDDataType::UInt8);
2965 assert_eq!(buf.len(), 4);
2966 assert_eq!(buf.as_u8_slice(), &[1, 2, 3, 4]);
2967 }
2968
2969 #[test]
2970 fn test_buffer_from_bytes_u16() {
2971 let original = vec![1000u16, 2000, 3000];
2972 let bytes: Vec<u8> = original.iter().flat_map(|v| v.to_ne_bytes()).collect();
2973 let buf = buffer_from_bytes(&bytes, NDDataType::UInt16).unwrap();
2974 assert_eq!(buf.data_type(), NDDataType::UInt16);
2975 assert_eq!(buf.len(), 3);
2976 if let NDDataBuffer::U16(v) = buf {
2977 assert_eq!(v, original);
2978 } else {
2979 panic!("wrong buffer type");
2980 }
2981 }
2982
2983 #[test]
2984 fn test_buffer_from_bytes_bad_alignment() {
2985 let data = vec![0u8; 3];
2987 assert!(buffer_from_bytes(&data, NDDataType::UInt16).is_none());
2988 }
2989
2990 #[test]
2991 fn test_buffer_from_bytes_f64_roundtrip() {
2992 let original = vec![1.5f64, -2.7, 3.14159];
2993 let bytes: Vec<u8> = original.iter().flat_map(|v| v.to_ne_bytes()).collect();
2994 let buf = buffer_from_bytes(&bytes, NDDataType::Float64).unwrap();
2995 if let NDDataBuffer::F64(v) = buf {
2996 assert_eq!(v, original);
2997 } else {
2998 panic!("wrong buffer type");
2999 }
3000 }
3001}