Spaces:
Sleeping
Sleeping
File size: 3,738 Bytes
cc2b2f6 35e05ba cc2b2f6 35e05ba cc2b2f6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | from __future__ import annotations
from collections import Counter
import unittest
import polars as pl
from backend.data import (
filter_tokens,
load_prefix_offsets,
normalize_word_query,
token_detail,
tokens_payload,
)
def sample_df() -> pl.DataFrame:
rows = []
specs = [
("AA", "AA_file_1", "ai", "kai", "primary", 1, 10.0),
("AA", "AA_file_2", "ae", "maikaʻi", "unstressed", 0, 20.0),
("DK", "DK_file_1", "ai", "kai", "unstressed", 1, 30.0),
("DK", "DK_file_2", "a", "hale", "secondary", 0, 40.0),
]
for original_order, (speaker, filename, vowel, word, stress, articles, start) in enumerate(specs):
rows.append(
{
"Speaker": speaker,
"filename": filename,
"vowel": vowel,
"word": word,
"stress": stress,
"previous_sound": "k",
"next_sound": "i",
"time": 5.0,
"f1": 500.0 + original_order,
"f2": 1500.0 + original_order,
"f1_normed": 0.1 + original_order,
"f2_normed": 0.2 + original_order,
"start": start,
"original_order": original_order,
"token_id": f"{speaker}|{filename}|{start}",
"word_search_key": normalize_word_query(word),
"articles": articles,
}
)
return pl.DataFrame(rows)
class DataFilteringTests(unittest.TestCase):
def test_verified_aa_interview_offsets_resolve_to_episode_clock(self) -> None:
offsets, loaded = load_prefix_offsets()
detail = token_detail(
pl.DataFrame(
[
{
"token_id": "AA|KLH057a_0001|42.8996707",
"Speaker": "AA",
"filename": "KLH057a_0001",
"word": "aloha",
"vowel": "a",
"stress": "primary",
"previous_sound": "k",
"next_sound": "l",
"start": 42.8996707,
}
]
),
offsets,
"AA|KLH057a_0001|42.8996707",
)
self.assertTrue(loaded)
self.assertEqual(offsets["KLH057a"], 0.0)
self.assertAlmostEqual(offsets["KLH057b"], 2600.5416780045352)
self.assertIsNotNone(detail)
self.assertTrue(detail["interview_offset_available"])
self.assertAlmostEqual(detail["interview_seconds"], 42.8996707)
def test_filter_tokens_combines_speaker_vowel_and_stress(self) -> None:
out = filter_tokens(
sample_df(),
speakers=["DK"],
vowels=["ai"],
stresses=["unstressed"],
)
self.assertEqual(out.height, 1)
self.assertEqual(out.item(0, "word"), "kai")
def test_filter_tokens_supports_function_include_and_word_query(self) -> None:
out = filter_tokens(
sample_df(),
speakers=None,
vowels=None,
stresses=None,
function_include=["articles"],
word_q="kai",
)
self.assertEqual(out.height, 2)
self.assertEqual(set(out.get_column("Speaker").to_list()), {"AA", "DK"})
def test_tokens_payload_round_robins_to_fill_limit(self) -> None:
payload = tokens_payload(sample_df(), limit=3)
self.assertEqual(payload["n_tokens"], 4)
self.assertEqual(payload["n_rows"], 3)
self.assertEqual(Counter(row["speaker"] for row in payload["rows"]), {"AA": 2, "DK": 1})
if __name__ == "__main__":
unittest.main()
|