()
| 932 | |
| 933 | |
| 934 | def test_old_custom_model_chain(): |
| 935 | compressed = np.array( |
| 936 | [0xa5dd25f7, 0xfaef49b5, 0xd5b12228, 0x156ceb98, 0x71a0a92b, |
| 937 | 0x99e6d365, 0x2eebfadb, 0x404a567b, 0xf6cbdc09, 0xe63f3848], |
| 938 | dtype=np.uint32) |
| 939 | |
| 940 | def fixed_model_params(): |
| 941 | model_scipy = scipy.stats.cauchy(loc=10.3, scale=5.8) |
| 942 | # Wrap the scipy-model in a `CustomModel`, which will implicitly |
| 943 | # quantize it to integers in the given range from -100 to 100 (both |
| 944 | # ends inclusively). |
| 945 | model = constriction.stream.model.CustomModel( |
| 946 | model_scipy.cdf, model_scipy.ppf, -100, 100) |
| 947 | |
| 948 | coder = constriction.stream.chain.ChainCoder(compressed, False, False) |
| 949 | symbols = coder.decode(model, 4) |
| 950 | assert np.all(symbols == np.array([18, 6, 33, 59])) |
| 951 | coder.encode_reverse(symbols, model) |
| 952 | assert np.all(np.hstack(coder.get_data()) == compressed) |
| 953 | |
| 954 | def variable_model_params(): |
| 955 | # The optional argument `params` will receive a 1-d python array when |
| 956 | # the model is used for encoding or decoding. |
| 957 | model = constriction.stream.model.CustomModel( |
| 958 | lambda x, loc, scale: scipy.stats.cauchy.cdf(x, loc, scale), |
| 959 | lambda x, loc, scale: scipy.stats.cauchy.ppf(x, loc, scale), |
| 960 | -100, 100) |
| 961 | |
| 962 | model_parameters = np.array([ |
| 963 | (7.3, 3.9), # Location and scale of entropy model for 1st symbol. |
| 964 | (11.5, 5.2), # Location and scale of entropy model for 2nd symbol. |
| 965 | (-3.2, 4.9), # and so on ... |
| 966 | (25.9, 7.1), |
| 967 | ]) |
| 968 | |
| 969 | coder = constriction.stream.chain.ChainCoder(compressed, False, False) |
| 970 | symbols = coder.decode( |
| 971 | model, model_parameters[:, 0].copy(), model_parameters[:, 1].copy()) |
| 972 | assert np.all(symbols == np.array([13, 7, 16, 85])) |
| 973 | coder.encode_reverse( |
| 974 | symbols, model, model_parameters[:, 0].copy(), model_parameters[:, 1].copy()) |
| 975 | assert np.all(np.hstack(coder.get_data()) == compressed) |
| 976 | |
| 977 | def discrete_distribution(): |
| 978 | model = constriction.stream.model.CustomModel( |
| 979 | lambda x, params: scipy.stats.binom.cdf(x, n=10, p=params), |
| 980 | lambda x, params: scipy.stats.binom.ppf(x, n=10, p=params), |
| 981 | 0, 10) |
| 982 | |
| 983 | success_probabilities = np.array([0.3, 0.7, 0.2, 0.6]) |
| 984 | |
| 985 | coder = constriction.stream.chain.ChainCoder(compressed, False, False) |
| 986 | symbols = coder.decode(model, success_probabilities) |
| 987 | assert np.all(symbols == np.array([4, 6, 4, 9])) |
| 988 | coder.encode_reverse( |
| 989 | symbols, model, success_probabilities) |
| 990 | assert np.all(np.hstack(coder.get_data()) == compressed) |
| 991 |
nothing calls this directly
no test coverage detected