Coverage for ccsds124/decompress.py: 100%

1""" 

2CCSDS 124.0-B-1 decompression algorithm implementation. 

3 

4Implements CCSDS 124.0-B-1 decompression (inverse of Section 5.3): 

5- Bit reader for parsing compressed packets 

6- COUNT and RLE decoding 

7- Packet decompression and mask reconstruction 

8 

9MicroPython compatible - no typing module imports. 

10""" 

11 

12# Import for type hints only 

13if False: # noqa: SIM108 

14 from ccsds124.bitvector import BitVector 

15 

16# Constants 

17MAX_ROBUSTNESS = 7 

18MAX_PACKET_LENGTH = 65535 # CCSDS 124.0-B-1 section 3.2: 1 <= F <= 2^16 - 1 

19 

20 

21class Decompressor: 

22 """CCSDS 124.0-B-1 decompressor state and operations.""" 

23 

24 def __init__( 

25 self, 

26 packet_length: int, 

27 robustness: int = 1, 

28 initial_mask: "BitVector | None" = None, 

29 ) -> None: 

30 """ 

31 Initialize decompressor. 

32 

33 Args: 

34 packet_length: F - Output vector length in bits 

35 robustness: Rt - Base robustness level (0-7) 

36 initial_mask: M0 - Initial mask vector (None = all zeros) 

37 """ 

38 from ccsds124.bitvector import BitVector 

39 

40 # CCSDS 124.0-B-1 section 3.2: 1 <= F <= 2^16 - 1 

41 if packet_length < 1 or packet_length > MAX_PACKET_LENGTH: 

42 raise ValueError( 

43 "packet_length must be in 1.." + str(MAX_PACKET_LENGTH) + " bits" 

44 ) 

45 

46 self.F = packet_length 

47 self.robustness = min(robustness, MAX_ROBUSTNESS) 

48 

49 # Initialize bit vectors 

50 self.mask = BitVector(packet_length) 

51 self.initial_mask = BitVector(packet_length) 

52 self.prev_output = BitVector(packet_length) 

53 self.Xt = BitVector(packet_length) # Positive changes tracker 

54 

55 # Set initial mask if provided 

56 if initial_mask is not None: 

57 self.initial_mask.copy_from(initial_mask) 

58 self.mask.copy_from(initial_mask) 

59 

60 # Time index 

61 self.t = 0 

62 

63 def reset(self) -> None: 

64 """Reset decompressor to initial state.""" 

65 self.t = 0 

66 self.mask.copy_from(self.initial_mask) 

67 self.prev_output.zero() 

68 self.Xt.zero() 

69 

70 def decompress_packet(self, reader: "BitReader") -> "BitVector": 

71 """ 

72 Decompress a single compressed packet. 

73 

74 Args: 

75 reader: BitReader positioned at packet start 

76 

77 Returns: 

78 Decompressed output packet (length F) 

79 """ 

80 from ccsds124.bitvector import BitVector 

81 from ccsds124.decode import bit_insert, count_decode, rle_decode 

82 

83 output = BitVector(self.F) 

84 

85 # Copy previous output as prediction base 

86 output.copy_from(self.prev_output) 

87 

88 # Clear positive changes tracker 

89 self.Xt.zero() 

90 

91 # ================================================================ 

92 # Parse ht: Mask change information 

93 # ht = RLE(Xt) || BIT4(Vt) || et || kt || ct || dt 

94 # ================================================================ 

95 

96 # Decode RLE(Xt) - mask changes 

97 Xt = rle_decode(reader, self.F) 

98 

99 # Read BIT4(Vt) - effective robustness 

100 Vt = reader.read_bits(4) 

101 

102 # Process et, kt, ct if Vt > 0 and there are changes 

103 ct = 0 

104 change_count = Xt.hamming_weight() 

105 

106 if Vt > 0 and change_count > 0: 

107 # Read et 

108 et = reader.read_bit() 

109 

110 if et == 1: 

111 # Read kt - determines positive/negative updates 

112 # kt has one bit per change in Xt 

113 kt_bits = [] 

114 

115 # Read kt bits (forward order) 

116 for i in range(self.F): 

117 if Xt.get_bit(i): 

118 bit_val = reader.read_bit() 

119 kt_bits.append(bit_val) 

120 

121 # Apply mask updates based on kt 

122 kt_idx = 0 

123 for i in range(self.F): 

124 if Xt.get_bit(i): 

125 # kt=1 means positive update (mask becomes 0) 

126 # kt=0 means negative update (mask becomes 1) 

127 if kt_bits[kt_idx]: 

128 self.mask.set_bit(i, 0) 

129 self.Xt.set_bit(i, 1) # Track positive change 

130 else: 

131 self.mask.set_bit(i, 1) 

132 kt_idx += 1 

133 

134 # Read ct 

135 ct = reader.read_bit() 

136 else: 

137 # et = 0: all updates are negative (mask bits become 1) 

138 for i in range(self.F): 

139 if Xt.get_bit(i): 

140 self.mask.set_bit(i, 1) 

141 

142 elif Vt == 0 and change_count > 0: 

143 # Vt = 0: toggle mask bits at change positions 

144 for i in range(self.F): 

145 if Xt.get_bit(i): 

146 current_val = self.mask.get_bit(i) 

147 self.mask.set_bit(i, 1 if current_val == 0 else 0) 

148 

149 # Read dt 

150 dt = reader.read_bit() 

151 

152 # ================================================================ 

153 # Parse qt: Optional full mask 

154 # ================================================================ 

155 

156 rt = 0 

157 

158 # dt=1 means both ft=0 and rt=0 (optimization per CCSDS Eq. 13) 

159 # dt=0 means we need to read ft and rt from the stream 

160 if dt == 0: 

161 # Read ft flag 

162 ft = reader.read_bit() 

163 

164 if ft == 1: 

165 # Full mask follows: decode RLE(M XOR (M<<)) 

166 mask_diff = rle_decode(reader, self.F) 

167 

168 # Reverse the horizontal XOR to get the actual mask. 

169 # HXOR encoding: HXOR[i] = M[i] XOR M[i+1], with HXOR[F-1] = M[F-1] 

170 # Reversal: start from LSB (position F-1) and work towards MSB 

171 # M[F-1] = HXOR[F-1] (just copy) 

172 # M[i] = HXOR[i] XOR M[i+1] for i < F-1 

173 

174 # Copy LSB bit directly (position F-1) 

175 current = mask_diff.get_bit(self.F - 1) 

176 self.mask.set_bit(self.F - 1, current) 

177 

178 # Process remaining bits from F-2 down to 0 

179 for i in range(self.F - 2, -1, -1): 

180 hxor_bit = mask_diff.get_bit(i) 

181 # M[i] = HXOR[i] XOR M[i+1] = HXOR[i] XOR current 

182 current = hxor_bit ^ current 

183 self.mask.set_bit(i, current) 

184 

185 # Read rt flag 

186 rt = reader.read_bit() 

187 

188 if rt == 1: 

189 # Full packet follows: COUNT(F) || It 

190 _ = count_decode(reader) # Read and discard packet length 

191 

192 # Read full packet 

193 for i in range(self.F): 

194 bit = reader.read_bit() 

195 output.set_bit(i, bit) 

196 else: 

197 # Compressed: extract unpredictable bits 

198 if ct == 1 and Vt > 0: 

199 # BE(It, (Xt OR Mt)) 

200 extraction_mask = self.mask.or_(self.Xt) 

201 else: 

202 # BE(It, Mt) 

203 extraction_mask = self.mask.copy() 

204 

205 # Insert unpredictable bits 

206 bit_insert(reader, output, extraction_mask) 

207 

208 # ================================================================ 

209 # Update state for next cycle 

210 # ================================================================ 

211 

212 self.prev_output.copy_from(output) 

213 self.t += 1 

214 

215 return output 

216 

217 

218# Import BitReader at module level for type hints 

219if False: # noqa: SIM108 

220 from ccsds124.bitreader import BitReader 

221 

222 

223def decompress( 

224 data: bytes, 

225 packet_size: int, 

226 robustness: int = 1, 

227 initial_mask: "BitVector | None" = None, 

228) -> bytes: 

229 """ 

230 Decompress data using CCSDS 124.0-B-1 algorithm. 

231 

232 High-level API that handles: 

233 - Sequential packet decompression 

234 - Byte boundary alignment between packets 

235 - Output accumulation 

236 

237 Args: 

238 data: Compressed input bytes 

239 packet_size: Packet length in bits 

240 robustness: Rt - Base robustness level (0-7) 

241 initial_mask: M0 initial mask (None = all zeros) 

242 

243 Returns: 

244 Decompressed data bytes 

245 """ 

246 from ccsds124.bitreader import BitReader 

247 

248 # Initialize decompressor 

249 decomp = Decompressor( 

250 packet_length=packet_size, 

251 robustness=robustness, 

252 initial_mask=initial_mask, 

253 ) 

254 

255 # Initialize bit reader 

256 reader = BitReader(data) 

257 

258 # Output accumulation 

259 output_bytes = bytearray() 

260 

261 # Decompress packets until input exhausted 

262 while reader.remaining > 0: 

263 output = decomp.decompress_packet(reader) 

264 

265 # Append to output 

266 output_bytes.extend(output.to_bytes()) 

267 

268 # Align to byte boundary for next packet 

269 reader.align_byte() 

270 

271 return bytes(output_bytes)