Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_compressor(self):
ts = get_example_ts(11, 17)
compressors = [
None,
formats.DEFAULT_COMPRESSOR,
blosc.Blosc(cname="zlib", clevel=1, shuffle=blosc.NOSHUFFLE),
]
with tempfile.TemporaryDirectory(prefix="tsinf_format_test") as tempdir:
for i, compressor in enumerate(compressors):
filename = os.path.join(tempdir, "samples_{}.tmp".format(i))
for path in [None, filename]:
with formats.SampleData(
sequence_length=ts.sequence_length,
path=path,
compressor=compressor,
) as samples:
self.verify_data_round_trip(ts, samples)
for _, array in samples.arrays():
self.assertEqual(array.compressor, compressor)
from numcodecs.gzip import GZip
register_codec(GZip)
from numcodecs.bz2 import BZ2
register_codec(BZ2)
try:
from numcodecs.lzma import LZMA
register_codec(LZMA)
except ImportError: # pragma: no cover
pass
try:
from numcodecs import blosc
from numcodecs.blosc import Blosc
register_codec(Blosc)
# initialize blosc
try:
ncores = multiprocessing.cpu_count()
except OSError: # pragma: no cover
ncores = 1
blosc.init()
blosc.set_nthreads(min(8, ncores))
atexit.register(blosc.destroy)
except ImportError: # pragma: no cover
pass
try:
from numcodecs import zstd
from numcodecs.zstd import Zstd
register_codec(Zstd)
except ImportError: # pragma: no cover