# Path: numcodecs/compat.py
# def ensure_text(s, encoding='utf-8'):
#     if not isinstance(s, str):
#         s = ensure_contiguous_ndarray(s)
#         s = codecs.decode(s, encoding)
#     return s
# 
# def ensure_bytes(buf):
#     """Obtain a bytes object from memory exposed by `buf`."""
# 
#     if not isinstance(buf, bytes):
# 
#         # go via numpy, for convenience
#         arr = ensure_ndarray(buf)
# 
#         # check for object arrays, these are just memory pointers,
#         # actual memory holding item data is scattered elsewhere
#         if arr.dtype == object:
#             raise TypeError('object arrays are not supported')
# 
#         # create bytes
#         buf = arr.tobytes(order='A')
# 
#     return buf
# 
# def ensure_contiguous_ndarray(buf, max_buffer_size=None):
#     """Convenience function to coerce `buf` to a numpy array, if it is not already a
#     numpy array. Also ensures that the returned value exports fully contiguous memory,
#     and supports the new-style buffer interface. If the optional max_buffer_size is
#     provided, raise a ValueError if the number of bytes consumed by the returned
#     array exceeds this value.
# 
#     Parameters
#     ----------
#     buf : array-like or bytes-like
#         A numpy array or any object exporting a buffer interface.
#     max_buffer_size : int
#         If specified, the largest allowable value of arr.nbytes, where arr
#         is the returned array.
# 
#     Returns
#     -------
#     arr : ndarray
#         A numpy array, sharing memory with `buf`.
# 
#     Notes
#     -----
#     This function will not create a copy under any circumstances, it is guaranteed to
#     return a view on memory exported by `buf`.
# 
#     """
# 
#     # ensure input is a numpy array
#     arr = ensure_ndarray(buf)
# 
#     # check for object arrays, these are just memory pointers, actual memory holding
#     # item data is scattered elsewhere
#     if arr.dtype == object:
#         raise TypeError('object arrays are not supported')
# 
#     # check for datetime or timedelta ndarray, the buffer interface doesn't support those
#     if arr.dtype.kind in 'Mm':
#         arr = arr.view(np.int64)
# 
#     # check memory is contiguous, if so flatten
#     if arr.flags.c_contiguous or arr.flags.f_contiguous:
#         # can flatten without copy
#         arr = arr.reshape(-1, order='A')
# 
#     else:
#         raise ValueError('an array with contiguous memory is required')
# 
#     if max_buffer_size is not None and arr.nbytes > max_buffer_size:
#         msg = "Codec does not support buffers of > {} bytes".format(max_buffer_size)
#         raise ValueError(msg)
# 
#     return arr


import array
import mmap
import numpy as np
import pytest
from numcodecs.compat import ensure_text, ensure_bytes, ensure_contiguous_ndarray
    typed_bufs = [
        ('u', 1, b'adsdasdas'),
        ('u', 1, bytes(20)),
        ('i', 8, np.arange(100, dtype=np.int64)),
        ('f', 8, np.linspace(0, 1, 100, dtype=np.float64)),
        ('i', 4, array.array('i', b'qwertyuiqwertyui')),
        ('u', 4, array.array('I', b'qwertyuiqwertyui')),
        ('f', 4, array.array('f', b'qwertyuiqwertyui')),
        ('f', 8, array.array('d', b'qwertyuiqwertyui')),
        ('i', 1, array.array('b', b'qwertyuiqwertyui')),
        ('u', 1, array.array('B', b'qwertyuiqwertyui')),
        ('u', 1, mmap.mmap(-1, 10))
    ]
    for expected_kind, expected_itemsize, buf in typed_bufs:
        a = ensure_contiguous_ndarray(buf)
        assert isinstance(a, np.ndarray)
        assert expected_kind == a.dtype.kind
        if isinstance(buf, array.array):
            assert buf.itemsize == a.dtype.itemsize
        else:
            assert expected_itemsize == a.dtype.itemsize
        assert np.shares_memory(a, memoryview(buf))


def test_ensure_bytes_invalid_inputs():

    # object array not allowed
    a = np.array(['Xin chào thế giới'], dtype=object)
    for e in [a, memoryview(a)]:
        with pytest.raises(TypeError):

# Path: numcodecs/compat.py
# def ensure_text(s, encoding='utf-8'):
#     if not isinstance(s, str):
#         s = ensure_contiguous_ndarray(s)
#         s = codecs.decode(s, encoding)
#     return s
# 
# def ensure_bytes(buf):
#     """Obtain a bytes object from memory exposed by `buf`."""
# 
#     if not isinstance(buf, bytes):
# 
#         # go via numpy, for convenience
#         arr = ensure_ndarray(buf)
# 
#         # check for object arrays, these are just memory pointers,
#         # actual memory holding item data is scattered elsewhere
#         if arr.dtype == object:
#             raise TypeError('object arrays are not supported')
# 
#         # create bytes
#         buf = arr.tobytes(order='A')
# 
#     return buf
# 
# def ensure_contiguous_ndarray(buf, max_buffer_size=None):
#     """Convenience function to coerce `buf` to a numpy array, if it is not already a
#     numpy array. Also ensures that the returned value exports fully contiguous memory,
#     and supports the new-style buffer interface. If the optional max_buffer_size is
#     provided, raise a ValueError if the number of bytes consumed by the returned
#     array exceeds this value.
# 
#     Parameters
#     ----------
#     buf : array-like or bytes-like
#         A numpy array or any object exporting a buffer interface.
#     max_buffer_size : int
#         If specified, the largest allowable value of arr.nbytes, where arr
#         is the returned array.
# 
#     Returns
#     -------
#     arr : ndarray
#         A numpy array, sharing memory with `buf`.
# 
#     Notes
#     -----
#     This function will not create a copy under any circumstances, it is guaranteed to
#     return a view on memory exported by `buf`.
# 
#     """
# 
#     # ensure input is a numpy array
#     arr = ensure_ndarray(buf)
# 
#     # check for object arrays, these are just memory pointers, actual memory holding
#     # item data is scattered elsewhere
#     if arr.dtype == object:
#         raise TypeError('object arrays are not supported')
# 
#     # check for datetime or timedelta ndarray, the buffer interface doesn't support those
#     if arr.dtype.kind in 'Mm':
#         arr = arr.view(np.int64)
# 
#     # check memory is contiguous, if so flatten
#     if arr.flags.c_contiguous or arr.flags.f_contiguous:
#         # can flatten without copy
#         arr = arr.reshape(-1, order='A')
# 
#     else:
#         raise ValueError('an array with contiguous memory is required')
# 
#     if max_buffer_size is not None and arr.nbytes > max_buffer_size:
#         msg = "Codec does not support buffers of > {} bytes".format(max_buffer_size)
#         raise ValueError(msg)
# 
#     return arr


# Path: numcodecs/tests/test_compat.py
import array
import mmap
import numpy as np
import pytest
from numcodecs.compat import ensure_text, ensure_bytes, ensure_contiguous_ndarray

    typed_bufs = [
        ('u', 1, b'adsdasdas'),
        ('u', 1, bytes(20)),
        ('i', 8, np.arange(100, dtype=np.int64)),
        ('f', 8, np.linspace(0, 1, 100, dtype=np.float64)),
        ('i', 4, array.array('i', b'qwertyuiqwertyui')),
        ('u', 4, array.array('I', b'qwertyuiqwertyui')),
        ('f', 4, array.array('f', b'qwertyuiqwertyui')),
        ('f', 8, array.array('d', b'qwertyuiqwertyui')),
        ('i', 1, array.array('b', b'qwertyuiqwertyui')),
        ('u', 1, array.array('B', b'qwertyuiqwertyui')),
        ('u', 1, mmap.mmap(-1, 10))
    ]
    for expected_kind, expected_itemsize, buf in typed_bufs:
        a = ensure_contiguous_ndarray(buf)
        assert isinstance(a, np.ndarray)
        assert expected_kind == a.dtype.kind
        if isinstance(buf, array.array):
            assert buf.itemsize == a.dtype.itemsize
        else:
            assert expected_itemsize == a.dtype.itemsize
        assert np.shares_memory(a, memoryview(buf))


def test_ensure_bytes_invalid_inputs():

    # object array not allowed
    a = np.array(['Xin chào thế giới'], dtype=object)
    for e in [a, memoryview(a)]:
        with pytest.raises(TypeError):

            ensure_bytes(e)