286 lines
9.7 KiB
Python
286 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
DATA_DIR = BASE_DIR / "data"
|
|
DB_PATH = DATA_DIR / "confivoice.db"
|
|
|
|
DB_DRIVER = os.getenv("CONFIVOICE_DB_DRIVER", "sqlite").lower()
|
|
MYSQL_HOST = os.getenv("CONFIVOICE_MYSQL_HOST", "127.0.0.1")
|
|
MYSQL_PORT = int(os.getenv("CONFIVOICE_MYSQL_PORT", "3306"))
|
|
MYSQL_USER = os.getenv("CONFIVOICE_MYSQL_USER", "root")
|
|
MYSQL_PASSWORD = os.getenv("CONFIVOICE_MYSQL_PASSWORD", "")
|
|
MYSQL_DATABASE = os.getenv("CONFIVOICE_MYSQL_DATABASE", "confivoice")
|
|
|
|
|
|
def get_database_label():
|
|
if DB_DRIVER == "mysql":
|
|
return f"mysql://{MYSQL_USER}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DATABASE}"
|
|
return str(DB_PATH)
|
|
|
|
|
|
def get_connection():
|
|
if DB_DRIVER == "mysql":
|
|
return get_mysql_connection()
|
|
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(DB_PATH)
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
|
|
def get_mysql_connection(database=MYSQL_DATABASE):
|
|
try:
|
|
import pymysql
|
|
except ImportError as error:
|
|
raise RuntimeError(
|
|
"PyMySQL belum terinstall. Jalankan: pip install PyMySQL"
|
|
) from error
|
|
|
|
return pymysql.connect(
|
|
host=MYSQL_HOST,
|
|
port=MYSQL_PORT,
|
|
user=MYSQL_USER,
|
|
password=MYSQL_PASSWORD,
|
|
database=database,
|
|
charset="utf8mb4",
|
|
cursorclass=pymysql.cursors.DictCursor,
|
|
autocommit=False,
|
|
)
|
|
|
|
|
|
def init_mysql_db():
|
|
with get_mysql_connection(database=None) as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
f"CREATE DATABASE IF NOT EXISTS `{MYSQL_DATABASE}` "
|
|
"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
|
|
)
|
|
connection.commit()
|
|
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS prediction_results (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
student_name VARCHAR(255) NOT NULL,
|
|
predicted_label VARCHAR(20),
|
|
description VARCHAR(255),
|
|
confidence DOUBLE NOT NULL DEFAULT 0,
|
|
probability_pd DOUBLE NOT NULL DEFAULT 0,
|
|
probability_tpd DOUBLE NOT NULL DEFAULT 0,
|
|
is_valid_audio TINYINT(1) NOT NULL DEFAULT 1,
|
|
error_message TEXT,
|
|
audio_duration DOUBLE NOT NULL DEFAULT 0,
|
|
volume_score DOUBLE NOT NULL DEFAULT 0,
|
|
intonation_score DOUBLE NOT NULL DEFAULT 0,
|
|
pause_score DOUBLE NOT NULL DEFAULT 0,
|
|
speech_activity_ratio DOUBLE NOT NULL DEFAULT 0,
|
|
silence_ratio DOUBLE NOT NULL DEFAULT 0,
|
|
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
connection.commit()
|
|
|
|
|
|
def init_db():
|
|
if DB_DRIVER == "mysql":
|
|
init_mysql_db()
|
|
return
|
|
|
|
with get_connection() as connection:
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS prediction_results (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
student_name TEXT NOT NULL,
|
|
predicted_label TEXT,
|
|
description TEXT,
|
|
confidence REAL NOT NULL DEFAULT 0,
|
|
probability_pd REAL NOT NULL DEFAULT 0,
|
|
probability_tpd REAL NOT NULL DEFAULT 0,
|
|
is_valid_audio INTEGER NOT NULL DEFAULT 1,
|
|
error_message TEXT,
|
|
audio_duration REAL NOT NULL DEFAULT 0,
|
|
volume_score REAL NOT NULL DEFAULT 0,
|
|
intonation_score REAL NOT NULL DEFAULT 0,
|
|
pause_score REAL NOT NULL DEFAULT 0,
|
|
speech_activity_ratio REAL NOT NULL DEFAULT 0,
|
|
silence_ratio REAL NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
connection.commit()
|
|
|
|
|
|
def save_prediction_result(student_name, result):
|
|
init_db()
|
|
audio_quality = result.get("audio_quality") or {}
|
|
indicators = result.get("voice_indicators") or {}
|
|
|
|
if DB_DRIVER == "mysql":
|
|
return save_prediction_result_mysql(student_name, result, audio_quality, indicators)
|
|
|
|
with get_connection() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO prediction_results (
|
|
student_name,
|
|
predicted_label,
|
|
description,
|
|
confidence,
|
|
probability_pd,
|
|
probability_tpd,
|
|
is_valid_audio,
|
|
error_message,
|
|
audio_duration,
|
|
volume_score,
|
|
intonation_score,
|
|
pause_score,
|
|
speech_activity_ratio,
|
|
silence_ratio
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
student_name,
|
|
result.get("predicted_label") or result.get("label"),
|
|
result.get("description"),
|
|
float(result.get("confidence") or 0),
|
|
float(result.get("probability_pd") or 0),
|
|
float(result.get("probability_tpd") or 0),
|
|
1 if result.get("is_valid_audio") is not False else 0,
|
|
result.get("error_message"),
|
|
float(audio_quality.get("duration") or 0),
|
|
float(indicators.get("volume_score") or 0),
|
|
float(indicators.get("intonation_score") or 0),
|
|
float(indicators.get("pause_score") or 0),
|
|
float(indicators.get("speech_activity_ratio") or 0),
|
|
float(indicators.get("silence_ratio") or 0),
|
|
),
|
|
)
|
|
connection.commit()
|
|
return cursor.lastrowid
|
|
|
|
|
|
def save_prediction_result_mysql(student_name, result, audio_quality, indicators):
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO prediction_results (
|
|
student_name,
|
|
predicted_label,
|
|
description,
|
|
confidence,
|
|
probability_pd,
|
|
probability_tpd,
|
|
is_valid_audio,
|
|
error_message,
|
|
audio_duration,
|
|
volume_score,
|
|
intonation_score,
|
|
pause_score,
|
|
speech_activity_ratio,
|
|
silence_ratio
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
""",
|
|
(
|
|
student_name,
|
|
result.get("predicted_label") or result.get("label"),
|
|
result.get("description"),
|
|
float(result.get("confidence") or 0),
|
|
float(result.get("probability_pd") or 0),
|
|
float(result.get("probability_tpd") or 0),
|
|
1 if result.get("is_valid_audio") is not False else 0,
|
|
result.get("error_message"),
|
|
float(audio_quality.get("duration") or 0),
|
|
float(indicators.get("volume_score") or 0),
|
|
float(indicators.get("intonation_score") or 0),
|
|
float(indicators.get("pause_score") or 0),
|
|
float(indicators.get("speech_activity_ratio") or 0),
|
|
float(indicators.get("silence_ratio") or 0),
|
|
),
|
|
)
|
|
prediction_id = cursor.lastrowid
|
|
connection.commit()
|
|
return prediction_id
|
|
|
|
|
|
def list_prediction_results(limit=200):
|
|
init_db()
|
|
if DB_DRIVER == "mysql":
|
|
return list_prediction_results_mysql(limit=limit)
|
|
|
|
with get_connection() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
student_name,
|
|
predicted_label,
|
|
description,
|
|
confidence,
|
|
probability_pd,
|
|
probability_tpd,
|
|
is_valid_audio,
|
|
error_message,
|
|
audio_duration,
|
|
volume_score,
|
|
intonation_score,
|
|
pause_score,
|
|
speech_activity_ratio,
|
|
silence_ratio,
|
|
created_at
|
|
FROM prediction_results
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def list_prediction_results_mysql(limit=200):
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
student_name,
|
|
predicted_label,
|
|
description,
|
|
confidence,
|
|
probability_pd,
|
|
probability_tpd,
|
|
is_valid_audio,
|
|
error_message,
|
|
audio_duration,
|
|
volume_score,
|
|
intonation_score,
|
|
pause_score,
|
|
speech_activity_ratio,
|
|
silence_ratio,
|
|
created_at
|
|
FROM prediction_results
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
rows = cursor.fetchall()
|
|
|
|
for row in rows:
|
|
row["created_at"] = str(row["created_at"])
|
|
return rows
|