Code / scripts/ml_training/essay_scoring/features/utils.py
scripts/ml_training/essay_scoring/features/utils.py
1
"""Shared text utilities for feature extraction.
2
3
Purpose:
4
Keep LanguageTool, spaCy/readability/syntax, paragraph, and embedding text
5
normalization explicit instead of routing every consumer through one
6
lossy whitespace stream.
7
8
Relationships:
9
- Tier 1 and Tier 2 feature extractors use the boundary-preserving stream.
10
- LanguageTool evidence clients may still use the compact whitespace stream
11
where service requests need it.
12
"""
13
14
from __future__ import annotations
15
16
import re
17
from typing import TYPE_CHECKING, Iterable
18
19
import numpy as np
20
from huleedu_nlp_shared.writing_issues.text_streams import (
21
derive_spacy_feature_text_stream,
22
)
23
24
if TYPE_CHECKING:
25
from spacy.language import Language
26
27
_WHITESPACE_RE = re.compile(r"\s+")
28
_PARAGRAPH_RE = re.compile(r"\n\s*\n")
29
30
31
def normalize_whitespace(text: str) -> str:
32
"""Normalize whitespace for consistent tokenization."""
33
34
return _WHITESPACE_RE.sub(" ", text).strip()
35
36
37
def normalize_boundary_preserving_text(text: str) -> str:
38
"""Normalize inline whitespace while preserving sentence/paragraph boundaries."""
39
40
return derive_spacy_feature_text_stream(text)
41
42
43
def split_paragraphs(text: str) -> list[str]:
44
"""Split text into paragraphs using blank lines."""
45
46
cleaned = normalize_boundary_preserving_text(text)
47
if not cleaned:
48
return []
49
paragraphs = [segment.strip() for segment in _PARAGRAPH_RE.split(cleaned) if segment.strip()]
50
return paragraphs or [cleaned]
51
52
53
def tokenize_words(text: str, nlp: Language) -> list[str]:
54
"""Tokenize words using spaCy."""
55
56
if nlp is None:
57
raise RuntimeError("spaCy pipeline required for tokenization.")
58
doc = nlp(text)
59
return [token.text for token in doc if not token.is_space]
60
61
62
def tokenize_sentences(text: str, nlp: Language) -> list[str]:
63
"""Tokenize sentences using spaCy."""
64
65
if nlp is None:
66
raise RuntimeError("spaCy pipeline required for sentence tokenization.")
67
doc = nlp(text)
68
return [sent.text.strip() for sent in doc.sents if sent.text.strip()]
69
70
71
def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
72
"""Safely divide two numbers with a default fallback for zero denominators."""
73
74
if denominator == 0:
75
return default
76
return numerator / denominator
77
78
79
def density_per_100_words(count: float, word_count: float) -> float:
80
"""Convert raw counts into densities per 100 words."""
81
82
return safe_divide(count, word_count, default=0.0) * 100.0
83
84
85
def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
86
"""Compute cosine similarity between two vectors."""
87
88
denom = np.linalg.norm(vec_a) * np.linalg.norm(vec_b)
89
if denom == 0:
90
return 0.0
91
return float(np.dot(vec_a, vec_b) / denom)
92
93
94
def variance(values: Iterable[float]) -> float:
95
"""Compute variance for a sequence of floats."""
96
97
values_list = list(values)
98
if len(values_list) < 2:
99
return 0.0
100
return float(np.var(np.array(values_list, dtype=float)))