|
|
#!/usr/bin/env python
|
|
|
# -*- coding: utf-8 -*-
|
|
|
import sys
|
|
|
import os
|
|
|
import time
|
|
|
import csv
|
|
|
import sqlite3
|
|
|
from datetime import datetime
|
|
|
import numpy as np
|
|
|
import joblib
|
|
|
import librosa
|
|
|
import pyaudio
|
|
|
import wave
|
|
|
from scipy.signal import hilbert
|
|
|
|
|
|
from PyQt5.QtWidgets import (
|
|
|
QApplication, QMainWindow, QPushButton, QLabel,
|
|
|
QLineEdit, QTextEdit, QFileDialog, QVBoxLayout,
|
|
|
QHBoxLayout, QWidget, QProgressBar, QStackedWidget,
|
|
|
QComboBox, QTableWidget, QTableWidgetItem, QMessageBox,
|
|
|
QDialog, QFormLayout, QDialogButtonBox
|
|
|
)
|
|
|
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 本地 SQLite 数据库（含 CRUD）
|
|
|
# ======================
|
|
|
class DatabaseManager:
|
|
|
def __init__(self, db_path="results.db"):
|
|
|
self.db_path = db_path
|
|
|
self._ensure_schema()
|
|
|
|
|
|
def _connect(self):
|
|
|
return sqlite3.connect(self.db_path)
|
|
|
|
|
|
def _ensure_schema(self):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
CREATE TABLE IF NOT EXISTS runs (
|
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
|
filename TEXT,
|
|
|
segments INTEGER,
|
|
|
mean_prob REAL,
|
|
|
final_label INTEGER,
|
|
|
created_at TEXT
|
|
|
);
|
|
|
""")
|
|
|
cur.execute("""
|
|
|
CREATE TABLE IF NOT EXISTS run_segments (
|
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
|
run_id INTEGER,
|
|
|
seg_index INTEGER,
|
|
|
label INTEGER,
|
|
|
proba REAL,
|
|
|
FOREIGN KEY(run_id) REFERENCES runs(id)
|
|
|
);
|
|
|
""")
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
|
|
|
# ---------- 原有写入 ----------
|
|
|
def insert_result(self, result_dict):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute(
|
|
|
"INSERT INTO runs(filename, segments, mean_prob, final_label, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
|
(
|
|
|
result_dict.get("filename"),
|
|
|
int(result_dict.get("segments", 0)),
|
|
|
float(result_dict.get("mean_prob", 0.0)),
|
|
|
int(result_dict.get("final_label", 0)),
|
|
|
datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
)
|
|
|
)
|
|
|
run_id = cur.lastrowid
|
|
|
preds = result_dict.get("predictions", [])
|
|
|
probas = result_dict.get("probabilities", [])
|
|
|
for i, (lab, pr) in enumerate(zip(preds, probas), start=1):
|
|
|
cur.execute(
|
|
|
"INSERT INTO run_segments(run_id, seg_index, label, proba) VALUES (?, ?, ?, ?)",
|
|
|
(run_id, i, int(lab), float(pr))
|
|
|
)
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
return run_id
|
|
|
|
|
|
# ---------- R（查询列表/单条） ----------
|
|
|
def fetch_recent_runs(self, limit=50):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
SELECT id, created_at, filename, segments, mean_prob, final_label
|
|
|
FROM runs ORDER BY id DESC LIMIT ?
|
|
|
""", (limit,))
|
|
|
rows = cur.fetchall()
|
|
|
con.close()
|
|
|
return rows
|
|
|
|
|
|
def get_run(self, run_id: int):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
SELECT id, created_at, filename, segments, mean_prob, final_label
|
|
|
FROM runs WHERE id = ?
|
|
|
""", (run_id,))
|
|
|
row = cur.fetchone()
|
|
|
con.close()
|
|
|
return row
|
|
|
|
|
|
def search_runs(self, keyword: str, limit=100):
|
|
|
kw = f"%{keyword}%"
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
SELECT id, created_at, filename, segments, mean_prob, final_label
|
|
|
FROM runs
|
|
|
WHERE filename LIKE ?
|
|
|
ORDER BY id DESC LIMIT ?
|
|
|
""", (kw, limit))
|
|
|
rows = cur.fetchall()
|
|
|
con.close()
|
|
|
return rows
|
|
|
|
|
|
# ---------- C/U/D（run） ----------
|
|
|
def create_run(self, filename: str, segments: int, mean_prob: float, final_label: int, created_at: str = None):
|
|
|
created_at = created_at or datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute(
|
|
|
"INSERT INTO runs(filename, segments, mean_prob, final_label, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
|
(filename, int(segments), float(mean_prob), int(final_label), created_at)
|
|
|
)
|
|
|
run_id = cur.lastrowid
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
return run_id
|
|
|
|
|
|
def update_run(self, run_id: int, filename: str, segments: int, mean_prob: float, final_label: int, created_at: str):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
UPDATE runs
|
|
|
SET filename=?, segments=?, mean_prob=?, final_label=?, created_at=?
|
|
|
WHERE id=?
|
|
|
""", (filename, int(segments), float(mean_prob), int(final_label), created_at, int(run_id)))
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
|
|
|
def delete_run(self, run_id: int):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
# 先删子表
|
|
|
cur.execute("DELETE FROM run_segments WHERE run_id=?", (int(run_id),))
|
|
|
cur.execute("DELETE FROM runs WHERE id=?", (int(run_id),))
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
|
|
|
# ---------- C/R/U/D（segments） ----------
|
|
|
def list_segments(self, run_id: int):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
SELECT id, seg_index, label, proba FROM run_segments
|
|
|
WHERE run_id=? ORDER BY seg_index ASC
|
|
|
""", (int(run_id),))
|
|
|
rows = cur.fetchall()
|
|
|
con.close()
|
|
|
return rows
|
|
|
|
|
|
def add_segment(self, run_id: int, seg_index: int, label: int, proba: float):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute(
|
|
|
"INSERT INTO run_segments(run_id, seg_index, label, proba) VALUES (?, ?, ?, ?)",
|
|
|
(int(run_id), int(seg_index), int(label), float(proba))
|
|
|
)
|
|
|
seg_id = cur.lastrowid
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
return seg_id
|
|
|
|
|
|
def update_segment(self, seg_id: int, seg_index: int, label: int, proba: float):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("""
|
|
|
UPDATE run_segments SET seg_index=?, label=?, proba=? WHERE id=?
|
|
|
""", (int(seg_index), int(label), float(proba), int(seg_id)))
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
|
|
|
def delete_segment(self, seg_id: int):
|
|
|
con = self._connect()
|
|
|
cur = con.cursor()
|
|
|
cur.execute("DELETE FROM run_segments WHERE id=?", (int(seg_id),))
|
|
|
con.commit()
|
|
|
con.close()
|
|
|
|
|
|
# ---------- 导出 ----------
|
|
|
def export_runs_to_csv(self, csv_path="results_export.csv", limit=1000):
|
|
|
rows = self.fetch_recent_runs(limit)
|
|
|
headers = ["id", "created_at", "filename", "segments", "mean_prob", "final_label"]
|
|
|
with open(csv_path, "w", newline="", encoding="utf-8") as f:
|
|
|
writer = csv.writer(f)
|
|
|
writer.writerow(headers)
|
|
|
for r in rows:
|
|
|
writer.writerow(r)
|
|
|
return csv_path
|
|
|
|
|
|
|
|
|
DB = DatabaseManager("results.db")
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 特征提取（同前）
|
|
|
# ======================
|
|
|
class FeatureExtractor:
|
|
|
WIN = 1024
|
|
|
OVERLAP = 512
|
|
|
THRESHOLD = 0.1
|
|
|
SEG_LEN_S = 0.2
|
|
|
STEP = WIN - OVERLAP
|
|
|
|
|
|
@staticmethod
|
|
|
def frame_energy(signal: np.ndarray) -> np.ndarray:
|
|
|
frames = librosa.util.frame(signal, frame_length=FeatureExtractor.WIN, hop_length=FeatureExtractor.STEP)
|
|
|
return np.sum(frames ** 2, axis=0)
|
|
|
|
|
|
@staticmethod
|
|
|
def detect_hits(energy: np.ndarray) -> np.ndarray:
|
|
|
idx = np.where(energy > FeatureExtractor.THRESHOLD)[0]
|
|
|
if idx.size == 0:
|
|
|
return np.array([])
|
|
|
flags = np.diff(np.concatenate(([0], idx))) > 5
|
|
|
return idx[flags]
|
|
|
|
|
|
@staticmethod
|
|
|
def extract_segments(signal: np.ndarray, sr: int, hit_starts: np.ndarray):
|
|
|
seg_len = int(FeatureExtractor.SEG_LEN_S * sr)
|
|
|
segments = []
|
|
|
for frame_idx in hit_starts:
|
|
|
s = frame_idx * FeatureExtractor.STEP
|
|
|
e = min(s + seg_len, len(signal))
|
|
|
segments.append(signal[s:e])
|
|
|
return segments
|
|
|
|
|
|
@staticmethod
|
|
|
def extract_features(x: np.ndarray, sr: int) -> np.ndarray:
|
|
|
x = x.flatten()
|
|
|
if len(x) == 0:
|
|
|
return np.zeros(5)
|
|
|
rms = np.sqrt(np.mean(x ** 2))
|
|
|
fft = np.fft.fft(x)
|
|
|
freq = np.fft.fftfreq(len(x), 1 / sr)
|
|
|
positive = freq >= 0
|
|
|
freq, fft_mag = freq[positive], np.abs(fft[positive])
|
|
|
main_freq = freq[np.argmax(fft_mag)] if len(fft_mag) > 0 else 0
|
|
|
spec_power = fft_mag
|
|
|
centroid = np.sum(freq * spec_power) / (np.sum(spec_power) + 1e-12)
|
|
|
spread = np.sqrt(np.sum(((freq - centroid) ** 2) * spec_power) / (np.sum(spec_power) + 1e-12))
|
|
|
skewness = np.sum(((freq - centroid) ** 3) * spec_power) / (
|
|
|
(np.sum(spec_power) + 1e-12) * (spread ** 3 + 1e-12)) if (spread > 0) else 0
|
|
|
mfcc = librosa.feature.mfcc(y=x, sr=sr, n_mfcc=13)
|
|
|
mfcc_mean = np.mean(mfcc[0]) if mfcc.size > 0 else 0
|
|
|
env_peak = np.max(np.abs(hilbert(x)))
|
|
|
return np.array([rms, main_freq, skewness, mfcc_mean, env_peak])
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 录音线程（同前）
|
|
|
# ======================
|
|
|
class RecordThread(QThread):
|
|
|
update_signal = pyqtSignal(str)
|
|
|
finish_signal = pyqtSignal(str)
|
|
|
progress_signal = pyqtSignal(int)
|
|
|
level_signal = pyqtSignal(int)
|
|
|
|
|
|
def __init__(self, max_duration=60, device_index=None):
|
|
|
super().__init__()
|
|
|
self.max_duration = max_duration
|
|
|
self.is_recording = False
|
|
|
self.temp_file = "temp_recording.wav"
|
|
|
self.device_index = device_index
|
|
|
|
|
|
def run(self):
|
|
|
FORMAT = pyaudio.paInt16
|
|
|
CHANNELS = 1
|
|
|
RATE = 44100
|
|
|
CHUNK = 1024
|
|
|
p, stream = None, None
|
|
|
try:
|
|
|
p = pyaudio.PyAudio()
|
|
|
stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True,
|
|
|
input_device_index=self.device_index, frames_per_buffer=CHUNK)
|
|
|
self.is_recording = True
|
|
|
start_time = time.time()
|
|
|
frames = []
|
|
|
max_int16 = 32767.0
|
|
|
while self.is_recording:
|
|
|
elapsed = time.time() - start_time
|
|
|
if elapsed >= self.max_duration:
|
|
|
self.update_signal.emit(f"已达最大时长 {self.max_duration} 秒")
|
|
|
break
|
|
|
self.progress_signal.emit(int((elapsed / self.max_duration) * 100))
|
|
|
data = stream.read(CHUNK, exception_on_overflow=False)
|
|
|
frames.append(data)
|
|
|
chunk_np = np.frombuffer(data, dtype=np.int16).astype(np.float32) / max_int16
|
|
|
rms = np.sqrt(np.mean(chunk_np ** 2))
|
|
|
self.level_signal.emit(int(np.clip(rms * 500, 0, 100)))
|
|
|
stream.stop_stream(); stream.close(); p.terminate()
|
|
|
if frames:
|
|
|
wf = wave.open(self.temp_file, 'wb')
|
|
|
wf.setnchannels(CHANNELS)
|
|
|
wf.setsampwidth(pyaudio.PyAudio().get_sample_size(FORMAT))
|
|
|
wf.setframerate(RATE)
|
|
|
wf.writeframes(b''.join(frames))
|
|
|
wf.close()
|
|
|
self.finish_signal.emit(self.temp_file)
|
|
|
else:
|
|
|
self.update_signal.emit("未录到有效音频。")
|
|
|
except Exception as e:
|
|
|
self.update_signal.emit(f"录制错误: {e}")
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 推理处理线程（写入 DB）
|
|
|
# ======================
|
|
|
class ProcessThread(QThread):
|
|
|
update_signal = pyqtSignal(str)
|
|
|
finish_signal = pyqtSignal(dict)
|
|
|
|
|
|
def __init__(self, wav_path, model_path, scaler_path):
|
|
|
super().__init__()
|
|
|
self.wav_path, self.model_path, self.scaler_path = wav_path, model_path, scaler_path
|
|
|
|
|
|
def run(self):
|
|
|
try:
|
|
|
self.update_signal.emit("加载模型和标准化器...")
|
|
|
model, scaler = joblib.load(self.model_path), joblib.load(self.scaler_path)
|
|
|
self.update_signal.emit(f"读取音频: {os.path.basename(self.wav_path)}")
|
|
|
sig, sr = librosa.load(self.wav_path, sr=None, mono=True)
|
|
|
sig = sig / max(np.max(np.abs(sig)), 1e-9)
|
|
|
ene = FeatureExtractor.frame_energy(sig)
|
|
|
hits = FeatureExtractor.detect_hits(ene)
|
|
|
segs = FeatureExtractor.extract_segments(sig, sr, hits)
|
|
|
if not segs:
|
|
|
self.update_signal.emit("未检测到有效片段！")
|
|
|
return
|
|
|
feats = [FeatureExtractor.extract_features(seg, sr) for seg in segs]
|
|
|
X = np.vstack(feats)
|
|
|
X_std = scaler.transform(X)
|
|
|
y_pred = model.predict(X_std)
|
|
|
if hasattr(model, "predict_proba"):
|
|
|
y_proba = model.predict_proba(X_std)[:, 1]
|
|
|
else:
|
|
|
y_proba = (y_pred.astype(float) + 0.0)
|
|
|
result = {
|
|
|
"filename": os.path.basename(self.wav_path),
|
|
|
"segments": len(segs),
|
|
|
"predictions": y_pred.tolist(),
|
|
|
"probabilities": [round(float(p), 4) for p in y_proba],
|
|
|
"mean_prob": round(float(np.mean(y_proba)), 4),
|
|
|
"final_label": int(np.mean(y_proba) >= 0.5)
|
|
|
}
|
|
|
DB.insert_result(result)
|
|
|
self.finish_signal.emit(result)
|
|
|
except Exception as e:
|
|
|
self.update_signal.emit(f"错误: {e}")
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 对话框：编辑 run
|
|
|
# ======================
|
|
|
class RunEditDialog(QDialog):
|
|
|
def __init__(self, parent=None, run_row=None):
|
|
|
"""
|
|
|
run_row: (id, created_at, filename, segments, mean_prob, final_label) 或 None（新增）
|
|
|
"""
|
|
|
super().__init__(parent)
|
|
|
self.setWindowTitle("编辑记录" if run_row else "新增记录")
|
|
|
self.resize(380, 240)
|
|
|
self.run_row = run_row
|
|
|
|
|
|
self.ed_id = QLineEdit(); self.ed_id.setReadOnly(True)
|
|
|
self.ed_created = QLineEdit(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
|
|
self.ed_filename = QLineEdit()
|
|
|
self.ed_segments = QLineEdit("0")
|
|
|
self.ed_mean_prob = QLineEdit("0.0")
|
|
|
self.ed_final_label = QLineEdit("0") # 0=实心 1=空心
|
|
|
|
|
|
form = QFormLayout()
|
|
|
form.addRow("ID：", self.ed_id)
|
|
|
form.addRow("时间(created_at)：", self.ed_created)
|
|
|
form.addRow("文件名(filename)：", self.ed_filename)
|
|
|
form.addRow("片段数(segments)：", self.ed_segments)
|
|
|
form.addRow("平均概率(mean_prob)：", self.ed_mean_prob)
|
|
|
form.addRow("最终标签(0=实 1=空)：", self.ed_final_label)
|
|
|
|
|
|
if run_row:
|
|
|
self.ed_id.setText(str(run_row[0]))
|
|
|
self.ed_created.setText(str(run_row[1]))
|
|
|
self.ed_filename.setText(str(run_row[2]))
|
|
|
self.ed_segments.setText(str(run_row[3]))
|
|
|
self.ed_mean_prob.setText(str(run_row[4]))
|
|
|
self.ed_final_label.setText(str(run_row[5]))
|
|
|
|
|
|
btns = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
|
|
|
btns.accepted.connect(self.accept)
|
|
|
btns.rejected.connect(self.reject)
|
|
|
|
|
|
lay = QVBoxLayout()
|
|
|
lay.addLayout(form)
|
|
|
lay.addWidget(btns)
|
|
|
self.setLayout(lay)
|
|
|
|
|
|
def values(self):
|
|
|
return dict(
|
|
|
id=(int(self.ed_id.text()) if self.ed_id.text().strip() else None),
|
|
|
created_at=self.ed_created.text().strip(),
|
|
|
filename=self.ed_filename.text().strip(),
|
|
|
segments=int(float(self.ed_segments.text() or 0)),
|
|
|
mean_prob=float(self.ed_mean_prob.text() or 0.0),
|
|
|
final_label=int(float(self.ed_final_label.text() or 0))
|
|
|
)
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 对话框：管理 segments
|
|
|
# ======================
|
|
|
class SegmentManagerDialog(QDialog):
|
|
|
def __init__(self, parent=None, run_id=None):
|
|
|
super().__init__(parent)
|
|
|
self.setWindowTitle(f"片段管理 - run_id={run_id}")
|
|
|
self.resize(560, 420)
|
|
|
self.run_id = run_id
|
|
|
|
|
|
self.table = QTableWidget(0, 4)
|
|
|
self.table.setHorizontalHeaderLabels(["ID", "seg_index", "label", "proba"])
|
|
|
|
|
|
self.btn_add = QPushButton("新增片段")
|
|
|
self.btn_edit = QPushButton("编辑所选")
|
|
|
self.btn_del = QPushButton("删除所选")
|
|
|
self.btn_close = QPushButton("关闭")
|
|
|
|
|
|
hl = QHBoxLayout()
|
|
|
for b in (self.btn_add, self.btn_edit, self.btn_del, self.btn_close):
|
|
|
hl.addWidget(b)
|
|
|
|
|
|
lay = QVBoxLayout()
|
|
|
lay.addWidget(self.table)
|
|
|
lay.addLayout(hl)
|
|
|
self.setLayout(lay)
|
|
|
|
|
|
self.btn_add.clicked.connect(self.add_segment)
|
|
|
self.btn_edit.clicked.connect(self.edit_selected)
|
|
|
self.btn_del.clicked.connect(self.delete_selected)
|
|
|
self.btn_close.clicked.connect(self.accept)
|
|
|
|
|
|
self.load_segments()
|
|
|
|
|
|
def load_segments(self):
|
|
|
rows = DB.list_segments(self.run_id)
|
|
|
self.table.setRowCount(len(rows))
|
|
|
for i, (sid, idx, label, proba) in enumerate(rows):
|
|
|
self.table.setItem(i, 0, QTableWidgetItem(str(sid)))
|
|
|
self.table.setItem(i, 1, QTableWidgetItem(str(idx)))
|
|
|
self.table.setItem(i, 2, QTableWidgetItem(str(label)))
|
|
|
self.table.setItem(i, 3, QTableWidgetItem(str(proba)))
|
|
|
self.table.resizeColumnsToContents()
|
|
|
|
|
|
def _get_selected_seg_id(self):
|
|
|
r = self.table.currentRow()
|
|
|
if r < 0:
|
|
|
return None
|
|
|
return int(self.table.item(r, 0).text())
|
|
|
|
|
|
def add_segment(self):
|
|
|
d = QDialog(self)
|
|
|
d.setWindowTitle("新增片段")
|
|
|
form = QFormLayout()
|
|
|
ed_index = QLineEdit("1")
|
|
|
ed_label = QLineEdit("0")
|
|
|
ed_proba = QLineEdit("0.0")
|
|
|
form.addRow("seg_index：", ed_index)
|
|
|
form.addRow("label(0/1)：", ed_label)
|
|
|
form.addRow("proba(0~1)：", ed_proba)
|
|
|
btns = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
|
|
|
btns.accepted.connect(d.accept); btns.rejected.connect(d.reject)
|
|
|
v = QVBoxLayout(); v.addLayout(form); v.addWidget(btns); d.setLayout(v)
|
|
|
if d.exec_() == QDialog.Accepted:
|
|
|
DB.add_segment(self.run_id, int(float(ed_index.text() or 1)), int(float(ed_label.text() or 0)),
|
|
|
float(ed_proba.text() or 0.0))
|
|
|
self.load_segments()
|
|
|
|
|
|
def edit_selected(self):
|
|
|
seg_id = self._get_selected_seg_id()
|
|
|
if not seg_id:
|
|
|
QMessageBox.information(self, "提示", "请先选择一行片段。")
|
|
|
return
|
|
|
# 读当前值
|
|
|
row = self.table.currentRow()
|
|
|
cur_idx = self.table.item(row, 1).text()
|
|
|
cur_label = self.table.item(row, 2).text()
|
|
|
cur_proba = self.table.item(row, 3).text()
|
|
|
|
|
|
d = QDialog(self)
|
|
|
d.setWindowTitle("编辑片段")
|
|
|
form = QFormLayout()
|
|
|
ed_index = QLineEdit(cur_idx)
|
|
|
ed_label = QLineEdit(cur_label)
|
|
|
ed_proba = QLineEdit(cur_proba)
|
|
|
form.addRow("seg_index：", ed_index)
|
|
|
form.addRow("label(0/1)：", ed_label)
|
|
|
form.addRow("proba(0~1)：", ed_proba)
|
|
|
btns = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
|
|
|
btns.accepted.connect(d.accept); btns.rejected.connect(d.reject)
|
|
|
v = QVBoxLayout(); v.addLayout(form); v.addWidget(btns); d.setLayout(v)
|
|
|
if d.exec_() == QDialog.Accepted:
|
|
|
DB.update_segment(seg_id, int(float(ed_index.text() or cur_idx)),
|
|
|
int(float(ed_label.text() or cur_label)),
|
|
|
float(ed_proba.text() or cur_proba))
|
|
|
self.load_segments()
|
|
|
|
|
|
def delete_selected(self):
|
|
|
seg_id = self._get_selected_seg_id()
|
|
|
if not seg_id:
|
|
|
QMessageBox.information(self, "提示", "请先选择一行片段。")
|
|
|
return
|
|
|
if QMessageBox.question(self, "确认", "确认删除该片段？此操作不可恢复！",
|
|
|
QMessageBox.Yes | QMessageBox.No) == QMessageBox.Yes:
|
|
|
DB.delete_segment(seg_id)
|
|
|
self.load_segments()
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 主菜单
|
|
|
# ======================
|
|
|
class MainMenuWidget(QWidget):
|
|
|
def __init__(self, parent):
|
|
|
super().__init__(parent)
|
|
|
layout = QVBoxLayout()
|
|
|
title = QLabel("音频分类器"); title.setAlignment(Qt.AlignCenter)
|
|
|
title.setStyleSheet("font-size: 24px; font-weight: bold; margin: 20px;")
|
|
|
layout.addWidget(title)
|
|
|
btn1 = QPushButton("采集音频"); btn1.clicked.connect(lambda: parent.switch_to_input("record"))
|
|
|
btn2 = QPushButton("上传WAV文件"); btn2.clicked.connect(lambda: parent.switch_to_input("upload"))
|
|
|
btn3 = QPushButton("历史记录 / 导出 / CRUD"); btn3.clicked.connect(parent.switch_to_history)
|
|
|
for b in [btn1, btn2, btn3]:
|
|
|
b.setMinimumHeight(60); layout.addWidget(b)
|
|
|
layout.addStretch(1); self.setLayout(layout)
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 输入界面（录音 / 上传）
|
|
|
# ======================
|
|
|
class InputWidget(QWidget):
|
|
|
def __init__(self, parent, mode="record"):
|
|
|
super().__init__(parent)
|
|
|
self.parent, self.mode = parent, mode
|
|
|
self.wav_path, self.record_thread = "", None
|
|
|
self.device_index = None
|
|
|
self.init_ui()
|
|
|
|
|
|
def init_ui(self):
|
|
|
layout = QVBoxLayout()
|
|
|
back = QPushButton("返回"); back.clicked.connect(self.parent.switch_to_main_menu)
|
|
|
layout.addWidget(back)
|
|
|
if self.mode == "record": self.setup_record(layout)
|
|
|
else: self.setup_upload(layout)
|
|
|
self.model_path, self.scaler_path = QLineEdit("svm_model.pkl"), QLineEdit("scaler.pkl")
|
|
|
path_layout = QHBoxLayout()
|
|
|
path_layout.addWidget(QLabel("模型:")); path_layout.addWidget(self.model_path)
|
|
|
path_layout.addWidget(QLabel("标准化器:")); path_layout.addWidget(self.scaler_path)
|
|
|
layout.addLayout(path_layout)
|
|
|
self.process_btn = QPushButton("开始处理"); self.process_btn.setEnabled(False)
|
|
|
self.process_btn.clicked.connect(self.start_process)
|
|
|
layout.addWidget(self.process_btn)
|
|
|
self.log = QTextEdit(); self.log.setReadOnly(True); layout.addWidget(self.log)
|
|
|
self.setLayout(layout)
|
|
|
|
|
|
def setup_record(self, layout):
|
|
|
title = QLabel("音频采集"); title.setAlignment(Qt.AlignCenter)
|
|
|
layout.addWidget(title)
|
|
|
dev_layout = QHBoxLayout(); dev_layout.addWidget(QLabel("设备:"))
|
|
|
self.dev_combo = QComboBox(); self.refresh_devices()
|
|
|
self.dev_combo.currentIndexChanged.connect(lambda: setattr(self, "device_index", self.dev_combo.currentData()))
|
|
|
dev_layout.addWidget(self.dev_combo)
|
|
|
btn_ref = QPushButton("刷新设备"); btn_ref.clicked.connect(self.refresh_devices)
|
|
|
dev_layout.addWidget(btn_ref); layout.addLayout(dev_layout)
|
|
|
self.rec_btn = QPushButton("按住录音"); self.rec_btn.setMinimumHeight(80)
|
|
|
self.rec_btn.mousePressEvent = self.start_rec; self.rec_btn.mouseReleaseEvent = self.stop_rec
|
|
|
self.level, self.progress, self.dur = QProgressBar(), QProgressBar(), QLabel("0s")
|
|
|
self.level.setFormat("电平:%p%"); self.progress.setRange(0,100)
|
|
|
for w in [self.rec_btn,self.level,self.progress,self.dur]: layout.addWidget(w)
|
|
|
self.timer = QTimer(self); self.timer.timeout.connect(self.update_dur)
|
|
|
|
|
|
def setup_upload(self, layout):
|
|
|
h = QHBoxLayout(); self.file = QLineEdit(); self.file.setReadOnly(True)
|
|
|
b = QPushButton("浏览"); b.clicked.connect(self.browse)
|
|
|
h.addWidget(self.file); h.addWidget(b); layout.addLayout(h)
|
|
|
|
|
|
def refresh_devices(self):
|
|
|
self.dev_combo.clear(); p = pyaudio.PyAudio()
|
|
|
for i in range(p.get_device_count()):
|
|
|
info = p.get_device_info_by_index(i)
|
|
|
if int(info.get('maxInputChannels',0))>0:
|
|
|
self.dev_combo.addItem(f"[{i}] {info.get('name')}", i)
|
|
|
if self.dev_combo.count()>0:
|
|
|
self.dev_combo.setCurrentIndex(0)
|
|
|
self.device_index = self.dev_combo.currentData()
|
|
|
p.terminate()
|
|
|
|
|
|
def start_rec(self,e):
|
|
|
if e.button()==Qt.LeftButton:
|
|
|
self.record_thread=RecordThread(device_index=self.device_index)
|
|
|
self.record_thread.update_signal.connect(self.log.append)
|
|
|
self.record_thread.finish_signal.connect(self.finish_rec)
|
|
|
self.record_thread.progress_signal.connect(self.progress.setValue)
|
|
|
self.record_thread.level_signal.connect(self.level.setValue)
|
|
|
self.record_thread.start(); self.start=time.time(); self.timer.start(100)
|
|
|
|
|
|
def stop_rec(self,e):
|
|
|
if e.button()==Qt.LeftButton and self.record_thread:
|
|
|
self.record_thread.is_recording=False; self.timer.stop(); self.progress.setValue(0); self.level.setValue(0)
|
|
|
|
|
|
def update_dur(self): self.dur.setText(f"{time.time()-self.start:.1f}s")
|
|
|
|
|
|
def finish_rec(self,file): self.wav_path=file; self.process_btn.setEnabled(True)
|
|
|
|
|
|
def browse(self):
|
|
|
f,_=QFileDialog.getOpenFileName(self,"选择WAV","","WAV文件 (*.wav)")
|
|
|
if f: self.file.setText(f); self.wav_path=f; self.process_btn.setEnabled(True)
|
|
|
|
|
|
def start_process(self):
|
|
|
if self.mode=="upload": self.wav_path=self.file.text()
|
|
|
if not os.path.exists(self.wav_path): self.log.append("无效文件"); return
|
|
|
self.thread=ProcessThread(self.wav_path,self.model_path.text(),self.scaler_path.text())
|
|
|
self.thread.update_signal.connect(self.log.append)
|
|
|
self.thread.finish_signal.connect(self.parent.switch_to_result)
|
|
|
self.thread.start()
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 历史记录（含 CRUD 按钮）
|
|
|
# ======================
|
|
|
class HistoryWidget(QWidget):
|
|
|
def __init__(self, parent):
|
|
|
super().__init__(parent)
|
|
|
self.parent=parent
|
|
|
self.init_ui()
|
|
|
|
|
|
def init_ui(self):
|
|
|
l=QVBoxLayout()
|
|
|
back=QPushButton("返回"); back.clicked.connect(self.parent.switch_to_main_menu); l.addWidget(back)
|
|
|
|
|
|
top=QHBoxLayout()
|
|
|
self.search_edit=QLineEdit(); self.search_edit.setPlaceholderText("按文件名搜索...")
|
|
|
btn_search=QPushButton("搜索"); btn_search.clicked.connect(self.search)
|
|
|
btn_refresh=QPushButton("刷新"); btn_refresh.clicked.connect(self.load)
|
|
|
btn_export=QPushButton("导出CSV"); btn_export.clicked.connect(self.export_csv)
|
|
|
btn_add=QPushButton("新增记录"); btn_add.clicked.connect(self.add_run)
|
|
|
btn_edit=QPushButton("编辑所选"); btn_edit.clicked.connect(self.edit_selected)
|
|
|
btn_del=QPushButton("删除所选"); btn_del.clicked.connect(self.delete_selected)
|
|
|
btn_seg=QPushButton("片段管理"); btn_seg.clicked.connect(self.open_segments)
|
|
|
for w in [self.search_edit, btn_search, btn_refresh, btn_export, btn_add, btn_edit, btn_del, btn_seg]:
|
|
|
top.addWidget(w)
|
|
|
l.addLayout(top)
|
|
|
|
|
|
self.table=QTableWidget(0,6)
|
|
|
self.table.setHorizontalHeaderLabels(["ID","时间","文件名","片段数","平均概率","最终标签"])
|
|
|
l.addWidget(self.table)
|
|
|
self.setLayout(l)
|
|
|
self.load()
|
|
|
|
|
|
def _current_run_id(self):
|
|
|
r=self.table.currentRow()
|
|
|
if r<0: return None
|
|
|
return int(self.table.item(r,0).text())
|
|
|
|
|
|
# ---- 数据加载 / 搜索 ----
|
|
|
def load(self):
|
|
|
rows=DB.fetch_recent_runs(200)
|
|
|
self._fill(rows)
|
|
|
|
|
|
def search(self):
|
|
|
kw=self.search_edit.text().strip()
|
|
|
rows=DB.search_runs(kw, limit=200) if kw else DB.fetch_recent_runs(200)
|
|
|
self._fill(rows)
|
|
|
|
|
|
def _fill(self, rows):
|
|
|
self.table.setRowCount(len(rows))
|
|
|
for r_idx, (rid, created_at, fn, segs, mean_prob, final_label) in enumerate(rows):
|
|
|
self.table.setItem(r_idx, 0, QTableWidgetItem(str(rid)))
|
|
|
self.table.setItem(r_idx, 1, QTableWidgetItem(str(created_at)))
|
|
|
self.table.setItem(r_idx, 2, QTableWidgetItem(str(fn)))
|
|
|
self.table.setItem(r_idx, 3, QTableWidgetItem(str(segs)))
|
|
|
self.table.setItem(r_idx, 4, QTableWidgetItem(str(mean_prob)))
|
|
|
self.table.setItem(r_idx, 5, QTableWidgetItem("空心" if int(final_label) else "实心"))
|
|
|
self.table.resizeColumnsToContents()
|
|
|
|
|
|
# ---- 导出 ----
|
|
|
def export_csv(self):
|
|
|
p=DB.export_runs_to_csv()
|
|
|
QMessageBox.information(self,"导出完成",f"已保存到 {os.path.abspath(p)}")
|
|
|
|
|
|
# ---- CRUD: run ----
|
|
|
def add_run(self):
|
|
|
dlg=RunEditDialog(self, run_row=None)
|
|
|
if dlg.exec_()==QDialog.Accepted:
|
|
|
v=dlg.values()
|
|
|
DB.create_run(v["filename"], v["segments"], v["mean_prob"], v["final_label"], v["created_at"])
|
|
|
self.load()
|
|
|
|
|
|
def edit_selected(self):
|
|
|
rid=self._current_run_id()
|
|
|
if not rid:
|
|
|
QMessageBox.information(self,"提示","请先选择一行记录。")
|
|
|
return
|
|
|
row=DB.get_run(rid)
|
|
|
dlg=RunEditDialog(self, run_row=row)
|
|
|
if dlg.exec_()==QDialog.Accepted:
|
|
|
v=dlg.values()
|
|
|
DB.update_run(rid, v["filename"], v["segments"], v["mean_prob"], v["final_label"], v["created_at"])
|
|
|
self.load()
|
|
|
|
|
|
def delete_selected(self):
|
|
|
rid=self._current_run_id()
|
|
|
if not rid:
|
|
|
QMessageBox.information(self,"提示","请先选择一行记录。")
|
|
|
return
|
|
|
if QMessageBox.question(self,"确认","确认删除所选记录及其所有片段？此操作不可恢复！",
|
|
|
QMessageBox.Yes|QMessageBox.No)==QMessageBox.Yes:
|
|
|
DB.delete_run(rid)
|
|
|
self.load()
|
|
|
|
|
|
# ---- 片段管理 ----
|
|
|
def open_segments(self):
|
|
|
rid=self._current_run_id()
|
|
|
if not rid:
|
|
|
QMessageBox.information(self,"提示","请先选择一行记录。")
|
|
|
return
|
|
|
dlg=SegmentManagerDialog(self, run_id=rid)
|
|
|
dlg.exec_()
|
|
|
# 片段变更后可按需同步更新片段数/平均概率，此处保持手动编辑为主
|
|
|
self.load()
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 结果页（简）
|
|
|
# ======================
|
|
|
class ResultWidget(QWidget):
|
|
|
def __init__(self, parent, result=None):
|
|
|
super().__init__(parent); self.parent=parent
|
|
|
layout=QVBoxLayout(); b=QPushButton("返回"); b.clicked.connect(self.parent.switch_to_main_menu); layout.addWidget(b)
|
|
|
t=QLabel("处理结果"); t.setAlignment(Qt.AlignCenter); layout.addWidget(t)
|
|
|
area=QTextEdit(); area.setReadOnly(True)
|
|
|
if result:
|
|
|
txt=f"文件:{result['filename']}\n片段:{result['segments']}\n平均概率:{result['mean_prob']}\n最终:{'空心' if result['final_label'] else '实心'}"
|
|
|
area.setText(txt)
|
|
|
layout.addWidget(area); self.setLayout(layout)
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 主窗口
|
|
|
# ======================
|
|
|
class AudioClassifierGUI(QMainWindow):
|
|
|
def __init__(self):
|
|
|
super().__init__()
|
|
|
self.current_input_mode = "record"
|
|
|
self.process_result = None
|
|
|
self.init_ui()
|
|
|
|
|
|
def init_ui(self):
|
|
|
self.setWindowTitle("音频分类器")
|
|
|
self.setGeometry(100, 100, 1000, 700)
|
|
|
|
|
|
self.stacked_widget = QStackedWidget()
|
|
|
self.setCentralWidget(self.stacked_widget)
|
|
|
|
|
|
self.main_menu_widget = MainMenuWidget(self)
|
|
|
self.record_input_widget = InputWidget(self, "record")
|
|
|
self.upload_input_widget = InputWidget(self, "upload")
|
|
|
self.result_widget = ResultWidget(self)
|
|
|
self.history_widget = HistoryWidget(self)
|
|
|
|
|
|
self.stacked_widget.addWidget(self.main_menu_widget) # 0
|
|
|
self.stacked_widget.addWidget(self.record_input_widget) # 1
|
|
|
self.stacked_widget.addWidget(self.upload_input_widget) # 2
|
|
|
self.stacked_widget.addWidget(self.result_widget) # 3
|
|
|
self.stacked_widget.addWidget(self.history_widget) # 4
|
|
|
|
|
|
self.stacked_widget.setCurrentWidget(self.main_menu_widget)
|
|
|
|
|
|
def switch_to_input(self, mode):
|
|
|
self.current_input_mode = mode
|
|
|
if mode == "record":
|
|
|
self.stacked_widget.setCurrentWidget(self.record_input_widget)
|
|
|
else:
|
|
|
self.stacked_widget.setCurrentWidget(self.upload_input_widget)
|
|
|
|
|
|
def switch_to_main_menu(self):
|
|
|
self.stacked_widget.setCurrentWidget(self.main_menu_widget)
|
|
|
|
|
|
def switch_to_result(self, result):
|
|
|
self.process_result = result
|
|
|
self.result_widget = ResultWidget(self, result)
|
|
|
self.stacked_widget.removeWidget(self.stacked_widget.widget(3))
|
|
|
self.stacked_widget.addWidget(self.result_widget)
|
|
|
self.stacked_widget.setCurrentWidget(self.result_widget)
|
|
|
|
|
|
def switch_to_history(self):
|
|
|
self.history_widget.load()
|
|
|
self.stacked_widget.setCurrentWidget(self.history_widget)
|
|
|
|
|
|
|
|
|
# ======================
|
|
|
# 上传/录音 界面（放最后，避免打断阅读）
|
|
|
# ======================
|
|
|
class InputWidget(QWidget):
|
|
|
def __init__(self, parent, mode="record"):
|
|
|
super().__init__(parent)
|
|
|
self.parent, self.mode = parent, mode
|
|
|
self.wav_path, self.record_thread = "", None
|
|
|
self.device_index = None
|
|
|
self.init_ui()
|
|
|
|
|
|
def init_ui(self):
|
|
|
layout = QVBoxLayout()
|
|
|
back = QPushButton("返回"); back.clicked.connect(self.parent.switch_to_main_menu)
|
|
|
layout.addWidget(back)
|
|
|
if self.mode == "record": self.setup_record(layout)
|
|
|
else: self.setup_upload(layout)
|
|
|
self.model_path, self.scaler_path = QLineEdit("svm_model.pkl"), QLineEdit("scaler.pkl")
|
|
|
path_layout = QHBoxLayout()
|
|
|
path_layout.addWidget(QLabel("模型:")); path_layout.addWidget(self.model_path)
|
|
|
path_layout.addWidget(QLabel("标准化器:")); path_layout.addWidget(self.scaler_path)
|
|
|
layout.addLayout(path_layout)
|
|
|
self.process_btn = QPushButton("开始处理"); self.process_btn.setEnabled(False)
|
|
|
self.process_btn.clicked.connect(self.start_process)
|
|
|
layout.addWidget(self.process_btn)
|
|
|
self.log = QTextEdit(); self.log.setReadOnly(True); layout.addWidget(self.log)
|
|
|
self.setLayout(layout)
|
|
|
|
|
|
def setup_record(self, layout):
|
|
|
title = QLabel("音频采集"); title.setAlignment(Qt.AlignCenter)
|
|
|
layout.addWidget(title)
|
|
|
dev_layout = QHBoxLayout(); dev_layout.addWidget(QLabel("设备:"))
|
|
|
self.dev_combo = QComboBox(); self.refresh_devices()
|
|
|
self.dev_combo.currentIndexChanged.connect(lambda: setattr(self, "device_index", self.dev_combo.currentData()))
|
|
|
dev_layout.addWidget(self.dev_combo)
|
|
|
btn_ref = QPushButton("刷新设备"); btn_ref.clicked.connect(self.refresh_devices)
|
|
|
dev_layout.addWidget(btn_ref); layout.addLayout(dev_layout)
|
|
|
self.rec_btn = QPushButton("按住录音"); self.rec_btn.setMinimumHeight(80)
|
|
|
self.rec_btn.mousePressEvent = self.start_rec; self.rec_btn.mouseReleaseEvent = self.stop_rec
|
|
|
self.level, self.progress, self.dur = QProgressBar(), QProgressBar(), QLabel("0s")
|
|
|
self.level.setFormat("电平:%p%"); self.progress.setRange(0,100)
|
|
|
for w in [self.rec_btn,self.level,self.progress,self.dur]: layout.addWidget(w)
|
|
|
self.timer = QTimer(self); self.timer.timeout.connect(self.update_dur)
|
|
|
|
|
|
def setup_upload(self, layout):
|
|
|
h = QHBoxLayout(); self.file = QLineEdit(); self.file.setReadOnly(True)
|
|
|
b = QPushButton("浏览"); b.clicked.connect(self.browse)
|
|
|
h.addWidget(self.file); h.addWidget(b); layout.addLayout(h)
|
|
|
|
|
|
def refresh_devices(self):
|
|
|
self.dev_combo.clear(); p = pyaudio.PyAudio()
|
|
|
for i in range(p.get_device_count()):
|
|
|
info = p.get_device_info_by_index(i)
|
|
|
if int(info.get('maxInputChannels',0))>0:
|
|
|
self.dev_combo.addItem(f"[{i}] {info.get('name')}", i)
|
|
|
if self.dev_combo.count()>0:
|
|
|
self.dev_combo.setCurrentIndex(0)
|
|
|
self.device_index = self.dev_combo.currentData()
|
|
|
p.terminate()
|
|
|
|
|
|
def start_rec(self,e):
|
|
|
if e.button()==Qt.LeftButton:
|
|
|
self.record_thread=RecordThread(device_index=self.device_index)
|
|
|
self.record_thread.update_signal.connect(self.log.append)
|
|
|
self.record_thread.finish_signal.connect(self.finish_rec)
|
|
|
self.record_thread.progress_signal.connect(self.progress.setValue)
|
|
|
self.record_thread.level_signal.connect(self.level.setValue)
|
|
|
self.record_thread.start(); self.start=time.time(); self.timer.start(100)
|
|
|
|
|
|
def stop_rec(self,e):
|
|
|
if e.button()==Qt.LeftButton and self.record_thread:
|
|
|
self.record_thread.is_recording=False; self.timer.stop(); self.progress.setValue(0); self.level.setValue(0)
|
|
|
|
|
|
def update_dur(self): self.dur.setText(f"{time.time()-self.start:.1f}s")
|
|
|
|
|
|
def finish_rec(self,file): self.wav_path=file; self.process_btn.setEnabled(True)
|
|
|
|
|
|
def browse(self):
|
|
|
f,_=QFileDialog.getOpenFileName(self,"选择WAV","","WAV文件 (*.wav)")
|
|
|
if f: self.file.setText(f); self.wav_path=f; self.process_btn.setEnabled(True)
|
|
|
|
|
|
def start_process(self):
|
|
|
if self.mode=="upload": self.wav_path=self.file.text()
|
|
|
if not os.path.exists(self.wav_path): self.log.append("无效文件"); return
|
|
|
self.thread=ProcessThread(self.wav_path,self.model_path.text(),self.scaler_path.text())
|
|
|
self.thread.update_signal.connect(self.log.append)
|
|
|
self.thread.finish_signal.connect(self.parent.switch_to_result)
|
|
|
self.thread.start()
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
try:
|
|
|
import pyaudio
|
|
|
except ImportError:
|
|
|
print("请先安装pyaudio: pip install pyaudio")
|
|
|
sys.exit(1)
|
|
|
|
|
|
app = QApplication(sys.argv)
|
|
|
window = AudioClassifierGUI()
|
|
|
window.show()
|
|
|
sys.exit(app.exec_())
|
