|
Flask后端开发(二) - 功能实现和项目总结
目录
前情回顾
Flask后端开发(一)-基础知识和前期准备
1. 功能1:修改文件参数值
针对文件参数值的修改,具体流程如下:
- 前端接收用户修改的数据,传递给后端;
- 后端接口接收数据之后,读取对应文件;
- 定位修改位置,替换数据;
- 返回修改结果给前端。
1.1. 获取网页端传参
前端传递参数的方式有两种,一种是GET,一种是POST,具体可参考Flask后端开发(一)-基础知识和前期准备
后端接收数据主要使用flask中的request模块,具体代码如下:- #包导入
- from flask import request
- #前后端协商好传递数据的名称之后,后端根据参数名进行接收
- if request.method == "POST":
- userID= str(request.form.get("userID"))
- elif request.method == "GET":
- userID= str(request.args.get("userID"))
- #如果需要额外处理,例如字符串"a,b,c"需要转换为列表["a","b","c"],可以使用split函数
- BSD= (request.form.get("BSD")).split(",")
- #注:上述是代码片段,而非完整代码,一般后端接收数据写在接口函数中
复制代码 1.2. 读取文件
1.2.1. 一般文件读取方式
一般文件包括.txt、.c、.log等文件,其内容读取主要使用python中file模块的open函数,具体代码如下:- path= "文件路径"
- with open(path, "r",encoding='utf8') as file:
- file_content = file.read()
- #整个文件内容存储在file_content中
复制代码 关于file模块的具体使用,可参考我的这篇博客:【python技巧】文本文件的读写操作。
1.2.2. 特殊文件 —— mlx文件
本项目的一个特殊之处就是需要处理.mlx文件(实时脚本文件),这是matlab中的一种文件格式,其内容是二进制的,无法直接读取。因此,在本文当中,解决方案是将.mlx文件手动转换为.m文件,然后再读取.m文件的内容(真的很笨蛋,但是有效)。
对于.m文件,则可以按照一般文件的读取方式进行读取。
1.2.3. 特殊文件 —— .xlx文件
本项目中还会涉及到表格文件的数据读写,这里使用的是xlrd模块,具体代码如下:- #包导入
- import xlrd
- #查找对应文件内容
- file_path="文件路径"
- #打开表格
- wb = xlrd.open_workbook(file_path)
- ws = wb.sheet_by_name('Sheet1')
- #按行读取,返回给前端一个行列表:
- n_rows = ws.nrows#获取行数
- for i in range(2,n_rows):#按行读取,进行筛选,第一行是表头,第二行开始是数据
- get_value=ws.cell(i,3).value#获取第i行第3列的数据
复制代码 1.3. 查找数据修改位置,替换数据
本项目的需求是修改文件中的对应参数,涉及很多代码行的参数修改,因此,需要根据变量名查找相关位置。
- # 其中file_content是文件内容,变量名是需要查找的变量名
- ## 1. 调度类型
- start_index_1 = file_content.find(
- "simParameters.SchedulingType ="
- )
- end_index_1 = file_content.find(
- "simParameters.NumUEs =",
- start_index_1,
- ) # 这之间修改schedulingtype的取值
- ## 2. UESpeed
- start_index_2 = file_content.find(
- "simParameters.UESpeed =", end_index_1
- )
- end_index_2 = file_content.find(
- "% Validate the UE positions",
- start_index_2,
- )
- ## 3. max_RB
- start_index_3 = file_content.find(
- "simParameters.NumRBs =", end_index_2
- )
- end_index_3 = file_content.find(
- "simParameters.SCS =",
- start_index_3,
- )
- ## 4. SCS
- start_index_4 = file_content.find(
- "simParameters.SCS =", end_index_3
- )
- end_index_4 = file_content.find(
- "simParameters.DLCarrierFreq =",
- start_index_4,
- )
复制代码 这部分的下标定位情况,可参考我的此篇博客:【python技巧】替换文件中的某几行
- 替换
在本项目中使用的全文替换,具体代码结构如下:
- # 1. 读取文件
- path = "文件路径"
- with open(path, "r",encoding='utf-8') as file:
- file_content = file.read()
- # 2. 定位
- start_index_1 = file_content.find("simParameters.UEPosition =")
- end_index_1 = file_content.find("simParameters.UESpeed =",start_index_1) # 这之间修改ue_position的取值
- start_index_2 = file_content.find("simParameters.Position = ", end_index_1)
- end_index_2 = file_content.find("csirsConfig = nrCSIRSConfig", start_index_2)
- if (start_index_1 == -1 or end_index_1 == -1 or start_index_2 == -1 or end_index_1 == -1):
- return jsonify({"Error": "找不到对应的参数位置"})
- # 3.更新参数值
- updated_content = (
- file_content[:start_index_1]
- + "simParameters.UEPosition = "
- + str(UE_position)
- + ";\n")
- updated_content += file_content[end_index_1:start_index_2]
- updated_content += "simParameters.Position = "+str(gNB_position)+";% Position of gNB in (x,y,z) coordinates"
- updated_content += file_content[end_index_2:]
- # 4. 更新文件
- if updated_content != "":
- with open(path, "w",encoding="utf-8") as file:
- file.write(updated_content)
- msg = "成功改变相关文件参数\n"
- return jsonify({"Sueecess": msg})
复制代码 2. 功能2:读取结果数据
2.1. 实时数据展示如何存储相关数据?
本项目中matlab会使用作图程序实时展示每个时隙的运行结果,但是这个作图程序无法显示在网页端,因此,考虑将数据存储在文件中,然后通过网页端读取文件中的数据,进行展示。
实时数据的存储代码需要手动添加,在matlab每次作图调用的函数中添加如下matlab代码:- date_str=datestr(now,31);
- new_str = replace(date_str,' ','_');
- new_str=replace(new_str,':','');#添加时间戳
- filename=sprintf('./文件夹名/file_name_%s.mat',new_str);#定义文件存储相对位置
- save(filename,"需要存储的变量名");#存储变量
复制代码 2.2. 读取相关数据,整理、打包、传递
2.2.1. 读取.mat文件
根据添加的代码,将会得到若干个.mat文件,需要读取.mat文件的内容并整理成前端需要的格式。
在本文中,是定义一个文件夹名为A,每个时隙的实时数据存储为一个.mat文件,文件名为file_name_时间戳.mat,因此需要读取文件夹A下的所有.mat文件。
在python中读取.mat文件的具体代码如下:- #1.包导入
- import os
- import scipy.io as sio
- #2. 读取数据
- file_list=os.listdir("./文件夹名/")#读取文件夹下所有文件名称,形成列表
- list_1=[]
- list_2=[]
- for file in file_list:#遍历文件列表
- file_content= sio.loadmat(f"./文件夹/{file}")#读取文件内容
- #这里的写法是根据我的.mat文件结构来的,如果不一样,需要根据自己的文件结构进行修改
- list_1.append(file_content["之前存储的变量名"][0].tolist())
- list_2.append(file_content["之前存储的变量名"][1].tolist())
- #之后得到的list_1和list_2就是前端需要的数据
- #3. 传递给前端
- result={
- "list_1或者其他变量名":list_1,
- "list_2或者其他变量名":list_2,
- }
- return jsonify(result)#数据打包为json格式,传递给前端
复制代码 2.2.2. 读取.xlsx文件
在项目中,还需要读取.xlsx文件,这里使用的是xlrd模块,具体代码如下:- ##包导入
- import xlrd
- # 1. 读取文件
- file_path="文件路径"
- #打开表格
- wb = xlrd.open_workbook(file_path)
- ws = wb.sheet_by_name('Sheet1')
- #数据读取
- list_1=[]
- list_2=[]
- n_rows = ws.nrows#获取行数
- for i in range(2,n_rows):#按行读取,进行筛选
- list_1.append(float(ws.cell(i,7).value))
- list_2.append(float(ws.cell(i,7).value))
- #之后得到的list_1和list_2就是前端需要的数据
- #3. 传递给前端
- result={
- "list_1或者其他变量名":list_1,
- "list_2或者其他变量名":list_2,
- }
- return jsonify(result)#数据打包为json格式,传递给前端
复制代码 2.2.3. 读取.txt/.log文件
在项目中,还需要读取某些文本文件,例如日志文件存储的相关数据,这里使用的是file模块,具体数据读取需要用到正则表达式相关知识,可参考博客【python技巧】文本处理-re库字符匹配,具体代码如下:- ##方式1:要查找的内容为:“serveraddr = xxx",需要提取xxx
- path= "文件路径"
- with open(path, "r") as file:
- for line in file:
- if "serveraddr" in line:
- serveraddr_match = re.search(r'serveraddr\s*=\s*"([^"]+)"', line)
- serveraddr = serveraddr_match.group(1) if serveraddr_match else ""
- elif "serverport" in line:
- serverport_match = re.search(r'serverport\s*=\s*"(\d+)"', line)
- serverport = (
- int(serverport_match.group(1)) if serverport_match else 0
- )
- ##方式2:要查找的内容为:“itemxx: 数据1;数据2;数据3;”,需要提取xxx
- path= "文件路径"
- with open(path, "r") as file:
- lines = file.readlines() # 一次读取并存入lines中,行列表
- for line in lines: # 查找lines中包含item的行
- if item in line: # 待查找的条目,提取关键数据
- line = line.strip("\n") # 去掉换行符
- dataInOneLine = line.split(";") # 分割数据
- dataInOneLine[0] = dataInOneLine[0].split(":")[1] # 去掉item部分
- for i in range(0, len(dataInOneLine)): # 去掉空格
- dataInOneLine[i] = dataInOneLine[i].strip()
- break # 找到目标行,跳出循环
- # 行遍历完成,得到行遍历结果dataInOneLine列表,存储关键数据
- line_content_list = {
- "itemName": item,
- "数据项1": dataInOneLine[0],
- "数据项2": dataInOneLine[1],
- "数据项3": dataInOneLine[2],
- }
- # 将字典存入result_list,result_list中存储多条目的关键数据
- result_list.append(line_content_list)
复制代码 3. 功能3:运行liunx命令行
这是本项目较难的一个功能点,需要使用python程序连接服务器,然后执行liunx命令,运行编译指令和matlab代码,并且需要将liunx命令的输出结果实时返回给前端。
3.1. 远程连接服务器
远程服务器连接使用的是paramiko模块,先使用pip install paramiko下载模块,具体代码如下:- import paramiko
- #创建ssh连接,可以复用的公共函数
- def create_ssh_client(ip, port, username, password):
- ssh_client = paramiko.SSHClient()
- ssh_client.load_system_host_keys()
- ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
- ssh_client.connect(ip, port, username, password)#连接服务器
- return ssh_client
- #调用方式:
- gnb_ssh_client = create_ssh_client(gnb_ip, gnb_port, gnb_username, gnb_password)
- ue_ssh_client = create_ssh_client(ue_ip, ue_port, ue_username, ue_password)
复制代码 【注意事项】:为了避免之后命令行执行可能存在的权限问题和密码输入问题,推荐使用root身份进行ssh连接。
3.2. 执行liunx命令
一般来说,使用paramiko的invoke_shell函数,然后使用send函数发送命令,使用recv函数接收命令执行结果,具体代码如下:- import paramiko
- def execute_command(ssh_client, command, output_lines):
- channel = ssh_client.invoke_shell()
- channel.send(command + '\n')
- while not channel.exit_status_ready():
- time.sleep(1)
复制代码 而关于command的具体写法,这里我简单介绍一下:
- cmd1="cd /home"
- cmd2="ls"
- cmd3="其他指令"
- cmd=cmd1+";"+cmd2+";"+cmd3+"\n"#这样拼接之后,一次性发送给服务器就能按顺序执行多条命令
复制代码
- 特殊命令——matlab脚本文件运行
pyhton文件通过远程服务器连接执行matlab脚本文件,这里使用的是matlab -r命令,具体命令如下:
- cmd="cd 脚本文件对应文件夹"+";"+"matlab -nodesktop -nosplash -r 脚本文件名,不需要带.m后缀"
复制代码
- 特殊命令——服务器文件复制
在服务器A的命令端,需要复制得到服务器B的相关文件,这里使用的是scp命令,具体命令如下:
- cmd="sshpass -p 服务器B的密码 scp -P 服务器B的端口 root@服务器B的IP:服务器内的文件路径(例如 /home/user/copy_file.txt) 想要复制在服务器A中的文件位置,例如`/home/user/`"
- #sshpass是为了避免scp命令需要输入密码,这里直接将密码写在命令中
复制代码
- 特殊命令——文件权限修改
在服务器A的命令端修改相关文件权限,从而能够被读写,这里使用的是chmod命令,具体命令如下:
3.3. 多线程执行
为了实现下文的实时读取输出和随时终止命令,都需要使用多线程方式,从而让终端在执行命令行的时候,还能够接收python发送的新请求,返回终端信息或者执行终止操作。
实现多线程需要用到threading模块,具体代码如下:- #包导入
- import threading
- #全局变量定义
- gnb_ssh_client = None
- ue_ssh_client = None
- gnb_output_lines = []
- ue_output_lines = []
- execution_in_progress = False # 用于标识执行是否正在进行中
- @model_name.route("/start_process", methods=["POST"])
- #需要异步多线程的处理方式
- def start_process():
- global gnb_ssh_client, ue_ssh_client, execution_in_progress,gnb_output_lines,ue_output_lines
- #前端传递参数
- #cmd定义
- #cmd拼接
- gnb_command="("+cmd1_gnb+";"+cmd2_gnb+";)"
- ue_command="(+"scp2ue_cmd_2+";"+cmd0_ue+";"+cmd1_ue+";"+cmd2_ue+";)"
- #连接ssh
- if gnb_ssh_client is None:
- gnb_ssh_client = create_ssh_client(gnb_ip, gnb_port, gnb_username, gnb_password)
- if ue_ssh_client is None:
- ue_ssh_client = create_ssh_client(ue_ip, ue_port, ue_username, ue_password)
- #执行命令
- gnb_output_lines=[]#先清空命令行输出
- ue_output_lines=[]
- gnb_thread = threading.Thread(target=execute_command, args=(gnb_ssh_client, gnb_command, gnb_output_lines, "gnb"))
- ue_thread = threading.Thread(target=execute_command, args=(ue_ssh_client, ue_command, ue_output_lines, "ue"))
- execution_in_progress = True
- gnb_thread.start()
- ue_thread.start()
- return jsonify({"success": "Execution started.","execution_in_progress":execution_in_progress})
- #ssh连接的公用函数
- def create_ssh_client(ip, port, username, password):
- ssh_client = paramiko.SSHClient()
- ssh_client.load_system_host_keys()
- ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
- ssh_client.connect(ip, port, username, password)
- return ssh_client
- #命令执行的公用函数:注意和前文的代码区别,添加了多线程相关代码
- def execute_command(ssh_client, command, output_lines, identifier):
- channel = ssh_client.invoke_shell()
- output_thread = threading.Thread(target=get_output, args=(channel, output_lines, identifier))
- output_thread.start()
- channel.send(command + '\n')
- while not channel.exit_status_ready():
- time.sleep(1)
- # 等待输出线程完成
- output_thread.join()
复制代码 3.4. 实时读取liunx命令的输出数据
按照多线程的方式执行代码之后,如何才能实时获取命令行的输出数据呢?
首先我们介绍一下整个命令行执行过程中,前后端所进行的操作:
- step1:
- 前端:发送命令行执行请求;
- 后端:调用命令行执行的接口(例如/start_process)
- 服务器:建立ssh连接,执行命令行;
- step2:
- 前端:按照一定的时间间隔,发送数据请求;
- 后端:调用数据请求的接口(例如/get_output)
- 服务器:继续执行命令,命令行输出数据被存储在全局变量中;
- step3:
- 前端:发送终止命令行执行请求;
- 后端:调用终止命令行执行的接口(例如/stop_process)
- 服务器:终止命令行执行,关闭ssh连接。
因此,此处的实时读取命令行输出数据,就是在step2中,前端按照一定的时间间隔,发送数据请求,后端调用数据请求的接口,返回命令行输出数据。具体代码如下:- @model_name.route('/get_output', methods=['POST', 'GET'])
- def get_output():#读取全局变量内容即可
- global gnb_output_lines, ue_output_lines
- gnb_output = "\n".join(gnb_output_lines)
- ue_output = "\n".join(ue_output_lines)
- return jsonify({
- "gnb_output": gnb_output,
- "ue_output": ue_output
- })
- def execute_command(ssh_client, command, output_lines, identifier):
- channel = ssh_client.invoke_shell()
- output_thread = threading.Thread(target=get_output, args=(channel, output_lines, identifier))#这一步是关键,将命令行输出数据存储在全局变量中
- output_thread.start()
- channel.send(command + '\n')
- while not channel.exit_status_ready():
- time.sleep(1)
- # 等待输出线程完成
- output_thread.join()
- def get_output(channel, output_lines, identifier):#核心代码,将命令行执行的输出数据存储在全局变量中,需要和上文的`execute_command`函数配合使用
- while not channel.exit_status_ready():
- if channel.recv_ready():
- output = channel.recv(1024).decode('utf-8')
- lines = output.split('\n')
- for line in lines:
- if line.strip():
- formatted_line = f"[{identifier}]:{line.strip()}"
- #print(formatted_line)
- output_lines.append(formatted_line)
复制代码 3.5. 随时终止liunx命令的执行
这个功能是为了避免指令执行无法自行终止,需要用户手动选择结束仿真。
在多线程的命令行执行中,前端发送请求,而后端在原有ssh连接的基础上,发送终止命令,具体代码如下:- @model_name.route('/stop_process', methods=['POST'])
- def stop_process():
- global gnb_ssh_client, ue_ssh_client, execution_in_progress
- if not execution_in_progress:#如果没有命令行执行,返回错误信息
- return jsonify({"error": "No execution in progress.","execution_in_progress":execution_in_progress})
- if ue_ssh_client is not None:#断开ued的ssh连接
- stop_execution(ue_ssh_client)
- ue_ssh_client.close()
- ue_ssh_client = None
- if gnb_ssh_client is not None:#断开gnb的ssh连接
- stop_execution(gnb_ssh_client)
- gnb_ssh_client.close()
- gnb_ssh_client = None
- execution_in_progress = False
- return jsonify({"success": "Execution stopped.","execution_in_progress":execution_in_progress})
- #终止指令执行的公用函数
- def stop_execution(ssh_client):
- ssh_client.invoke_shell().send('\x03') # 发送Ctrl+C来终止命令
复制代码 4. 其他收获
4.1. 异常处理
在后端程序执行过程中,可能会出现各种checked exeption,这类异常需要程序员进行捕获,不然则会影响程序的运行,产生报错,而这里我们可以将这些异常捕获之后,将信息作为返回值传递给前端,从而让用户知道程序运行成功与否。
具体代码结构如下:- @model_name.route('/function_name', methods=['POST'])
- def function_name():
- try:
- #程序运行代码
- except Exception as e:
- return jsonify({"error": str(e)})
- finally:
- #程序运行结束后的代码,例如return jsonify({"success": "Execution stopped."})
复制代码 4.2 日志生成
在上一篇博客中,我们提到了分模块的flask项目结构,而日志部分的处理,需要在主文件app.py中。添加如下代码:- import logging
- log_filename = 'app.log' # 日志文件名
- log_level = logging.DEBUG # 日志级别
- log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' # 日志格式
- # 配置日志
- logging.basicConfig(filename=os.path.join(os.path.dirname(__file__), log_filename), level=log_level, format=log_format)
复制代码 则可以在程序运行过程中,将日志信息存储在app.log文件中,相关报错信息也会存储在该文件中,方便程序员进行调试。
4.3 环境部署
当我们在本地进行程序开发之后,需要将程序部署到服务器上,而迁移到服务器上或者别人的电脑上,需要安装相关的环境。在这种情况下,我们可以将自己开发用到的包进行打包,然后在服务器上进行安装,具体步骤如下:
- 将当前环境中的包信息存储在requirements.txt文件中,存储在当前文件夹目录下
pip freeze > requirements.txt
- 将整个项目文件打包之后,其他人可使用如下命令进行安装:
pip install -r requirements.txt
4.4. vscode远程连接服务器
当我们在服务器上部署完项目代码之后,可能会经过多次测试和调试,因此代码会不断被修改,且会被无数次的运行,如果每次都是本地修改+服务器部署运行,实在有些不够优雅,因此,我们可以试着把自己的本地编辑器连接到服务器上,这样就可以在本地进行服务器端代码的修改,然后直接在服务器上运行,这样就可以避免每次修改都需要部署的麻烦。
我在项目中使用的是vscode,具体步骤如下:
- 插件安装:Remote-SSH
- 选择新建远程,输入ssh连接信息
- 连接之后选择想要编辑的项目文件夹,即可在本地编辑器中进行服务器端代码的修改;
- 调试的时候可使用vscode进行运行、部署;
具体的配置细节,可参考博客
5. 项目感受
- 项目的文件读写、服务器连接等都是非常有趣的功能,多线程解决实时输出算是一个较大的技术难点;
- 项目本身难度不大,主要难点在于理解业务本身,如果有相对应的业务文档,可能开发效率会更高一点;
- 后端开发部分也需要了解基本的前端知识,例如前后端数据传递、前端页面的渲染等,这样才能更好的理解前端的需求,从而更好的进行后端开发;
- 项目对接需要良好的文档编辑能力和沟通能力;
- 相关业务的具体实现,也是体现个人智慧的时刻,具体的开发技术和解决问题的智慧都是必不可少的,开发技术是工具,而我们需要用自己已有的工具去实现某些功能,这是一种思维方式,也是一种能力。,当然,技术会的越多,能用的工具越多,解决问题的思路就不会被限制;
- 总会有开发者不会的东西,这是一个学习的过程,重点在于不断地学习,虚心请教和快速掌握。
6. 后记
如果觉得我写得还算不错,不妨点赞关注一波走起~
想看更多博文,请访问我的各平台主页:博客园 / CSDN / 51CTO / 掘金论坛 / 知乎
来源:https://www.cnblogs.com/CrazyPixel/p/17796888.html
免责声明:由于采集信息均来自互联网,如果侵犯了您的权益,请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容,谢谢合作! |
本帖子中包含更多资源
您需要 登录 才可以下载或查看,没有账号?立即注册
x
|