512 lines
21 KiB
Python
512 lines
21 KiB
Python
import sqlite3
|
||
import datetime
|
||
import json
|
||
import csv
|
||
from typing import List, Dict, Optional, Tuple, Any
|
||
from contextlib import contextmanager
|
||
import logging
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BatteryDatabase:
|
||
def __init__(self, db_path="battery_data.db"):
|
||
self.db_path = db_path
|
||
# self.init_database()
|
||
|
||
@contextmanager
|
||
def get_connection(self):
|
||
"""数据库连接上下文管理器"""
|
||
conn = sqlite3.connect(self.db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
try:
|
||
yield conn
|
||
conn.commit()
|
||
except Exception as e:
|
||
conn.rollback()
|
||
raise e
|
||
finally:
|
||
conn.close()
|
||
|
||
def init_database(self):
|
||
"""初始化数据库和表结构"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
# 设置数据库使用本地时间
|
||
cursor.execute("PRAGMA foreign_keys = ON")
|
||
# 创建批次记录表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS batch_records (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
batch_name TEXT NOT NULL,
|
||
created_time TIMESTAMP DEFAULT (datetime('now', 'localtime')),
|
||
description TEXT,
|
||
status TEXT DEFAULT 'active'
|
||
)
|
||
''')
|
||
|
||
# 创建弹夹表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS magazines (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
magazine_code TEXT NOT NULL,
|
||
created_time TIMESTAMP DEFAULT (datetime('now', 'localtime')),
|
||
description TEXT
|
||
)
|
||
''')
|
||
|
||
# 创建箱表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS boxes (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
magazine_id INTEGER NOT NULL,
|
||
box_code TEXT NOT NULL,
|
||
created_time TIMESTAMP DEFAULT (datetime('now', 'localtime')),
|
||
description TEXT,
|
||
FOREIGN KEY (magazine_id) REFERENCES magazines(id)
|
||
)
|
||
''')
|
||
|
||
# 创建电池数据表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS battery_data (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
box_id INTEGER NOT NULL,
|
||
time TIMESTAMP NOT NULL,
|
||
code TEXT NOT NULL,
|
||
top_qr_code TEXT,
|
||
bottom_qr_code TEXT,
|
||
h2_value REAL CHECK (h2_value >= 0),
|
||
o2_value REAL CHECK (o2_value >= 0),
|
||
symbol TEXT,
|
||
thickness REAL CHECK (thickness >= 0),
|
||
status TEXT DEFAULT 'normal',
|
||
notes TEXT,
|
||
box_index INTEGER,
|
||
magazine_index INTEGER,
|
||
FOREIGN KEY (box_id) REFERENCES boxes(id)
|
||
)
|
||
''')
|
||
|
||
# 创建保存电池数据表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS save_battery_data (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
batch_id INTEGER NOT NULL,
|
||
time TIMESTAMP NOT NULL,
|
||
code TEXT NOT NULL,
|
||
top_qr_code TEXT,
|
||
bottom_qr_code TEXT,
|
||
h2_value REAL CHECK (h2_value >= 0),
|
||
o2_value REAL CHECK (o2_value >= 0),
|
||
symbol TEXT,
|
||
thickness REAL CHECK (thickness >= 0),
|
||
status TEXT DEFAULT 'normal',
|
||
notes TEXT,
|
||
group_id INTEGER,
|
||
data_index INTEGER,
|
||
box_code INTEGER,
|
||
box_index INTEGER,
|
||
magazine_code INTEGER,
|
||
magazine_index INTEGER
|
||
)
|
||
''')
|
||
|
||
# 创建索引以提高查询性能
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_box_id ON battery_data(box_id)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_time ON battery_data(time)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_code ON battery_data(code)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_box_magazine ON boxes(magazine_id)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_batch_status ON batch_records(status)')
|
||
|
||
# 为save_battery_data表也创建索引
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_save_time ON save_battery_data(time)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_save_code ON save_battery_data(code)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_save_group ON save_battery_data(group_id)')
|
||
|
||
logger.info("数据库初始化完成")
|
||
|
||
# ========== 增 ==========
|
||
def create_batch(self, batch_name: str, description: str = "") -> int:
|
||
"""创建新的批次记录"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO batch_records (batch_name, description)
|
||
VALUES (?, ?)
|
||
''', (batch_name, description))
|
||
batch_id = cursor.lastrowid
|
||
logger.info(f"创建批次成功: {batch_name} (ID: {batch_id})")
|
||
return batch_id
|
||
except sqlite3.IntegrityError:
|
||
logger.error(f"批次名称已存在: {batch_name}")
|
||
raise ValueError(f"批次名称 '{batch_name}' 已存在")
|
||
except Exception as e:
|
||
logger.error(f"创建批次失败: {e}")
|
||
raise
|
||
|
||
def create_magazine(self, magazine_code: str, description: str = "") -> int:
|
||
"""创建弹夹"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO magazines (magazine_code, description)
|
||
VALUES (?, ?)
|
||
''', (magazine_code, description))
|
||
magazine_id = cursor.lastrowid
|
||
logger.info(f"创建弹夹成功: {magazine_code} (ID: {magazine_id})")
|
||
return magazine_id
|
||
except sqlite3.IntegrityError:
|
||
logger.error(f"弹夹代码已存在: {magazine_code}")
|
||
raise ValueError(f"弹夹代码 '{magazine_code}' 已存在")
|
||
except Exception as e:
|
||
logger.error(f"创建弹夹失败: {e}")
|
||
raise
|
||
|
||
def create_box(self, magazine_id: int, box_code: str, description: str = "") -> int:
|
||
"""创建箱"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO boxes (magazine_id, box_code, description)
|
||
VALUES (?, ?, ?)
|
||
''', (magazine_id, box_code, description))
|
||
box_id = cursor.lastrowid
|
||
logger.info(f"创建箱成功: {box_code} (ID: {box_id})")
|
||
return box_id
|
||
except sqlite3.IntegrityError:
|
||
logger.error(f"箱代码已存在: {box_code}")
|
||
raise ValueError(f"箱代码 '{box_code}' 已存在")
|
||
except Exception as e:
|
||
logger.error(f"创建箱失败: {e}")
|
||
raise
|
||
|
||
def _validate_battery_data(self, battery_data: Dict):
|
||
"""验证电池数据"""
|
||
required_fields = ['code']
|
||
for field in required_fields:
|
||
if field not in battery_data or not battery_data[field]:
|
||
raise ValueError(f"缺少必要字段: {field}")
|
||
|
||
# 验证数值范围
|
||
if battery_data.get('h2_value') is not None and battery_data['h2_value'] < 0:
|
||
raise ValueError("H2值不能为负数")
|
||
if battery_data.get('o2_value') is not None and battery_data['o2_value'] < 0:
|
||
raise ValueError("O2值不能为负数")
|
||
if battery_data.get('thickness') is not None and battery_data['thickness'] < 0:
|
||
raise ValueError("厚度不能为负数")
|
||
|
||
def add_batch_batteries(self, box_id: int, batteries: List[Dict]) -> bool:
|
||
|
||
"""批量添加电池数据"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
for battery in batteries:
|
||
cursor.execute('''
|
||
INSERT INTO battery_data
|
||
(box_id, time, code, top_qr_code, bottom_qr_code, h2_value, o2_value,
|
||
symbol, thickness, status, notes,
|
||
box_index, magazine_index)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?)
|
||
''', (
|
||
box_id,
|
||
battery.get('time', datetime.datetime.now()),
|
||
battery['serial_number'],
|
||
battery.get('Below_QR'),
|
||
battery.get('Up_QR'),
|
||
battery.get('H2'),
|
||
battery.get('02'),
|
||
battery.get('Cell_ran'),
|
||
battery.get('thick'),
|
||
battery.get('status', 'normal'),
|
||
battery.get('source_code'),
|
||
battery.get('box_index'),
|
||
battery.get('magazine_index')
|
||
))
|
||
|
||
logger.info(f"批量添加电池数据成功，共 {len(batteries)} 条记录")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"批量添加电池数据失败: {e}")
|
||
return False
|
||
|
||
def add_save_battery_data_batch_with_groups(self, batch_id, battery_data_with_groups: List[Tuple[Dict, int]]) -> \
|
||
List[int]:
|
||
"""批量添加保存电池数据（带分组信息）"""
|
||
try:
|
||
record_ids = []
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
for battery_data, group_id in battery_data_with_groups:
|
||
self._validate_battery_data(battery_data)
|
||
cursor.execute('''
|
||
INSERT INTO save_battery_data
|
||
(batch_id, time, code, top_qr_code, bottom_qr_code, h2_value,
|
||
o2_value, symbol, thickness, status, notes, group_id, data_index,
|
||
box_code, box_index, magazine_code, magazine_index)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
''', (
|
||
batch_id,
|
||
battery_data.get('time', datetime.datetime.now()),
|
||
battery_data['code'],
|
||
battery_data.get('top_qr_code'),
|
||
battery_data.get('bottom_qr_code'),
|
||
battery_data.get('h2_value'),
|
||
battery_data.get('o2_value'),
|
||
battery_data.get('symbol'),
|
||
battery_data.get('thickness'),
|
||
battery_data.get('status', 'normal'),
|
||
battery_data.get('notes'),
|
||
group_id,
|
||
battery_data.get('data_index'),
|
||
battery_data.get('box_code'),
|
||
battery_data.get('box_index'),
|
||
battery_data.get('magazine_code'),
|
||
battery_data.get('magazine_index'),
|
||
))
|
||
record_ids.append(cursor.lastrowid)
|
||
|
||
conn.commit()
|
||
logger.info(f"批量添加保存电池数据成功: 共 {len(record_ids)} 条记录")
|
||
return record_ids
|
||
|
||
except sqlite3.IntegrityError as e:
|
||
logger.error(f"电池代码已存在: {e}")
|
||
raise ValueError(f"电池代码已存在")
|
||
except Exception as e:
|
||
logger.error(f"批量添加保存电池数据失败: {e}")
|
||
raise
|
||
|
||
# ========== 查 ==========
|
||
def get_latest_battery_data(self, limit: int = 2000, status: Optional[str] = None) -> List[Dict]:
|
||
"""
|
||
获取电池数据表的最新数据
|
||
|
||
Args:
|
||
limit: 返回的记录数量，默认为2000
|
||
status: 筛选状态（如'normal'），默认为None表示不过滤状态
|
||
|
||
Returns:
|
||
包含电池数据的字典列表
|
||
"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
# 构建SQL查询
|
||
query = '''
|
||
SELECT bd.id, bd.time, bd.code, bd.top_qr_code, bd.bottom_qr_code,
|
||
bd.h2_value, bd.o2_value, bd.symbol, bd.thickness,
|
||
bd.status, bd.notes, bd.box_id,
|
||
bx.box_code, m.magazine_code,
|
||
bx.box_code, bd.box_index,
|
||
m.magazine_code, bd.magazine_index
|
||
FROM battery_data bd
|
||
LEFT JOIN boxes bx ON bd.box_id = bx.id
|
||
LEFT JOIN magazines m ON bx.magazine_id = m.id
|
||
'''
|
||
params = []
|
||
|
||
# 添加状态筛选条件
|
||
if status:
|
||
query += ' WHERE bd.status = ?'
|
||
params.append(status)
|
||
|
||
# 添加排序和限制
|
||
query += ' ORDER BY bd.time DESC, bd.id DESC LIMIT ?'
|
||
params.append(limit)
|
||
|
||
# 执行查询
|
||
cursor.execute(query, params)
|
||
|
||
# 转换为字典列表
|
||
results = [dict(row) for row in cursor.fetchall()]
|
||
|
||
logger.info(f"获取最新电池数据成功，共 {len(results)} 条记录")
|
||
return results
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取最新电池数据失败: {e}")
|
||
return []
|
||
|
||
def get_batches_by_time_range_with_daily_sort(self, start_time: str, end_time: str) -> List[Dict]:
|
||
"""根据时间范围查询批次记录表所对应的保存电池数据表的批次，并按天排序
|
||
|
||
Args:
|
||
start_time: 开始时间 (格式: 'YYYY-MM-DD HH:MM:SS')
|
||
end_time: 结束时间 (格式: 'YYYY-MM-DD HH:MM:SS')
|
||
|
||
Returns:
|
||
包含批次信息的字典列表，按天分组排序
|
||
"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT DISTINCT
|
||
br.id,
|
||
br.batch_name,
|
||
br.created_time,
|
||
br.description,
|
||
br.status,
|
||
DATE(br.created_time) as batch_date,
|
||
COUNT(sbd.id) as total_data_count,
|
||
COUNT(DISTINCT sbd.group_id) as group_count,
|
||
MIN(sbd.time) as earliest_data_time,
|
||
MAX(sbd.time) as latest_data_time
|
||
FROM batch_records br
|
||
INNER JOIN save_battery_data sbd ON br.id = sbd.batch_id
|
||
WHERE sbd.time BETWEEN ? AND ?
|
||
GROUP BY br.id, br.batch_name, br.created_time, br.description, br.status, batch_date
|
||
ORDER BY batch_date DESC, br.created_time DESC
|
||
''', (start_time, end_time))
|
||
|
||
batches = [dict(row) for row in cursor.fetchall()]
|
||
logger.info(f"根据时间范围 {start_time} 到 {end_time} 查询到 {len(batches)} 个批次")
|
||
print(batches)
|
||
return batches
|
||
except Exception as e:
|
||
logger.error(f"根据时间范围查询批次失败: {e}")
|
||
return []
|
||
|
||
def get_batch_data(self, batch_id: int, group_id: int = None) -> List[Dict]:
|
||
"""根据批次ID和组ID查询数据
|
||
|
||
Args:
|
||
batch_id: 批次ID
|
||
group_id: 组ID（可选），如果提供则返回该组的详细数据，否则返回批次下所有组的信息
|
||
|
||
Returns:
|
||
包含组信息或详细数据的字典列表
|
||
"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
if group_id is not None:
|
||
# 查询指定组的详细数据
|
||
cursor.execute('''
|
||
SELECT
|
||
sbd.id, sbd.time, sbd.code, sbd.top_qr_code, sbd.bottom_qr_code,
|
||
sbd.h2_value, sbd.o2_value, sbd.symbol, sbd.thickness,
|
||
sbd.status, sbd.notes, sbd.batch_id, sbd.group_id, sbd.data_index,
|
||
sbd.box_code, sbd.box_index, sbd.magazine_code, sbd.magazine_index,
|
||
br.batch_name
|
||
FROM save_battery_data sbd
|
||
LEFT JOIN batch_records br ON sbd.batch_id = br.id
|
||
WHERE sbd.batch_id = ? AND sbd.group_id = ?
|
||
ORDER BY sbd.data_index ASC
|
||
''', (batch_id, group_id))
|
||
|
||
data = [dict(row) for row in cursor.fetchall()]
|
||
logger.info(f"查询批次 {batch_id} 组 {group_id} 的数据成功，共 {len(data)} 条记录")
|
||
return data
|
||
else:
|
||
# 查询批次下所有组的信息
|
||
cursor.execute('''
|
||
SELECT
|
||
group_id,
|
||
COUNT(*) as data_count,
|
||
MIN(data_index) as min_index,
|
||
MAX(data_index) as max_index,
|
||
MIN(time) as start_time,
|
||
MAX(time) as end_time,
|
||
AVG(h2_value) as avg_h2,
|
||
AVG(o2_value) as avg_o2,
|
||
AVG(thickness) as avg_thickness,
|
||
COUNT(CASE WHEN status = 'normal' THEN 1 END) as normal_count,
|
||
COUNT(CASE WHEN status != 'normal' THEN 1 END) as abnormal_count
|
||
FROM save_battery_data
|
||
WHERE batch_id = ?
|
||
GROUP BY group_id
|
||
ORDER BY group_id ASC
|
||
''', (batch_id,))
|
||
|
||
groups = []
|
||
rows = cursor.fetchall()
|
||
print(rows) # 调试信息
|
||
|
||
for row in rows:
|
||
group_data = dict(row)
|
||
# 处理平均值，保留4位小数
|
||
for key in ['avg_h2', 'avg_o2', 'avg_thickness']:
|
||
if group_data[key] is not None:
|
||
group_data[key] = round(group_data[key], 4)
|
||
groups.append(group_data)
|
||
|
||
logger.info(f"查询批次 {batch_id} 的组成功，共 {len(groups)} 个组")
|
||
return groups
|
||
|
||
except Exception as e:
|
||
if group_id is not None:
|
||
logger.error(f"查询批次 {batch_id} 组 {group_id} 数据失败: {e}")
|
||
else:
|
||
logger.error(f"查询批次 {batch_id} 组信息失败: {e}")
|
||
return []
|
||
|
||
def get_group_by_qr_codes_exact(self, batch_id: int, top_qr_code: str = None, bottom_qr_code: str = None) -> List[Dict]:
|
||
"""通过上面二维码和/或下面二维码精确查找对应的组信息
|
||
|
||
Args:
|
||
batch_id: 批次ID
|
||
top_qr_code: 上面二维码（可选）
|
||
bottom_qr_code: 下面二维码（可选）
|
||
|
||
Returns:
|
||
包含组信息的字典列表
|
||
"""
|
||
try:
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
# 构建查询条件
|
||
conditions = ["batch_id = ?"]
|
||
params = [batch_id]
|
||
|
||
if top_qr_code and top_qr_code.strip():
|
||
conditions.append("top_qr_code = ?")
|
||
params.append(top_qr_code.strip())
|
||
|
||
if bottom_qr_code and bottom_qr_code.strip():
|
||
conditions.append("bottom_qr_code = ?")
|
||
params.append(bottom_qr_code.strip())
|
||
|
||
# 如果两个二维码都没有提供，返回空列表
|
||
if not top_qr_code and not bottom_qr_code:
|
||
return []
|
||
|
||
query = f'''
|
||
SELECT DISTINCT group_id
|
||
FROM save_battery_data
|
||
WHERE {' AND '.join(conditions)}
|
||
ORDER BY group_id ASC
|
||
'''
|
||
|
||
cursor.execute(query, params)
|
||
results = [dict(row) for row in cursor.fetchall()]
|
||
|
||
logger.info(f"通过二维码精确查询批次 {batch_id} 成功，找到 {len(results)} 个匹配的组")
|
||
return results
|
||
|
||
except Exception as e:
|
||
logger.error(f"通过二维码精确查询组信息失败: {e}")
|
||
return []
|
||
# # 使用示例和测试
|
||
# def main():
|
||
# # 创建数据库实例
|
||
# db = BatteryDatabase("../battery_data.db")
|
||
# full_start_time = "2025-11-06 11:38:51"
|
||
# full_end_time = "2025-11-06 11:38:51"
|
||
# batches = db.get_batches_by_time_range_with_daily_sort(full_start_time,full_end_time)
|
||
# print(batches)
|
||
# if __name__ == "__main__":
|
||
# main()
|