numpy.load из потока io.BytesIO

Question

У меня есть пустые массивы, сохраненные в хранилище BLOB-объектов Azure, и я загружаю их в поток следующим образом:

stream = io.BytesIO()
store.get_blob_to_stream(container, 'cat.npy', stream)

Из stream.getvalue() я знаю, что поток содержит метаданные для восстановления массива. Это первые 150 байтов:

b"\x93NUMPY\x01\x00v\x00{'descr': '|u1', 'fortran_order': False, 'shape': (720, 1280, 3), }                                                  \n\xc1\xb0\x94\xc2\xb1\x95\xc3\xb2\x96\xc4\xb3\x97\xc5\xb4\x98\xc6\xb5\x99\xc7\xb6\x9a\xc7"

Можно ли загрузить поток байтов с помощью numpy.load или другим простым способом?

Вместо этого я мог бы сохранить массив на диск и загрузить его с диска, но я бы хотел этого избежать по нескольким причинам ...

РЕДАКТИРОВАТЬ: просто чтобы подчеркнуть, выходные данные должны быть массивом Numpy с формой и dtype, указанным в 128 первых байтах потока.

Answer 1

Я пытался использовать несколько способов реализовать ваши потребности.

Вот мои примеры кодов.

from azure.storage.blob.baseblobservice import BaseBlobService
import numpy as np

account_name = '<your account name>'
account_key = '<your account key>'
container_name = '<your container name>'
blob_name = '<your blob name>'

blob_service = BaseBlobService(
    account_name=account_name,
    account_key=account_key
)

Пример 1. Создание URL-адреса BLOB-объекта с токеном sas для получения содержимого через requests

from azure.storage.blob import BlobPermissions
from datetime import datetime, timedelta
import requests

sas_token = blob_service.generate_blob_shared_access_signature(container_name, blob_name, permission=BlobPermissions.READ, expiry=datetime.utcnow() + timedelta(hours=1))
print(sas_token)
url_with_sas = blob_service.make_blob_url(container_name, blob_name, sas_token=sas_token)
print(url_with_sas)

r = requests.get(url_with_sas)
dat = np.frombuffer(r.content)
print('from requests', dat)

Пример 2. Для загрузки содержимого блоба в память через BytesIO

import io
stream = io.BytesIO()
blob_service.get_blob_to_stream(container_name, blob_name, stream)
dat = np.frombuffer(stream.getbuffer())
print('from BytesIO', dat)

Пример 3. Используйте numpy.fromfile с DataSource, чтобы открыть URL-адрес блоба с токеном sas, он фактически загрузит файл блоба в локальную файловую систему.

ds = np.DataSource()
# ds = np.DataSource(None)  # use with temporary file
# ds = np.DataSource(path) # use with path like `data/`
f = ds.open(url_with_sas)
dat = np.fromfile(f)
print('from DataSource', dat)

Я думаю, что образцы 1 и 2 лучше для вас.

Answer 2

Это немного хакерский способ, которым я придумал, который в основном просто получает метаданные из первых 128 байтов:

def load_npy_from_stream(stream_):
    """Experimental, may not work!

    :param stream_: io.BytesIO() object obtained by e.g. calling BlockBlobService().get_blob_to_stream() containing
        the binary stream of a standard format .npy file.
    :return: numpy.ndarray
    """
    stream_.seek(0)
    prefix_ = stream_.read(128)  # first 128 bytes seem to be the metadata
    dict_string = re.search('\{(.*?)\}', prefix_[1:].decode())[0]
    metadata_dict = eval(dict_string)

    array = np.frombuffer(stream_.read(), dtype=metadata_dict['descr']).reshape(metadata_dict['shape'])

    return array

Может быть неудачным во многих отношениях, но я выкладываю это здесь, если кто-то хочет дать ему шанс. Я буду выполнять тесты с этим и вернусь, как я знаю больше.