对Qpython+Flask实现对小孩学习的监控-CSDN博客中html页面进行改造，利用Ajax，提交一段文字，发送到数据库，再在服务器，发送该段文件给手机端，然手机端TTS朗读出来，增加了父母监控小孩学习，自定义提醒小孩的功能。
一、index.html的更改。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>实时**学习情况图</title>
<script type="text/javascript">
// 设置定时器，每20秒（20000毫秒）刷新一次页面
setInterval(function(){
location.reload();
}, 20000);
</script>
<script>
function submitForm() {
// 获取文本输入框的值
var text = document.getElementById('textInput').value;
// 创建一个XMLHttpRequest对象
var xhr = new XMLHttpRequest();
// 配置请求：POST方法，请求的URL，以及是否异步
xhr.open("POST", "/submit", true);
// 设置请求头，指定内容类型为表单数据
xhr.setRequestHeader("Content-Type", "application/x-www-form-urlencoded");
// 定义当请求完成并成功时的回调函数
xhr.onreadystatechange = function () {
if (xhr.readyState === XMLHttpRequest.DONE && xhr.status === 200) {
// 解析并处理服务器的响应
var response = xhr.responseText;
document.getElementById('response').innerText = response;
}
};
// 编码表单数据并发送
var formData = "textInput=" + encodeURIComponent(text);
xhr.send(formData);
// 阻止表单的默认提交行为
return false;
}
</script>
</head>
<body>
<h1>学习情况，每20秒刷新一次页面</h1>
<div>
<img src="{
{ image_url }}" alt="Latest Image">
</div>
<h1>状态：{
{ qk }} </h1>
<form onsubmit="return submitForm();">
<label for="textInput">输入文字发给手机说:</label><br><br>
<textarea id="textInput" name="textInput" rows="1" required></textarea><br><br>
<input type="submit" value="提交">
</form>
<p id="response"></p>
<p>
<a href="http://192.168.99.235:5000/qk" target="_self">查看实时学习情况</a>
</p>
</body>
</html>
二、flask服务器端
# -*- coding: utf-8 -*-
"""
Created on Wed Jan 15 22:39:02 2025
@author: Ybk
"""
from flask import Flask, jsonify, send_from_directory, render_template, request
import time
import os
from pathlib import Path
import sqlite3
import datetime
import torch
import torch.nn as nn
from torchvision import transforms, models
from PIL import Image
import broadlink
import plotly.graph_objects as go
# 加载模型
# 加载模型并设置为评估模式
model = models.resnet18(pretrained=False) # 假设使用的是resnet18
num_ftrs = model.fc.in_features # 获取输入特征数量
model.fc = nn.Linear(num_ftrs, 6) # 将输出特征数量改为4
model.load_state_dict(torch.load('best66_resnet18_model.pth')) # 加载模型参数
model.to('cuda' if torch.cuda.is_available() else 'cpu') # 移动到设备上
model.eval() # 设置为评估模式
# 定义预处理步骤
preprocess = transforms.Compose([
transforms.Resize(224),
transforms.CenterCrop(224),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])
app = Flask(__name__)
db_filepath = Path(__file__).joinpath("../jk.db").resolve()
# 设置上传文件的目录和允许的文件扩展名
now = datetime.datetime.now()
date_str = now.strftime("%Y%m%d")
UPLOAD_FOLDER = fr'd:\upload\{date_str}'
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
# 创建上传目录如果不存在
if not os.path.exists(UPLOAD_FOLDER):
os.makedirs(UPLOAD_FOLDER)
def insertdb(sj,furl,xl,zt,qk): #插入一行数据zt1为有人0为无人
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
insert_query = "INSERT INTO jkme(sj,furl,xl,zt,qk) VALUES(?,?,?,?,?);"
insert_data = (sj,furl,xl,zt,qk)
c.execute(insert_query,insert_data)
conn.commit()
c.close()
conn.close
def insertmsg(msg): #插入一行数据zt1为有人0为无人
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
insert_query = "INSERT INTO jkmsg(sj,msg,zt) VALUES(?,?,0);"
now = datetime.datetime.now()
sj = now.strftime("%Y-%m-%d %H:%M:%S")
insert_data = (sj,msg)
c.execute(insert_query,insert_data)
conn.commit()
c.close()
conn.close
# 检查文件是否是允许的类型
def allowed_file(filename):
# 检查文件名是否包含点，并且文件扩展名（转换为小写后）是否在允许的扩展名集合中
if '.' in filename:
# 获取文件扩展名，并转换为小写
file_extension = filename.rsplit('.', 1)[-1].lower()
# 检查扩展名是否在允许的集合中
return file_extension in ALLOWED_EXTENSIONS
else:
# 如果文件名不包含点，则扩展名不存在或不合法
return False
def sc_xxqkpng():
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
#lastxxid,bsj,esj
cursor = c.execute("SELECT id,sj FROM jkme WHERE qk = 4 and sj > datetime('now', 'localtime', '-120 minute') order by id desc")
row = cursor.fetchone()
if row:
lastxxid = row[0]
lastxxsj = row[1]
cursor = c.execute(f"SELECT count(id) FROM jkme WHERE id > {lastxxid}")
row = cursor.fetchone()
if row[0] > 50:
fig = go.Figure()
# 定义颜色列表
colors = ['cyan','blue','yellow', 'green']
cursor = c.execute(f"SELECT qk FROM jkme WHERE id > {lastxxid}")
rows = cursor.fetchall()
list0 = [row[0] for row in rows]
# 初始化一个空列表来存储结果
result = []
# 初始化变量来跟踪当前数字及其起始位置
current_number = None
start_index = None
alltime=0
rztime=0
lktime=0
# 遍历数字序列
for i, number in enumerate(list0):
# 如果当前数字与之前的数字不同（或者是第一个数字）
if number == 0:
lktime = lktime + 1
if number in (1,2):
alltime = alltime + 1
if number == 1:
rztime = rztime + 1
if current_number is None or number != current_number:
# 如果之前已经有过数字，则记录其范围
if current_number is not None:
result.append([current_number, start_index, i - 1])
# 更新当前数字和起始位置
current_number = number
start_index = i
# 处理序列中的最后一个数字范围
if current_number is not None:
result.append([current_number, start_index, len(list0) - 1])
all_data_tj = len(list0)
for xlist in result:
# print(xlist)
# 添加矩形
# 添加一个矩形形状，没有边框
fig.add_shape(
type="rect",
x0=1+xlist[1]*2, # 矩形左下角的x坐标
y0=1, # 矩形左下角的y坐标
x1=1+(xlist[2]+1)*2, # 矩形右上角的x坐标
y1=6, # 矩形右上角的y坐标
fillcolor=colors[xlist[0]], # 矩形的填充颜色
line=dict(width=0) # 设置边框宽度为0，即没有边框
)
i = i + 1
xxtime = int(alltime / 6)
rztime = round(rztime / alltime * 100,2)
lktime = round(lktime / all_data_tj * 100,2)
all_time = int(all_data_tj / 6)
# 设置图的标题和坐标轴的限制
cursor = c.execute(f"SELECT sj FROM jkme WHERE id > {lastxxid} order by id asc")
row = cursor.fetchone()
bsj = row[0]
cursor = c.execute("SELECT sj FROM jkme order by id desc")
row = cursor.fetchone()
esj = row[0]
sjstr = bsj.split()[0] + ' ' + bsj.split()[1] + '-' + esj.split()[1]
# 设置布局以隐藏背景和坐标轴
# 更新图形的布局，隐藏坐标轴和背景
fig.update_layout(
xaxis=dict(
range=[0,(all_data_tj+1)*2],
showgrid=False, # 隐藏网格线
zeroline=False, # 隐藏零轴线
visible=False # 隐藏x轴
),
yaxis=dict(
range=[0,5],
showgrid=False, # 隐藏网格线
zeroline=False, # 隐藏零轴线
visible=False # 隐藏y轴
),
plot_bgcolor='white', # 设置背景颜色为透明
paper_bgcolor='white', # 设置整个图形的背景颜色为透明
title=dict(
text=f'{sjstr} 学习情况图', # 标题文本
font=dict(
size=16, # 字体大小
color='black' # 字体颜色
),
x=0.5, # 标题水平居中
xanchor='center', # 相对于x位置的锚点设置为中心
y=0.8, # 标题的垂直位置（可选，根据需要调整）
yanchor='top' # 相对于y位置的锚点设置为顶部
),
annotations=[
dict(
text=f'总{all_time}分钟，学习{xxtime}分钟，认真学习时间占{rztime}%，离开学习占总时间{lktime}%', # 注释的文本内容
x=0.5, # 注释的水平位置（相对于图表宽度的比例）
y=-0.2, # 注释的垂直位置（相对于图表高度的比例，负值表示在图表下方）
xref='paper', # 参照物为整个图表区域（'paper'）而非数据坐标（'x'）
yref='paper', # 参照物为整个图表区域（'paper'）而非数据坐标（'y'）
showarrow=False, # 不显示箭头
font=dict(
size=14, # 字体大小
color='black' # 字体颜色
),
align='center' # 文本对齐方式
)
],
width=800, # 设置图表宽度（可选）
height=240 # 设置图表高度（可选）
)
pngfilename = esj.replace('-','').replace(' ','').replace(':','')
# 显示图表
# fig.show()
fig.write_image(rf'D:\sb\xx\{pngfilename}.png')
insert_query = "INSERT INTO qkpng(url) VALUES(?);"
insert_data = (rf'D:\sb\xx\{pngfilename}.png',)
c.execute(insert_query,insert_data)
conn.commit()
print(rf'D:\sb\xx\{pngfilename}.png')
c.close()
conn.close()
return f'学习{xxtime}分钟，认真学习时间占{rztime}%'
else:
c.close()
conn.close()
return '学习情况不能统计！'
else:
c.close()
conn.close()
return '学习情况不能统计！'
def getprexl(): #获取最后一个xl
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
cursor = c.execute("select xl from jkme order by id desc limit 0,1;")
row = cursor.fetchone()
if row:
xl = row[0]
else:
xl = 0
c.close()
conn.close
return xl
def turntolight(turn):#turn=true为开灯，turn=false为关灯
try:
device = broadlink.sp4(host=('192.168.99.50',80), mac=bytearray.fromhex('E81656CA934B'), devtype=32021)
device.auth()
state = device.check_power()
if turn:
if state == True:
print(" ON pass")
elif state == False:
device.set_power(True)
print(" ON")
else:
if state == False:
print(" OFF pass")
elif state == True:
device.set_power(False)
print(" OFF")
except Exception as e:
print(e)
# turntolight(False)
@app.route('/upload', methods=['POST'])
def upload_file():
now = datetime.datetime.now()
# 检查请求中是否包含'description'参数
description = request.form.get('description')
if not description:
return jsonify({'error': 'No description provided'}), 400
# 检查是否有文件在请求中
if 'file' not in request.files:
return jsonify({'error': 'No file part in the request'}), 400
file = request.files['file']
# 如果用户没有选择文件，返回错误
if file.filename == '':
return jsonify({'error': 'No selected file'}), 400
# 如果文件合法，尝试保存文件
if file and allowed_file(file.filename):
filename = file.filename
filepath = os.path.join(app.config['UPLOAD_FOLDER'], description.replace(' ','').replace('-','').replace(':','') + '.' + filename.rsplit('.', 1)[-1])
file.save(filepath)
res = 'ok'
xl = 0
tj = 0
tjx = 0
qkz = 0
msg = ''
if filepath:
img = Image.open(filepath) # 加载图片
img_tensor = preprocess(img) # 预处理图片
img_tensor = img_tensor.unsqueeze(0) # 增加batch维度
img_tensor = img_tensor.to('cuda' if torch.cuda.is_available() else 'cpu') # 移动到设备上
# 进行推理
with torch.no_grad():
outputs = model(img_tensor)
_, predicted = torch.max(outputs, 1)
# 解释结果
result = int(predicted.item())
if result == 0:
res = '**没在学习'
elif result == 1:
res = 'ok'
elif result == 2:
res = '**要认真学习'
elif result == 3:
res = '是**'
elif result == 4:
res = '是爸爸'
elif result == 5:
res = '是其他人'
if result > 2:
result = 3
#正常情况
xl = 0
zt = 0
#如果26分钟内1和2，已经超过138个，那么设置为2，2
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
cursor = c.execute("SELECT count(*) FROM jkme WHERE zt = 0 and (qk > 0 and qk < 4) and sj > datetime('now', 'localtime', '-26 minute')")
row = cursor.fetchone()
tj = row[0]
print(f'学习统计={tj}个10秒')
if tj > 137 or getprexl() == 2:
xl = 2
zt = 2
result = 4
if tj == 138:
res = sc_xxqkpng()
turntolight(True)
#如果6分钟内2，2已经超过30个，那么设置为0，0
cursor = c.execute("SELECT count(*) FROM jkme WHERE xl = 2 and zt = 2 and sj > datetime('now', 'localtime', '-6 minute')")
row = cursor.fetchone()
tjx = row[0]
print(f'休息统计={tjx}个10秒')
if tjx > 30:
xl = 0
zt = 0
res = '休息时间完了。'
if tjx == 31:
turntolight(False)
#如果上一个为2，这一个也是2，那么设置为1，0
cursor = c.execute("SELECT qk FROM jkme order by id desc limit 0,1;")
row = cursor.fetchone()
qkz = row[0]
print(f'上一个={qkz}')
if qkz == 2:
xl = 1 #1为提醒响铃
zt = 0
if result == 0 and zt != 2:
xl = 3 #3为警报响铃
cursor = c.execute("SELECT id,msg FROM jkmsg WHERE zt = 0 limit 0,1;")
row = cursor.fetchone()
if row:
msg = row[1]
mid = row[0]
c.execute(f"UPDATE jkmsg SET zt = 1 WHERE id = {mid}")
conn.commit()
c.close()
conn.close
insertdb(now.strftime("%Y-%m-%d %H:%M:%S"),filepath,xl,zt,result)
# 返回包含接收到的描述和文件路径的响应
if result == 4:
res = 'ok'
if int(now.strftime("%H")) in (12,13,14):
res = 'ok'
if msg != '':
res = msg
return jsonify({
'message': f'学习={tj}，' + f'休息={tjx}，' + f'上个={qkz}。' + f'{msg}',
'speak': res,
'sj': now.strftime("%Y-%m-%d %H:%M:%S"),
'xl': xl
}), 201
else:
return jsonify({'error': 'Allowed file types are png, jpg, jpeg, gif'}), 400
@app.route('/')
def index():
# 获取最新的图片
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
cursor = c.execute("select furl,qk from jkme order by id desc limit 0,1;")
row = cursor.fetchone()
upic = row[0]
result = row[1]
res = 0
if result == 0:
res = '没在学习'
elif result == 1:
res = '认真学习'
elif result == 2:
res = '没有认真学习'
elif result == 3:
res = '是其他人'
elif result == 4:
res = '休息时间'
c.close
conn.close
furl = os.path.basename(upic)
image_url = f'http://192.168.99.235:5000/static/upimages/{furl}'
return render_template('index.html', image_url=image_url, qk = res)
@app.route('/qk')
def index0():
urls = []
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
cursor = c.execute("SELECT url FROM qkpng order by id desc limit 0,5;")
rows = cursor.fetchall()
for row in rows:
urls.append(os.path.basename(row[0]))
c.close()
conn.close
return render_template('index0.html', image_urls=urls)
@app.route('/static/images/<path:filename>')
def custom_static(filename):
return send_from_directory(r'D:\sb\xx', filename)
@app.route('/static/upimages/<path:filename>')
def custom_static_gz(filename):
now = datetime.datetime.now()
date_str = now.strftime("%Y%m%d")
return send_from_directory(fr'D:\upload\{date_str}', filename)
@app.route('/submit', methods=['POST'])
def handle_form_submit():
# 从POST请求中获取名为'textInput'的参数
text_input = request.form.get('textInput')
insertmsg(text_input)
# 处理接收到的数据（这里只是简单地返回它）
response = {
'status': 'success',
'receivedText': text_input
}
# 返回JSON格式的响应
return jsonify(response)
@app.route('/get_data', methods=['GET'])
def get_data():
# 在这里，你可以根据需要动态生成数据
# 例如，从数据库查询、调用其他API等
conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)
c = conn.cursor()
cursor = c.execute("SELECT sj,furl,xl,zt,qk FROM jkme order by id desc limit 0,1")
row = cursor.fetchone()
sj = row[0]
url = row[1]
xl = row[2]
zt = row[3]
getqk = row[4]
c.close
conn.close
if getqk == 0:
qk = '无人'
elif getqk == 1:
qk = '认真'
elif getqk == 2:
qk = '不认真'
else:
qk = '其他情况'
furl = os.path.basename(url)
data = {
'sj':sj,
'image_url': f'http://192.168.99.235:5000/static/upimages/{furl}',
'str_value': qk,
'xl': xl,
'zt': zt,
'qk': getqk
}
# 返回JSON响应
return jsonify(data)
if __name__ == '__main__':
app.run(host='192.168.99.235')
三、数据库：
CREATE TABLE jkmsg (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sj TEXT,
msg TEXT,
zt INTEGER
);运行效果：
