wulili/hms_cattle_research
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: wulili:master_a
Branches Tags
wulili:master
wulili:master_a
..
compare: wulili:master
Branches Tags
wulili:master_a
wulili:master

No commits in common. "master_a" and "master" have entirely different histories.
master_a ... master

9 changed files with 94 additions and 1276 deletions
Show Stats Expand all files Collapse all files

6
.idea/sqldialects.xml
View File

@ -1,6 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="SqlDialectMappings">
<file url="file://$PROJECT_DIR$/models.py" dialect="GenericSQL" />
</component>
</project>

49
config.py
View File

@ -1,49 +0,0 @@
# Author: Lee Wu Love Lele
# Datetime: 2024/12/6 15:31
# 远程数据库配置
RDB_HOST = "rm-bp1442up0e0nlz95szo.mysql.rds.aliyuncs.com"
RDB_USER = "rhm_read_only"
RDB_PASSWORD = "aA123456+"
RDB_NAME = "rhm_insure_dev"
# 本地数据库配置
LDB_HOST = "127.0.0.1"
LDB_PORT = "3306"
LDB_USER = "root"
LDB_PASSWORD = "1QAZ9ol."
LDB_NAME = "hm_data"
# 数据库日期时间格式
DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S"
# tqdm执行进度条开关
TQDM_DISABLE = False
# 计算牛只体重波动时，变化率小于此值视为稳定
STABILITY_THRESHOLD = 3
# 提取体重日志时的最小时间窗口，体重波小于10s的不提取
MIN_TIME_WINDOW = 10
# 所有牛只购入体重最大值
BUY_WEIGHT_MAX = '609'
# 所有牛只购入体重最小值
BUY_WEIGHT_MIN = '363'
# 体重日志截取最小值，BUY_WEIGHT_AVG 的百分比
MIN_TOTAL_BUY_WEIGHT_PERCENT = '0.7'
MIN_SINGLE_BUY_WEIGHT_PERCENT = '0.7'
# 体重日志截取最大值，BUY_WEIGHT_AVG 的百分比
MAX_TOTAL_BUY_WEIGHT_PERCENT = '1.5'
# 根据特定牛只的购入体重截取最大值比例，在目前的数据中，有一头牛出栏体重比购入体重大1.8倍，已这个为参考值定义2倍。
MAX_SINGLE_BUY_WEIGHT_PERCENT = '2'
TOTAL_BUY_WEIGHT_AVG = '439.24'
# 判断一个波的时间周期内，RFID频率的阈值（大于）
RFID_FREQUENCY_THRESHOLD_VALUE = 0.8

295
controller.py
View File

@ -1,295 +0,0 @@
# Author: Lee Wu Love Lele
# Datetime: 2024/12/6 15:24
from util import calc_one_cattle, smooth_data
from models import WeightWave, DAO, ExecLog, DailyWeight
from decimal import Decimal, getcontext
import copy
from datetime import datetime
from config import DATETIME_FORMAT, TQDM_DISABLE, MIN_TIME_WINDOW, BUY_WEIGHT_MAX, BUY_WEIGHT_MIN, \
MIN_TOTAL_BUY_WEIGHT_PERCENT, \
MAX_TOTAL_BUY_WEIGHT_PERCENT, RFID_FREQUENCY_THRESHOLD_VALUE, MIN_SINGLE_BUY_WEIGHT_PERCENT, \
MAX_SINGLE_BUY_WEIGHT_PERCENT
from tqdm import tqdm
import numpy as np
import pandas as pd
from scipy.signal import savgol_filter
def clear_all_data():
DAO.delete_weight_wave_all(False)
DAO.delete_exec_log_all(True)
def clear_data_by_time_std(time_std):
DAO.delete_weight_wave_by_time_std(time_std)
DAO.delete_exec_log_by_time_std(time_std)
def init_data(time_std):
"""
初始化数据，从远程数据库中截取地磅的连续波，同时记录连续波中的体重值出现过的对应的RFID Reader ID
:return:
"""
search_start_time = '2024-09-02 00:00:00' # TODO: 从开始时间截取的波之前如果有日志，这个波的开始时间并不准确，因为之前可能存在连续值属于这个波。
search_end_time = "2024-12-19 23:59:59" # 查询远程日志的截止时间，当前系统时间
getcontext().prec = 5
buy_weight_max = Decimal(BUY_WEIGHT_MAX) # 609
buy_weight_min = Decimal(BUY_WEIGHT_MIN) # 363
ctrl_min_weight = buy_weight_min * Decimal(MIN_TOTAL_BUY_WEIGHT_PERCENT) # 254.10
ctrl_max_weight = buy_weight_max * Decimal(MAX_TOTAL_BUY_WEIGHT_PERCENT) # 791.70
# init_weight_wave(search_start_time, search_end_time, ctrl_min_weight, ctrl_max_weight, time_std, MIN_TIME_WINDOW)
init_weight_wave(search_start_time, search_end_time, ctrl_min_weight, ctrl_max_weight, time_std, MIN_TIME_WINDOW)
# def init_daily_weight(time_std):
# """
# 初始化牛只每日体重
# :param time_std:
# :return:
# """
# rfid_lst = DAO.get_rfid(time_std) # 分组查询所有体重波中的牛只RFID
# for rfid in rfid_lst:
# weight_wave_lst = DAO.get_weight_wave_by_rfid(rfid) # 根据RFID查询所有体重波
# for weight_wave in weight_wave_lst:
# pass
# pass
# return
def get_daily_weight(rfid, time_std):
"""
根据单个RFID获取按照每日计算的体重
:param rfid:
:param time_std:
:return:
"""
weight_wave_lst = DAO.get_weight_wave_by_rfid(rfid, time_std)
record_date = None
weight_lst = []
daily_weight_lst = []
index = 1
for weight_wave in weight_wave_lst:
if (record_date is not None and record_date != datetime.strptime(weight_wave.start_time,
DATETIME_FORMAT).date()) or index == len(
weight_wave_lst):
if len(weight_lst) > 2:
trimmed_lst = sorted(weight_lst)[1:-1]
daily_weight_lst.append(DailyWeight(
weight=float(Decimal(str(sum(trimmed_lst))) / Decimal(str(len(trimmed_lst)))),
date=datetime.strftime(record_date, DATETIME_FORMAT)
))
weight_lst.clear()
if record_date is None or record_date != datetime.strptime(weight_wave.start_time, DATETIME_FORMAT).date():
record_date = datetime.strptime(weight_wave.start_time, DATETIME_FORMAT).date()
weight_lst.append(weight_wave.weight)
index += 1
if not weight_lst and daily_weight_lst and len(weight_lst) > 2:
trimmed_lst = sorted(weight_lst)[1:-1]
daily_weight_lst.append(DailyWeight(
weight=float(Decimal(str(sum(trimmed_lst))) / Decimal(str(len(trimmed_lst)))),
date=datetime.strftime(record_date, DATETIME_FORMAT)
))
dict_lst = [obj.__dict__ for obj in daily_weight_lst]
sorted_lst = sorted(dict_lst, key=lambda x: datetime.strptime(x['date'], DATETIME_FORMAT))
return sorted_lst
def get_weight_wave(rfid, time_std):
"""
根据RFID获取体重波
:param rfid: str
:param time_std: int 1=设备时间，2=服务器时间
:return:
"""
weight_wave_lst = DAO.get_weight_wave_by_rfid(rfid, time_std)
# dict_lst = [obj.__dict__ for obj in weight_wave_lst]
sorted_lst = sorted(weight_wave_lst, key=lambda x: datetime.strptime(x.start_time, DATETIME_FORMAT))
return sorted_lst
def get_weight_wave_log(weight_wave, time_std):
params = {'floor_scale_id': (weight_wave.floor_scale_id,), 'start_time': weight_wave.start_time,
'end_time': weight_wave.end_time, 'time_std': time_std}
weight_wave_logs = DAO.get_weight_logs(**params)
return weight_wave_logs
def init_weight_wave(search_start_time, search_end_time, ctrl_min_weight, ctrl_max_weight, time_std=1,
min_time_window=10):
"""
初始化体重波
根据地磅列表从所有地磅日志中从开始时间逐秒轮询，截取在指定体重范围内的连续体重
:param search_start_time:
:param search_end_time:
:param ctrl_min_weight:
:param ctrl_max_weight:
:param time_std:
:param min_time_window
:return:
"""
devices = DAO.get_devices_id() # 获取所有地磅ID，从养殖场记录中获取。TODO: 是否时当前设备，如果设备更换了如何处理？
exec_logs = DAO.get_exec_log((time_std,)) # 根据所需分析的时间类型获取执行记录。
for device in tqdm(devices, desc="遍历地磅设备：", position=1, disable=TQDM_DISABLE):
floor_scale_id = device['reader_code'] # 取出地磅id
never_logged = True # 截取进程记录控制
# if floor_scale_id != '2405085412':
# continue
for exec_log_item in exec_logs:
if exec_log_item.get('floor_scale_id') == floor_scale_id:
print('-- 地磅：' + floor_scale_id + ' | ' + (
'设备时间' if time_std == 1 else '服务器时间 :') + ' 记录存在')
# 执行记录中存在地磅并且新查询的结束时间在执行记录的范围内，则不做处理。
if exec_log_item.get('end_time') < datetime.strptime(
search_end_time, DATETIME_FORMAT):
# 执行记录中存在地磅并且新查询的结束时间在执行记录之后，根据执行记录最后一个波的结束时间开始执行
print('---- 查询时间段在记录时间段外，补充时间段：' + exec_log_item.get('end_time').strftime(
DATETIME_FORMAT) + ' 到 ' + search_end_time)
weight_wave_lst = create_weight_wave(floor_scale_id,
exec_log_item.get('end_time').strftime(DATETIME_FORMAT),
search_end_time, ctrl_min_weight, ctrl_max_weight,
time_std, min_time_window)
exec_log = save_weight_wave(floor_scale_id, weight_wave_lst, time_std)
save_exec_log(exec_log)
never_logged = False
break
if never_logged: # 执行记录中没有出现过地磅数据，则根据查询起止时间全量查询
weight_wave_lst = create_weight_wave(floor_scale_id, search_start_time, search_end_time, ctrl_min_weight,
ctrl_max_weight,
time_std, min_time_window)
exec_log = save_weight_wave(floor_scale_id, weight_wave_lst, time_std)
save_exec_log(exec_log)
def create_weight_wave(floor_scale_id, search_start_time, search_end_time, ctrl_min_weight, ctrl_max_weight,
time_std=1, min_time_window=10):
"""
从远程服务器拉取一个地磅的日志数据，分析体重数据。
:param floor_scale_id: 地磅ID
:param search_start_time: 查询起始时间（包含）
:param search_end_time: 查询截止时间（包含）
:param ctrl_min_weight: 体重窗口最小值
:param ctrl_max_weight: 体重窗口最大值
:param time_std: 1 = 设备时间，2 = 服务器时间
:param min_time_window: 最小时间窗口（s）
:return:
"""
start_time = None # 一个波循环计数开始时间
prev_time = None # 上一条数据的时间指针
weight_wave_lst = [] # 存放波的结构
weight_lst = [] # 存放波的体重值
relation_reader_name_set = set() # 去重存放从体重日志中关联的RFID Reader ID
for weight_log in tqdm(DAO.get_weight_logs(
**{'floor_scale_id': (floor_scale_id,), 'start_time': search_start_time, 'end_time': search_end_time,
'time_std': time_std}),
desc="---- 根据地磅ID迭代日志：", position=2, disable=TQDM_DISABLE):
log_weight = weight_log.weight # 获取体重值
log_time = weight_log.time # 获得数据采集时间
log_relation_reader_name = weight_log.rfid_reader
is_out_of_range = ctrl_min_weight > Decimal(log_weight) or Decimal(log_weight) > ctrl_max_weight
is_out_of_interval = (log_time - prev_time).total_seconds() > 1 if prev_time is not None else False
is_in_of_time_window = (
prev_time - start_time).total_seconds() > min_time_window if start_time is not None else False
if start_time is not None and (is_out_of_interval or is_out_of_range) and is_in_of_time_window:
"""
一个波的结束，在符合波长时间超过最小波长时间窗口，两个临近值间隔超过1s或者体重范围超过阈值时触发截断，形成一个波，
但是如果该波段内没有识别到牛只RFID或者识别到的牛只RFID没有明显的频率（有一个RIFD频率超过75%），放弃获取这个波。
"""
rfid_count_lst = DAO.get_rfid_count(tuple(relation_reader_name_set), start_time.strftime(DATETIME_FORMAT),
prev_time.strftime(DATETIME_FORMAT))
if rfid_count_lst:
df = pd.DataFrame(rfid_count_lst)
df["percentage"] = df["repeat_count"] / df["repeat_count"].sum() # 计算所有RFID的频率
df_sorted = df.sort_values(by="percentage", ascending=False) # 按照频率从大到小排列
# df_sorted = df_sorted.reset_index(drop=True) # 重新排序index
# TODO: 这里可以考虑，虽然频率高但是如果一个10秒以上的波只有一个RFID频次出现，是否需要取这个波？在这里记录一下RFID的频次
max_frq = df_sorted.iloc[0]["percentage"] # 取最大频率的RFID对应的频率
max_rfid = df_sorted.iloc[0]["biz_data"] # 取最大频率的RFID
weight_lst_len = len(weight_lst)
# TODO：在这里增加一个判断，需要判断出现一定时常的连续稳定值
if max_frq > RFID_FREQUENCY_THRESHOLD_VALUE:
weight = calc_one_cattle( # TODO: 这里的稳定值计算与图表显示的不一致，这里相对准确（因为图表显示数据向上取整）。
smooth_data(weight_lst, max(3, int(weight_lst_len * (
12 / 100)) // 2 * 2 + 1))) # 取体重波长度的 12%处附近的奇数作为平滑窗口大小，最小窗口为3
if not np.isnan(weight): # weight可能为nan，当数据平均浮动过大时，可能找不到稳定值，则不取这个波
buy_weight = DAO.get_buy_weight_by_rfid(max_rfid) # 获取这个RFID牛只的购入体重
if buy_weight * Decimal(MAX_SINGLE_BUY_WEIGHT_PERCENT) > weight > buy_weight * Decimal(
MIN_SINGLE_BUY_WEIGHT_PERCENT): # 根据这一头牛的购入体重上下30%区间获取体重波
weight_wave = WeightWave(
id=-1,
floor_scale_id=floor_scale_id,
start_time=start_time.strftime(DATETIME_FORMAT),
end_time=prev_time.strftime(DATETIME_FORMAT),
time_std=time_std,
max_frq=max_frq,
max_frq_rfid=max_rfid,
weight=weight
)
weight_wave_lst.append(weight_wave) # 将一个体重波数据加入缓存列表
relation_reader_name_set.clear() # 清空RFID Reader集，按照每一个波的范围内识别Reader
weight_lst.clear() # 清空体重值存储列表
if (start_time is None and not is_out_of_range) or (
start_time is not None and is_out_of_interval and not is_out_of_range):
"""
一个新的波的开始，重置体重波开始时间，将最新时间作为开始时间。
当第一次迭代或者因为时间间隔超过1s，但是当前值在阈值范围内。
"""
start_time = log_time
if start_time is not None and is_out_of_range:
"""
在一个波的遍历过程中（已有开始时间）,
如果获取到的值超过了体重阈值，则无法将这个值作为一个新的波的开始，重置体重波开始时间，将开始时间清空待后续遍历值符合要求后再设定。
"""
start_time = None
if not is_out_of_range and log_relation_reader_name not in relation_reader_name_set:
"""
当一个有效波获取到时（波的开始节点，中间过程和结束节点），在一个波的形成过程中，将体重值存储并将RFID Reader ID去重存储。
"""
relation_reader_name_set.add(log_relation_reader_name) # 取出关联的RFID Reader ID去重存储
if not is_out_of_range:
"""
当一个有效波获取到时（波的开始节点，中间过程和结束节点），在一个波的形成过程中，将体重值缓存
"""
weight_lst.append(float(log_weight))
prev_time = log_time
return weight_wave_lst
def save_weight_wave(floor_scale_id, weight_wave_lst, time_std):
"""
将体重波存储本地
:param floor_scale_id
:param weight_wave_lst（已经存放了所有值）
:param time_std
:return:
"""
# 当一个地磅的波取完时，将波和波所属的地磅对应的RFID Reader存储进本地数据库
period_start_time = None # 一个地磅第一个波的开始时间
period_end_time = None # 一个地磅最后一个波的结束时间
for weight_wave_item in tqdm(weight_wave_lst, desc="---- 插入体重波", position=2,
disable=TQDM_DISABLE):
DAO.add_weight_wave(weight_wave_item, True) # 插入体重波数据
if period_start_time is None: # 判断exec_logs时间周期是否是截点
period_start_time = weight_wave_item.start_time
period_end_time = weight_wave_item.end_time
return ExecLog(floor_scale_id, period_start_time, period_end_time, time_std)
def save_exec_log(exec_log):
"""
储存进度日志，已存在的地磅更新，只往后更新不往前更新
:param exec_log:
:return:
"""
if exec_log.period_start_time is None or exec_log.period_end_time is None: # 地磅波列表是空的话不做记录
return
if exec_log.period_start_time == '': # 当地磅日志没有定义开始时间时，表示需要更新，仅更新该条地磅日志的结束时间
DAO.update_exec_log(exec_log)
else:
DAO.add_exec_log(exec_log)
def get_rfid_all():
return DAO.get_rfid_all()
def get_buy_weight(rfid):
return DAO.get_buy_weight_by_rfid(rfid)

14
db_connection.py Normal file
View File

@ -0,0 +1,14 @@
# Author: Lee Wu Love Lele
# Datetime: 2024/9/7 21:20
import mysql.connector
# 连接 MySQL 数据库
connection = mysql.connector.connect(
host="rm-bp1442up0e0nlz95szo.mysql.rds.aliyuncs.com",
user="rhm_read_only",
password="aA123456+",
database="rhm_insure_dev"
)
def get_connection():
return connection

2
db_dao.py
View File

@ -1,6 +1,6 @@
# Author: Lee Wu Love Lele # Author: Lee Wu Love Lele
# Datetime: 2024/9/7 21:28 # Datetime: 2024/9/7 21:28
import models import db_connection
# 创建游标对象 # 创建游标对象
connection = db_connection.get_connection() connection = db_connection.get_connection()

347
main.py
View File

@ -1,291 +1,102 @@
# This is a sample Python script.
# Press ⌃R to execute it or replace it with your code.
# Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings. # Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings.
import db_dao as dd
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
import numpy as np import numpy as np
import matplotlib.dates as mdates import matplotlib.dates as mdates
from itertools import product
import pandas as pd import pandas as pd
from datetime import datetime from datetime import datetime,timedelta
from config import DATETIME_FORMAT
import config
import sql_statements
import models
import sql_statements
import controller
from decimal import Decimal, getcontext
from util import smooth_data, calc_one_cattle
import util
import math
SPLIT_LINE = "----------------------------------------" m_weight = []
m_datetime = []
m_start_time = '2024-09-09 06:50:00'
m_end_time = '2024-09-09 07:10:00'
# m_reader = '2405085412' # 客户看不到视频
m_weighbridge_reader = '2405085474' # 客户可以看到设备
# m_reader = '2404049517'
# m_reader = '2405085489'
m_rfid_reader = '0A26AE180665' # 对应地磅的 2405085474
date_format = "%Y-%m-%d %H:%M:%S"
# Press the green button in the gutter to run the script.
# 创建查询时间范围的序列 # 创建查询时间范围的序列
# def create_timestamp_seq(): def create_timestamp_seq():
# # 将起止时间转换为 datetime 对象 # 将起止时间转换为 datetime 对象
# m_datetime_start_obj = datetime.strptime(m_start_time, date_format) m_datetime_start_obj = datetime.strptime(m_start_time, date_format)
# m_datetime_end_obj = datetime.strptime(m_end_time, date_format) m_datetime_end_obj = datetime.strptime(m_end_time, date_format)
#
# # 计算时间差并转为秒数
# time_diff = m_datetime_end_obj - m_datetime_start_obj
# seconds = time_diff.total_seconds() + 1
#
# # 创建时间序列，单位为秒
# start_time = pd.Timestamp(m_start_time)
# time_seconds_seq = pd.date_range(start=start_time, periods=int(seconds), freq='s') # 生成日期差秒数个时间点，每个时间点间隔1秒
# return time_seconds_seq
# 计算时间差并转为秒数
time_diff = m_datetime_end_obj - m_datetime_start_obj
seconds = time_diff.total_seconds()+1
# 创建时间序列，单位为秒
start_time = pd.Timestamp(m_start_time)
time_seconds_seq = pd.date_range(start=start_time, periods=int(seconds), freq='s') # 生成日期差秒数个时间点，每个时间点间隔1秒
return time_seconds_seq
# 补充缺失的时间秒数 # 补充缺失的时间秒数
# def time_data_suppl(timestamp, data): def time_data_suppl(timestamp,data):
# i = 1 i = 1
# sorted_data = [] sorted_data = []
# for ts in timestamp: for ts in timestamp:
# format_ts = ts.strftime(date_format) format_ts = ts.strftime(date_format)
# if format_ts not in data: if format_ts not in data:
# sorted_data.append(0) sorted_data.append(0)
# else: else:
# sorted_data.append(float(data[format_ts])) sorted_data.append(float(data[format_ts]))
# # print(str(i)+' 存在数据：'+format_ts+' ------> '+data[format_ts]) # print(str(i)+' 存在数据：'+format_ts+' ------> '+data[format_ts])
# i += 1 i += 1
# return sorted_data return sorted_data
# 对相同时间的数据去重复，取体重最大值 # 对相同时间的数据去重复，取体重最大值
# def time_data_dedup(data): def time_data_dedup(data):
# result_dict = {} result_dict = {}
# for row in data: for row in data:
# key = row[0].strftime(date_format) key = row[0].strftime(date_format)
# if key not in result_dict or float(row[1]) > float(result_dict[key]): if key not in result_dict or float(row[1]) > float(result_dict[key]):
# result_dict[key] = float(row[1]) result_dict[key] = float(row[1])
#
# return result_dict
return result_dict
def weight_wave_menu(rfid, buy_weight, time_std): # TODO: 需要修改为可以多个输入ID # rfid数据去重
weight_wave_lst = controller.get_weight_wave(rfid, time_std) def rfid_data_dedup(data):
if not weight_wave_lst: # 保留顺序的去重方法
print('未找到体重数据，请重新输入牛只ID') seen = set()
return -1 unique_tuple_list = []
print(SPLIT_LINE) for item in data:
for weight_wave in weight_wave_lst: if item not in seen:
print(str(weight_wave)) unique_tuple_list.append(item)
print(SPLIT_LINE) seen.add(item)
while True: return unique_tuple_list
weight_wave_id = input('请输入体重波ID（或输入-1返回）:')
if weight_wave_id == '-1':
cattle_menu(time_std)
elif not any(str(i.id) == weight_wave_id for i in weight_wave_lst):
print('输入不合法，请重试')
else:
item = next((obj for obj in weight_wave_lst if obj.id == int(weight_wave_id)), None)
diff_s = (datetime.strptime(item.end_time, DATETIME_FORMAT) - datetime.strptime(item.start_time,
DATETIME_FORMAT)).total_seconds()
if item is not None:
duration = item.start_time + " ~ " + item.end_time + " ( " + str(diff_s) + "s )"
print('正在获取体重日志...')
weight_wave_logs = controller.get_weight_wave_log(item, time_std)
print('正在显示图表...')
show_log_diagram(weight_wave_logs, buy_weight, daily_weight=item.weight, duration=duration,
rfid_frq=item.max_frq, id=item.id)
def show_log_diagram(weight_wave_logs, buy_weight, daily_weight, duration, rfid_frq, id):
index = 0
x = []
y = []
for item in weight_wave_logs:
y.append(float(item.weight))
x.append(item.time)
index += 1
y_smooth = smooth_data(y, max(5, int(len(weight_wave_logs) * (
12 / 100)) // 2 * 2 + 1))
m_daily_weight = calc_one_cattle(smooth_data(y, max(5, int(len(weight_wave_logs) * (
12 / 100)) // 2 * 2 + 1)))
plt.figure(1, figsize=(20, 4))
plt.plot(x, y, color='blue', linestyle='-', linewidth=1)
# 添加横向分割线
plt.axhline(y=math.ceil(float(buy_weight)), color='red', linestyle='--', linewidth=1,
label='buy_weight: ' + str(math.ceil(float(buy_weight))))
if not np.isnan(m_daily_weight): # TODO: 为什么 4 - 1307 - 8311 体重波没有稳定值但是记录了 486.0952
plt.axhline(y=math.ceil(m_daily_weight), color='green', linestyle='--', linewidth=1,
label='daily_weight: ' + str(math.ceil(m_daily_weight)))
plt.legend()
plt.xlabel('datetime')
plt.ylabel('weight(kg)')
plt.title(
"id: " + str(id) + " len: " + str(len(weight_wave_logs)) + " duration: " + duration + " rfid_frq: " + str(rfid_frq))
plt.figure(2, figsize=(20, 4))
plt.plot(x, y_smooth, color='blue', linestyle='-', linewidth=1)
plt.axhline(y=math.ceil(float(buy_weight)), color='red', linestyle='--', linewidth=1,
label='buy_weight: ' + str(math.ceil(float(buy_weight))))
if not np.isnan(m_daily_weight): # TODO: 为什么 4 - 1307 - 8311 体重波没有稳定值但是记录了 486.0952
plt.axhline(y=math.ceil(m_daily_weight), color='green', linestyle='--', linewidth=1,
label='daily_weight: ' + str(math.ceil(m_daily_weight)))
# 添加标签和图例
plt.legend()
plt.xlabel('datetime')
plt.ylabel('weight(kg)')
plt.title(
"id: " + str(id) + " len: " + str(len(weight_wave_logs)) + " duration: " + duration + " rfid_frq: " + str(rfid_frq))
plt.show()
return
def cattle_menu(time_std):
rfid_lst = sorted(controller.get_rfid_all(), key=lambda x: x.id) # 获取所有牛只的RFID，根据记录主键从小到达排序
print(SPLIT_LINE)
for item in rfid_lst:
print(vars(item))
print(SPLIT_LINE)
while True:
cattle_id = input('请输入id（返回输入-1）：')
if cattle_id == '-1':
main_menu()
elif not any(i.id == int(cattle_id) for i in rfid_lst):
print('输入无效请重新输入')
continue
else:
item = next((obj for obj in rfid_lst if obj.id == int(cattle_id)), None)
print('RFID：', item.rfid)
buy_weight = controller.get_buy_weight(item.rfid)
rt = weight_wave_menu(item.rfid, buy_weight, time_std)
if rt == -1:
continue
# daily_weight_lst = controller.get_daily_weight(item.rfid)
# if not daily_weight_lst:
# print('未找到体重数据，请重新输入牛只ID')
# continue
# for daily_weight in daily_weight_lst:
# print(daily_weight)
def main_menu():
while True:
print('主菜单')
print(SPLIT_LINE)
print('0 - 退出程序')
print('1 - 初始化体重数据（设备时间）')
print('2 - 查看牛只日体重（设备时间）')
print('3 - 初始化体重数据（服务器时间）')
print('4 - 查看牛只日体重（服务器时间）')
print('5 - 重置本地数据库')
print(SPLIT_LINE)
ctrl = input('请输入：')
if ctrl == '-1':
test()
break
elif ctrl == '0':
break
elif ctrl == '1':
controller.clear_data_by_time_std(1)
controller.init_data(1)
cattle_menu(1)
elif ctrl == '2':
cattle_menu(1)
elif ctrl == '3':
controller.clear_data_by_time_std(2)
controller.init_data(2)
cattle_menu(2)
elif ctrl == '4':
cattle_menu(2)
elif ctrl == '5':
confirm = input('确定要清除所有数据吗？y ：确定 / any key：退出')
if confirm == 'y' or 'Y':
controller.clear_all_data()
print('重置成功！')
continue
else:
print('输入错误，请重新输入')
continue
def test():
data = [1, 2, 4, 7, 2, 3, 7, 10, 12, 5, 6, 5, 5, 20, 23, 21, 20, 23, 22, 19]
util.calc_one_cattle(data)
return
if __name__ == '__main__': if __name__ == '__main__':
main_menu() timestamp_seq = create_timestamp_seq()
# timestamp_seq = create_timestamp_seq() mysql_weight_result = dd.get_cattle_weight_by_datetime(m_start_time, m_end_time, m_weighbridge_reader)
# mysql_weight_result = dd.get_cattle_weight_by_datetime(m_start_time, m_end_time, m_weighbridge_reader) mysql_rfid_result = dd.get_rfid_by_datetime(m_start_time, m_end_time, m_rfid_reader)
# mysql_rfid_result = dd.get_rfid_by_datetime(m_start_time, m_end_time, m_rfid_reader) dd.close()
# dd.close() clean_weight_result = time_data_dedup(mysql_weight_result)
# clean_weight_result = time_data_dedup(mysql_weight_result) clean_rfid_result = rfid_data_dedup(mysql_rfid_result)
# clean_rfid_result = rfid_data_dedup(mysql_rfid_result)
#
# final_weight_result = time_data_suppl(timestamp_seq, clean_weight_result)
# weight_wave_log = controller.get_weight_wave_log() final_weight_result = time_data_suppl(timestamp_seq, clean_weight_result)
# time_seq = [] plt.figure(figsize=(24, 4))
# weight_seq = [float(item.get('biz_data')) for item in weight_wave_log] plt.plot(timestamp_seq, final_weight_result, color='blue', linestyle='-', linewidth=1)
# # weight_seq = [100, 90, 110, 130, 150, 170, 190, 192, 195, 199, 210, 220, 230, 240, 248, 255, 245, 248, 250, 252,
# # 253, 253, 257, 254, 250, 255, 259, 265, 275, 285, 295, 310, 320, 305, 295, 283, 273, 263, 253, 240,
# # 229, 209, 190, 174, 162, 150, 139, 114, 100, 90, 100, 101, 102, 100, 95, 97]
# index_seq = [i for i in range(1, len(weight_seq) + 1)]
# print('weight_seq: ' + str(weight_seq) + '\nindex_seq: ' + str(index_seq))
# weight_seq_smooth = (smooth_data(weight_seq))
# plt.figure(figsize=(20, 4))
# plt.plot(index_seq, weight_seq_smooth, color='blue', linestyle='-', linewidth=1)
# plt.xlabel('sequence')
# plt.ylabel('weight(kg)')
# plt.title('weight_wave')
# plt.show()
# calc_one_cattle(weight_seq)
"""
x = np.linspace(0, 10, 100)
y1 = np.sin(x)
y2 = np.cos(x)
# 第1个图形窗口 for rst in clean_rfid_result:
plt.figure(1) # 添加纵向分割线
plt.plot(x, y1, color='r') plt.axvline(x=rst[0], color='r', linestyle='--', label=rst[1], linewidth = 0.5)
plt.title("Figure 1: Sine Function")
# 第2个图形窗口 # 添加图例
plt.figure(2) plt.legend()
plt.plot(x, y2, color='b')
plt.title("Figure 2: Cosine Function")
# 显示所有窗口
plt.show()
"""
# index = 1 plt.xlabel('datetime')
# for log in weight_wave_log: plt.ylabel('weight(kg)')
# m_dt = datetime.strftime(log.get('biz_time'), controller.DATETIME_FORMAT) plt.title('reader: ' + m_weighbridge_reader + '\n' + m_start_time + ' ---> ' + m_end_time)
# print(m_dt) plt.show()
# time_seq.append(m_dt)
# index_seq.append(index)
# weight_seq.append(log.get('biz_data'))
# index += 1
# for rst in clean_rfid_result:
# # 添加纵向分割线
# plt.axvline(x=rst[0], color='r', linestyle='--', label=rst[1], linewidth = 0.5)
#
# # 添加图例
# plt.legend()
#
#
# result = models.get_logs(floor_scale_id ="2404049517", start_time ="2024-09-01 00:00:00", end_time ="2024-09-02 00:00:00")
# result = models.get_ear_tag('9057970969300989')
# print(result)
# floor_scale_id, start_time, end_time, max_weight, max_weight_time, max_frq_ear_tag, max_frq_rfid, time_std
# weight_wave = ('floor_scale_id', '2024-12-01 00:00:00',
# '2024-12-07 00:00:00',
# 100.00, '2024-12-06 19:00:00', 'max_frq_ear_tag',
# 'max_frq_rfid', 1)
# parent_id = models.add_weight_wave(False, weight_wave, False)
# rfid_frq = []
# for i in range(1, 11):
# rfid_frq.append(
# (str(parent_id), str(i), 'rfid' + str(i), i, 1))
# models.add_rfid_frq(True, rfid_frq,True)
#
# print('parent_id:'+str(parent_id))
#
# models.delete_weight_wave(False)
# models.delete_rfid_frq(True)
# controller.init_data()
# controller.clean_data()

429
models.py
View File

@ -1,429 +0,0 @@
# Author: Lee Wu
# Datetime: 2024/12/6 14:44
import datetime
import decimal
from abc import abstractmethod
import mysql.connector
from config import RDB_HOST, RDB_USER, RDB_NAME, RDB_PASSWORD, LDB_HOST, LDB_PORT, LDB_NAME, LDB_USER, LDB_PASSWORD, \
DATETIME_FORMAT, TOTAL_BUY_WEIGHT_AVG
import sql_statements
from dataclasses import dataclass, astuple
from decimal import Decimal
# from datetime import datetime
@dataclass
class RFID:
id: int
rfid: str
ear_tag: str
@dataclass
class WeightWave:
id: int
weight: float
start_time: str
end_time: str
time_std: int
max_frq_rfid: str
max_frq: float
floor_scale_id: str
@dataclass
class Weight:
rfid: str
weight: float
time: str
time_std: int
rfid_reader: str = ""
@dataclass
class RFIDReader:
id_weight_wave: int
reader_name: str
@dataclass
class ExecLog:
floor_scale_id: str
period_start_time: str
period_end_time: str
time_std: int
@dataclass
class DailyWeight:
weight: float
date: str
# 数据库链接管理父类
class DataBaseModel:
connection = None
cursor = None
@classmethod
@abstractmethod
def get_connection_params(cls):
"""
返回数据库连接参数。
"""
pass
@classmethod
def init(cls):
if cls.connection is None: # 防止重复初始化
params = cls.get_connection_params()
cls.connection = mysql.connector.connect(**params)
cls.cursor = cls.connection.cursor(dictionary=True)
return cls.connection
@classmethod
def query(cls, sql, params=None):
"""
执行查询并返回结果
"""
if cls.connection is None:
cls.init()
cls.cursor.execute(sql, params)
return cls.cursor.fetchall()
@classmethod
def execute_trans(cls, sql, is_commit=True, params=None):
if cls.connection is None:
cls.init()
cls.cursor.executemany(sql, params)
if is_commit:
cls.connection.commit()
# try:
# # 检查事务是否已启动，未启动则启动
# if cls.connection.in_transaction is False:
# # 事务未启动
# cls.connection.start_transaction()
# if type(params) is tuple:
# cls.cursor.execute(sql, params)
# parent_id = cls.cursor.lastrowid
# elif type(params) is list:
# cls.cursor.executemany(sql, params)
# if is_commit:
# cls.connection.commit()
# except Exception as e:
# print(f"捕获错误: {e} \n --> " + cls.cursor.statement + '\n ----> ' + params)
# cls.connection.rollback()
# finally:
@classmethod
def execute(cls, sql, is_commit=True, params=None):
if cls.connection is None:
cls.init()
cls.cursor.execute(sql, params)
parent_id = cls.cursor.lastrowid
if is_commit:
cls.connection.commit()
return parent_id
@classmethod
def close(cls):
"""
关闭数据库连接
"""
if cls.cursor:
cls.cursor.close()
if cls.connection:
cls.connection.close()
cls.cursor = None
cls.connection = None
# 链接远程数据库，主要用户获取原始数据
class RemoteDataBaseModel(DataBaseModel):
"""
具体的子类，定义具体的数据库连接参数。
"""
@classmethod
def get_connection_params(cls):
"""
返回服务器数据库参数
"""
return {
"host": RDB_HOST,
"user": RDB_USER,
"password": RDB_PASSWORD,
"database": RDB_NAME
}
# 链接本地数据库，主要将分析结果储存到本地数据库
class LocalDataBaseModel(DataBaseModel):
"""
具体的子类，定义具体的数据库连接参数。
"""
@classmethod
def get_connection_params(cls):
"""
返回服务器数据库参数
"""
return {
"host": LDB_HOST,
'port': LDB_PORT,
"user": LDB_USER,
"password": LDB_PASSWORD,
"database": LDB_NAME
}
class DAO:
# 查询所有地磅ID和RFID Reader ID的关联数据
@staticmethod
def get_devices_id():
result = RemoteDataBaseModel.query(sql_statements.QUERY_DEVICES)
RemoteDataBaseModel.close()
return result
@staticmethod
def get_buy_weight_by_rfid(rfid):
result = RemoteDataBaseModel.query(sql_statements.sql_query_buy_weight_by_rfid(), (rfid,))
RemoteDataBaseModel.close()
if result:
return result[0].get('buy_weight')
else:
return Decimal(TOTAL_BUY_WEIGHT_AVG) # 如果没有购入体重的，则取所有牛只购入体重的平均值
# @staticmethod
# def get_rfid_reader_id(**kwargs):
# """
# 根据地磅ID、时间段获取对应RFIDReaderID，若多个返回值则表示设备ID关系变化，对应的RFIDReader设备识别的RFID都要参考。
# :param kwargs: floor_scale_id(tuple),start_time(str),end_time(str)
# :return:
# """
# if 'floor_scale_id' not in kwargs:
# return
# sql = sql_statements.sql_query_logs_group_by_relation_reader_name(kwargs['floor_scale_id'])
# params = kwargs['floor_scale_id'] + (kwargs['start_time'], kwargs['end_time'])
# result = RemoteDataBaseModel.query(sql, params)
# RemoteDataBaseModel.close()
# return result
# 根据地磅ID查询所有日志或者根据地磅ID和一个时间段查询日志
@staticmethod
def get_weight_logs(**kwargs):
"""
根据地磅ID或RFID读取器ID，结合开始时间和结束时间获取日志数据
:param kwargs: floor_scale_id(tuple),start_time(str),end_time(str),time_std(int)
:return:
"""
time_std = kwargs['time_std']
sql = sql_statements.sql_query_logs_by_reader_name_and_duartion(kwargs['floor_scale_id'],
time_std)
params = kwargs['floor_scale_id'] + (kwargs['start_time'], kwargs['end_time'])
result = RemoteDataBaseModel.query(sql, params)
RemoteDataBaseModel.close()
weight_lst = [Weight(
weight=item.get('biz_data'),
time=item.get('biz_time') if time_std == 1 else item.get('create_time'),
time_std=kwargs['time_std'],
rfid='',
rfid_reader=item.get('relation_reader_name')
) for item in result]
RemoteDataBaseModel.close()
return weight_lst
@staticmethod
def get_rfid_count(rfid_reader_id, start_time, end_time):
"""
根据多个RFID Reader ID查询其识别的RFID频次
:param rfid_reader_id: tuple
:param start_time: str
:param end_time: str
:return:
"""
sql = sql_statements.sql_query_logs_count_group_by_biz_data(rfid_reader_id)
params = rfid_reader_id + (start_time, end_time)
result = RemoteDataBaseModel.query(sql, params)
RemoteDataBaseModel.close()
return result
# 根据RFID查询对应的定位耳标外壳号
@staticmethod
def get_ear_tag(rfid):
result = RemoteDataBaseModel.query(sql_statements.QUERY_EAR_TAG_ID_BY_RFID, [rfid])
RemoteDataBaseModel.close()
return result
# 根据定位耳标外壳号和日期查询体重波,本地查询
@staticmethod
def get_weight_wave_by_floor_scale_id(floor_scale_id):
"""
获取体重波基本信息
:param floor_scale_id:
:return:
"""
result = LocalDataBaseModel.query(sql_statements.sql_query_weight_wave_by_floor_scale_id(), (floor_scale_id,))
LocalDataBaseModel.close()
return result
@staticmethod
def get_weight_wave_by_rfid(rfid, time_std):
"""
根据牛只RFID获取所有体重波
:param rfid:
:param time_std:
:return:
"""
result = LocalDataBaseModel.query(sql_statements.sql_query_weight_wave_by_rfid(), (rfid, time_std))
LocalDataBaseModel.close()
weight_wave_lst = [WeightWave(
floor_scale_id=item.get('floor_scale_id'),
max_frq_rfid=rfid,
start_time=item.get('start_time').strftime(DATETIME_FORMAT),
end_time=item.get('end_time').strftime(DATETIME_FORMAT),
time_std=item.get('time_std'),
max_frq=item.get('max_frq'),
weight=item.get('weight'),
id=item.get('id_weight_wave')
) for item in result]
return weight_wave_lst
@staticmethod
def get_reader(id_weight_wave):
"""
根据体重波ID查询对应的RFID Reader Name
:param id_weight_wave:
:return:
"""
result = LocalDataBaseModel.query(sql_statements.sql_query_relation_reader_by_id_weight_wave(),
(id_weight_wave,))
LocalDataBaseModel.close()
return result
@staticmethod
def get_rfid_all():
result = RemoteDataBaseModel.query(sql_statements.sql_query_rfid_all_in_cattle())
rfid_lst = [RFID(
id=item.get('id'),
rfid=item.get('rfid'),
ear_tag=item.get('ear_tag')
) for item in result]
RemoteDataBaseModel.close()
return rfid_lst
# @staticmethod
# def get_rfid(time_std):
# result = LocalDataBaseModel.query(sql_statements.sql_query_weight_wave_group_by_rfid(), (time_std,))
# LocalDataBaseModel.close()
# return result
# 插入一个体重波
@staticmethod
def add_weight_wave(weight_wave, is_commit):
"""
体重波插入值，不直接提交，等RFID Reader获取后统一提交 floor_scale_id, start_time, end_time, max_weight, max_weight_time, time_std, create_time, last_update_time
:param weight_wave:
:param is_commit: 是否提交查询
:return: 插入后的主键
"""
id_weight_wave = LocalDataBaseModel.execute(sql_statements.INSERT_WEIGHT_WAVE, is_commit, (
weight_wave.floor_scale_id, weight_wave.start_time, weight_wave.end_time, weight_wave.time_std,
weight_wave.max_frq_rfid, weight_wave.max_frq, weight_wave.weight))
return id_weight_wave
@staticmethod
def add_rfid_frq(is_commit=False, params=None, is_close=True):
"""
插入多条RFID频次值
:param is_commit: 是否继续事务
:param params: RFID频次插入值列表 id_weight_wave,ear_tag,rfid,rfid_frq,time_std
:param is_close: 是否关闭链接
:return:
"""
# if params is None:
# return
# LocalDataBaseModel.execute_trans(sql_statements.INSERT_RIFD_FRQ, params, is_commit)
# if is_close:
# LocalDataBaseModel.close()
pass
@staticmethod
def add_rfid_readers(rfid_reader_lst):
"""
插入Reader后关闭链接
:param rfid_reader_lst:
:return:
"""
LocalDataBaseModel.execute_trans(sql_statements.sql_insert_rfid_reader(), True,
[astuple(rfid_reader) for rfid_reader in rfid_reader_lst])
LocalDataBaseModel.close()
@staticmethod
def add_exec_log(exec_log, is_close=True):
LocalDataBaseModel.execute(sql_statements.sql_insert_exec_log(), True, (
exec_log.floor_scale_id, exec_log.period_start_time, exec_log.period_end_time, exec_log.time_std))
if is_close:
LocalDataBaseModel.close()
@staticmethod
def update_exec_log(exec_log, is_close=True):
LocalDataBaseModel.execute(sql_statements.sql_update_exec_log(), True,
(exec_log.period_end_time, exec_log.floor_scale_id))
if is_close:
LocalDataBaseModel.close()
@staticmethod
def delete_weight_wave_all(is_close=True):
rows = LocalDataBaseModel.execute(sql_statements.sql_delete_weight_wave_all())
if is_close:
LocalDataBaseModel.close()
return rows
@staticmethod
def delete_weight_wave_by_time_std(time_std):
rows = LocalDataBaseModel.execute(sql_statements.sql_delete_weight_wave_by_time_std(), True, (time_std,))
LocalDataBaseModel.close()
return rows
@staticmethod
def delete_rfid_frq(is_close=True):
rows = LocalDataBaseModel.execute(sql_statements.DELETE_RFID_FRQ)
if is_close:
LocalDataBaseModel.close()
return rows
@staticmethod
def get_exec_log(time_std):
result = LocalDataBaseModel.query(sql_statements.QUERY_EXEC_LOG, time_std)
LocalDataBaseModel.close()
return result
@staticmethod
def delete_exec_log_all(is_close=True):
rows = LocalDataBaseModel.execute(sql_statements.sql_delete_exec_log_all())
if is_close:
LocalDataBaseModel.close()
return rows
@staticmethod
def delete_exec_log_by_time_std(time_std):
rows = LocalDataBaseModel.execute(sql_statements.sql_delete_exec_log_by_time_std(), True, (time_std,))
LocalDataBaseModel.close()
return rows
@staticmethod
def delete_relation_reader(is_close=True):
"""
删除与体重波关联的RFID Reader
:param is_close:
:return:
"""
rows = LocalDataBaseModel.execute(sql_statements.sql_delete_relatioin_rfid_reader())
if is_close:
LocalDataBaseModel.close()
return rows

154
sql_statements.py
View File

@ -1,154 +0,0 @@
# Author: Lee Wu Love Lele
# Datetime: 2024/12/6 19:53
"""
SELECT ff.reader_code FROM farm_field AS ff
WHERE ff.is_deleted = 1
AND ff.reader_code IS NOT NULL AND ff.reader_code != '' AND ff.rfid_reader_code IS NOT NULL AND ff.rfid_reader_code != ''
GROUP BY ff.reader_code
"""
QUERY_DEVICES = "SELECT ff.reader_code FROM farm_field AS ff WHERE ff.is_deleted = 1 AND ff.reader_code IS NOT NULL AND ff.reader_code != '' AND ff.rfid_reader_code IS NOT NULL AND ff.rfid_reader_code != '' GROUP BY ff.reader_code;"
QUERY_EAR_TAG_ID_BY_RFID = """
SELECT sb.shell_subject_code
FROM subject_base AS sb
WHERE is_deleted = 1
AND lot_subject_code = %s
AND shell_subject_code != ''
"""
"""
查询本地日志中是否分析过地磅或者分析到了什么时间
"""
QUERY_EXEC_LOG = "SELECT * FROM exec_log AS el WHERE el.time_std = %s;"
INSERT_WEIGHT_WAVE = """
INSERT INTO weight_wave (floor_scale_id, start_time, end_time, time_std , max_frq_rfid , max_frq, weight, create_time,last_update_time)
VALUES (%s,%s,%s,%s,%s,%s,%s,NOW(),NOW())
"""
INSERT_RIFD_FRQ = """
INSERT INTO rfid_value_frq(id_weight_wave,ear_tag,rfid,rfid_frq,time_std,create_time,last_update_time)
VALUES (%s,%s,%s,%s,%s,%s,%s)
"""
def sql_delete_weight_wave_all():
return "TRUNCATE TABLE weight_wave"
def sql_delete_weight_wave_by_time_std():
return "DELETE FROM weight_wave WHERE time_std = %s;"
DELETE_RFID_FRQ = "TRUNCATE TABLE rfid_value_frq"
def sql_query_buy_weight_by_rfid():
return "SELECT buy_weight FROM subject_base WHERE lot_subject_code = %s;"
def sql_query_logs_count_group_by_biz_data(reader_names=None):
"""
耳标RFID识别结果频次，biz_data返回的是RFID
:param reader_names: RFID Reader ID (tuple)
:return:
"""
placeholders = ', '.join(['%s'] * len(reader_names))
return f"SELECT biz_data, COUNT(*) AS repeat_count FROM notify_log WHERE notify_group = 'rfidCallback' AND reader_name IN ({placeholders}) AND biz_time BETWEEN %s AND %s GROUP BY biz_data ORDER BY repeat_count DESC;"
# def sql_query_logs_by_reader_name(reader_names=None):
# """
# 根据设备ID查询全时间周期
# :param reader_name: (tuple)
# :return:
# """
# placeholders = ', '.join(['%s'] * len(reader_names))
# return f"SELECT nl.biz_time, nl.biz_data, nl.create_time, nl.relation_reader_name FROM notify_log AS nl WHERE reader_name IN ({placeholders}) AND is_deleted = 1 ORDER BY biz_time ASC, create_time ASC;"
def sql_query_logs_by_reader_name_and_duartion(reader_names=None, time_std=1):
"""
根据设备ID查询特定时间段
:param reader_name: (tuple)
:return:
"""
placeholders = ', '.join(['%s'] * len(reader_names))
first_order = 'biz_data'
second_order = 'create_time'
time = 'biz_time'
if time_std == 1:
pass
elif time_std == 2:
first_order = 'create_time'
second_order = 'biz_data'
time = 'create_time'
pass
return f"SELECT nl.notify_id, nl.biz_data ,nl.biz_time, nl.create_time, nl.relation_reader_name FROM notify_log AS nl WHERE reader_name IN({placeholders}) AND is_deleted = 1 AND {time} BETWEEN %s AND %s ORDER BY {first_order} ASC, {second_order} ASC, notify_id ASC;"
def sql_query_logs_group_by_relation_reader_name(reader_names=None):
"""
根据地磅ID查询日志中对应的RFIDReaderID出现次数，识别一个时间段中地磅ID与RFIDReaderID对应关系的变化。
:param reader_name: (tuple)
:return:
"""
placeholders = ', '.join(['%s'] * len(reader_names))
return f"SELECT relation_reader_name, COUNT(*) AS repeat_count FROM notify_log WHERE reader_name IN ({placeholders}) AND biz_time BETWEEN %s AND %s GROUP BY relation_reader_name ORDER BY repeat_count DESC;"
def sql_insert_rfid_reader():
return "INSERT INTO relation_rfid_reader(id_weight_wave,reader_name,create_time,last_update_time) VALUES (%s,%s,NOW(),NOW());"
def sql_insert_exec_log():
return "INSERT INTO exec_log(floor_scale_id,start_time,end_time,time_std,create_time,last_update_time) VALUES(%s,%s,%s,%s,NOW(),NOW());"
def sql_delete_exec_log_all():
return "TRUNCATE TABLE exec_log;"
def sql_delete_exec_log_by_time_std():
return "DELETE FROM weight_wave WHERE time_std = %s;"
def sql_delete_relatioin_rfid_reader():
return "TRUNCATE TABLE relation_rfid_reader;"
def sql_update_exec_log():
return "UPDATE exec_log SET end_time = %s, last_update_time = NOW() WHERE floor_scale_id = %s;"
def sql_query_weight_wave_by_floor_scale_id():
"""
通过地磅ID查询全部体重波
:return:
"""
return "SELECT * FROM weight_wave where floor_scale_id = %s"
def sql_query_weight_wave_by_rfid():
"""
最高出现频率的RFID查询全部体重波
:return:
"""
return "SELECT * FROM weight_wave where max_frq_rfid = %s"
def sql_query_relation_reader_by_id_weight_wave():
return "SELECT * FROM relation_rfid_reader AS rrr WHERE rrr.id_weight_wave = %s"
def sql_query_weight_wave_group_by_rfid():
return "SELECT max_frq_rfid, COUNT(*) AS repeat_count FROM weight_wave WHERE time_std = %s GROUP BY max_frq_rfid;"
def sql_query_weight_wave_by_rfid():
return "SELECT * FROM weight_wave WHERE max_frq_rfid = %s AND time_std = %s ORDER BY start_time ASC;"
def sql_query_rfid_all_in_cattle():
return "SELECT subject_id AS id ,lot_subject_code AS rfid, shell_subject_code AS ear_tag FROM subject_base WHERE is_deleted = 1 AND buy_weight != 0 AND lot_subject_code != '' AND shell_subject_code != '';"

74
util.py
View File

@ -1,74 +0,0 @@
# Author: Lee Wu Love Lele
# Datetime: 2024/12/8 12:58
from config import DATETIME_FORMAT
from datetime import datetime
from scipy.signal import savgol_filter
import numpy as np
from config import STABILITY_THRESHOLD
import pandas as pd
def smooth_data(data, window_length=5, polyorder=3):
"""
平滑数据降噪，使用Savitzky-Golay滤波
* 通过算法（Savitzky-Golay滤波），将短期波动平滑化，突出重量变化的主要趋势，从而更接近牛的真实体重曲线。
* 在牛走上或走下地磅时，体重数据会从小到大（走上）或从大到小（走下）发生变化。平滑处理可以让“从小到大”或“从大到小”的趋势更加清晰，便于后续算法找到牛完整通过地磅的时间段。
* 如果直接对噪声较大的数据进行分析（如提取最大值、计算均值等），可能需要大量的异常值处理或复杂的算法来排除干扰。
* 平滑数据可以简化后续计算过程，使得分析模型更加高效。
* 牛在地磅上可能出现诸如抬腿、摇晃等小幅动作，这些动作会导致短时间内体重数据发生波动。平滑能够消除这些短时间的小波动，使得体重数据更加稳定。
* 平滑可以让真正的异常值（如多头牛站上地磅）更加显著，便于识别。例如：没有平滑时，普通波动和异常值可能混淆在一起。平滑后，正常数据的波动范围变小，异常值更加容易识别。
:param data:
:param window_length: int
:param polyorder: int
:return:
"""
"""
window_length
* 含义：
* 表示滑动窗口的长度，即在每次滤波计算中使用的点的数量。
* 它必须是一个正奇数，如 3、5、7 等，因为 Savitzky-Golay 滤波需要确保窗口中心有一个对称点。
* 作用：
* 窗口越大：
* 数据被平滑得越明显，但可能会导致细节丢失。
* 更适合处理噪声较多但关注长期趋势的情况。
* 窗口越小：
* 保留更多的细节，但对噪声的抑制效果较差。
* 更适合处理数据变化较快且噪声较少的情况。
polyorder
* 含义：
* 表示拟合多项式的阶数，用于滑动窗口内的数据拟合。
* 它必须小于 window_length，因为阶数不能超过拟合点的数量（否则拟合就无意义）。
* 作用：
* 多项式阶数越高：
* 滤波器拟合数据的能力更强，可以保留更多复杂的细节，但同时可能引入噪声（过拟合）。
* 更适合处理具有快速变化趋势的数据。
* 多项式阶数越低：
* 滤波器更倾向于平滑数据，减少噪声，但可能会忽略一些数据细节。
* 更适合处理平稳的数据或去除高频噪声。
"""
print(f'window_len: {window_length}, polyorder: {polyorder}')
smoothed_data = savgol_filter(data, window_length=window_length, polyorder=polyorder)
return smoothed_data
def calc_one_cattle(data):
print('data: ', data)
# print('data len: ', len(data))
# 计算变化率
rate_of_change = np.abs(np.diff(data)) # 一阶差分后的绝对值，比data长度少1
print('np.diff: ', np.diff(data))
# print('np.diff len: ', len(np.diff(data)))
print('rate_of_change: ', rate_of_change)
# print('rate_of_change len: ', len(rate_of_change))
# 设置阈值，识别稳定时间段
stability_threshold = STABILITY_THRESHOLD # 变化率小于此值视为稳定
stable_indices = np.where(rate_of_change < stability_threshold)[0]
print('stable_indices: ', stable_indices)
# 找到稳定时间段的体重中值
stable_weights = [data[i] for i in stable_indices]
print('stable_weights: ', stable_weights)
stable_median_weight = np.median(stable_weights) # TODO:这里取了所有稳定值的中位数，这里包含了异常稳定值形成干扰，如何识别这些异常？
print('stable_median_weight: ', stable_median_weight)
return stable_median_weight