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_json_object_array(self):
for chunks in [1, 2, 5, 10, 100]:
n = 10
z = zarr.empty(
n, dtype=object, object_codec=numcodecs.JSON(), chunks=(chunks,)
)
for j in range(n):
z[j] = {str(k): k for k in range(j)}
self.filter_warnings_verify_round_trip({"z": z})
def test_empty_string_list(self):
z = zarr.empty(1, dtype=object, object_codec=numcodecs.JSON(), chunks=(2,))
z[0] = ["", ""]
self.filter_warnings_verify_round_trip({"z": z})
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.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
pass
try:
from numcodecs import lz4
from numcodecs.lz4 import LZ4
register_codec(LZ4)
except ImportError: # pragma: no cover
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
pass
try:
from numcodecs import lz4
from numcodecs.lz4 import LZ4
register_codec(LZ4)
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
from numcodecs.lz4 import LZ4
register_codec(LZ4)
except ImportError: # pragma: no cover
pass
from numcodecs.astype import AsType
register_codec(AsType)
from numcodecs.delta import Delta
register_codec(Delta)
from numcodecs.quantize import Quantize
register_codec(Quantize)
from numcodecs.fixedscaleoffset import FixedScaleOffset
register_codec(FixedScaleOffset)
from numcodecs.packbits import PackBits
register_codec(PackBits)
from numcodecs.categorize import Categorize
register_codec(Categorize)
from numcodecs.pickles import Pickle
register_codec(Pickle)
try:
from numcodecs.msgpacks import MsgPack
register_codec(MsgPack)
except ImportError: # pragma: no cover
pass
from numcodecs.categorize import Categorize
register_codec(Categorize)
from numcodecs.pickles import Pickle
register_codec(Pickle)
try:
from numcodecs.msgpacks import MsgPack
register_codec(MsgPack)
except ImportError: # pragma: no cover
pass
from numcodecs.checksum32 import CRC32, Adler32
register_codec(CRC32)
register_codec(Adler32)
from numcodecs.json import JSON
register_codec(JSON)
try:
from numcodecs import vlen
from numcodecs.vlen import VLenUTF8, VLenBytes, VLenArray
register_codec(VLenUTF8)
register_codec(VLenBytes)
register_codec(VLenArray)
except ImportError: # pragma: no cover
pass
`_.
"""
import multiprocessing
import atexit
from numcodecs.version import version as __version__
from numcodecs.registry import get_codec, register_codec
from numcodecs.zlib import Zlib
register_codec(Zlib)
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:
pass
from numcodecs.astype import AsType
register_codec(AsType)
from numcodecs.delta import Delta
register_codec(Delta)
from numcodecs.quantize import Quantize
register_codec(Quantize)
from numcodecs.fixedscaleoffset import FixedScaleOffset
register_codec(FixedScaleOffset)
from numcodecs.packbits import PackBits
register_codec(PackBits)
from numcodecs.categorize import Categorize
register_codec(Categorize)
from numcodecs.pickles import Pickle
register_codec(Pickle)
try:
from numcodecs.msgpacks import MsgPack
register_codec(MsgPack)
except ImportError: # pragma: no cover
pass
from numcodecs.checksum32 import CRC32, Adler32
register_codec(CRC32)
register_codec(Adler32)