HEX
Server: Apache/2.4.46 (Win64) OpenSSL/1.1.1j PHP/8.4.25
System: Windows NT DESKTOP-4TAV2RJ 10.0 build 19045 (Windows 10) AMD64
User: fred (0)
PHP: 8.4.25
Disabled: NONE
Upload Files
File: C:/Users/fred/anaconda3/Lib/site-packages/distributed/protocol/utils_test.py
from __future__ import annotations

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    import numpy


def get_host_array(a: numpy.ndarray) -> numpy.ndarray | bytes | bytearray:
    """Given a numpy array, find the underlying memory allocated by either
    distributed.protocol.utils.host_array or internally by numpy
    """
    import numpy

    assert isinstance(a, numpy.ndarray)
    o: object = a
    while True:
        if isinstance(o, memoryview):
            o = o.obj
        elif isinstance(o, numpy.ndarray):
            if o.base is not None:
                o = o.base
            else:
                return o
        elif isinstance(o, (bytes, bytearray)):
            return o
        else:  # pragma: nocover
            raise TypeError(f"Unexpected numpy buffer: {o!r}")