arjun10g's picture
Add natural language filters and optimize visualizer
cc2b2f6 verified
Raw
History Blame Contribute Delete
14.8 kB
from __future__ import annotations
import json
import logging
import os
import unicodedata
from pathlib import Path
import polars as pl
from .schemas import classify_vowel
log = logging.getLogger(__name__)
REPO_ROOT = Path(__file__).resolve().parent.parent
# Prefer the column-pruned, stress-filtered parquet (~5.5MB) when present —
# 16× smaller than the raw CSV (88MB), loads ~10× faster, and ships in the
# Docker image / git repo directly. Falls back to the original CSV in dev
# when only the upstream file is on disk.
DEFAULT_PARQUET_PATH = REPO_ROOT / "all_data_18Nov2023.parquet"
DEFAULT_CSV_PATH = REPO_ROOT / "all_data_18Nov2023.csv"
DEFAULT_PREFIX_OFFSETS_PATH = REPO_ROOT / "data" / "prefix_offsets.json"
GIT_LFS_POINTER_PREFIX = "version https://git-lfs.github.com/spec/v1"
# Subset of CSV columns we actually use. Saves memory on the 88MB file.
USED_COLUMNS: tuple[str, ...] = (
"Speaker",
"filename",
"vowel",
"word",
"stress",
"previous_sound",
"next_sound",
"previous_word",
"next_word",
"time",
"f1",
"f2",
"f3",
"f1_normed",
"f2_normed",
"f3_normed",
"original_order",
"Moras",
"Syllabification",
"aole",
"kaika",
"maikai",
"Kinney1956",
"articles",
"demons",
"particles",
"directionals",
"pronouns",
"interrogatives",
"funct",
"manawa",
"mea",
"ae",
"start",
"word_start",
"word_end",
)
FUNCTION_WORD_COLUMNS: tuple[str, ...] = (
"aole",
"kaika",
"maikai",
"Kinney1956",
"articles",
"demons",
"particles",
"directionals",
"pronouns",
"interrogatives",
"funct",
"manawa",
"mea",
"ae",
)
INTERVIEW_URL_SUFFIX = (
"e=-------en-20--1--txt-txIN%7CtxTI%7CtxTA%7CtxCO%7CtxTY%7CtxLA%7CtxKE%7C"
"txPR%7CtxSG%7CtxTO%7CtxTG%7CtxSM%7CtxTR%7CtxSP%7CtxCT%7CtxET%7CtxHT"
"--------------------"
)
SPEAKER_EPISODES: dict[str, str] = {
"IN": "013",
"RM": "014",
"JM": "016",
"HM": "021",
"LV": "032",
"SB": "033",
"AA": "057",
"DK": "063",
}
class DataStore:
"""In-memory dataset. Loaded once at app startup."""
df: pl.DataFrame
prefix_offsets: dict[str, float]
prefix_offsets_loaded: bool
def __init__(self, df: pl.DataFrame, prefix_offsets: dict[str, float], loaded: bool) -> None:
self.df = df
self.prefix_offsets = prefix_offsets
self.prefix_offsets_loaded = loaded
def _data_path() -> Path:
"""Resolve the source file. Honors KLH_DATA_PATH override, otherwise
prefers parquet → csv in that order."""
override = os.environ.get("KLH_DATA_PATH") or os.environ.get("KLH_CSV_PATH")
if override:
return Path(override)
if DEFAULT_PARQUET_PATH.exists() and not _is_git_lfs_pointer(DEFAULT_PARQUET_PATH):
return DEFAULT_PARQUET_PATH
if DEFAULT_PARQUET_PATH.exists():
log.warning(
"%s is a Git LFS pointer, not parquet data — falling back to CSV",
DEFAULT_PARQUET_PATH,
)
return DEFAULT_CSV_PATH
def _is_git_lfs_pointer(path: Path) -> bool:
if not path.is_file() or path.stat().st_size > 1024:
return False
try:
return path.read_text(errors="ignore").startswith(GIT_LFS_POINTER_PREFIX)
except OSError:
return False
def _prefix_offsets_path() -> Path:
override = os.environ.get("KLH_PREFIX_OFFSETS_PATH")
return Path(override) if override else DEFAULT_PREFIX_OFFSETS_PATH
def normalize_word_query(value: str) -> str:
"""Normalize user/corpus words for lookup without changing display text."""
normalized = unicodedata.normalize("NFC", value.strip())
normalized = normalized.replace("'", "ʻ").replace("`", "ʻ").replace("’", "ʻ")
return normalized.casefold()
def _valid_function_columns(columns: list[str] | None) -> list[str]:
if not columns:
return []
allowed = set(FUNCTION_WORD_COLUMNS)
return [col for col in columns if col in allowed]
def _function_flags(row: dict) -> dict[str, bool]:
return {col: bool(row.get(col) or 0) for col in FUNCTION_WORD_COLUMNS}
def load_prefix_offsets(path: Path | None = None) -> tuple[dict[str, float], bool]:
"""Load prefix → seconds map. Returns ({}, False) when file is missing/invalid.
Per CLAUDE.md: do not crash if the file is absent — the jump-link feature
should silently disable instead.
"""
target = path or _prefix_offsets_path()
if not target.exists():
log.warning("prefix_offsets.json not found at %s — interview jump-link disabled", target)
return {}, False
try:
raw = json.loads(target.read_text())
if not isinstance(raw, dict):
log.error("prefix_offsets.json is not a JSON object — disabling jump-link")
return {}, False
offsets = {str(k): float(v) for k, v in raw.items()}
return offsets, True
except (json.JSONDecodeError, ValueError, OSError) as e:
log.error("Failed to parse prefix_offsets.json: %s — disabling jump-link", e)
return {}, False
def load_dataframe(path: Path | None = None) -> pl.DataFrame:
"""Load the formant data into a Polars DataFrame, keeping only used columns.
Reads either parquet (preferred — pre-pruned, pre-filtered, 5.5MB) or the
raw CSV (88MB, requires column-subset + stress filter at load time).
Adds a stable `token_id` derived from (Speaker, filename, word_start).
"""
target = path or _data_path()
if target.suffix == ".parquet" and _is_git_lfs_pointer(target):
if DEFAULT_CSV_PATH.exists():
log.warning("%s is a Git LFS pointer — loading %s instead", target, DEFAULT_CSV_PATH)
target = DEFAULT_CSV_PATH
else:
raise FileNotFoundError(
f"{target} is a Git LFS pointer and {DEFAULT_CSV_PATH} is missing"
)
log.info("Loading data from %s", target)
if target.suffix == ".parquet":
df = pl.read_parquet(target)
missing = [col for col in USED_COLUMNS if col not in df.columns]
if missing and path is None and DEFAULT_CSV_PATH.exists():
log.warning(
"%s is missing columns needed by current filters (%s) — loading %s instead",
target,
", ".join(missing[:6]) + ("..." if len(missing) > 6 else ""),
DEFAULT_CSV_PATH,
)
target = DEFAULT_CSV_PATH
df = pl.read_csv(
target,
columns=list(USED_COLUMNS),
null_values=["NA", ""],
)
before = df.height
df = df.filter(pl.col("stress") != "0")
log.info("Filtered stress==0: dropped %d rows", before - df.height)
else:
df = pl.read_csv(
target,
columns=list(USED_COLUMNS),
null_values=["NA", ""],
)
# CSV path — apply the stress filter that's already baked into parquet.
before = df.height
df = df.filter(pl.col("stress") != "0")
log.info("Filtered stress==0: dropped %d rows", before - df.height)
df = df.with_columns(
(
pl.col("Speaker").cast(pl.Utf8)
+ pl.lit("|")
+ pl.col("filename").cast(pl.Utf8)
+ pl.lit("|")
+ pl.col("word_start").cast(pl.Utf8)
).alias("token_id"),
)
df = df.with_columns(
pl.col("filename").str.split("_").list.get(0).alias("filename_prefix"),
pl.col("word").map_elements(normalize_word_query, return_dtype=pl.Utf8).alias(
"word_search_key"
),
)
df = df.with_columns(
(
pl.col("Speaker").cast(pl.Utf8)
+ pl.lit("|")
+ pl.col("filename_prefix").cast(pl.Utf8)
+ pl.lit("|")
+ pl.col("word_start").cast(pl.Utf8)
+ pl.lit("|")
+ pl.col("word").cast(pl.Utf8)
).alias("word_occurrence_id"),
)
log.info("Loaded %d rows, %d unique tokens", df.height, df.select("token_id").n_unique())
return df
# Audio file location in the KLHData repo. Verified 2026-04-23.
# Use raw.githubusercontent.com directly (the destination of the github.com/raw
# 302 redirect) — native <audio> elements don't reliably follow cross-origin
# redirects for media, which made the audio fail silently in the browser even
# though curl could fetch it.
AUDIO_URL_BASE = "https://raw.githubusercontent.com/tkettig/KLHData/main"
# Per-speaker subfolder under {Speaker}/. Most speakers store sounds under
# output/sounds/; LV uses output_dissertation/sounds/. Verified by HEAD-checking
# every speaker against a real filename from each speaker's tokens.
SPEAKER_SOUNDS_SUBPATH: dict[str, str] = {
"AA": "output/sounds",
"DK": "output/sounds",
"HM": "output/sounds",
"IN": "output/sounds",
"JM": "output/sounds",
"LV": "output_dissertation/sounds",
"RM": "output/sounds",
"SB": "output/sounds",
}
DEFAULT_SOUNDS_SUBPATH = "output/sounds"
def audio_url(speaker: str, filename: str) -> str:
sub = SPEAKER_SOUNDS_SUBPATH.get(speaker, DEFAULT_SOUNDS_SUBPATH)
return f"{AUDIO_URL_BASE}/{speaker}/{sub}/{filename}.wav"
def interview_url(speaker: str) -> str | None:
episode = SPEAKER_EPISODES.get(speaker)
if not episode:
return None
return f"https://ulukau.org/kaniaina/?a=d&d=A-KLH-HV24-{episode}&{INTERVIEW_URL_SUFFIX}"
def filename_prefix(filename: str) -> str:
"""Prefix used to look up an interview offset.
Filenames look like 'KLH057a_1805' — the underscore-prefixed segment is
the per-utterance index, and everything before it identifies the full
interview file.
"""
return filename.split("_", 1)[0]
def token_detail(df: pl.DataFrame, prefix_offsets: dict[str, float], token_id: str) -> dict | None:
"""Return enriched metadata for a single token, or None if not found."""
rows = df.filter(pl.col("token_id") == token_id)
if rows.height == 0:
return None
# Token-level fields are constant across rows; pick the first.
first = rows.head(1).to_dicts()[0]
speaker = first["Speaker"]
filename = first["filename"]
prefix = filename_prefix(filename)
offset = prefix_offsets.get(prefix)
interview_seconds = (offset + float(first["start"])) if offset is not None else None
return {
"token_id": token_id,
"speaker": speaker,
"filename": filename,
"word": first["word"],
"vowel": first["vowel"],
"stress": first["stress"],
"previous_sound": first.get("previous_sound"),
"next_sound": first.get("next_sound"),
"start": float(first["start"]),
"audio_url": audio_url(speaker, filename),
"interview_url": interview_url(speaker),
"interview_seconds": interview_seconds,
"interview_offset_available": offset is not None,
}
def build_store() -> DataStore:
df = load_dataframe()
offsets, loaded = load_prefix_offsets()
return DataStore(df=df, prefix_offsets=offsets, loaded=loaded)
def metadata_payload(df: pl.DataFrame) -> dict:
speakers = sorted(df.get_column("Speaker").drop_nulls().unique().to_list())
vowels = sorted(df.get_column("vowel").drop_nulls().unique().to_list())
stresses = sorted(df.get_column("stress").drop_nulls().unique().to_list())
prev_sounds = sorted(df.get_column("previous_sound").drop_nulls().unique().to_list())
next_sounds = sorted(df.get_column("next_sound").drop_nulls().unique().to_list())
vowel_types = {v: t for v in vowels if (t := classify_vowel(v)) is not None}
return {
"speakers": speakers,
"vowels": vowels,
"stresses": stresses,
"prev_sounds": prev_sounds,
"next_sounds": next_sounds,
"vowel_types": vowel_types,
"function_word_columns": list(FUNCTION_WORD_COLUMNS),
}
def filter_tokens(
df: pl.DataFrame,
*,
speakers: list[str] | None,
vowels: list[str] | None,
stresses: list[str] | None,
function_include: list[str] | None = None,
function_exclude: list[str] | None = None,
word_q: str | None = None,
) -> pl.DataFrame:
out = df
if speakers:
out = out.filter(pl.col("Speaker").is_in(speakers))
if vowels:
out = out.filter(pl.col("vowel").is_in(vowels))
if stresses:
out = out.filter(pl.col("stress").is_in(stresses))
for col in _valid_function_columns(function_include):
if col not in out.columns:
continue
out = out.filter(pl.col(col).fill_null(0).cast(pl.Int64) == 1)
for col in _valid_function_columns(function_exclude):
if col not in out.columns:
continue
out = out.filter(pl.col(col).fill_null(0).cast(pl.Int64) != 1)
if word_q:
query_key = normalize_word_query(word_q)
if query_key:
out = out.filter(pl.col("word_search_key").str.contains(query_key, literal=True))
return out
def tokens_payload(filtered: pl.DataFrame, *, limit: int | None) -> dict:
n_tokens = filtered.select("token_id").n_unique()
out = filtered
if limit is not None and n_tokens > limit:
# Stratify across speakers so multi-speaker selections don't get
# silently dominated by whichever speaker appears first in the CSV.
# Round-robin fill so uneven speaker counts still use the full budget.
speaker_ids = out.select(["Speaker", "token_id"]).unique(maintain_order=True)
speakers = sorted(str(speaker) for speaker in speaker_ids.get_column("Speaker").unique().to_list())
ids_by_speaker: dict[str, list[str]] = {}
max_len = 0
for sp in speakers:
ids = [
str(token_id)
for token_id in (
speaker_ids.filter(pl.col("Speaker") == sp)
.get_column("token_id")
.to_list()
)
]
ids_by_speaker[sp] = ids
max_len = max(max_len, len(ids))
keep_ids: list[str] = []
for idx in range(max_len):
for sp in speakers:
ids = ids_by_speaker[sp]
if idx < len(ids):
keep_ids.append(ids[idx])
if len(keep_ids) >= limit:
break
if len(keep_ids) >= limit:
break
out = out.filter(pl.col("token_id").is_in(keep_ids))
rows = out.rename({"Speaker": "speaker"}).to_dicts()
for row in rows:
row["audio_url"] = audio_url(str(row["speaker"]), str(row["filename"]))
row["function_flags"] = _function_flags(row)
return {"n_tokens": n_tokens, "n_rows": len(rows), "rows": rows}