# 何为bytes？





是Python3的数据类型，详见：[http://c.biancheng.net/view/2175.html](http://c.biancheng.net/view/2175.html)。

# np.frombuffer

返回一维数组：

![在这里插入图片描述](/static/img/images/00321.png)

# cv2.imdecode

cv2.imdecode()函数从指定的内存缓存中读取数据，并把数据转换(解码)成图像格式;主要用于从网络传输数据中恢复出图像。

cv2.imencode()函数是将图片格式转换(编码)成流数据，赋值到内存缓存中;主要用于图像数据格式的压缩，方便网络传输。

# bytes转图片numpy矩阵
```python
img = cv2.imdecode(np.frombuffer(file, dtype=np.uint8), cv2.IMREAD_COLOR)
```

# 逆运算 numpy矩阵转bytes

这里的img_bytes 就是bytes的file
```python
# 对数组的图片格式进行编码
success, encoded_image = cv2.imencode(".jpg", img)
# 将数组转为bytes
img_bytes = encoded_image.tostring()
```

# bytes转图片numpy矩阵


# 完整代码：

```python
# -*- coding:utf-8 -*-
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, File, Form, UploadFile
import cv2
import numpy as np
from io import BytesIO
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.post("/files/")
async def alg_file(
        file: bytes = File(...),
        token: str = Form(...)
):
    img = cv2.imdecode(np.frombuffer(file, dtype=np.uint8), 1)
    return {
        "token": token,
        "size": img.shape
    }
if __name__ == '__main__':
    uvicorn.run('run1:app', host='0.0.0.0', port=8002, reload=True, debug=True, workers=1)
```

FastAPI bytes转矩阵 bytes转图片 bytes opencv URL传文件 POST FORM Python

# 1 安装FastAPI





```powershell
# python 3.8
pip install fastapi uvicorn python-multipart
```

# 2 简单的例子 启动FastAPI服务

需要学习：

- 路由组织

- CORS跨域问题

- uvicorn启动

`run1.py  : `

```python
# -*- coding:utf-8 -*-
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from test_ok import app03
app = FastAPI(
    title='FastAPI',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.include_router(app03, prefix='/c3', tags=['test 3'])
@app.get("/hello")
async def path_params01():
    return {"hello world"}
if __name__ == '__main__':
    uvicorn.run('run1:app', host='0.0.0.0', port=8001, reload=True, debug=True, workers=1)
```

`test_ok.py :`

```python
# -*- coding:utf-8 -*-
from fastapi import APIRouter
app03 = APIRouter()
@app03.get("/path/parameters")
def path_params01():
    return {"message": "This is a message"}
```

`启动：`

```powershell
sudo /home/kevin_xie/miniconda3/envs/fastapi/bin/python run1.py
```


# 3 启动服务 单一文件

`run1.py :`

```python
# -*- coding:utf-8 -*-
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.get("/hello")
async def path_params01():
    return {"hello world"}
if __name__ == '__main__':
    uvicorn.run('run1:app', host='0.0.0.0', port=8001, reload=True, debug=True, workers=1)
```

`启动：`

```powershell
sudo /home/kevin_xie/miniconda3/envs/fastapi/bin/python run1.py
```

# 3 使用UploadFile上传图片和表单

```python
# -*- coding:utf-8 -*-
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, File, Form, UploadFile
import cv2
import numpy as np
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.post("/files/")
async def alg_file(
        fileb: UploadFile = File(...),
        token: str = Form(...)
):
    img = cv2.imdecode(np.fromstring(fileb.file.read(), np.uint8), cv2.IMREAD_COLOR)
    return {
        "token": token,
        "filename": fileb.filename,
        "content_type": fileb.content_type,
        "size": img.shape
    }
if __name__ == '__main__':
    uvicorn.run('run1:app', host='0.0.0.0', port=8001, reload=True, debug=True, workers=1)
```

![在这里插入图片描述](/static/img/images/00324.png)


# 4 使用File上传图片和表单 

```python
# -*- coding:utf-8 -*-
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, File, Form, UploadFile
import cv2
import numpy as np
from io import BytesIO
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.post("/files/")
async def alg_file(
        file: bytes = File(...),
        token: str = Form(...)
):
    img = cv2.imdecode(np.frombuffer(file, dtype=np.uint8), 1)
    return {
        "token": token,
        "size": img.shape
    }
if __name__ == '__main__':
    uvicorn.run('run1:app', host='0.0.0.0', port=8002, reload=True, debug=True, workers=1)
```


# 5 特点对比，为什么选FastAPI 

![在这里插入图片描述](/static/img/images/00325.png)

![在这里插入图片描述](/static/img/images/00326.png)

![在这里插入图片描述](/static/img/images/00327.png)

# 参考

FastAPI中文文档：[https://fastapi.tiangolo.com/zh/features/](https://fastapi.tiangolo.com/zh/features/)

FastAPI的一个项目：[https://github.com/liaogx/fastapi-tutorial](https://github.com/liaogx/fastapi-tutorial)

FastAPI 学习笔记 demo FastAPI对比特点 为什么选FastAPI 开启服务

Python





```python
# coding:utf-8
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
    return "HELLO WORLD"
if __name__ == '__main__':
    app.run(host="0.0.0.0", debug=True, port=2020)
```

gunicorn.conf
```python
workers = 5  # 定义同时开启的处理请求的进程数量，根据网站流量适当调整
worker_class = "gevent"  # 采用gevent库，支持异步处理请求，提高吞吐量
bind = "0.0.0.0:2333"
```


运行sh

```python
nohup /home/kevin_xie/miniconda3/envs/py37kevin/bin/gunicorn app:app -c gunicorn.conf.py >> flask.log 2>&1 &
```

flask 拉起 gunicorn

初始化数据库信息





```
# coding:utf-8
import sqlite3
# 没有就建立数据库 有的话就打开
conn = sqlite3.connect('wulian.db')
cursor = conn.cursor()
cursor.execute('create table tempbiao (id varchar(20) primary key, tempvalue float(32),kaiguan int(16))')
cursor.execute('insert into tempbiao (id, tempvalue,kaiguan) values (\'1\', \'2.365\', \'0\')')
cursor.close()
conn.commit()
conn.close()
```

开启服务

```
# coding:utf-8
from flask import Flask
from gevent.pywsgi import WSGIServer
import sqlite3
app = Flask(__name__)
# 接受浮点数字 传感器访问
@app.route('/mcum/<float:revNo>')
def mcum(revNo):
    # 没有就建立数据库 有的话就打开
    conn = sqlite3.connect('wulian.db')
    cursor = conn.cursor()
    sql = "Update tempbiao Set tempvalue='{0}' Where ID='{1}'".format(revNo, '1')
    cursor.execute(sql)
    conn.commit()
    cursor.close()
    conn.close()
    return '写入的结果是 %f' % revNo
@app.route('/androidm/<int:NO>')
def androidmodifykaiguan(NO):
    # 没有就建立数据库 有的话就打开
    conn = sqlite3.connect('wulian.db')
    cursor = conn.cursor()
    sql = "Update tempbiao Set kaiguan='{0}' Where ID='{1}'".format(NO, '1')
    cursor.execute(sql)
    conn.commit()
    cursor.close()
    conn.close()
    return "OK"
# 手机查询数据库变量 返回回去
@app.route('/androidcheck')
def checkdata():
    # 没有就建立数据库 有的话就打开
    conn = sqlite3.connect('wulian.db')
    cursor = conn.cursor()
    cursor.execute('select * from tempbiao where id=?', ('1',))
    checkresult = cursor.fetchall()
    cursor.close()
    conn.close()
    checkresult=str(list(checkresult[0])[1])
    return checkresult
@app.route('/mcucheck')
def mcugetkaigaun():
    # 没有就建立数据库 有的话就打开
    conn = sqlite3.connect('wulian.db')
    cursor = conn.cursor()
    cursor.execute('select * from tempbiao where id=?', ('1',))
    checkresult = cursor.fetchall()
    print(checkresult)
    cursor.close()
    conn.close()
    checkresult=str(list(checkresult[0])[2])
    return checkresult
if __name__ == '__main__':
    # http_server = WSGIServer(('0.0.0.0', 5000), app)
    # http_server.serve_forever()
    app.run(port=5000)
```

Flask sqlite3 服务器脚本

安装





> apt update apt upgrade apt install gcc wget

> http://www.python.org/ftp/python/3.7.0/Python-3.7.0.tgz tar -xvzf

> Python-3.7.0.tgz cd Python-3.7.0 ./configure --with-ssl make

> 

> sudo apt install build-essential python-dev python-setuptools

> python-pip python-smbus build-essential libncursesw5-dev libgdbm-dev

> libc6-dev zlib1g-dev libsqlite3-dev tk-dev libssl-dev openssl

> libffi-dev

> 

> sudo make install

安装Python编译外部模块文件使用的Python-dev

> sudo apt install aptitude 

> sudo apt install python-dev

升级

> sudo pip3 install --upgrade pip

阿里云服务器Ubuntu18.04，安装Python版本Python3.7

我是一个专注技术分享的博主,尤其爱搞深度学习和单片机电子。你现在看到的是我的个人博客网站。