(1, 0): ('<H', 'latin1'),

(2, 0): ('<I', 'latin1'),

(3, 0): ('<I', 'utf8'),

""" Return the magic string for the given file format version.

if major < 0 or major > 255:

raise ValueError("major version must be 0 <= major < 256")

if minor < 0 or minor > 255:

raise ValueError("minor version must be 0 <= minor < 256")

""" Read the magic string to get the version of the file format.

magic_str = _read_bytes(fp, MAGIC_LEN, "magic string")

if magic_str[:-2] != MAGIC_PREFIX:

msg = "the magic string is not correct; expected %r, got %r"

raise ValueError(msg % (MAGIC_PREFIX, magic_str[:-2]))

major, minor = magic_str[-2:]

"""

# NOTE: that drop_metadata may not return the right dtype e.g. for user

# dtypes. In that case our code below would fail the same, though.

new_dtype = drop_metadata(dtype)

if new_dtype is not dtype:

warnings.warn("metadata on a dtype is not saved to an npy/npz. "

"Use another format (such as pickle) to store it.",

UserWarning, stacklevel=2)

dtype = new_dtype

if dtype.names is not None:

# This is a record array. The .descr is fine. XXX: parts of the

# record array with an empty name, like padding bytes, still get

# fiddled with. This needs to be fixed in the C implementation of

# dtype().

return dtype.descr

elif not type(dtype)._legacy:

# this must be a user-defined dtype since numpy does not yet expose any

# non-legacy dtypes in the public API

#

# non-legacy dtypes don't yet have __array_interface__

# support. Instead, as a hack, we use pickle to save the array, and lie

# that the dtype is object. When the array is loaded, the descriptor is

# unpickled with the array and the object dtype in the header is

# discarded.

#

# a future NEP should define a way to serialize user-defined

# descriptors and ideally work out the possible security implications

warnings.warn("Custom dtypes are saved as python objects using the "

"pickle protocol. Loading this file requires "

"allow_pickle=True to be set.",

UserWarning, stacklevel=2)

return "|O"

else:

"""

if isinstance(descr, str):

# No padding removal needed

return numpy.dtype(descr)

elif isinstance(descr, tuple):

# subtype, will always have a shape descr[1]

dt = descr_to_dtype(descr[0])

return numpy.dtype((dt, descr[1]))

titles = []

names = []

formats = []

offsets = []

offset = 0

for field in descr:

if len(field) == 2:

name, descr_str = field

dt = descr_to_dtype(descr_str)

else:

name, descr_str, shape = field

dt = numpy.dtype((descr_to_dtype(descr_str), shape))

# Ignore padding bytes, which will be void bytes with '' as name

# Once support for blank names is removed, only "if name == ''" needed)

is_pad = (name == '' and dt.type is numpy.void and dt.names is None)

if not is_pad:

title, name = name if isinstance(name, tuple) else (None, name)

titles.append(title)

names.append(name)

formats.append(dt)

offsets.append(offset)

offset += dt.itemsize

return numpy.dtype({'names': names, 'formats': formats, 'titles': titles,

""" Get the dictionary of header metadata from a numpy.ndarray.

d = {'shape': array.shape}

if array.flags.c_contiguous:

d['fortran_order'] = False

elif array.flags.f_contiguous:

d['fortran_order'] = True

else:

# Totally non-contiguous data. We will have to make it C-contiguous

# before writing. Note that we need to test for C_CONTIGUOUS first

# because a 1-D array is both C_CONTIGUOUS and F_CONTIGUOUS.

d['fortran_order'] = False

d['descr'] = dtype_to_descr(array.dtype)

"""

import struct

assert version is not None

fmt, encoding = _header_size_info[version]

header = header.encode(encoding)

hlen = len(header) + 1

padlen = ARRAY_ALIGN - ((MAGIC_LEN + struct.calcsize(fmt) + hlen) % ARRAY_ALIGN)

try:

header_prefix = magic(*version) + struct.pack(fmt, hlen + padlen)

except struct.error:

msg = f"Header length {hlen} too big for version={version}"

raise ValueError(msg) from None

# Pad the header with spaces and a final newline such that the magic

# string, the header-length short and the header are aligned on a

# ARRAY_ALIGN byte boundary. This supports memory mapping of dtypes

# aligned up to ARRAY_ALIGN on systems like Linux where mmap()

# offset must be page-aligned (i.e. the beginning of the file).

"""

try:

return _wrap_header(header, (1, 0))

except ValueError:

pass

try:

ret = _wrap_header(header, (2, 0))

except UnicodeEncodeError:

pass

else:

warnings.warn("Stored array in format 2.0. It can only be"

"read by NumPy >= 1.9", UserWarning, stacklevel=2)

return ret

header = _wrap_header(header, (3, 0))

warnings.warn("Stored array in format 3.0. It can only be "

"read by NumPy >= 1.17", UserWarning, stacklevel=2)

""" Write the header for an array and returns the version used

header = ["{"]

for key, value in sorted(d.items()):

# Need to use repr here, since we eval these when reading

header.append(f"'{key}': {repr(value)}, ")

header.append("}")

header = "".join(header)

# Add some spare space so that the array header can be modified in-place

# when changing the array size, e.g. when growing it by appending data at

# the end.

shape = d['shape']

header += " " * ((GROWTH_AXIS_MAX_DIGITS - len(repr(

shape[-1 if d['fortran_order'] else 0]

))) if len(shape) > 0 else 0)

if version is None:

header = _wrap_header_guess_version(header)

else:

header = _wrap_header(header, version)

""" Write the header for an array using the 1.0 format.

""" Write the header for an array using the 2.0 format.

"""

return _read_array_header(

"""

return _read_array_header(

"""Clean up 'L' in npz header ints.

import tokenize

from io import StringIO

tokens = []

last_token_was_number = False

for token in tokenize.generate_tokens(StringIO(s).readline):

token_type = token[0]

token_string = token[1]

if (last_token_was_number and

token_type == tokenize.NAME and

token_string == "L"):

continue

else:

tokens.append(token)

last_token_was_number = (token_type == tokenize.NUMBER)

"""

# Read an unsigned, little-endian short int which has the length of the

# header.

import ast

import struct

hinfo = _header_size_info.get(version)

if hinfo is None:

raise ValueError(f"Invalid version {version!r}")

hlength_type, encoding = hinfo

hlength_str = _read_bytes(fp, struct.calcsize(hlength_type), "array header length")

header_length = struct.unpack(hlength_type, hlength_str)[0]

header = _read_bytes(fp, header_length, "array header")

header = header.decode(encoding)

if len(header) > max_header_size:

raise ValueError(

f"Header info length ({len(header)}) is large and may not be safe "

"to load securely.\n"

"To allow loading, adjust `max_header_size` or fully trust "

"the `.npy` file using `allow_pickle=True`.\n"

"For safety against large resource use or crashes, sandboxing "

"may be necessary.")

# The header is a pretty-printed string representation of a literal

# Python dictionary with trailing newlines padded to an ARRAY_ALIGN byte

# boundary. The keys are strings.

# "shape" : tuple of int

# "fortran_order" : bool

# "descr" : dtype.descr

# Versions (2, 0) and (1, 0) could have been created by a Python 2

# implementation before header filtering was implemented.

#

# For performance reasons, we try without _filter_header first though

try:

d = ast.literal_eval(header)

except SyntaxError as e:

if version <= (2, 0):

header = _filter_header(header)

try:

d = ast.literal_eval(header)

except SyntaxError as e2:

msg = "Cannot parse header: {!r}"

raise ValueError(msg.format(header)) from e2

else:

warnings.warn(

"Reading `.npy` or `.npz` file required additional "

"header parsing as it was created on Python 2. Save the "

"file again to speed up loading and avoid this warning.",

UserWarning, stacklevel=4)

else:

msg = "Cannot parse header: {!r}"

raise ValueError(msg.format(header)) from e

if not isinstance(d, dict):

msg = "Header is not a dictionary: {!r}"

raise ValueError(msg.format(d))

if EXPECTED_KEYS != d.keys():

keys = sorted(d.keys())

msg = "Header does not contain the correct keys: {!r}"

raise ValueError(msg.format(keys))

# Sanity-check the values.

if (not isinstance(d['shape'], tuple) or

not all(isinstance(x, int) for x in d['shape'])):

msg = "shape is not valid: {!r}"

raise ValueError(msg.format(d['shape']))

if not isinstance(d['fortran_order'], bool):

msg = "fortran_order is not a valid bool: {!r}"

raise ValueError(msg.format(d['fortran_order']))

try:

dtype = descr_to_dtype(d['descr'])

except TypeError as e:

msg = "descr is not a valid dtype descriptor: {!r}"

raise ValueError(msg.format(d['descr'])) from e

"""

_check_version(version)

_write_array_header(fp, header_data_from_array_1_0(array), version)

if array.itemsize == 0:

buffersize = 0

else:

# Set buffer size to 16 MiB to hide the Python loop overhead.

buffersize = max(16 * 1024 ** 2 // array.itemsize, 1)

dtype_class = type(array.dtype)

if array.dtype.hasobject or not dtype_class._legacy:

# We contain Python objects so we cannot write out the data

# directly. Instead, we will pickle it out

if not allow_pickle:

if array.dtype.hasobject:

raise ValueError("Object arrays cannot be saved when "

"allow_pickle=False")

if not dtype_class._legacy:

raise ValueError("User-defined dtypes cannot be saved "

"when allow_pickle=False")

if pickle_kwargs is None:

pickle_kwargs = {}

pickle.dump(array, fp, protocol=4, **pickle_kwargs)

elif array.flags.f_contiguous and not array.flags.c_contiguous:

if isfileobj(fp):

array.T.tofile(fp)

else:

for chunk in numpy.nditer(

array, flags=['external_loop', 'buffered', 'zerosize_ok'],

buffersize=buffersize, order='F'):

fp.write(chunk.tobytes('C'))

elif isfileobj(fp):

array.tofile(fp)

else:

for chunk in numpy.nditer(

array, flags=['external_loop', 'buffered', 'zerosize_ok'],

buffersize=buffersize, order='C'):

max_header_size=_MAX_HEADER_SIZE):

"""

if allow_pickle:

# Effectively ignore max_header_size, since `allow_pickle` indicates

# that the input is fully trusted.

max_header_size = 2**64

version = read_magic(fp)

_check_version(version)

shape, fortran_order, dtype = _read_array_header(

fp, version, max_header_size=max_header_size)

if len(shape) == 0:

count = 1

else:

count = numpy.multiply.reduce(shape, dtype=numpy.int64)

# Now read the actual data.

if dtype.hasobject:

# The array contained Python objects. We need to unpickle the data.

if not allow_pickle:

raise ValueError("Object arrays cannot be loaded when "

"allow_pickle=False")

if pickle_kwargs is None:

pickle_kwargs = {}

try:

array = pickle.load(fp, **pickle_kwargs)

except UnicodeError as err:

# Friendlier error message

raise UnicodeError(

f"Unpickling a python object failed: {err!r}\n"

"You may need to pass the encoding= option "

"to numpy.load"

) from err

else:

if isfileobj(fp):

# We can use the fast fromfile() function.

array = numpy.fromfile(fp, dtype=dtype, count=count)

else:

# This is not a real file. We have to read it the

# memory-intensive way.

# crc32 module fails on reads greater than 2 ** 32 bytes,

# breaking large reads from gzip streams. Chunk reads to

# BUFFER_SIZE bytes to avoid issue and reduce memory overhead

# of the read. In non-chunked case count < max_read_count, so

# only one read is performed.

# Use np.ndarray instead of np.empty since the latter does

# not correctly instantiate zero-width string dtypes; see

# https://github.com/numpy/numpy/pull/6430

array = numpy.ndarray(count, dtype=dtype)

if dtype.itemsize > 0:

# If dtype.itemsize == 0 then there's nothing more to read

max_read_count = BUFFER_SIZE // min(BUFFER_SIZE, dtype.itemsize)

for i in range(0, count, max_read_count):

read_count = min(max_read_count, count - i)

read_size = int(read_count * dtype.itemsize)

data = _read_bytes(fp, read_size, "array data")

array[i:i + read_count] = numpy.frombuffer(data, dtype=dtype,

count=read_count)

if array.size != count:

raise ValueError(

"Failed to read all data for array. "

f"Expected {shape} = {count} elements, "

f"could only read {array.size} elements. "

"(file seems not fully written?)"

)

if fortran_order:

array = array.reshape(shape[::-1])

array = array.transpose()

else:

array = array.reshape(shape)

fortran_order=False, version=None, *,

max_header_size=_MAX_HEADER_SIZE):

"""

if isfileobj(filename):

raise ValueError("Filename must be a string or a path-like object."

" Memmap cannot use existing file handles.")

if 'w' in mode:

# We are creating the file, not reading it.

# Check if we ought to create the file.

_check_version(version)

# Ensure that the given dtype is an authentic dtype object rather

# than just something that can be interpreted as a dtype object.

dtype = numpy.dtype(dtype)

if dtype.hasobject:

msg = "Array can't be memory-mapped: Python objects in dtype."

raise ValueError(msg)

d = {

"descr": dtype_to_descr(dtype),

"fortran_order": fortran_order,

"shape": shape,

}

# If we got here, then it should be safe to create the file.

with open(os.fspath(filename), mode + 'b') as fp:

_write_array_header(fp, d, version)

offset = fp.tell()

else:

# Read the header of the file first.

with open(os.fspath(filename), 'rb') as fp:

version = read_magic(fp)

_check_version(version)

shape, fortran_order, dtype = _read_array_header(

fp, version, max_header_size=max_header_size)

if dtype.hasobject:

msg = "Array can't be memory-mapped: Python objects in dtype."

raise ValueError(msg)

offset = fp.tell()

if fortran_order:

order = 'F'

else:

order = 'C'

# We need to change a write-only mode to a read-write mode since we've

# already written data to the file.

if mode == 'w+':

mode = 'r+'

marray = numpy.memmap(filename, dtype=dtype, shape=shape, order=order,

mode=mode, offset=offset)