Đề bài: Xây dựng hệ CSDL lưu trữ và tìm kiếm tiếng...

Dưới đây là hướng làm cho phần 4, 5 và web demo mình đã chuẩn bị sẵn cho bạn.

File code web demo: Tải bird_demo_web.zip

1. Hướng viết báo cáo phần 4

4. Xây dựng hệ thống tìm kiếm tiếng chim

Hệ thống nhận đầu vào là một file âm thanh tiếng chim bất kỳ, có thể thuộc loài đã có hoặc chưa có trong CSDL. Đầu ra là danh sách 5 file âm thanh giống nhất, được sắp xếp giảm dần theo độ tương đồng. Yêu cầu này bám sát đề bài trong báo cáo hiện tại của nhóm ở trang 2. ([D22- Lớp 4] Nhóm 6.pdf)

4.a. Sơ đồ khối hệ thống

Bạn có thể đưa sơ đồ này vào báo cáo:

text
File audio truy vấn
        ↓
Tiền xử lý âm thanh
- Load audio
- Chuyển sample rate về 22050 Hz
- Cắt khoảng lặng
- Chuẩn hóa độ dài 10 giây
        ↓
Phân đoạn âm thanh
- Chia file thành 10 đoạn
        ↓
Trích xuất đặc trưng
- MFCC 13 chiều
- Spectral Centroid
- Zero Crossing Rate
- Silence Ratio
- Spectral Bandwidth
- Pitch/F0
        ↓
Tạo ma trận đặc trưng truy vấn
- Kích thước: 10 x 18
        ↓
Đọc feature_vector từ SQLite
        ↓
Chuẩn hóa Z-score
        ↓
Lọc các đoạn im lặng
        ↓
Tính khoảng cách Weighted Euclidean
        ↓
Quy đổi sang độ tương đồng
similarity = exp(-distance)
        ↓
Sắp xếp giảm dần similarity
        ↓
Trả về Top 5 file giống nhất

Trong code hiện tại của bạn, mỗi file được chia thành 10 phân đoạn, mỗi phân đoạn có 18 đặc trưng, gồm 13 MFCC và 5 đặc trưng phổ/thời gian khác. Phần này đã được triển khai trong search_birds.py. (search_birds.py)

4.b. Quy trình tìm kiếm

Bạn có thể viết như sau:

Bước 1: Nhận file truy vấn

Người dùng tải lên một file âm thanh mới. File có thể là .wav, .mp3, .ogg hoặc .flac.

Bước 2: Tiền xử lý

Âm thanh được đưa về cùng chuẩn với dữ liệu trong CSDL:

text
sample_rate = 22050 Hz
duration = 10 giây
số phân đoạn = 10

Nếu file dài hơn 10 giây thì cắt lấy 10 giây đầu sau khi loại bỏ khoảng lặng. Nếu file ngắn hơn thì thêm padding bằng 0.

Bước 3: Trích xuất đặc trưng

Mỗi đoạn âm thanh được biểu diễn bằng vector 18 chiều:

Nhóm đặc trưng	Ý nghĩa
MFCC 1–13	Mô tả hình dạng phổ âm, thường dùng trong nhận dạng âm thanh
Spectral Centroid	Cho biết âm thanh thiên về tần số thấp hay cao
Zero Crossing Rate	Phản ánh mức biến thiên nhanh của tín hiệu
Silence Ratio	Tỷ lệ khoảng lặng trong đoạn âm thanh
Spectral Bandwidth	Độ rộng phổ tần số
Pitch/F0	Độ cao cơ bản của âm thanh

Kết quả sau bước này là ma trận:

text
Query Feature Matrix = 10 x 18

Bước 4: Đọc CSDL

Hệ thống đọc bảng bird_metadata trong SQLite. Bảng này lưu các metadata như file_name, common_name, scientific_name, author, file_path, duration_seconds, sample_rate và feature_vector. Cấu trúc này đã được định nghĩa trong build_db.py. (build_db.py)

Bước 5: Tính độ tương đồng

Với mỗi file trong CSDL:

text
distance = Weighted Euclidean Distance
similarity = exp(-distance)

Các đoạn có Silence_Ratio > 0.95 được bỏ qua để tránh so sánh phần im lặng. Trong code hiện tại, hệ thống dùng StandardScaler để chuẩn hóa Z-score, sau đó tính khoảng cách Euclidean có trọng số. (search_birds.py)

Bước 6: Trả về Top 5

Sau khi tính similarity với toàn bộ file trong CSDL, hệ thống sắp xếp giảm dần và lấy 5 file có điểm cao nhất.

Với loài chưa có trong CSDL, hệ thống vẫn trả về 5 file gần nhất. Tuy nhiên, nếu similarity cao nhất nhỏ hơn một ngưỡng, ví dụ 0.60, có thể kết luận rằng file truy vấn có khả năng thuộc loài chưa có trong CSDL.

2. Các kết quả trung gian nên trình bày

Trong phần 4.b hoặc phần demo, bạn nên chụp màn hình các kết quả sau:

Kết quả trung gian	Ý nghĩa
Waveform	Quan sát dạng sóng sau tiền xử lý
Mel Spectrogram	Quan sát phân bố tần số theo thời gian
Ma trận đặc trưng 10 x 18	Chứng minh hệ thống đã trích xuất đặc trưng
Số segment hợp lệ	Cho biết có bao nhiêu đoạn không bị coi là im lặng
Bảng Top 5	Kết quả tìm kiếm cuối cùng
Similarity và Distance	Căn cứ để xếp hạng độ giống nhau

3. Hướng viết phần 5: Demo và đánh giá

5.1. Demo hệ thống

Bạn có thể viết:

Hệ thống demo được xây dựng dưới dạng web app bằng Streamlit. Người dùng có thể chọn một trong hai chế độ:

text
1. Upload file âm thanh mới từ máy tính
2. Chọn một file mẫu có sẵn trong CSDL

Sau khi nhấn nút tìm kiếm, hệ thống hiển thị:

text
- File âm thanh truy vấn
- Waveform sau tiền xử lý
- Mel Spectrogram
- Ma trận đặc trưng 10 x 18
- Bảng 5 file giống nhất
- Điểm similarity
- Khoảng cách distance
- Metadata của từng file kết quả
- Audio player để nghe lại từng file

5.2. Đánh giá kết quả

Trong web demo mình đã thêm phần Đánh giá nhanh trên CSDL.

Cách đánh giá:

text
- Chọn ngẫu nhiên N file trong database làm truy vấn
- Loại chính file đó khỏi tập ứng viên
- Tìm Top 5 file giống nhất
- So sánh nhãn loài của file truy vấn với kết quả trả về

Chỉ số đánh giá:

Chỉ số	Ý nghĩa
Top-1 Accuracy	Tỷ lệ trường hợp kết quả đầu tiên đúng loài
Recall@5	Tỷ lệ trường hợp trong Top 5 có ít nhất một file đúng loài

Bạn có thể chạy thử với 50, 100 hoặc toàn bộ 500 mẫu rồi chụp bảng kết quả đưa vào báo cáo.

4. Cách chạy web demo

Giải nén file zip, đặt thư mục theo cấu trúc:

text
bird_demo_web/
├── app.py
├── requirements.txt
├── bird_database.db
└── bird_prepared_500/
    ├── bird_001_asbfly_XC134896.wav
    ├── bird_002_...
    └── ...

Cài thư viện:

bash
pip install -r requirements.txt

Chạy web:

bash
streamlit run app.py

Sau đó mở trình duyệt theo đường dẫn Streamlit hiển thị, thường là:

text
http://localhost:8501

5. Lưu ý quan trọng

Trong search_birds.py hiện tại, phần in tên loài đang lấy:

python
species = res['file_name'].split('_')[-1]

Với tên file dạng bird_001_asbfly_XC134896.wav, cách này sẽ lấy ra XC134896 thay vì asbfly. Trong web demo mình đã sửa bằng cách lấy trực tiếp common_name từ SQLite, đúng hơn cho báo cáo và demo.

Mình viết lại thành 2 file đầy đủ:

• app.py: web demo bằng Streamlit.
• search_birds.py: module tìm kiếm, trích xuất đặc trưng, tính similarity, đánh giá Top-1/Recall@5.

Tải nhanh 2 file tại đây: bird_web_code_files.zip

Code dưới đây dựa trên cấu trúc CSDL bird_metadata trong build_db.py và luồng tìm kiếm hiện có trong search_birds.py của bạn. (search_birds.py) (build_db.py)

---

1. File app.py

Lưu file này đúng tên là app.py.

python
"""
app.py
Web demo tìm kiếm tiếng chim bằng Streamlit.

Cách chạy:
    pip install streamlit librosa soundfile numpy pandas scikit-learn matplotlib
    streamlit run app.py

Cấu trúc thư mục khuyến nghị:
    project/
    ├── app.py
    ├── search_birds.py
    ├── bird_database.db
    └── bird_prepared_500/
        ├── bird_001_....wav
        └── ...
"""

import os
import tempfile
from typing import Optional

import librosa
import librosa.display
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import streamlit as st

from search_birds import (
    AUDIO_DIR,
    DB_PATH,
    SR,
    evaluate_database_sample,
    extract_features_segmented,
    feature_matrix_to_dataframe,
    get_database_summary,
    load_database_features,
    resolve_audio_path,
    results_to_dataframe,
    search_similar_birds,
)


# ==========================================
# CẤU HÌNH GIAO DIỆN
# ==========================================
st.set_page_config(
    page_title="Bird Song Search Demo",
    page_icon="🐦",
    layout="wide",
)


@st.cache_data(show_spinner=False)
def cached_database_summary(db_path: str):
    return get_database_summary(db_path)


@st.cache_data(show_spinner=False)
def cached_records(db_path: str):
    return load_database_features(db_path)


# ==========================================
# HÀM PHỤ TRỢ VẼ BIỂU ĐỒ
# ==========================================
def plot_waveform(file_path: str):
    y, sr = librosa.load(file_path, sr=SR, mono=True)
    fig, ax = plt.subplots(figsize=(10, 3))
    librosa.display.waveshow(y, sr=sr, ax=ax)
    ax.set_title("Waveform của file truy vấn")
    ax.set_xlabel("Thời gian (giây)")
    ax.set_ylabel("Biên độ")
    fig.tight_layout()
    return fig


def plot_mel_spectrogram(file_path: str):
    y, sr = librosa.load(file_path, sr=SR, mono=True)
    mel = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=128)
    mel_db = librosa.power_to_db(mel, ref=np.max)

    fig, ax = plt.subplots(figsize=(10, 3))
    img = librosa.display.specshow(mel_db, sr=sr, x_axis="time", y_axis="mel", ax=ax)
    ax.set_title("Mel Spectrogram của file truy vấn")
    fig.colorbar(img, ax=ax, format="%+2.0f dB")
    fig.tight_layout()
    return fig


def save_uploaded_file(uploaded_file) -> Optional[str]:
    if uploaded_file is None:
        return None

    suffix = os.path.splitext(uploaded_file.name)[1] or ".wav"
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(uploaded_file.read())
        return tmp.name


def show_audio_player(file_path: str, label: str = "Audio"):
    if file_path and os.path.exists(file_path):
        st.caption(label)
        st.audio(file_path)
    else:
        st.warning(f"Không tìm thấy file audio: {file_path}")


def format_similarity(value: float) -> str:
    try:
        return f"{float(value):.6f}"
    except Exception:
        return ""


# ==========================================
# SIDEBAR
# ==========================================
st.sidebar.title("Cấu hình demo")

db_path = st.sidebar.text_input("Đường dẫn database SQLite", value=DB_PATH)
audio_dir = st.sidebar.text_input("Thư mục chứa audio", value=AUDIO_DIR)

top_k = st.sidebar.slider(
    "Số kết quả trả về",
    min_value=1,
    max_value=10,
    value=5,
    step=1,
)

unknown_threshold = st.sidebar.slider(
    "Ngưỡng gợi ý loài chưa có trong CSDL",
    min_value=0.0,
    max_value=1.0,
    value=0.60,
    step=0.05,
)

exclude_self = st.sidebar.checkbox(
    "Loại chính file truy vấn nếu chọn từ database",
    value=True,
)

st.title("Hệ thống tìm kiếm tiếng chim hót")
st.write(
    "Web demo nhận một file âm thanh đầu vào, trích xuất đặc trưng âm học, "
    "so sánh với CSDL SQLite và trả về các file giống nhất."
)

if not os.path.exists(db_path):
    st.error(f"Không tìm thấy database: {db_path}")
    st.stop()

try:
    summary = cached_database_summary(db_path)
except Exception as exc:
    st.error(f"Không đọc được database. Lỗi: {exc}")
    st.stop()

col_a, col_b, col_c = st.columns(3)
col_a.metric("Số file trong CSDL", summary["n_files"])
col_b.metric("Số loài", summary["n_species"])
col_c.metric("Sample rate chuẩn", f"{SR} Hz")


# ==========================================
# TABS CHỨC NĂNG
# ==========================================
tab_search, tab_eval, tab_report = st.tabs(
    [
        "Tìm kiếm",
        "Đánh giá nhanh",
        "Gợi ý viết báo cáo",
    ]
)


# ==========================================
# TAB 1: TÌM KIẾM
# ==========================================
with tab_search:
    st.subheader("1. Chọn file truy vấn")

    mode = st.radio(
        "Nguồn file truy vấn",
        options=["Upload file mới", "Chọn file trong database"],
        horizontal=True,
    )

    query_path = None
    query_label = ""

    if mode == "Upload file mới":
        uploaded = st.file_uploader(
            "Tải lên file tiếng chim",
            type=["wav", "mp3", "ogg", "flac", "m4a"],
        )

        if uploaded is not None:
            query_path = save_uploaded_file(uploaded)
            query_label = uploaded.name

    else:
        records = cached_records(db_path)

        if not records:
            st.warning("Database không có bản ghi hợp lệ.")
        else:
            options = [
                f"{r.file_name} | loài: {r.common_name}"
                for r in records
            ]

            selected = st.selectbox("Chọn file mẫu", options=options)
            selected_index = options.index(selected)
            selected_record = records[selected_index]

            query_path = resolve_audio_path(selected_record.file_path, audio_dir)
            query_label = selected_record.file_name

    if query_path:
        st.info(f"File truy vấn: {query_label}")
        show_audio_player(query_path, "Nghe file truy vấn")

    run_search = st.button(
        "Tìm kiếm Top K",
        type="primary",
        disabled=query_path is None,
    )

    if run_search and query_path:
        if not os.path.exists(query_path):
            st.error(f"Không tìm thấy file truy vấn: {query_path}")
            st.stop()

        with st.spinner("Đang trích xuất đặc trưng và tìm kiếm..."):
            try:
                exclude_path = query_path if exclude_self else None

                results, query_matrix, query_info = search_similar_birds(
                    query_path,
                    top_k=top_k,
                    db_path=db_path,
                    audio_dir=audio_dir,
                    exclude_file_path=exclude_path,
                    return_query_features=True,
                )

                feature_df = feature_matrix_to_dataframe(query_matrix)
                result_df = results_to_dataframe(results)

            except Exception as exc:
                st.error(f"Tìm kiếm thất bại. Lỗi: {exc}")
                st.stop()

        st.success("Tìm kiếm hoàn tất")

        best_score = float(results[0]["similarity"]) if results else 0.0

        if best_score < unknown_threshold:
            st.warning(
                f"Similarity cao nhất = {best_score:.4f}, thấp hơn ngưỡng "
                f"{unknown_threshold:.2f}. File truy vấn có thể thuộc loài "
                "chưa có trong CSDL."
            )
        else:
            st.info(
                f"Similarity cao nhất = {best_score:.4f}. "
                f"Loài gần nhất theo CSDL: {results[0]['common_name']}"
            )

        st.subheader("2. Kết quả trung gian")

        m1, m2, m3, m4 = st.columns(4)
        m1.metric("Thời lượng gốc", f"{query_info['original_duration_seconds']:.2f}s")
        m2.metric("Thời lượng xử lý", f"{query_info['processed_duration_seconds']:.2f}s")
        m3.metric("Số segment", query_info["n_segments"])
        m4.metric("Số đặc trưng/segment", query_info["n_features"])

        fig_col1, fig_col2 = st.columns(2)

        with fig_col1:
            try:
                st.pyplot(plot_waveform(query_path))
            except Exception as exc:
                st.warning(f"Không vẽ được waveform: {exc}")

        with fig_col2:
            try:
                st.pyplot(plot_mel_spectrogram(query_path))
            except Exception as exc:
                st.warning(f"Không vẽ được Mel Spectrogram: {exc}")

        st.markdown("**Ma trận đặc trưng của file truy vấn, kích thước 10 x 18**")
        st.dataframe(feature_df, use_container_width=True)

        segment_df = pd.DataFrame(query_info["segment_infos"])
        st.markdown("**Thông tin từng segment và lọc khoảng lặng**")
        st.dataframe(segment_df, use_container_width=True)

        st.subheader("3. Top file giống nhất")

        if not result_df.empty:
            display_df = result_df.copy()
            display_df["similarity"] = display_df["similarity"].map(format_similarity)
            display_df["distance"] = display_df["distance"].map(
                lambda x: "inf" if np.isinf(x) else f"{x:.6f}"
            )
            st.dataframe(display_df, use_container_width=True)

        for item in results:
            with st.expander(
                f"#{item['rank']} - {item['common_name']} - "
                f"similarity {item['similarity']:.6f}",
                expanded=item["rank"] == 1,
            ):
                c1, c2 = st.columns([1, 2])

                with c1:
                    show_audio_player(item["resolved_file_path"], "Nghe file kết quả")

                with c2:
                    st.write(f"**File:** {item['file_name']}")
                    st.write(f"**Loài:** {item['common_name']}")
                    st.write(f"**Tên khoa học:** {item['scientific_name']}")
                    st.write(f"**Similarity:** {item['similarity']:.6f}")

                    if np.isinf(item["distance"]):
                        st.write("**Distance:** inf")
                    else:
                        st.write(f"**Distance:** {item['distance']:.6f}")

                    st.write(f"**Số segment hợp lệ:** {item['valid_segment_count']}")
                    st.write(f"**Các segment hợp lệ:** {item['valid_segments']}")
                    st.write(f"**Đường dẫn:** `{item['resolved_file_path']}`")


# ==========================================
# TAB 2: ĐÁNH GIÁ NHANH
# ==========================================
with tab_eval:
    st.subheader("Đánh giá nhanh trên CSDL")
    st.write(
        "Chức năng này chọn ngẫu nhiên một số file trong CSDL làm truy vấn, "
        "loại chính file đó khỏi tập ứng viên, sau đó kiểm tra kết quả "
        "Top-1 và Recall@K."
    )

    eval_col1, eval_col2, eval_col3 = st.columns(3)

    sample_size = eval_col1.slider(
        "Số mẫu đánh giá",
        5,
        min(200, summary["n_files"]),
        50,
        5,
    )

    eval_top_k = eval_col2.slider(
        "K trong Recall@K",
        1,
        10,
        5,
        1,
    )

    seed = eval_col3.number_input(
        "Seed",
        value=42,
        step=1,
    )

    if st.button("Chạy đánh giá"):
        with st.spinner("Đang đánh giá..."):
            try:
                metrics, detail_df = evaluate_database_sample(
                    db_path=db_path,
                    sample_size=sample_size,
                    top_k=eval_top_k,
                    seed=int(seed),
                )
            except Exception as exc:
                st.error(f"Đánh giá thất bại. Lỗi: {exc}")
                st.stop()

        r1, r2, r3 = st.columns(3)
        r1.metric("Số mẫu", metrics["sample_size"])
        r2.metric("Top-1 Accuracy", f"{metrics['top1_accuracy']:.2%}")
        r3.metric(f"Recall@{eval_top_k}", f"{metrics[f'recall@{eval_top_k}']:.2%}")

        st.markdown("**Chi tiết từng truy vấn đánh giá**")
        st.dataframe(detail_df, use_container_width=True)

        csv = detail_df.to_csv(index=False).encode("utf-8-sig")

        st.download_button(
            "Tải kết quả đánh giá CSV",
            data=csv,
            file_name="bird_search_evaluation.csv",
            mime="text/csv",
        )


# ==========================================
# TAB 3: GỢI Ý VIẾT BÁO CÁO
# ==========================================
with tab_report:
    st.subheader("Nội dung có thể đưa vào phần 4 và 5 của báo cáo")

    st.markdown(
        """
### Sơ đồ khối hệ thống

```text
File audio truy vấn
    ↓
Tiền xử lý: load audio, resample 22050 Hz, cắt khoảng lặng, chuẩn hóa 10 giây
    ↓
Chia thành 10 segment
    ↓
Trích xuất 18 đặc trưng cho mỗi segment
    ↓
Tạo ma trận đặc trưng 10 x 18
    ↓
Đọc feature_vector từ SQLite
    ↓
Chuẩn hóa Z-score
    ↓
Bỏ qua các segment có Silence Ratio > 0.95
    ↓
Tính Weighted Euclidean Distance
    ↓
Quy đổi similarity = exp(-distance)
    ↓
Sắp xếp giảm dần và trả về Top 5 file giống nhất

Kết quả trung gian cần chụp màn hình

• File truy vấn và audio player.
• Waveform.
• Mel Spectrogram.
• Ma trận đặc trưng 10 x 18.
• Bảng segment hợp lệ sau khi lọc khoảng lặng.
• Bảng Top 5 kết quả gồm similarity, distance, tên file và loài.
• Kết quả đánh giá Top-1 Accuracy và Recall@5. """ )

text

---

## 2. File `search_birds.py`

> Lưu file này đúng tên là `search_birds.py`, đặt cùng thư mục với `app.py`.

```python
"""
search_birds.py
Module tìm kiếm tiếng chim dựa trên đặc trưng âm thanh lưu trong SQLite.

Cách dùng nhanh:
    python search_birds.py --query path/to/bird.wav --top_k 5

Yêu cầu thư mục chạy có:
    bird_database.db
    bird_prepared_500/
"""

import argparse
import json
import os
import sqlite3
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Sequence, Tuple

import librosa
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler


# ==========================================
# CẤU HÌNH
# ==========================================
DB_PATH = os.environ.get("BIRD_DB_PATH", "bird_database.db")
AUDIO_DIR = os.environ.get("BIRD_AUDIO_DIR", "bird_prepared_500")

SR = 22050
N_MFCC = 13
N_SEGMENTS = 10

# Thứ tự đặc trưng:
# MFCC(0-12), centroid(13), zcr(14), silence(15), bandwidth(16), pitch(17)
SILENCE_IDX = 15

FEATURE_WEIGHTS = np.array(
    [0.05] * 13 + [0.10, 0.05, 0.05, 0.10, 0.05],
    dtype=float,
)

FEATURE_COLUMNS = (
    [f"mfcc_{i:02d}" for i in range(1, N_MFCC + 1)]
    + [
        "spectral_centroid",
        "zcr",
        "silence_ratio",
        "spectral_bandwidth",
        "pitch",
    ]
)


@dataclass
class BirdRecord:
    id: int
    file_name: str
    file_path: str
    common_name: str
    scientific_name: str
    duration_seconds: Optional[float]
    sample_rate: Optional[int]
    feature_matrix: List[List[float]]


def normalize_path(path: str) -> str:
    return path.replace("\\", os.sep).replace("/", os.sep)


def resolve_audio_path(file_path: str, audio_dir: str = AUDIO_DIR) -> str:
    """
    Tìm đường dẫn thật của file âm thanh.
    DB thường lưu dạng bird_prepared_500\\file.wav trên Windows.
    Hàm này giúp web chạy ổn trên cả Windows/Mac/Linux.
    """
    if not file_path:
        return ""

    candidates = []

    normalized = normalize_path(file_path)
    candidates.append(normalized)
    candidates.append(os.path.abspath(normalized))

    base_name = os.path.basename(normalized)
    candidates.append(os.path.join(audio_dir, base_name))
    candidates.append(os.path.abspath(os.path.join(audio_dir, base_name)))

    here = os.path.dirname(os.path.abspath(__file__))
    candidates.append(os.path.join(here, normalized))
    candidates.append(os.path.join(here, audio_dir, base_name))

    for p in candidates:
        if p and os.path.exists(p):
            return p

    return normalized


# ==========================================
# 1. TRÍCH XUẤT ĐẶC TRƯNG ÂM THANH
# ==========================================
def extract_features_segmented(
    file_path: str,
    n_segments: int = N_SEGMENTS,
    sr: int = SR,
    return_intermediate: bool = False,
) -> Any:
    """
    Trích xuất ma trận đặc trưng cho file âm thanh.

    Output mặc định:
        matrix_features: List[List[float]], kích thước n_segments x 18

    Nếu return_intermediate=True:
        return (matrix_features, info_dict)
    """
    try:
        y_original, loaded_sr = librosa.load(file_path, sr=sr, mono=True)

        original_duration = len(y_original) / float(sr) if sr else 0.0

        y_trimmed, trim_index = librosa.effects.trim(y_original, top_db=25)

        target_len = sr * 10

        if len(y_trimmed) > target_len:
            y_processed = y_trimmed[:target_len]
        else:
            y_processed = np.pad(
                y_trimmed,
                (0, target_len - len(y_trimmed)),
                mode="constant",
            )

        segment_len = len(y_processed) // n_segments

        matrix_features: List[List[float]] = []
        segment_infos: List[Dict[str, Any]] = []

        for i in range(n_segments):
            start = i * segment_len
            end = (i + 1) * segment_len if i < n_segments - 1 else len(y_processed)

            y_seg = y_processed[start:end]

            mfcc = np.mean(
                librosa.feature.mfcc(
                    y=y_seg,
                    sr=sr,
                    n_mfcc=N_MFCC,
                ),
                axis=1,
            )

            centroid = float(
                np.mean(
                    librosa.feature.spectral_centroid(
                        y=y_seg,
                        sr=sr,
                    )
                )
            )

            zcr = float(
                np.mean(
                    librosa.feature.zero_crossing_rate(y_seg)
                )
            )

            silence_ratio = float(
                np.sum(np.abs(y_seg) < 0.01) / max(len(y_seg), 1)
            )

            bandwidth = float(
                np.mean(
                    librosa.feature.spectral_bandwidth(
                        y=y_seg,
                        sr=sr,
                    )
                )
            )

            pitches, magnitudes = librosa.piptrack(
                y=y_seg,
                sr=sr,
            )

            pitch = float(np.mean(pitches[pitches > 0])) if np.any(pitches > 0) else 0.0

            vector_seg = np.concatenate(
                [
                    mfcc.astype(float),
                    np.array(
                        [
                            centroid,
                            zcr,
                            silence_ratio,
                            bandwidth,
                            pitch,
                        ],
                        dtype=float,
                    ),
                ]
            )

            matrix_features.append(vector_seg.tolist())

            segment_infos.append(
                {
                    "segment": i + 1,
                    "start_second": round(start / sr, 3),
                    "end_second": round(end / sr, 3),
                    "silence_ratio": silence_ratio,
                    "is_silent": silence_ratio > 0.95,
                }
            )

        if not return_intermediate:
            return matrix_features

        info = {
            "file_path": file_path,
            "sample_rate": sr,
            "loaded_sample_rate": loaded_sr,
            "original_duration_seconds": original_duration,
            "processed_duration_seconds": len(y_processed) / float(sr),
            "trim_start_sample": int(trim_index[0]) if len(trim_index) > 0 else 0,
            "trim_end_sample": int(trim_index[1]) if len(trim_index) > 1 else len(y_original),
            "n_segments": n_segments,
            "n_features": len(FEATURE_COLUMNS),
            "feature_columns": FEATURE_COLUMNS,
            "segment_infos": segment_infos,
        }

        return matrix_features, info

    except Exception as exc:
        raise RuntimeError(
            f"Không trích xuất được đặc trưng từ file: {file_path}. Lỗi: {exc}"
        ) from exc


# ==========================================
# 2. ĐỌC CSDL
# ==========================================
def load_database_features(db_path: str = DB_PATH) -> List[BirdRecord]:
    if not os.path.exists(db_path):
        raise FileNotFoundError(f"Không tìm thấy database: {db_path}")

    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    cursor.execute(
        """
        SELECT id, file_name, file_path, common_name, scientific_name,
               duration_seconds, sample_rate, feature_vector
        FROM bird_metadata
        WHERE feature_vector IS NOT NULL AND feature_vector != ''
        """
    )

    rows = cursor.fetchall()
    conn.close()

    records: List[BirdRecord] = []

    for row in rows:
        try:
            feature_matrix = json.loads(row["feature_vector"])
            arr = np.asarray(feature_matrix, dtype=float)

            if arr.ndim != 2 or arr.shape[1] != len(FEATURE_COLUMNS):
                continue

            records.append(
                BirdRecord(
                    id=int(row["id"]),
                    file_name=row["file_name"],
                    file_path=row["file_path"],
                    common_name=row["common_name"],
                    scientific_name=row["scientific_name"] or "Unknown",
                    duration_seconds=row["duration_seconds"],
                    sample_rate=row["sample_rate"],
                    feature_matrix=feature_matrix,
                )
            )

        except Exception:
            continue

    return records


def get_database_summary(db_path: str = DB_PATH) -> Dict[str, Any]:
    records = load_database_features(db_path)
    species = sorted({r.common_name for r in records})

    return {
        "n_files": len(records),
        "n_species": len(species),
        "species": species,
    }


# ==========================================
# 3. TÍNH ĐỘ TƯƠNG ĐỒNG
# ==========================================
def calculate_combined_similarity(
    query_matrix: Sequence[Sequence[float]],
    db_matrix_list: Sequence[Sequence[Sequence[float]]],
    return_details: bool = False,
) -> Any:
    """
    Quy trình:
        1. Chuẩn hóa Z-score trên toàn bộ query + database.
        2. Bỏ qua segment im lặng nếu silence_ratio > 0.95.
        3. Tính Weighted Euclidean Distance.
        4. Quy đổi similarity = exp(-avg_distance).
    """
    query_matrix_np = np.asarray(query_matrix, dtype=float)
    db_matrices_np = np.asarray(db_matrix_list, dtype=float)

    if query_matrix_np.ndim != 2:
        raise ValueError("query_matrix phải có dạng 2 chiều: n_segments x n_features")

    if db_matrices_np.ndim != 3:
        raise ValueError("db_matrix_list phải có dạng 3 chiều: n_files x n_segments x n_features")

    if len(db_matrices_np) == 0:
        empty = np.array([], dtype=float)
        return (empty, []) if return_details else empty

    n_segments, n_features = query_matrix_np.shape

    if n_features != len(FEATURE_WEIGHTS):
        raise ValueError(
            f"Số đặc trưng không hợp lệ: {n_features}. Cần {len(FEATURE_WEIGHTS)}"
        )

    if db_matrices_np.shape[1:] != query_matrix_np.shape:
        raise ValueError(
            "Kích thước feature không khớp. "
            f"Query: {query_matrix_np.shape}, DB: {db_matrices_np.shape[1:]}"
        )

    all_data = np.vstack(
        [
            query_matrix_np,
            db_matrices_np.reshape(-1, n_features),
        ]
    )

    scaler = StandardScaler()
    all_data_scaled = scaler.fit_transform(all_data)

    query_scaled = all_data_scaled[:n_segments]
    db_scaled = all_data_scaled[n_segments:].reshape(
        len(db_matrices_np),
        n_segments,
        n_features,
    )

    scores: List[float] = []
    details: List[Dict[str, Any]] = []

    for file_idx in range(len(db_matrices_np)):
        segment_distances: List[float] = []
        valid_segments: List[int] = []

        for seg_idx in range(n_segments):
            query_silence = query_matrix_np[seg_idx, SILENCE_IDX]
            db_silence = db_matrices_np[file_idx, seg_idx, SILENCE_IDX]

            if query_silence <= 0.95 and db_silence <= 0.95:
                diff = (
                    query_scaled[seg_idx] - db_scaled[file_idx, seg_idx]
                ) * FEATURE_WEIGHTS

                dist = float(np.sqrt(np.sum(diff ** 2)))

                segment_distances.append(dist)
                valid_segments.append(seg_idx + 1)

        if not segment_distances:
            avg_distance = float("inf")
            similarity = 0.0
        else:
            avg_distance = float(np.mean(segment_distances))
            similarity = float(np.exp(-avg_distance))

        scores.append(similarity)

        details.append(
            {
                "avg_distance": avg_distance,
                "similarity": similarity,
                "valid_segment_count": len(valid_segments),
                "valid_segments": valid_segments,
                "segment_distances": segment_distances,
            }
        )

    scores_np = np.asarray(scores, dtype=float)

    return (scores_np, details) if return_details else scores_np


# ==========================================
# 4. TÌM KIẾM TOP K
# ==========================================
def search_similar_birds(
    query_file_path: str,
    top_k: int = 5,
    db_path: str = DB_PATH,
    audio_dir: str = AUDIO_DIR,
    exclude_file_path: Optional[str] = None,
    return_query_features: bool = False,
) -> Any:
    query_matrix, query_info = extract_features_segmented(
        query_file_path,
        return_intermediate=True,
    )

    records = load_database_features(db_path)

    if not records:
        raise RuntimeError("CSDL không có bản ghi hợp lệ hoặc feature_vector rỗng.")

    filtered_records: List[BirdRecord] = []

    query_abs = os.path.abspath(query_file_path)
    exclude_abs = os.path.abspath(exclude_file_path) if exclude_file_path else None

    for rec in records:
        rec_audio_path = resolve_audio_path(rec.file_path, audio_dir)
        rec_abs = os.path.abspath(rec_audio_path)

        should_exclude = False

        if exclude_abs and rec_abs == exclude_abs:
            should_exclude = True

        if os.path.exists(rec_audio_path) and rec_abs == query_abs:
            should_exclude = True

        if not should_exclude:
            filtered_records.append(rec)

    if not filtered_records:
        raise RuntimeError("Không còn ứng viên nào để so sánh sau khi loại file truy vấn.")

    db_matrices = [r.feature_matrix for r in filtered_records]

    scores, sim_details = calculate_combined_similarity(
        query_matrix,
        db_matrices,
        return_details=True,
    )

    top_indices = np.argsort(scores)[::-1][:top_k]

    results: List[Dict[str, Any]] = []

    for rank, idx in enumerate(top_indices, start=1):
        rec = filtered_records[int(idx)]
        detail = sim_details[int(idx)]
        audio_path = resolve_audio_path(rec.file_path, audio_dir)

        results.append(
            {
                "rank": rank,
                "id": rec.id,
                "file_name": rec.file_name,
                "file_path": rec.file_path,
                "resolved_file_path": audio_path,
                "common_name": rec.common_name,
                "scientific_name": rec.scientific_name,
                "duration_seconds": rec.duration_seconds,
                "sample_rate": rec.sample_rate,
                "similarity": float(scores[int(idx)]),
                "distance": detail["avg_distance"],
                "valid_segment_count": detail["valid_segment_count"],
                "valid_segments": detail["valid_segments"],
            }
        )

    if return_query_features:
        return results, query_matrix, query_info

    return results


# ==========================================
# 5. ĐÁNH GIÁ NHANH TRÊN DATABASE
# ==========================================
def search_by_feature_matrix(
    query_matrix: Sequence[Sequence[float]],
    records: Sequence[BirdRecord],
    top_k: int = 5,
    exclude_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    candidates = [r for r in records if r.id != exclude_id]

    db_matrices = [r.feature_matrix for r in candidates]

    scores, sim_details = calculate_combined_similarity(
        query_matrix,
        db_matrices,
        return_details=True,
    )

    top_indices = np.argsort(scores)[::-1][:top_k]

    output: List[Dict[str, Any]] = []

    for rank, idx in enumerate(top_indices, start=1):
        rec = candidates[int(idx)]
        detail = sim_details[int(idx)]

        output.append(
            {
                "rank": rank,
                "id": rec.id,
                "file_name": rec.file_name,
                "common_name": rec.common_name,
                "scientific_name": rec.scientific_name,
                "similarity": float(scores[int(idx)]),
                "distance": detail["avg_distance"],
                "valid_segment_count": detail["valid_segment_count"],
            }
        )

    return output


def evaluate_database_sample(
    db_path: str = DB_PATH,
    sample_size: int = 50,
    top_k: int = 5,
    seed: int = 42,
) -> Tuple[Dict[str, Any], pd.DataFrame]:
    """
    Đánh giá nhanh bằng cách lấy ngẫu nhiên sample_size file trong CSDL làm query.
    File query bị loại khỏi danh sách ứng viên.

    Metrics:
        Top-1 Accuracy: kết quả hạng 1 có cùng common_name với query.
        Recall@K: trong Top K có ít nhất 1 kết quả cùng common_name với query.
    """
    records = load_database_features(db_path)

    if not records:
        raise RuntimeError("Không có dữ liệu để đánh giá.")

    rng = np.random.default_rng(seed)

    sample_size = min(sample_size, len(records))

    sampled_indices = rng.choice(
        len(records),
        size=sample_size,
        replace=False,
    )

    detail_rows: List[Dict[str, Any]] = []

    top1_correct = 0
    recall_k_correct = 0

    for idx in sampled_indices:
        query_rec = records[int(idx)]

        top_results = search_by_feature_matrix(
            query_rec.feature_matrix,
            records,
            top_k=top_k,
            exclude_id=query_rec.id,
        )

        top1_species = top_results[0]["common_name"] if top_results else ""
        topk_species = [r["common_name"] for r in top_results]

        is_top1_correct = top1_species == query_rec.common_name
        is_recall_k_correct = query_rec.common_name in topk_species

        top1_correct += int(is_top1_correct)
        recall_k_correct += int(is_recall_k_correct)

        detail_rows.append(
            {
                "query_file": query_rec.file_name,
                "query_species": query_rec.common_name,
                "top1_file": top_results[0]["file_name"] if top_results else "",
                "top1_species": top1_species,
                "top1_similarity": top_results[0]["similarity"] if top_results else 0.0,
                "top1_correct": is_top1_correct,
                f"recall@{top_k}": is_recall_k_correct,
                "top_k_species": ", ".join(topk_species),
            }
        )

    metrics = {
        "sample_size": sample_size,
        "top_k": top_k,
        "top1_accuracy": top1_correct / sample_size if sample_size else 0.0,
        f"recall@{top_k}": recall_k_correct / sample_size if sample_size else 0.0,
    }

    return metrics, pd.DataFrame(detail_rows)


# ==========================================
# 6. TIỆN ÍCH HIỂN THỊ
# ==========================================
def feature_matrix_to_dataframe(matrix: Sequence[Sequence[float]]) -> pd.DataFrame:
    df = pd.DataFrame(matrix, columns=FEATURE_COLUMNS)
    df.insert(0, "segment", np.arange(1, len(df) + 1))
    return df


def results_to_dataframe(results: Sequence[Dict[str, Any]]) -> pd.DataFrame:
    rows = []

    for r in results:
        rows.append(
            {
                "rank": r.get("rank"),
                "common_name": r.get("common_name"),
                "scientific_name": r.get("scientific_name"),
                "file_name": r.get("file_name"),
                "similarity": r.get("similarity"),
                "distance": r.get("distance"),
                "valid_segment_count": r.get("valid_segment_count"),
                "duration_seconds": r.get("duration_seconds"),
                "sample_rate": r.get("sample_rate"),
                "file_path": r.get("resolved_file_path") or r.get("file_path"),
            }
        )

    return pd.DataFrame(rows)


# ==========================================
# 7. CLI DEMO
# ==========================================
def main() -> None:
    parser = argparse.ArgumentParser(description="Tìm kiếm tiếng chim tương đồng")

    parser.add_argument(
        "--query",
        required=True,
        help="Đường dẫn file audio truy vấn",
    )

    parser.add_argument(
        "--db",
        default=DB_PATH,
        help="Đường dẫn SQLite database",
    )

    parser.add_argument(
        "--audio_dir",
        default=AUDIO_DIR,
        help="Thư mục audio trong database",
    )

    parser.add_argument(
        "--top_k",
        type=int,
        default=5,
        help="Số kết quả cần trả về",
    )

    parser.add_argument(
        "--exclude_self",
        action="store_true",
        help="Loại file query nếu nó nằm trong database",
    )

    args = parser.parse_args()

    query_path = resolve_audio_path(args.query, args.audio_dir)

    if not os.path.exists(query_path):
        raise FileNotFoundError(f"Không tìm thấy file query: {args.query}")

    results = search_similar_birds(
        query_path,
        top_k=args.top_k,
        db_path=args.db,
        audio_dir=args.audio_dir,
        exclude_file_path=query_path if args.exclude_self else None,
    )

    df = results_to_dataframe(results)

    pd.set_option("display.max_columns", None)
    pd.set_option("display.width", 180)

    print(df.to_string(index=False))


if __name__ == "__main__":
    main()

---

3. Cài thư viện và chạy

Tạo thêm file requirements.txt:

txt
streamlit
librosa
soundfile
numpy
pandas
scikit-learn
matplotlib

Cài thư viện:

bash
pip install -r requirements.txt

Chạy web:

bash
streamlit run app.py

Cấu trúc thư mục nên là:

text
project/
├── app.py
├── search_birds.py
├── bird_database.db
├── requirements.txt
└── bird_prepared_500/
    ├── bird_001_asbfly_XC134896.wav
    ├── bird_002_asbfly_XC164848.wav
    └── ...

Lưu ý: file tải về trong zip đang tên là search_birds_full.py; khi dùng thật, bạn đổi tên thành search_birds.py.