()
| 202 | |
| 203 | |
| 204 | def test_stack1(): |
| 205 | # Define the two parts of the message and their respective entropy models: |
| 206 | message_part1 = np.array([1, 2, 0, 3, 2, 3, 0], dtype=np.int32) |
| 207 | probabilities_part1 = np.array([0.2, 0.4, 0.1, 0.3], dtype=np.float64) |
| 208 | model_part1 = constriction.stream.model.Categorical(probabilities_part1, perfect=False) |
| 209 | # `model_part1` is a categorical distribution over the (implied) alphabet |
| 210 | # {0,1,2,3} with P(X=0) = 0.2, P(X=1) = 0.4, P(X=2) = 0.1, and P(X=3) = 0.3; |
| 211 | # we will use it below to encode each of the 7 symbols in `message_part1`. |
| 212 | |
| 213 | message_part2 = np.array([6, 10, -4, 2], dtype=np.int32) |
| 214 | means_part2 = np.array([2.5, 13.1, -1.1, -3.0], dtype=np.float64) |
| 215 | stds_part2 = np.array([4.1, 8.7, 6.2, 5.4], dtype=np.float64) |
| 216 | model_family_part2 = constriction.stream.model.QuantizedGaussian(-100, 100) |
| 217 | # `model_family_part2` is a *family* of Gaussian distributions, quantized to |
| 218 | # bins of width 1 centered at the integers -100, -99, ..., 100. We could |
| 219 | # have provided a fixed mean and standard deviation to the constructor of |
| 220 | # `QuantizedGaussian` but we'll instead provide individual means and standard |
| 221 | # deviations for each symbol when we encode and decode `message_part2` below. |
| 222 | |
| 223 | print( |
| 224 | f"Original message: {np.concatenate([message_part1, message_part2])}") |
| 225 | |
| 226 | # Encode both parts of the message in sequence (in reverse order): |
| 227 | coder = constriction.stream.stack.AnsCoder() |
| 228 | coder.encode_reverse( |
| 229 | message_part2, model_family_part2, means_part2, stds_part2) |
| 230 | coder.encode_reverse(message_part1, model_part1) |
| 231 | |
| 232 | # Get and print the compressed representation: |
| 233 | compressed = coder.get_compressed() |
| 234 | print(f"compressed representation: {compressed}") |
| 235 | print(f"(in binary: {[bin(word) for word in compressed]})") |
| 236 | |
| 237 | # You could save `compressed` to a file using `compressed.tofile("filename")`, |
| 238 | # read it back in: `compressed = np.fromfile("filename", dtype=np.uint32) and |
| 239 | # then re-create `coder = constriction.stream.stack.AnsCoder(compressed)`. |
| 240 | |
| 241 | # Decode the message: |
| 242 | decoded_part1 = coder.decode(model_part1, 7) # (decodes 7 symbols) |
| 243 | decoded_part2 = coder.decode(model_family_part2, means_part2, stds_part2) |
| 244 | print(f"Decoded message: {np.concatenate([decoded_part1, decoded_part2])}") |
| 245 | assert np.all(decoded_part1 == message_part1) |
| 246 | assert np.all(decoded_part2 == message_part2) |
| 247 | |
| 248 | |
| 249 | def test_stack2(): |
nothing calls this directly
no test coverage detected