# 1 Flask 路由阻塞





当启动一个Flask APP如下，在同一浏览器开2个窗口访问这个服务（我的是http://172.18.43.32:8001/hello），前一个请求没返回时，另外一个窗口的相同请求不会有反应。

这是浏览器的锅，同一请求会顺序执行，此时可以用同一台电脑的不同浏览器测试，可以发现不会存在路由阻塞。
```python
import time
from flask import Flask
app = Flask(__name__)
@app.route('/hello')
def hello_world():
    print("receive !")
    time.sleep(20)
    return 'Hello World!'
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8001)
```

# 2 requests访问

- 换个方式，使用requests访问，线程池中100个线程同时执行访问：

```python
import requests
from concurrent.futures import ThreadPoolExecutor
def task():
    res = requests.request(method="GET", url="http://172.18.43.32:8001/hello", timeout=None)
executor = ThreadPoolExecutor(max_workers=100)
for i in range(100):
    executor.submit(task)
```

还是可以看见不会阻塞。


# 3 FastAPI 路由阻塞

下面的情况下，访问就会阻塞。


```python
# -*- coding:utf-8 -*-
import asyncio
import time
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.get("/hello")
async def hl():
    print("receive !")
    time.sleep(20)
    return {
        "token": "ads"
    }
def fastweb():
    uvicorn.run('deal:app', host='0.0.0.0', port=8001, reload=False, debug=False, workers=1)
if __name__ == '__main__':
    fastweb()
```

原因：在async标识的函数中使用同步方法time.sleep(20)，这导致了一个异步程序之间的同步问题，出现了错误！这种写法本身就是错误的，加async 是想利用Python的协程机制 将路由变成异步请求， 进一步提高并发量！只要考虑加async ，那么就得好好设计函数内部， 多用一些异步方法，比如可以写成：

```python
@app.get("/hello")
async def hl():
    print("receive !")
    await asyncio.sleep(20)
    return {
        "token": "ads"
    }
```

设计合理就会极大利用起协程异步，从而极大提高并发量，设计不合理就是呆瓜。若是分不清，还是老老实实和Flask一样写法：

```python
@app.get("/hello")
def hl():
    print("receive !")
    time.sleep(20)
    return {
        "token": "ads"
    }
```

这样也是出现不了路由阻塞的。

# 4 总结：同步并发 异步并发

1 Flask 中的属于同步并发，可以无阻塞访问！但要注意别让浏览器欺骗你！

2 FastAPI 支持异步框架，支持是支持，前提是你得用好它，不然就呆瓜。用好了并发量就提升了！

3 FastAPI 中同样支持Flask 那样子的同步并发，和Flask 一样一样的，但要注意写法！

4 FastAPI 貌似比 Flask 更强大！

# 参考

java web

golang 协程

Flask FastAPI 路由阻塞 Web Python 访问同一url出现阻塞 同步 异步 并发

# 何为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

Opencv 读取的图片是BGR 矩阵：




```python
import cv2 
import matplotlib.pyplot as plt
img=cv2.imread('./res.jpg')
plt.imshow(img)
plt.show()
```

plt显示方法imshow是默认矩阵是RGB的，

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

想要正确显示，需要：

```python
imgrgb= cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
plt.imshow(imgrgb)
plt.show()
```

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

Opencv Python 图片通道 BGR 转 RGB imgrgb= cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

读取




```python
import cv2
import numpy as np 
img = cv2.imdecode(np.fromfile("./人物-1.jpg", dtype=np.uint8), 1) # img是矩阵
```

写入：

```python
import cv2
cv2.imencode('.jpg', img)[1].tofile(os.path.join(self.keyword, self.keyword + str(count) + ".jpg"))
```

Python cv2.imread 无法打开中文路径的图像

当有字典x和字典y，想合并为字典z。





-------------------------------

Python 3.9 + 提供了最简单的方法：

`z = x | y          # NOTE: 3.9+ ONLY`

-------------------------------

Python 3.5 + 的方法：

`z = {**x, **y}`

-------------------------------

Python 3.4 or lower  的方法，使用update方法写一个合并方法：

```python
def merge_dicts(*dict_args):
    """
    Given any number of dictionaries, shallow copy and merge into a new dict,
    precedence goes to key-value pairs in latter dictionaries.
    """
    result = {}
    for dictionary in dict_args:
        result.update(dictionary)
    return result
```

或者直接：

`z = x.update( y )`

-------------------------------

参考1 ： [stackoverflow](https://stackoverflow.com/questions/38987/how-do-i-merge-two-dictionaries-in-a-single-expression-taking-union-of-dictiona)

Python 字典合并 几种方式 优雅的字典合并 如何优雅地合并python字典

是文件，不会获取路径。





```python
def listPathAllfiles( dirname):
    result = []
    for maindir, subdir, file_name_list in os.walk(dirname):
        for filename in file_name_list:
            apath = os.path.join(maindir, filename)
            result.append(apath)
    return result
```

【python】获取路径下所有的文件名称列表，绝对路径返回

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





```python
def get_EMC(temperature_c, humidity1):
    """
    temperature_c  unit C
    humidity1 unit %
    """
    humidity = humidity1 / 100
    temperature_f = 32 + (9.0 / 5.0 * temperature_c)
    EMC_W = 330 + 0.0452 * temperature_f + 0.00415 * temperature_f
    EMC_K = 0.791 + 0.000463 * temperature_f - 0.000000844 * temperature_f * temperature_f
    EMC_K1 = 6.34 + 0.000775 * temperature_f - 0.0000935 * temperature_f * temperature_f
    EMC_K2 = 1.09 + 0.0284 * temperature_f - 0.0000904 * temperature_f * temperature_f
    EMC = 1800 / EMC_W * ((EMC_K * humidity / (1 - EMC_K * humidity)) + (
            EMC_K1 * EMC_K * humidity + 2 * EMC_K1 * EMC_K2 * EMC_K * EMC_K * humidity * humidity) / (
                                  1 + EMC_K1 * EMC_K * humidity + EMC_K1 * EMC_K2 * EMC_K * EMC_K * humidity * humidity))
    return EMC
```

python 平衡含水率计算公式

# 解决办法





一句话，给搜索路径，代码在调用路人甲，就把路人甲的父亲路径加入即可，可以采取的办法1是绝对路径：

```python
import os
import sys
sys.path.append("/home/kevin_xie/yifeinfs/pycharm_k/python_test/")
```

办法2是相对路径+sys的一些方法：

```python
import os
import sys
__dir__ = os.path.dirname(os.path.abspath(__file__))
sys.path.append(__dir__)
sys.path.append(os.path.abspath(os.path.join(__dir__, '../..'))) # to get "/home/kevin_xie/yifeinfs/pycharm_k/python_test/"
```

# 详细说明

python提示找不到包怎么办？

比如我有如下的工程，里面有reco1包和reco2包，每个包里面还有sub包。在prfunction.py中有一个pr函数，我想在test1.py中去运行这个函数。

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

如果在pythcharm中，你可以这么写，没有问题，能找到包。

```powershell
from reco1.sub.prfunction import pr
pr()
```

但你在终端执行，在/home/kevin_xie/yifeinfs/pycharm_k/python_test/reco2/sub2执行是执行不了的：

```powershell
(py37) [kevin_xie@localhost sub2]$ pwd
/home/kevin_xie/yifeinfs/pycharm_k/python_test/reco2/sub2
(py37) [kevin_xie@localhost sub2]$ python test1.py
Traceback (most recent call last):
  File "test1.py", line 2, in <module>
    from reco1.sub.prfunction import pr
ModuleNotFoundError: No module named 'reco1'
```

或者你可以去/home/kevin_xie/yifeinfs/pycharm_k/python_test执行，也是不行的：

```powershell
(py37) [kevin_xie@localhost python_test]$ pwd
/home/kevin_xie/yifeinfs/pycharm_k/python_test
(py37) [kevin_xie@localhost python_test]$ python reco2/sub2/test1.py
Traceback (most recent call last):
  File "reco2/sub2/test1.py", line 2, in <module>
    from reco1.sub.prfunction import pr
ModuleNotFoundError: No module named 'reco1'
```

推出方法，如果这么写，将想调用的包的父亲路径加入，就能寻找到了。


```powershell
import os
import sys
__dir__ = os.path.dirname(os.path.abspath(__file__))
sys.path.append(__dir__)
sys.path.append(os.path.abspath(os.path.join(__dir__, '../..'))) # to get "/home/kevin_xie/yifeinfs/pycharm_k/python_test/"
from reco1.sub.prfunction import pr
pr()
```

python 导包找不到路径终极解决办法 ModuleNotFoundError: No module named ‘xxx’

# 问题





预测模型执行代码过慢，想要快速预测。

每次都是将历史数据作为训练数据喂给模型，然后让模型输出预测数据。这个模型的训练过程太慢。

# 解决

（1）讲训练的模型存储起来，每次预测的时候直接用存储的模型进行预测，躲开训练环节；

（2）有新数据的时候，可以在不紧急的时候训练模型并存储起来；依照执行速度，可以设定更新频率，比如2个小时更新一次模型，1天更新一次模型。

# 实现

建立路径：

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

quikly_predict.py 内容
```python
import os
from fbprophet import Prophet
import pandas as pd
import matplotlib.pyplot as plt
import math
import pickle
class data_quikly_predict:
    pwd = os.path.dirname(os.path.abspath(__file__))
    data_dir = os.path.join(pwd, "data")
    def __init__(self):
        if not os.path.exists(self.data_dir):
            os.mkdir(self.data_dir)
        self.model_list = self.find_all_file(self.data_dir)
    def update_model_list(self):
        self.model_list = self.find_all_file(self.data_dir)
    def find_all_file(self, dirname):
        result = []
        for maindir, subdir, file_name_list in os.walk(dirname):
            for filename in file_name_list:
                apath = os.path.join(maindir, filename)
                result.append(apath)
        return result
    def create_new_model(self, df: pd.DataFrame, new_model_name: str):
        """
        :param df: 训练新模型所需要的数据
        :param new_model_name: 模型名称
        :return:
        """
        assert ("ds" in df.columns) and ("y" in df.columns)
        df['ds'] = df['ds'].astype('datetime64[ns]')
        model = Prophet(yearly_seasonality=True, weekly_seasonality=True)
        model.fit(df)  # 训练模型m
        model_file = os.path.join(self.data_dir, new_model_name)
        if model_file in self.model_list:
            print("模型名称已经被使用,但即将被置换\r\n")
        with open(model_file, 'wb') as f:
            pickle.dump(model, f)
        self.update_model_list()
    def predict(self, future_time: pd.DataFrame, model_name: str):
        """
        :param future_time:未来需要预测的时间点DataFrame，只有一行'ds'
        :param model_name:想要使用的模型名称
        :return:
        """
        model_file = os.path.join(self.data_dir, model_name)
        if model_file not in self.model_list:
            raise print("没有在data里找到这个模型文件，请确保用数据训练出了可使用的模型\r\n")
        with open(model_file, 'rb') as f:
            model = pickle.load(f)
            forecast = model.predict(future_time)  # 开始预测
            forecast = forecast[['ds', 'yhat']]
        return forecast
    def dummy_predict(self, periods, freq: str, model_name: str):
        """
        傻瓜化预测
        :param periods: 需要预测多少个数据结果
        :param freq: 从训练数据末尾调用频率
        :param model_name: 模型名称
        :return:
        """
        model_file = os.path.join(self.data_dir, model_name)
        if model_file not in self.model_list:
            raise print("没有在data里找到这个模型文件，请确保用数据训练出了可使用的模型\r\n")
        with open(model_file, 'rb') as f:
            model = pickle.load(f)
        future_time = model.make_future_dataframe(periods=periods, freq=freq, include_history=False)  # 预测的设置 还没预测
        forecast = model.predict(future_time)  # 开始预测
        forecast = forecast[['ds', 'yhat']]
        return forecast
    def quikly_plot_forecast(self, forecast: pd.DataFrame):
        plt.plot(list(forecast['ds']), list(forecast['yhat']), color='b')
        plt.show()
if __name__ == '__main__':
    dataqp = data_quikly_predict()
    timelist = list(pd.date_range(start='2021-01-01 00:00:00', end='2022-01-01 00:00:00', freq='H'))
    y = [math.sin(data.hour) for k, data in enumerate(timelist)]
    data_df = pd.DataFrame({'ds': timelist, 'y': y})
    dataqp.create_new_model(data_df, "camera1")
    future = pd.DataFrame({'ds': list(pd.date_range(start='2022-05-01 00:00:00',
                                                    end='2022-05-05 01:00:00',
                                                    freq='H'))})  # 预测的设置 还没预测
    forecast_res = dataqp.predict(future, model_name="camera1")
    dataqp.quikly_plot_forecast(forecast_res)
```

# 如何使用

下面代码是正常过程，先执行训练，然后预测数据。
```python
import math
import pandas as pd
from fbprophet_persistent_prodict.quikly_predict import data_quikly_predict
dataqp = data_quikly_predict()
timelist = list(pd.date_range(start='2021-01-01 00:00:00', end='2022-01-01 00:00:00', freq='H'))
y = [math.sin(data.hour) for k, data in enumerate(timelist)]
data_df = pd.DataFrame({'ds': timelist, 'y': y})
dataqp.create_new_model(data_df, "camera6")
future = pd.DataFrame({'ds': list(pd.date_range(start='2022-05-01 00:00:00',
                                            end='2022-05-05 01:00:00',
                                            freq='H'))})  # 预测的设置 还没预测
forecast_res = dataqp.predict(future, model_name="camera6")
dataqp.quikly_plot_forecast(forecast_res)
```

先预测，后用训练更新模型：

```csharp
import math
import pandas as pd
from fbprophet_persistent_prodict.quikly_predict import data_quikly_predict
dataqp = data_quikly_predict()
future = pd.DataFrame({'ds': list(pd.date_range(start='2022-05-01 00:00:00',
                                            end='2022-05-05 01:00:00',
                                            freq='H'))})  # 预测的设置 还没预测
forecast_res = dataqp.predict(future, model_name="camera6")
timelist = list(pd.date_range(start='2021-01-01 00:00:00', end='2022-01-01 00:00:00', freq='H'))
y = [math.sin(data.hour) for k, data in enumerate(timelist)]
data_df = pd.DataFrame({'ds': timelist, 'y': y})
dataqp.create_new_model(data_df, "camera6")
```

python 时序预测模型 fbprophet 持久化存储 大数据快速预测

利用python解释器对文件内置的__file__属性获取：




```python
os.path.dirname(os.path.abspath(__file__))
```

os.path.abspath(__file__)是当前py文件在操作系统的绝对路径文件名，os.path.dirname获取文件所在路径。