在 FastAPI 中渲染 NumPy 数组

Question

我发现 如何使用 FastAPI 将 numpy 数组作为图像返回？ 然而，我仍然在努力显示图像，它看起来只是一个白色方块。

我将一个数组读入 io.BytesIO ，如下所示：

def iterarray(array):
    output = io.BytesIO()
    np.savez(output, array)
    yield output.get_value()

在我的端点中，我的返回是 StreamingResponse(iterarray(), media_type='application/octet-stream')

当我将 media_type 留空以推断下载了 zip 文件。

如何将数组显示为图像？

Answer 1

选项 1 - 以字节形式返回图像

下面的示例展示了如何将从磁盘加载的图像或内存中的图像（以 numpy 数组的形式）转换为字节（使用 PIL 或 < code>OpenCV 库）并使用自定义 直接响应。出于本演示的目的，以下代码用于创建内存中示例图像（numpy 数组），该图像基于此回答。

# Function to create a sample RGB image
def create_img():
    w, h = 512, 512
    arr = np.zeros((h, w, 3), dtype=np.uint8)
    arr[0:256, 0:256] = [255, 0, 0] # red patch in upper left
    return arr

使用 PIL

服务器端：

您可以使用 Image.open 从磁盘加载图像，或使用 Image.fromarray 加载内存中的图像（注意：出于演示目的，当该案例正在从磁盘加载图像，如下演示了路由内的操作。但是，如果要多次提供同一个图像，则只能在 startup 和 将其存储到 app 实例，如"="">这个答案和这个答案）。接下来，将图像写入缓冲流，即 BytesIO ，并使用 getvalue() 方法获取缓冲区的全部内容。尽管缓冲流在超出范围时会被垃圾收集，但通常更好的是调用 close() 或使用with 语句，如此处和下面的示例所示。

from fastapi import Response
from PIL import Image
import numpy as np
import io

@app.get('/image', response_class=Response)
def get_image():
    # loading image from disk
    # im = Image.open('test.png')
    
    # using an in-memory image
    arr = create_img()
    im = Image.fromarray(arr)
    
    # save image to an in-memory bytes buffer
    with io.BytesIO() as buf:
        im.save(buf, format='PNG')
        im_bytes = buf.getvalue()
        
    headers = {'Content-Disposition': 'inline; filename="test.png"'}
    return Response(im_bytes, headers=headers, media_type='image/png')

客户端：

下面演示如何使用Python requests模块向上述端点发送请求，并将接收到的字节写入文件，或将字节转换回PIL Image，如所述="https://stackoverflow.com/a/32908899/17865804">这里。

import requests
from PIL import Image

url = 'http://127.0.0.1:8000/image'
r = requests.get(url=url)

# write raw bytes to file
with open('test.png', 'wb') as f:
    f.write(r.content)

# or, convert back to PIL Image
# im = Image.open(io.BytesIO(r.content))
# im.save('test.png') 

使用 OpenCV

服务器端：

您可以使用 < 从磁盘加载图像code>cv2.imread() 函数，或者使用内存中的图像，如果它是 RGB 顺序，如下例所示，则需要转换，如 OpenCV 使用 BGR 作为图像的默认颜色顺序。接下来，使用 cv2.imencode() 函数，它压缩图像数据（基于您传递的文件扩展名）定义输出格式，即 .png、.jpg 等）并将其存储在内存缓冲区中，该缓冲区用于通过网络传输数据。

import cv2

@app.get('/image', response_class=Response)
def get_image():
    # loading image from disk
    # arr = cv2.imread('test.png', cv2.IMREAD_UNCHANGED)
    
    # using an in-memory image
    arr = create_img()
    arr = cv2.cvtColor(arr, cv2.COLOR_RGB2BGR)
    # arr = cv2.cvtColor(arr, cv2.COLOR_RGBA2BGRA) # if dealing with 4-channel RGBA (transparent) image

    success, im = cv2.imencode('.png', arr)
    headers = {'Content-Disposition': 'inline; filename="test.png"'}
    return Response(im.tobytes(), headers=headers, media_type='image/png')

客户端：

在客户端，您可以将原始字节写入文件，或使用 "="">numpy.frombuffer() 函数和 cv2.imdecode()函数将缓冲区解压缩为图像格式（类似于this) - cv2.imdecode() 不需要文件扩展名，因为正确的编解码器将是从缓冲区中压缩图像的第一个字节推导出来。

url = 'http://127.0.0.1:8000/image'
r = requests.get(url=url) 

# write raw bytes to file
with open('test.png', 'wb') as f:
    f.write(r.content)

# or, convert back to image format    
# arr = np.frombuffer(r.content, np.uint8)
# img_np = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED)
# cv2.imwrite('test.png', img_np)

---

有用信息

既然您注意到您希望显示的图像类似于 FileResponse ，使用自定义 Response 返回字节应该是执行此操作的方法，而不是使用 StreamingResponse （如您的问题所示）。要指示应在浏览器中查看图像，HTTP 响应应包含以下Content-Disposition 标头，如所述 此处 并如上面的示例所示（如果 文件名 两边的引号是必需的， >文件名包含特殊字符）：

headers = {'Content-Disposition': 'inline; filename="test.png"'}

然而，要下载图像而不是查看图像（使用附件而不是inline)：

headers = {'Content-Disposition': 'attachment; filename="test.png"'}

如果您想使用 JavaScript 接口（例如 Fetch API 或 Axios）显示（或下载）图像，请查看答案 此处和此处。

对于StreamingResponse，如果整个 numpy 数组/图像已加载到内存中，根本不需要 StreamingResponse （当然，不应该是将已加载到内存中的数据返回到这 客户）。 StreamingResponse 通过迭代 iter() 函数提供的块来进行流式传输。如StreamingResponse的实现所示class，如果您传递的迭代器/生成器不是AsyncIterable，来自外部线程池的线程 - 有关该线程池的更多详细信息，请参阅此答案 -将使用 Starlette 的 ="">同步迭代器href="https://github.com/encode/starlette/blob/827de43b97ad58c7b8daae12ad17c1848a4f1341/starlette/concurrency.py#L54" rel="nofollow noreferrer">iterate_in_threadpool() 函数，以避免阻塞事件循环。还应该注意的是，使用 StreamingResponse 时未设置 Content-Length 响应标头（这是有道理的，因为 StreamingResponse< /code> 应该在您事先不知道响应大小的情况下使用），与 FastAPI/Starlette 的其他 Response 类不同，它们为您设置该标头，以便浏览器知道数据结束的地方。应保持这种方式，就好像包含 Content-Length 标头（其值应与整体响应正文大小（以字节为单位）匹配），然后发送到服务器 StreamingResponse 看起来与 Response 相同，因为在这种情况下服务器不会使用 transfer-encoding: chunked （即使在应用程序级别两者仍然不同） -看看Uvicorn 有关响应标头的文档 和 MDN 有关 Transfer-Encoding 的文档：分块以获取更多详细信息。即使您事先知道主体大小，但仍然需要使用 StreamingResponse，因为它允许您通过指定您选择的块大小来加载和传输数据，这与 FileResponse 不同。 /code> （请参阅稍后的详细信息），您应该确保不自行设置 Content-Length 标头，例如 StreamingResponse(iterfile() , headers={'内容长度': str(content_length)})，因为这会导致服务器不使用transfer-encoding: chunked（无论应用程序将数据传递到网络）服务器以块的形式，如 相关实现）。

如此答案中所述：

当您不知道数据的大小时，分块传输编码是有意义的
提前输出，并且您不想等待收集它
在开始将其发送给客户之前，请先了解所有内容。那可以
适用于提供慢速数据库查询结果之类的东西，但是
它通常不适用于提供图像。

即使您想流式传输保存在磁盘上的图像文件，类文件对象，例如由open()创建的对象，是普通迭代器；因此，您可以直接在 StreamingResponse 中返回它们，如 文档 并如下所示（如果您发现 yield from f 相当慢，当使用StreamingResponse，请查看这个答案，了解如何使用以下命令分块读取文件您选择的块大小 - 应根据您的需求以及服务器的资源进行设置）。应该注意的是，使用 FileResponse 也会将文件内容以块的形式读入内存，而不是一次读取整个内容。但是，如 FileResponse 的实现所示 class，使用的块大小是预先定义的并设置为64KB。因此，根据需求，他们应该决定使用两个 Response 类中的哪一个。

@app.get('/image')
def get_image():
    def iterfile():  
        with open('test.png', mode='rb') as f:  
            yield from f  
            
    return StreamingResponse(iterfile(), media_type='image/png')

或者，如果图像已经加载到内存中，然后保存到 BytesIO 缓冲流，因为 BytesIO 是一个 "="">类文件对象（就像io 模块），您可以直接在 StreamingResponse 中返回它（或者，最好简单地调用 buf.getvalue() 来获取整个图像字节并使用自定义返回它们直接响应，如前所示）。如果返回缓冲流，如下例所示，请记住调用 buf.seek(0)，以便将光标倒回到缓冲区的开头，并调用close() 在后台任务中，以便在响应发送到客户端后丢弃缓冲区。

from fastapi import BackgroundTasks

@app.get('/image')
def get_image(background_tasks: BackgroundTasks):
    # supposedly, the buffer already existed in memory
    arr = create_img()
    im = Image.fromarray(arr)
    buf = BytesIO()
    im.save(buf, format='PNG')

    # rewind the cursor to the start of the buffer
    buf.seek(0)
    # discard the buffer, after the response is returned
    background_tasks.add_task(buf.close)
    return StreamingResponse(buf, media_type='image/png')

因此，在您的情况下，最适合的方法是返回自定义 直接响应，包括您的自定义内容和media_type，以及如前所述，设置 Content-Disposition 标头，以便在浏览器中查看图像。

选项 2 - 将图像返回为 JSON 编码的 numpy 数组

下面不应该用于在浏览器中显示图像，但为了完整起见，在此处添加了它，展示了如何将图像转换为 numpy 数组（最好是， 使用 asarray() 函数)，然后 以JSON格式返回数据，最后在客户端将数据转换回图像，如这个和这个答案。如需标准 Python json 库的更快替代方案，请参阅此答案。

使用 PIL

服务器端：

from PIL import Image
import numpy as np
import json

@app.get('/image')
def get_image():
    im = Image.open('test.png')
    # im = Image.open('test.png').convert('RGBA') # if dealing with 4-channel RGBA (transparent) image 
    arr = np.asarray(im)
    return json.dumps(arr.tolist())

客户端：

import requests
from PIL import Image
import numpy as np
import json

url = 'http://127.0.0.1:8000/image'
r = requests.get(url=url) 
arr = np.asarray(json.loads(r.json())).astype(np.uint8)
im = Image.fromarray(arr)
im.save('test_received.png')

使用 OpenCV

服务器端：

import cv2
import json

@app.get('/image')
def get_image():
    arr = cv2.imread('test.png', cv2.IMREAD_UNCHANGED)
    return json.dumps(arr.tolist())

客户端：

import requests
import numpy as np
import cv2
import json

url = 'http://127.0.0.1:8000/image'
r = requests.get(url=url) 
arr = np.asarray(json.loads(r.json())).astype(np.uint8)
cv2.imwrite('test_received.png', arr)