code: def deserialize_numpy_ndarray(header, frames):
      if header.get("pickle"):
          return pickle.loads(frames[0], buffers=frames[1:])
  
      (frame,) = frames
      (writeable,) = header["writeable"]
  
      is_custom, dt = header["dtype"]
      if is_custom:
          dt = pickle.loads(dt)
      else:
          dt = np.dtype(dt)
  
      if header.get("broadcast_to"):
          shape = header["broadcast_to"]
      else:
          shape = header["shape"]
  
      x = np.ndarray(
          shape, dtype=dt, buffer=frame, strides=header["strides"]
      )
      if not writeable:
          x.flags.writeable = False
      elif not x.flags.writeable:
          # This should exclusively happen when the underlying buffer is read-only, e.g.
          # a read-only mmap.mmap or a bytes object.
          # The only known case is:
          #    decompression with a library that does not support output to
          #    bytearray (lz4 does; snappy, zlib, and zstd don't). Note that this
          #    only applies to buffers whose uncompressed size was small enough
          #    that they weren't sharded (distributed.comm.shard); for larger
          #    buffers the decompressed output is deep-copied beforehand into a
          #    bytearray in order to merge it.
          x = np.require(x, requirements=["W"])
  
      return x
  
file_path: distributed/distributed/protocol/numpy.py
imports: from __future__ import annotations
  import math
  import numpy
  from packaging.version import parse
  from distributed.protocol import pickle
  from distributed.protocol.serialize import dask_deserialize, dask_serialize
  from distributed.utils import log_errors
label_desc: Use image node_id 3 for calling a global function with example usage: deserialize_numpy_ndarray(header, frames) and returns: x, pickle