()
| 871 | |
| 872 | |
| 873 | def test_custom_model_range(): |
| 874 | def fixed_model_params(): |
| 875 | model_scipy = scipy.stats.cauchy(loc=10.3, scale=5.8) |
| 876 | # Wrap the scipy-model in a `CustomModel`, which will implicitly |
| 877 | # quantize it to integers in the given range from -100 to 100 (both |
| 878 | # ends inclusively). |
| 879 | model = constriction.stream.model.CustomModel( |
| 880 | model_scipy.cdf, model_scipy.ppf, -100, 100) |
| 881 | |
| 882 | symbols = np.array([5, 14, -1, 21], dtype=np.int32) |
| 883 | encoder = constriction.stream.queue.RangeEncoder() |
| 884 | encoder.encode(symbols, model) |
| 885 | compressed = encoder.get_compressed() |
| 886 | decoder = constriction.stream.queue.RangeDecoder(compressed) |
| 887 | assert np.all(decoder.decode(model, 4) == symbols) |
| 888 | |
| 889 | def variable_model_params(): |
| 890 | # The optional argument `params` will receive a 1-d python array when |
| 891 | # the model is used for encoding or decoding. |
| 892 | model = constriction.stream.model.CustomModel( |
| 893 | lambda x, loc, scale: scipy.stats.cauchy.cdf(x, loc, scale), |
| 894 | lambda x, loc, scale: scipy.stats.cauchy.ppf(x, loc, scale), |
| 895 | -100, 100) |
| 896 | |
| 897 | model_parameters = np.array([ |
| 898 | (7.3, 3.9), # Location and scale of entropy model for 1st symbol. |
| 899 | (11.5, 5.2), # Location and scale of entropy model for 2nd symbol. |
| 900 | (-3.2, 4.9), # and so on ... |
| 901 | (25.9, 7.1), |
| 902 | ]) |
| 903 | |
| 904 | symbols = np.array([5, 14, -1, 21], dtype=np.int32) |
| 905 | encoder = constriction.stream.queue.RangeEncoder() |
| 906 | encoder.encode( |
| 907 | symbols, model, model_parameters[:, 0].copy(), model_parameters[:, 1].copy()) |
| 908 | compressed = encoder.get_compressed() |
| 909 | decoder = constriction.stream.queue.RangeDecoder(compressed) |
| 910 | assert np.all( |
| 911 | decoder.decode(model, model_parameters[:, 0].copy(), model_parameters[:, 1].copy()) == symbols) |
| 912 | |
| 913 | def discrete_distribution(): |
| 914 | model = constriction.stream.model.CustomModel( |
| 915 | lambda x, params: scipy.stats.binom.cdf(x, n=10, p=params), |
| 916 | lambda x, params: scipy.stats.binom.ppf(x, n=10, p=params), |
| 917 | 0, 10) |
| 918 | |
| 919 | success_probabilities = np.array([0.3, 0.7, 0.2, 0.6]) |
| 920 | |
| 921 | symbols = np.array([4, 8, 1, 5], dtype=np.int32) |
| 922 | encoder = constriction.stream.queue.RangeEncoder() |
| 923 | encoder.encode(symbols, model, success_probabilities) |
| 924 | compressed = encoder.get_compressed() |
| 925 | decoder = constriction.stream.queue.RangeDecoder(compressed) |
| 926 | assert np.all( |
| 927 | decoder.decode(model, success_probabilities) == symbols) |
| 928 | |
| 929 | fixed_model_params() |
| 930 | variable_model_params() |
nothing calls this directly
no test coverage detected