()
| 88 | |
| 89 | |
| 90 | def test_module_example3(): |
| 91 | # Same message as above, but a complex entropy model consisting of two parts: |
| 92 | message = np.array( |
| 93 | [6, 10, -4, 2, 5, 2, 1, 0, 2], dtype=np.int32) |
| 94 | means = np.array([2.3, 6.1, -8.5, 4.1, 1.3], dtype=np.float64) |
| 95 | stds = np.array([6.2, 5.3, 3.8, 3.2, 4.7], dtype=np.float64) |
| 96 | entropy_model1 = constriction.stream.model.QuantizedGaussian(-50, 50) |
| 97 | entropy_model2 = constriction.stream.model.Categorical( |
| 98 | np.array([0.2, 0.5, 0.3], dtype=np.float64), # Probabilities of the symbols 0,1,2. |
| 99 | perfect=False |
| 100 | ) |
| 101 | |
| 102 | # Simply encode both parts in sequence with their respective models: |
| 103 | encoder = constriction.stream.queue.RangeEncoder() |
| 104 | # per-symbol params. |
| 105 | encoder.encode(message[0:5], entropy_model1, means, stds) |
| 106 | encoder.encode(message[5:9], entropy_model2) |
| 107 | |
| 108 | compressed = encoder.get_compressed() |
| 109 | print(f"compressed representation: {compressed}") |
| 110 | print(f"(in binary: {[bin(word) for word in compressed]})") |
| 111 | assert np.all(compressed == np.array([3176507208], dtype=np.uint32)) |
| 112 | |
| 113 | decoder = constriction.stream.queue.RangeDecoder(compressed) |
| 114 | decoded_part1 = decoder.decode(entropy_model1, means, stds) |
| 115 | decoded_part2 = decoder.decode(entropy_model2, 4) |
| 116 | assert np.all(np.concatenate((decoded_part1, decoded_part2)) == message) |
| 117 | |
| 118 | |
| 119 | def test_chain1(): |
nothing calls this directly
no test coverage detected