Coverage for ccsds124/compress.py: 100%
193 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-29 15:01 +0000
1"""
2CCSDS 124.0-B-1 compression algorithm implementation.
4Implements CCSDS 124.0-B-1 Section 5.3 (Encoding Step):
5- Compressor initialization and state management
6- Main compression algorithm
7- Output packet encoding: ot = ht || qt || ut
9MicroPython compatible - no typing module imports.
10"""
12# Import for type hints only
13if False: # noqa: SIM108
14 from ccsds124.bitvector import BitVector
16# Constants
17MAX_HISTORY = 16
18MAX_VT_HISTORY = 16
19MAX_ROBUSTNESS = 7
20MAX_PACKET_LENGTH = 65535 # CCSDS 124.0-B-1 section 3.2: 1 <= F <= 2^16 - 1
23class CompressParams:
24 """Compression parameters for a single packet."""
26 def __init__(
27 self,
28 min_robustness: int = 1,
29 new_mask_flag: bool = False,
30 send_mask_flag: bool = False,
31 uncompressed_flag: bool = False,
32 ) -> None:
33 """
34 Initialize compression parameters.
36 Args:
37 min_robustness: Rt - Minimum robustness level (0-7)
38 new_mask_flag: pt - Update mask from build vector
39 send_mask_flag: ft - Include mask in output
40 uncompressed_flag: rt - Send uncompressed
41 """
42 self.min_robustness = min_robustness
43 self.new_mask_flag = new_mask_flag
44 self.send_mask_flag = send_mask_flag
45 self.uncompressed_flag = uncompressed_flag
48class Compressor:
49 """CCSDS 124.0-B-1 compressor state and operations."""
51 def __init__(
52 self,
53 packet_length: int,
54 robustness: int = 1,
55 initial_mask: "BitVector | None" = None,
56 pt_limit: int = 0,
57 ft_limit: int = 0,
58 rt_limit: int = 0,
59 ) -> None:
60 """
61 Initialize compressor.
63 Args:
64 packet_length: F - Input vector length in bits
65 robustness: Rt - Base robustness level (0-7)
66 initial_mask: M0 - Initial mask vector (None = all zeros)
67 pt_limit: New mask period (0 = manual control)
68 ft_limit: Send mask period (0 = manual control)
69 rt_limit: Uncompressed period (0 = manual control)
70 """
71 from ccsds124.bitvector import BitVector
73 # CCSDS 124.0-B-1 section 3.2: 1 <= F <= 2^16 - 1
74 if packet_length < 1 or packet_length > MAX_PACKET_LENGTH:
75 raise ValueError(
76 "packet_length must be in 1.." + str(MAX_PACKET_LENGTH) + " bits"
77 )
79 self.F = packet_length
80 self.robustness = min(robustness, MAX_ROBUSTNESS)
82 # Period limits for automatic parameter management
83 self.pt_limit = pt_limit
84 self.ft_limit = ft_limit
85 self.rt_limit = rt_limit
87 # Initialize bit vectors
88 self.mask = BitVector(packet_length)
89 self.prev_mask = BitVector(packet_length)
90 self.build = BitVector(packet_length)
91 self.prev_input = BitVector(packet_length)
92 self.initial_mask = BitVector(packet_length)
94 # Set initial mask if provided
95 if initial_mask is not None:
96 self.initial_mask.copy_from(initial_mask)
97 self.mask.copy_from(initial_mask)
99 # Change history (circular buffer)
100 self.change_history = [BitVector(packet_length) for _ in range(MAX_HISTORY)]
101 self.history_index = 0
103 # Flag history for ct calculation
104 self.new_mask_flag_history = [0] * MAX_VT_HISTORY
105 self.flag_history_index = 0
107 # Time index
108 self.t = 0
110 # Countdown counters
111 self.pt_counter = pt_limit
112 self.ft_counter = ft_limit
113 self.rt_counter = rt_limit
115 def reset(self) -> None:
116 """Reset compressor to initial state."""
117 self.t = 0
118 self.history_index = 0
120 # Reset mask to initial
121 self.mask.copy_from(self.initial_mask)
122 self.prev_mask.zero()
123 self.build.zero()
124 self.prev_input.zero()
126 # Clear change history
127 for ch in self.change_history:
128 ch.zero()
130 # Reset countdown counters
131 self.pt_counter = self.pt_limit
132 self.ft_counter = self.ft_limit
133 self.rt_counter = self.rt_limit
135 def compress_packet(
136 self, input_vec: "BitVector", params: "CompressParams | None" = None
137 ) -> bytes:
138 """
139 Compress a single input packet.
141 Args:
142 input_vec: Input packet It (must be length F)
143 params: Compression parameters (None = use defaults)
145 Returns:
146 Compressed output bytes
147 """
148 from ccsds124.bitbuffer import BitBuffer
149 from ccsds124.bitvector import BitVector
150 from ccsds124.encode import (
151 bit_extract,
152 bit_extract_forward,
153 count_encode,
154 rle_encode,
155 )
156 from ccsds124.mask import compute_change, update_build, update_mask
158 # Use default params if none provided
159 if params is None:
160 params = CompressParams(min_robustness=self.robustness)
162 output = BitBuffer()
164 # ================================================================
165 # STEP 1: Update Mask and Build Vectors (CCSDS Section 4)
166 # ================================================================
168 # Save previous mask
169 prev_mask = self.mask.copy()
171 # Save previous build
172 prev_build = self.build.copy()
174 # Update build vector (Equation 6)
175 if self.t > 0:
176 update_build(
177 self.build,
178 input_vec,
179 self.prev_input,
180 params.new_mask_flag,
181 self.t,
182 )
184 # Update mask vector (Equation 7)
185 if self.t > 0:
186 update_mask(
187 self.mask,
188 input_vec,
189 self.prev_input,
190 prev_build,
191 params.new_mask_flag,
192 )
194 # Compute change vector (Equation 8)
195 change = BitVector(self.F)
196 compute_change(change, self.mask, prev_mask, self.t)
198 # Store change in history (circular buffer)
199 self.change_history[self.history_index].copy_from(change)
201 # ================================================================
202 # STEP 2: Encode Output Packet (CCSDS Section 5.3)
203 # ot = ht || qt || ut
204 # ================================================================
206 # Calculate Xt (robustness window)
207 Xt = compute_robustness_window(self, change)
209 # Calculate Vt (effective robustness)
210 Vt = compute_effective_robustness(self, change)
212 # Calculate dt flag
213 dt = 1 if (not params.send_mask_flag and not params.uncompressed_flag) else 0
215 # ================================================================
216 # Component ht: Mask change information
217 # ht = RLE(Xt) || BIT4(Vt) || et || kt || ct || dt
218 # ================================================================
220 # 1. RLE(Xt) - Run-length encode the robustness window
221 rle_encode(output, Xt)
223 # 2. BIT4(Vt) - 4-bit effective robustness level
224 for i in range(3, -1, -1):
225 output.append_bit((Vt >> i) & 1)
227 # 3. et, kt, ct - Only if Vt > 0 and there are mask changes
228 if Vt > 0 and Xt.hamming_weight() > 0:
229 # Calculate et
230 et = has_positive_updates(Xt, self.mask)
231 output.append_bit(et)
233 if et:
234 # kt - Output '1' for positive updates (mask=0), '0' for negative
235 # Extract INVERTED mask values in forward order
236 inverted_mask = BitVector(self.F)
237 for j in range(self.mask.length):
238 mask_bit = self.mask.get_bit(j)
239 inverted_mask.set_bit(j, 1 if mask_bit == 0 else 0)
241 bit_extract_forward(output, inverted_mask, Xt)
243 # Calculate and encode ct
244 ct = compute_ct_flag(self, Vt, params.new_mask_flag)
245 output.append_bit(ct)
247 # 4. dt - Flag indicating if both ft and rt are zero
248 output.append_bit(dt)
250 # ================================================================
251 # Component qt: Optional full mask
252 # qt = empty if dt=1, '1' || RLE(<(Mt XOR (Mt<<))>) if ft=1, '0' otherwise
253 # ================================================================
255 if dt == 0: # Only if dt = 0
256 if params.send_mask_flag:
257 output.append_bit(1) # Flag: mask follows
259 # Encode mask as RLE(M XOR (M<<))
260 mask_shifted = self.mask.left_shift()
261 mask_diff = self.mask.xor(mask_shifted)
262 rle_encode(output, mask_diff)
263 else:
264 output.append_bit(0) # Flag: no mask
266 # ================================================================
267 # Component ut: Unpredictable bits or full input
268 # ================================================================
270 if params.uncompressed_flag:
271 # '1' || COUNT(F) || It
272 output.append_bit(1) # Flag: full input follows
273 count_encode(output, self.F)
274 output.append_bitvector(input_vec)
275 else:
276 if dt == 0:
277 # '0' || BE(...)
278 output.append_bit(0) # Flag: compressed
280 # Determine extraction mask based on ct
281 ct = compute_ct_flag(self, Vt, params.new_mask_flag)
283 if ct and Vt > 0:
284 # BE(It, (Xt OR Mt)) - extract bits where mask OR changes are set
285 extraction_mask = self.mask.or_(Xt)
286 bit_extract(output, input_vec, extraction_mask)
287 else:
288 # BE(It, Mt) - extract only unpredictable bits
289 bit_extract(output, input_vec, self.mask)
291 # ================================================================
292 # STEP 3: Update State for Next Cycle
293 # ================================================================
295 # Save current input and mask as previous for next iteration
296 self.prev_input.copy_from(input_vec)
297 self.prev_mask.copy_from(self.mask)
299 # Track new_mask_flag for ct calculation
300 self.new_mask_flag_history[self.flag_history_index] = (
301 1 if params.new_mask_flag else 0
302 )
303 self.flag_history_index = (self.flag_history_index + 1) % MAX_VT_HISTORY
305 # Advance time
306 self.t += 1
308 # Advance history index (circular buffer)
309 self.history_index = (self.history_index + 1) % MAX_HISTORY
311 return output.to_bytes()
314def compute_robustness_window(
315 comp: Compressor, current_change: "BitVector"
316) -> "BitVector":
317 """
318 Compute robustness window Xt.
320 Xt = <(Dt-Rt OR Dt-Rt+1 OR ... OR Dt)>
321 where <a> means reverse the bit order.
323 Args:
324 comp: Compressor with change history
325 current_change: Current change vector Dt
327 Returns:
328 Robustness window Xt
329 """
330 from ccsds124.bitvector import BitVector
332 Xt = BitVector(comp.F)
334 if comp.robustness == 0 or comp.t == 0:
335 # Xt = Dt (no reversal - RLE processes LSB to MSB directly)
336 Xt.copy_from(current_change)
337 else:
338 # OR together changes from t-Rt to t
339 combined = current_change.copy()
341 # Determine how many historical changes to include
342 num_changes = min(comp.t, comp.robustness)
344 # OR with historical changes (going backwards from current)
345 for i in range(1, num_changes + 1):
346 # Calculate index of change from i iterations ago
347 hist_idx = (comp.history_index + MAX_HISTORY - i) % MAX_HISTORY
348 combined = combined.or_(comp.change_history[hist_idx])
350 # Don't reverse - RLE will process from LSB to MSB directly
351 Xt.copy_from(combined)
353 return Xt
356def compute_effective_robustness(comp: Compressor, current_change: "BitVector") -> int:
357 """
358 Compute effective robustness Vt.
360 Vt = Rt + Ct (CCSDS Section 5.3.2.2)
361 where Ct = number of consecutive iterations with no mask changes
363 Args:
364 comp: Compressor with change history
365 current_change: Current change vector Dt
367 Returns:
368 Effective robustness Vt (0-15)
369 """
370 _ = current_change # Unused - Ct is computed from history
372 Rt = comp.robustness
373 Vt = Rt
375 # For t > Rt, compute Ct
376 if comp.t > Rt:
377 # Count backwards through history starting from Rt+1 positions back
378 Ct = 0
380 for i in range(Rt + 1, min(16, comp.t + 1)):
381 hist_idx = (comp.history_index + MAX_HISTORY - i) % MAX_HISTORY
382 if comp.change_history[hist_idx].hamming_weight() > 0:
383 break # Found a change, stop counting
384 Ct += 1
385 if Ct >= 15 - Rt:
386 break # Cap at maximum Ct value
388 Vt = Rt + Ct
389 # Note: Vt cannot exceed 15 since Ct is capped at (15 - Rt) above
391 return Vt
394def has_positive_updates(Xt: "BitVector", mask: "BitVector") -> int:
395 """
396 Check for positive mask updates (et flag).
398 et = 1 if any changed bits (in Xt) are predictable (mask bit = 0)
400 Args:
401 Xt: Robustness window
402 mask: Current mask Mt
404 Returns:
405 1 if positive updates exist, 0 otherwise
406 """
407 # Word-level check: Xt AND (NOT mask) - any bit set means positive update
408 for i in range(min(Xt.num_words, mask.num_words)):
409 if Xt._data[i] & ~mask._data[i] & 0xFFFFFFFF:
410 return 1
411 return 0
414def compute_ct_flag(comp: Compressor, Vt: int, current_new_mask_flag: bool) -> int:
415 """
416 Compute ct flag for multiple mask updates.
418 ct = 1 if new_mask_flag was set 2+ times in last Vt+1 iterations
419 (including current packet)
421 Args:
422 comp: Compressor with flag history
423 Vt: Effective robustness level
424 current_new_mask_flag: Current packet's pt value
426 Returns:
427 1 if multiple updates, 0 otherwise
428 """
429 if Vt == 0:
430 return 0
432 # Count how many times new_mask_flag was set
433 count = 0
435 # Include current packet's flag
436 if current_new_mask_flag:
437 count += 1
439 # Check history for Vt previous entries
440 iterations_to_check = min(Vt, comp.t)
442 for i in range(iterations_to_check):
443 # Calculate history index going backwards from previous
444 hist_idx = (comp.flag_history_index + MAX_VT_HISTORY - 1 - i) % MAX_VT_HISTORY
445 if comp.new_mask_flag_history[hist_idx]:
446 count += 1
448 return 1 if count >= 2 else 0
451def compress(
452 data: bytes,
453 packet_size: int,
454 robustness: int = 1,
455 pt_limit: int = 10,
456 ft_limit: int = 20,
457 rt_limit: int = 50,
458 initial_mask: "BitVector | None" = None,
459) -> bytes:
460 """
461 Compress data using CCSDS 124.0-B-1 algorithm.
463 High-level API that handles:
464 - Splitting input into F-bit packets
465 - Automatic pt/ft/rt parameter management
466 - CCSDS init phase (first Rt+1 packets)
467 - Output accumulation with byte padding
469 Args:
470 data: Input data bytes (must be multiple of packet_size/8)
471 packet_size: Packet length in bits
472 robustness: Rt - Base robustness level (0-7)
473 pt_limit: New mask period (default 10)
474 ft_limit: Send mask period (default 20)
475 rt_limit: Uncompressed period (default 50)
476 initial_mask: M0 initial mask (None = all zeros)
478 Returns:
479 Compressed data bytes
480 """
481 from ccsds124.bitvector import BitVector
483 # Calculate packet size in bytes
484 packet_bytes = (packet_size + 7) // 8
486 # Verify input size is a multiple of packet size
487 if len(data) % packet_bytes != 0:
488 raise ValueError(
489 f"Input size {len(data)} is not a multiple of packet size {packet_bytes}"
490 )
492 # Calculate number of packets
493 num_packets = len(data) // packet_bytes
495 # Initialize compressor
496 comp = Compressor(
497 packet_length=packet_size,
498 robustness=robustness,
499 initial_mask=initial_mask,
500 pt_limit=pt_limit,
501 ft_limit=ft_limit,
502 rt_limit=rt_limit,
503 )
505 # Output accumulation
506 output_bytes = bytearray()
508 # Process each packet
509 for i in range(num_packets):
510 # Load packet data
511 input_vec = BitVector(packet_size)
512 packet_data = data[i * packet_bytes : (i + 1) * packet_bytes]
513 input_vec.from_bytes(packet_data)
515 # Compute parameters
516 params = CompressParams(min_robustness=robustness)
518 # Automatic parameter management
519 if pt_limit > 0 and ft_limit > 0 and rt_limit > 0:
520 if i == 0:
521 # First packet: fixed init values
522 params.send_mask_flag = True
523 params.uncompressed_flag = True
524 params.new_mask_flag = False
525 else:
526 # Packets 1+: check and update countdown counters
527 # ft counter
528 if comp.ft_counter == 1:
529 params.send_mask_flag = True
530 comp.ft_counter = ft_limit
531 else:
532 comp.ft_counter -= 1
533 params.send_mask_flag = False
535 # pt counter
536 if comp.pt_counter == 1:
537 params.new_mask_flag = True
538 comp.pt_counter = pt_limit
539 else:
540 comp.pt_counter -= 1
541 params.new_mask_flag = False
543 # rt counter
544 if comp.rt_counter == 1:
545 params.uncompressed_flag = True
546 comp.rt_counter = rt_limit
547 else:
548 comp.rt_counter -= 1
549 params.uncompressed_flag = False
551 # Override for remaining init packets
552 # CCSDS requires first Rt+1 packets to have ft=1, rt=1, pt=0
553 if i <= robustness:
554 params.send_mask_flag = True
555 params.uncompressed_flag = True
556 params.new_mask_flag = False
558 # Compress packet
559 packet_output = comp.compress_packet(input_vec, params)
560 output_bytes.extend(packet_output)
562 return bytes(output_bytes)