Signlink / mapping.py
gaurannggg7's picture
feat: add three-tier retrieval (exact → semantic → fuzzy) to ASLDictionary
4653d09
Raw History Blame Contribute Delete
6.94 kB
import os
import re
import difflib
import zipfile
import pandas as pd
def ensure_dictionary_extracted(
zip_path: str = "content/asl_app_data/dictionary.zip",
extract_to: str = "content/asl_app_data",
) -> None:
marker = os.path.join(extract_to, "dictionary")
if not os.path.isdir(marker):
print(f"Extracting {zip_path}…")
with zipfile.ZipFile(zip_path) as z:
z.extractall(extract_to)
print("Extraction complete.")
class ASLDictionary:
"""
Load token→video-filename map from CSV and resolve paths under dictionary_dir.
Lookup chain: exact/phrase → lemmatization → semantic (VectorIndex) → fuzzy → fingerspelling.
"""
LEMMA_RULES = [
(r'ING$', ''),
(r'INGS$', ''),
(r'ED$', ''),
(r'LY$', ''),
(r'ER$', ''),
(r'EST$', ''),
(r'ION$', ''),
(r'TION$', ''),
(r'S$', ''),
]
def __init__(
self,
csv_path: str = "content/asl_app_data/asl_video_index_final_with_path_cleaned.csv",
dictionary_dir: str = "content/asl_app_data/dictionary",
fingerspelling_dir: str = "content/asl_app_data/Letters",
vector_index=None,
):
ensure_dictionary_extracted()
self.directory = dictionary_dir
self.fingerspelling_dir = fingerspelling_dir
df = pd.read_csv(csv_path)
self.map: dict[str, str] = {}
for _, row in df.iterrows():
filepath = row["path"]
fname = os.path.basename(filepath)
for col in ("token", "phrase", "word"):
val = row.get(col, "")
if pd.notna(val) and str(val).strip():
key = str(val).upper().strip()
self.map[key] = fname
self.keys = sorted(self.map.keys(), key=lambda x: len(x.split()), reverse=True)
# Walk directory for actual file paths (handles subdirectories)
self.actual_file_paths: dict[str, str] = {}
if os.path.exists(self.directory):
for root, dirs, files in os.walk(self.directory):
for file in files:
if not file.startswith('.'):
self.actual_file_paths[file] = os.path.join(root, file)
# Fingerspelling index: A-Z letter paths
self.letter_paths: dict[str, str] = {}
if os.path.exists(self.fingerspelling_dir):
for letter in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ':
path = os.path.join(self.fingerspelling_dir, f"{letter}.mp4")
if os.path.exists(path):
self.letter_paths[letter] = path
if vector_index is None:
from vector_index import get_vector_index
self._vector_index = get_vector_index(csv_path=csv_path)
else:
self._vector_index = vector_index
print(f"✅ Dictionary loaded: {len(self.map)} tokens")
def _resolve(self, token: str) -> str | None:
"""Look up token in map and resolve to absolute file path."""
fname = self.map.get(token)
if fname:
full = self.actual_file_paths.get(fname) or os.path.join(self.directory, fname)
if os.path.exists(full):
return full
return None
def _lemmatize(self, token: str) -> str | None:
"""Try stripping suffixes to find a base form in vocabulary."""
for pattern, replacement in self.LEMMA_RULES:
candidate = re.sub(pattern, replacement, token)
if candidate != token and len(candidate) > 2:
if candidate in self.map:
return candidate
return None
def _fingerspell(self, token: str) -> list[str]:
"""Break token into individual letter video paths."""
paths = []
for char in token.upper():
if char in self.letter_paths:
paths.append(self.letter_paths[char])
elif char != ' ':
print(f"⚠️ No fingerspelling for: '{char}'")
return paths
def get_paths(self, gloss_tokens: list[str]) -> list[str]:
fuzzy_cutoff = float(os.getenv("FUZZY_CUTOFF", "0.6"))
out_paths: list[str] = []
tokens = [t.upper() for t in gloss_tokens]
n = len(tokens)
i = 0
while i < n:
# Tier 1a: greedy multi-word phrase exact match
match = None
match_len = 0
for key in self.keys:
parts = key.split()
L = len(parts)
if L > 1 and i + L <= n and tokens[i : i + L] == parts:
match, match_len = key, L
break
if match:
full = self._resolve(match)
if full:
print(f"[retrieval: exact] {' '.join(tokens[i:i+match_len])} → {match}")
out_paths.append(full)
else:
print(f"⚠️ File not found for phrase '{match}'")
i += match_len
continue
tok = tokens[i]
# Tier 1b: single token exact match
full = self._resolve(tok)
if full:
print(f"[retrieval: exact] {tok}")
out_paths.append(full)
i += 1
continue
# Tier 2: lemmatization
lemma = self._lemmatize(tok)
if lemma:
full = self._resolve(lemma)
if full:
print(f"[retrieval: lemma] {tok} → {lemma}")
out_paths.append(full)
i += 1
continue
# Tier 3: semantic vector search
results = self._vector_index.query_gloss(tok, top_k=1)
if results:
matched_key, score, fname = results[0]
full = self.actual_file_paths.get(fname) or os.path.join(self.directory, fname)
if os.path.exists(full):
print(f"[retrieval: semantic] {tok} → {matched_key} ({score:.2f})")
out_paths.append(full)
i += 1
continue
# Tier 4: fuzzy match
fuzzy = difflib.get_close_matches(tok, self.map.keys(), n=1, cutoff=fuzzy_cutoff)
if fuzzy:
full = self._resolve(fuzzy[0])
if full:
print(f"[retrieval: fuzzy] {tok} → {fuzzy[0]}")
out_paths.append(full)
i += 1
continue
# Tier 5: fingerspelling fallback
print(f"⚠️ No mapping for '{tok}' — fingerspelling...")
spelled = self._fingerspell(tok)
if spelled:
out_paths.extend(spelled)
else:
print(f"⚠️ Could not fingerspell '{tok}'")
i += 1
return out_paths