diff --git a/openmed/core/decoding/__init__.py b/openmed/core/decoding/__init__.py index 12d9170d..ca952e23 100644 --- a/openmed/core/decoding/__init__.py +++ b/openmed/core/decoding/__init__.py @@ -17,10 +17,12 @@ edge_f1, ) from .spans import ( + IndicSpanRefinement, TokenClassificationSpan, TokenClassificationStreamEvent, coerce_token_classification_spans, reconcile_stream_spans, + refine_indic_name_span, refine_privacy_filter_span, stable_span_id, stable_span_key, @@ -43,6 +45,7 @@ "EdgeDecisionTrace", "GraphExplainReport", "IncrementalViterbiState", + "IndicSpanRefinement", "SpanEdge", "SpanGraph", "SpanGraphConstraints", @@ -57,6 +60,7 @@ "edge_f1", "labels_to_token_spans", "reconcile_stream_spans", + "refine_indic_name_span", "refine_privacy_filter_span", "resolve_viterbi_biases", "stable_span_key", diff --git a/openmed/core/decoding/spans.py b/openmed/core/decoding/spans.py index e99d3a18..8da9d0d6 100644 --- a/openmed/core/decoding/spans.py +++ b/openmed/core/decoding/spans.py @@ -10,9 +10,12 @@ import hashlib import re +from collections.abc import Iterable from dataclasses import dataclass, replace from typing import Any, Final +from openmed.core.labels import supports_name_boundary_refinement + def trim_span_whitespace(start: int, end: int, text: str) -> tuple[int, int]: """Strip leading and trailing whitespace from ``text[start:end]``. @@ -34,11 +37,189 @@ def trim_span_whitespace(start: int, end: int, text: str) -> tuple[int, int]: ) +@dataclass(frozen=True) +class IndicSpanRefinement: + """An optional name-boundary refinement with absolute source offsets.""" + + original_start: int + original_end: int + start: int + end: int + applied: bool + reason: str + offset_map: tuple[tuple[int, int], ...] + grapheme_origins: tuple[tuple[int, int], ...] + rule: str | None = None + + def to_source_span(self, start: int, end: int) -> tuple[int, int]: + """Map a span relative to the refined text back to source offsets.""" + if not (0 <= start <= end <= len(self.offset_map)): + raise ValueError("span must fit within the refined output") + if start == end: + if start < len(self.offset_map): + anchor = self.offset_map[start][0] + elif self.offset_map: + anchor = self.offset_map[-1][1] + else: + anchor = self.start + return anchor, anchor + return self.offset_map[start][0], self.offset_map[end - 1][1] + + +def refine_indic_name_span( + label: str, + start: int, + end: int, + text: str, + *, + enabled: bool = False, + language: str | None = None, + confidence: float = 0.0, + allowed_stems: Iterable[str] = (), + minimum_confidence: float = 0.9, + minimum_stem_graphemes: int = 2, +) -> IndicSpanRefinement: + """Optionally tighten an Indic person-name span to an allow-listed stem. + + The disabled path returns the input offsets exactly. When enabled, the + span changes only if the label is name-like, confidence and allow-list + gates pass, and both the model span and the proposed end are grapheme + boundaries in the full source text. + + Args: + label: Source entity label. + start: Inclusive source code-point offset. + end: Exclusive source code-point offset. + text: Full source text. + enabled: Opt-in switch; false preserves the original offsets. + language: Supported Indic language code or name. + confidence: Upstream entity confidence. + allowed_stems: Caller-approved exact name stems. + minimum_confidence: Minimum confidence required to refine. + minimum_stem_graphemes: Minimum graphemes retained in the name. + + Returns: + An :class:`IndicSpanRefinement` carrying absolute offset mappings. + """ + if not (0 <= start <= end <= len(text)): + raise ValueError("span must satisfy 0 <= start <= end <= len(text)") + + unchanged = _indic_refinement_result( + text, + original_start=start, + original_end=end, + start=start, + end=end, + applied=False, + reason="disabled" if not enabled else "guard_rejected", + ) + if not enabled: + return unchanged + if language is None or not supports_name_boundary_refinement(label): + return unchanged + + # Import lazily to keep decoding helpers independent during processing + # package initialization. + from openmed.processing.morphology import grapheme_spans, stem_token + + source_graphemes = grapheme_spans(text) + source_boundaries = {0, *(cluster_end for _, cluster_end in source_graphemes)} + if start not in source_boundaries or end not in source_boundaries: + return _indic_refinement_result( + text, + original_start=start, + original_end=end, + start=start, + end=end, + applied=False, + reason="unaligned_source_span", + ) + + analysis = stem_token( + text[start:end], + language, + confidence=confidence, + allowed_stems=allowed_stems, + minimum_confidence=minimum_confidence, + minimum_stem_graphemes=minimum_stem_graphemes, + ) + if not analysis.applied: + return unchanged + + local_start, local_end = analysis.stem_span + refined_start = start + local_start + refined_end = start + local_end + mapped = analysis.offset_map.to_source_span(0, len(analysis.stem)) + if ( + local_start != 0 + or mapped != analysis.stem_span + or refined_start not in source_boundaries + or refined_end not in source_boundaries + or text[refined_start:refined_end] != analysis.stem + ): + return _indic_refinement_result( + text, + original_start=start, + original_end=end, + start=start, + end=end, + applied=False, + reason="offset_safety_rejected", + ) + + return _indic_refinement_result( + text, + original_start=start, + original_end=end, + start=refined_start, + end=refined_end, + applied=True, + reason="allowlisted_suffix", + rule=analysis.rule, + ) + + +def _indic_refinement_result( + text: str, + *, + original_start: int, + original_end: int, + start: int, + end: int, + applied: bool, + reason: str, + rule: str | None = None, +) -> IndicSpanRefinement: + from openmed.processing.morphology import grapheme_spans + + return IndicSpanRefinement( + original_start=original_start, + original_end=original_end, + start=start, + end=end, + applied=applied, + reason=reason, + offset_map=tuple((index, index + 1) for index in range(start, end)), + grapheme_origins=tuple( + (cluster_start, cluster_end) + for cluster_start, cluster_end in grapheme_spans(text) + if start <= cluster_start and cluster_end <= end + ), + rule=rule, + ) + + def refine_privacy_filter_span( label: str, start: int, end: int, text: str, + *, + indic_morphology: bool = False, + language: str | None = None, + confidence: float = 0.0, + morphology_allowlist: Iterable[str] = (), + minimum_morphology_confidence: float = 0.9, ) -> tuple[int, int]: """Tighten obvious structured-PII spans when the model absorbs glue words. @@ -62,7 +243,21 @@ def refine_privacy_filter_span( if span_text.lower().endswith(suffix): end -= len(suffix) break - return trim_span_whitespace(start, end, text) + start, end = trim_span_whitespace(start, end, text) + if indic_morphology: + refinement = refine_indic_name_span( + label, + start, + end, + text, + enabled=True, + language=language, + confidence=confidence, + allowed_stems=morphology_allowlist, + minimum_confidence=minimum_morphology_confidence, + ) + return refinement.start, refinement.end + return start, end def _byte_offset(text: str, char_offset: int) -> int: @@ -325,10 +520,12 @@ def stable_span_key(span: Any) -> tuple[int, int, str, str]: __all__ = [ + "IndicSpanRefinement", "TokenClassificationSpan", "TokenClassificationStreamEvent", "coerce_token_classification_spans", "reconcile_stream_spans", + "refine_indic_name_span", "refine_privacy_filter_span", "stable_span_id", "stable_span_key", diff --git a/openmed/core/labels.py b/openmed/core/labels.py index 49df9a4c..30ec1ca4 100644 --- a/openmed/core/labels.py +++ b/openmed/core/labels.py @@ -283,6 +283,13 @@ } ) +# Boundary morphology is intentionally limited to labels that unambiguously +# represent a person's name. Prefixes and usernames are excluded because a +# suffix-like surface may be part of the identifier itself. +NAME_BOUNDARY_REFINEMENT_LABELS: Final[FrozenSet[str]] = frozenset( + {PERSON, FIRST_NAME, LAST_NAME, MIDDLE_NAME} +) + # --------------------------------------------------------------------------- # Policy metadata @@ -1080,6 +1087,11 @@ def normalize_label(label: str, lang: str = "en") -> str: return OTHER +def supports_name_boundary_refinement(label: str, lang: str = "en") -> bool: + """Return whether ``label`` is eligible for conservative name stemming.""" + return normalize_label(label, lang=lang) in NAME_BOUNDARY_REFINEMENT_LABELS + + def id_subtype_for(label: str, lang: str = "en") -> str | None: """Return optional ID_NUM subtype metadata for a source label. @@ -1120,7 +1132,9 @@ def hipaa_class_for(label: str, lang: str = "en") -> str: __all__ = [ "CANONICAL_LABELS", + "NAME_BOUNDARY_REFINEMENT_LABELS", "normalize_label", + "supports_name_boundary_refinement", "id_subtype_for", "ID_ALIAS_SUBTYPES", "ID_SUBTYPES", diff --git a/openmed/processing/morphology.py b/openmed/processing/morphology.py new file mode 100644 index 00000000..f296c0bd --- /dev/null +++ b/openmed/processing/morphology.py @@ -0,0 +1,550 @@ +"""Conservative Indic morphology helpers with source-offset preservation. + +The rules in this module intentionally cover only frequent case markers and +postpositions whose surface form can be removed at an existing grapheme +boundary. They are not a general-purpose stemmer. A rule is applied only +when both a confidence threshold and a caller-supplied stem allow-list agree, +which keeps the default behavior precision-first for PHI name boundaries. +""" + +from __future__ import annotations + +import unicodedata +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import Final + +SUPPORTED_INDIC_MORPHOLOGY_LANGUAGES: Final[tuple[str, ...]] = ( + "hi", + "mr", + "ta", + "te", + "ml", + "kn", +) + + +@dataclass(frozen=True) +class _SuffixRule: + surface: str + separated: str + separated_char_origins: tuple[tuple[int, int], ...] + + +def _suffix_rule(surface: str) -> _SuffixRule: + return _SuffixRule( + surface=surface, + separated=surface, + separated_char_origins=tuple( + (index, index + 1) for index in range(len(surface)) + ), + ) + + +def _tamil_glide_rule(surface: str, separated: str) -> _SuffixRule: + """Reverse an initial Tamil ``வ`` glide and retain its source alignment.""" + return _SuffixRule( + surface=surface, + separated=separated, + separated_char_origins=( + (0, 2), + *((index + 1, index + 2) for index in range(1, len(separated))), + ), + ) + + +# Longest forms precede their suffixes so matching is deterministic. These +# tables are deliberately small: expanding them requires precision fixtures. +_SUFFIX_RULES: Final[Mapping[str, tuple[_SuffixRule, ...]]] = { + "hi": tuple( + _suffix_rule(surface) + for surface in ("में", "पर", "को", "से", "ने", "का", "की", "के") + ), + "mr": ( + *( + _suffix_rule(surface) + for surface in ( + "पासून", + "मध्ये", + "बरोबर", + "साठी", + "कडे", + "वरून", + "ला", + "ने", + "चा", + "ची", + "चे", + "वर", + ) + ), + ), + "ta": ( + _tamil_glide_rule("விடமிருந்து", "இடமிருந்து"), + _tamil_glide_rule("விடம்", "இடம்"), + _tamil_glide_rule("வுடன்", "உடன்"), + _tamil_glide_rule("வுக்கு", "உக்கு"), + _tamil_glide_rule("வால்", "ஆல்"), + _tamil_glide_rule("வில்", "இல்"), + _tamil_glide_rule("வின்", "இன்"), + _tamil_glide_rule("வை", "ஐ"), + ), + "te": tuple( + _suffix_rule(surface) + for surface in ("నుంచి", "యొక్క", "దగ్గర", "తో", "లో", "కు", "కి", "ని", "ను") + ), + "ml": tuple( + _suffix_rule(surface) + for surface in ( + "യിൽനിന്ന്", + "യോടൊപ്പം", + "യുടെ", + "യിൽ", + "യോട്", + "യ്ക്ക്", + "യെ", + "ൽ", + ) + ), + "kn": tuple( + _suffix_rule(surface) for surface in ("ಜೊತೆಗೆ", "ದಿಂದ", "ದಲ್ಲಿ", "ಕ್ಕೆ", "ನ್ನು", "ಗೆ") + ), +} +INDIC_SUFFIX_TABLES: Final[Mapping[str, tuple[str, ...]]] = { + language: tuple(rule.surface for rule in rules) + for language, rules in _SUFFIX_RULES.items() +} + +_LANGUAGE_ALIASES: Final[Mapping[str, str]] = { + "hi": "hi", + "hin": "hi", + "hindi": "hi", + "mr": "mr", + "mar": "mr", + "marathi": "mr", + "ta": "ta", + "tam": "ta", + "tamil": "ta", + "te": "te", + "tel": "te", + "telugu": "te", + "ml": "ml", + "mal": "ml", + "malayalam": "ml", + "kn": "kn", + "kan": "kn", + "kannada": "kn", +} + +_VIRAMAS: Final[frozenset[str]] = frozenset( + { + "\u094d", # Devanagari sign virama + "\u0bcd", # Tamil sign virama + "\u0c4d", # Telugu sign virama + "\u0ccd", # Kannada sign virama + "\u0d4d", # Malayalam sign virama + } +) +_JOINERS: Final[frozenset[str]] = frozenset({"\u200c", "\u200d"}) + + +@dataclass(frozen=True) +class MorphologyOffsetMap: + """Map each output code point to its source code-point span. + + ``char_origins[i]`` is the ``[start, end)`` source span that produced + output character ``text[i]``. Inserted sandhi separators use a zero-width + source span at the split boundary. + """ + + text: str + source: str + char_origins: tuple[tuple[int, int], ...] + + def __post_init__(self) -> None: + if len(self.char_origins) != len(self.text): + raise ValueError("char_origins must contain one entry per output character") + for start, end in self.char_origins: + if not (0 <= start <= end <= len(self.source)): + raise ValueError("char origin falls outside the source text") + + def to_source_span(self, start: int, end: int) -> tuple[int, int]: + """Map an output ``[start, end)`` span back to source offsets.""" + if not (0 <= start <= end <= len(self.text)): + raise ValueError("span must satisfy 0 <= start <= end <= len(text)") + if start == end: + if start < len(self.char_origins): + anchor = self.char_origins[start][0] + elif self.char_origins: + anchor = self.char_origins[-1][1] + else: + anchor = 0 + return anchor, anchor + return self.char_origins[start][0], self.char_origins[end - 1][1] + + +@dataclass(frozen=True) +class StemResult: + """Result of conservative suffix stripping for one token.""" + + token: str + stem: str + language: str + confidence: float + stem_span: tuple[int, int] + stripped_suffix_spans: tuple[tuple[int, int], ...] + offset_map: MorphologyOffsetMap + rule: str | None = None + applied: bool = False + + @property + def stripped_suffixes(self) -> tuple[str, ...]: + """Return source suffix surfaces selected by the rule.""" + return tuple(self.token[start:end] for start, end in self.stripped_suffix_spans) + + +@dataclass(frozen=True) +class SandhiSplitResult: + """Offset-preserving split of a fused or already-separated suffix.""" + + token: str + text: str + language: str + confidence: float + parts: tuple[str, ...] + part_spans: tuple[tuple[int, int], ...] + offset_map: MorphologyOffsetMap + rule: str | None = None + applied: bool = False + already_split: bool = False + + +def grapheme_spans(text: str) -> tuple[tuple[int, int], ...]: + """Return tailored extended-grapheme spans for the supported Indic scripts. + + The implementation keeps combining marks, virama-linked consonants, and + optional joiners in one cluster. This covers the boundary safety needed by + the Devanagari and Dravidian rules without adding a runtime dependency. + + Args: + text: Source Unicode text. + + Returns: + Ordered ``[start, end)`` code-point spans for each grapheme. + """ + spans: list[tuple[int, int]] = [] + index = 0 + while index < len(text): + start = index + index += 1 + links_next_letter = False + while index < len(text): + char = text[index] + category = unicodedata.category(char) + if category in {"Mc", "Me", "Mn"}: + links_next_letter = links_next_letter or char in _VIRAMAS + index += 1 + continue + if char in _JOINERS: + index += 1 + continue + if links_next_letter and category.startswith("L"): + links_next_letter = False + index += 1 + continue + break + spans.append((start, index)) + return tuple(spans) + + +def grapheme_boundaries(text: str) -> tuple[int, ...]: + """Return all safe code-point boundaries for ``text``.""" + spans = grapheme_spans(text) + return (0, *(end for _, end in spans)) + + +def stem_token( + token: str, + language: str, + *, + confidence: float, + allowed_stems: Iterable[str], + minimum_confidence: float = 0.9, + minimum_stem_graphemes: int = 2, +) -> StemResult: + """Strip one allow-listed Indic inflectional suffix conservatively. + + Both the confidence and exact stem allow-list gates are mandatory. The + returned stem is always an exact prefix of ``token`` and its offset map + therefore reconstructs directly from the source. + + Args: + token: One potentially inflected name surface. + language: Supported ISO language code or language name. + confidence: Upstream entity confidence in the inclusive range 0..1. + allowed_stems: Caller-approved name stems. An empty collection is a + fail-closed no-op. + minimum_confidence: Minimum upstream confidence required to strip. + minimum_stem_graphemes: Minimum number of graphemes retained. + + Returns: + A :class:`StemResult` with explicit source offsets. + """ + normalized_language = _normalize_language(language) + _validate_thresholds(confidence, minimum_confidence, minimum_stem_graphemes) + unchanged = _unchanged_stem(token, normalized_language, confidence) + if not token or any(char.isspace() for char in token): + return unchanged + match = _joined_suffix_match( + token, + normalized_language, + confidence=confidence, + minimum_confidence=minimum_confidence, + allowed_stems=allowed_stems, + minimum_stem_graphemes=minimum_stem_graphemes, + ) + if match is None: + return unchanged + + stem, rule, boundary = match + origins = tuple((index, index + 1) for index in range(boundary)) + return StemResult( + token=token, + stem=stem, + language=normalized_language, + confidence=confidence, + stem_span=(0, boundary), + stripped_suffix_spans=((boundary, len(token)),), + offset_map=MorphologyOffsetMap( + text=stem, + source=token, + char_origins=origins, + ), + rule=rule.surface, + applied=True, + ) + + +def split_sandhi( + token: str, + language: str, + *, + confidence: float, + allowed_stems: Iterable[str], + minimum_confidence: float = 0.9, + minimum_stem_graphemes: int = 2, +) -> SandhiSplitResult: + """Split a frequent fused case marker while preserving source offsets. + + A fused input is rendered with one inserted ASCII space. Re-applying the + function to that rendered text detects the existing split and leaves it + byte-for-byte unchanged. + + Args: + token: Fused or already-separated surface. + language: Supported ISO language code or language name. + confidence: Upstream entity confidence in the inclusive range 0..1. + allowed_stems: Caller-approved name stems. + minimum_confidence: Minimum confidence required to split. + minimum_stem_graphemes: Minimum number of graphemes in the stem. + + Returns: + A :class:`SandhiSplitResult` with part spans and an output offset map. + """ + normalized_language = _normalize_language(language) + _validate_thresholds(confidence, minimum_confidence, minimum_stem_graphemes) + allowed = _normalized_allowlist(allowed_stems) + + separated = _already_split_match( + token, + normalized_language, + confidence=confidence, + minimum_confidence=minimum_confidence, + allowed_stems=allowed, + minimum_stem_graphemes=minimum_stem_graphemes, + ) + if separated is not None: + stem, rule, stem_end, suffix_start = separated + return SandhiSplitResult( + token=token, + text=token, + language=normalized_language, + confidence=confidence, + parts=(stem, rule.separated), + part_spans=((0, stem_end), (suffix_start, len(token))), + offset_map=_identity_offset_map(token), + rule=rule.surface, + already_split=True, + ) + + joined = _joined_suffix_match( + token, + normalized_language, + confidence=confidence, + minimum_confidence=minimum_confidence, + allowed_stems=allowed, + minimum_stem_graphemes=minimum_stem_graphemes, + ) + if joined is None: + return SandhiSplitResult( + token=token, + text=token, + language=normalized_language, + confidence=confidence, + parts=(token,), + part_spans=((0, len(token)),), + offset_map=_identity_offset_map(token), + ) + + stem, rule, boundary = joined + rendered = f"{stem} {rule.separated}" + origins = ( + *((index, index + 1) for index in range(boundary)), + (boundary, boundary), + *( + (boundary + source_start, boundary + source_end) + for source_start, source_end in rule.separated_char_origins + ), + ) + return SandhiSplitResult( + token=token, + text=rendered, + language=normalized_language, + confidence=confidence, + parts=(stem, rule.separated), + part_spans=((0, boundary), (boundary, len(token))), + offset_map=MorphologyOffsetMap( + text=rendered, + source=token, + char_origins=tuple(origins), + ), + rule=rule.surface, + applied=True, + ) + + +def _normalize_language(language: str) -> str: + normalized = _LANGUAGE_ALIASES.get(language.strip().casefold()) + if normalized is None: + supported = ", ".join(SUPPORTED_INDIC_MORPHOLOGY_LANGUAGES) + raise ValueError( + f"unsupported morphology language; expected one of: {supported}" + ) + return normalized + + +def _validate_thresholds( + confidence: float, + minimum_confidence: float, + minimum_stem_graphemes: int, +) -> None: + if not 0.0 <= confidence <= 1.0: + raise ValueError("confidence must be between 0 and 1") + if not 0.0 <= minimum_confidence <= 1.0: + raise ValueError("minimum_confidence must be between 0 and 1") + if minimum_stem_graphemes < 1: + raise ValueError("minimum_stem_graphemes must be at least 1") + + +def _normalized_allowlist(allowed_stems: Iterable[str]) -> frozenset[str]: + return frozenset( + unicodedata.normalize("NFC", stem).casefold() for stem in allowed_stems if stem + ) + + +def _allowed(stem: str, allowed_stems: frozenset[str]) -> bool: + return unicodedata.normalize("NFC", stem).casefold() in allowed_stems + + +def _joined_suffix_match( + token: str, + language: str, + *, + confidence: float, + minimum_confidence: float, + allowed_stems: Iterable[str], + minimum_stem_graphemes: int, +) -> tuple[str, _SuffixRule, int] | None: + if confidence < minimum_confidence: + return None + allowed = ( + allowed_stems + if isinstance(allowed_stems, frozenset) + else _normalized_allowlist(allowed_stems) + ) + if not allowed: + return None + boundaries = frozenset(grapheme_boundaries(token)) + for rule in _SUFFIX_RULES[language]: + if not token.endswith(rule.surface) or len(token) <= len(rule.surface): + continue + boundary = len(token) - len(rule.surface) + stem = token[:boundary] + if boundary not in boundaries: + continue + if len(grapheme_spans(stem)) < minimum_stem_graphemes: + continue + if _allowed(stem, allowed): + return stem, rule, boundary + return None + + +def _already_split_match( + token: str, + language: str, + *, + confidence: float, + minimum_confidence: float, + allowed_stems: frozenset[str], + minimum_stem_graphemes: int, +) -> tuple[str, _SuffixRule, int, int] | None: + if confidence < minimum_confidence or not allowed_stems: + return None + for rule in _SUFFIX_RULES[language]: + if not token.endswith(rule.separated): + continue + suffix_start = len(token) - len(rule.separated) + stem_end = suffix_start + while stem_end > 0 and token[stem_end - 1].isspace(): + stem_end -= 1 + if stem_end == suffix_start: + continue + stem = token[:stem_end] + if len(grapheme_spans(stem)) >= minimum_stem_graphemes and _allowed( + stem, allowed_stems + ): + return stem, rule, stem_end, suffix_start + return None + + +def _identity_offset_map(text: str) -> MorphologyOffsetMap: + return MorphologyOffsetMap( + text=text, + source=text, + char_origins=tuple((index, index + 1) for index in range(len(text))), + ) + + +def _unchanged_stem(token: str, language: str, confidence: float) -> StemResult: + return StemResult( + token=token, + stem=token, + language=language, + confidence=confidence, + stem_span=(0, len(token)), + stripped_suffix_spans=(), + offset_map=_identity_offset_map(token), + ) + + +__all__ = [ + "INDIC_SUFFIX_TABLES", + "SUPPORTED_INDIC_MORPHOLOGY_LANGUAGES", + "MorphologyOffsetMap", + "SandhiSplitResult", + "StemResult", + "grapheme_boundaries", + "grapheme_spans", + "split_sandhi", + "stem_token", +] diff --git a/tests/fixtures/indic_morphology.json b/tests/fixtures/indic_morphology.json new file mode 100644 index 00000000..c21f0f04 --- /dev/null +++ b/tests/fixtures/indic_morphology.json @@ -0,0 +1,55 @@ +{ + "cases": [ + { + "language": "hi", + "context": "रोगी रामको आज देखा गया।", + "surface": "रामको", + "stem": "राम", + "suffix": "को" + }, + { + "language": "mr", + "context": "मीराला औषध देण्यात आले.", + "surface": "मीराला", + "stem": "मीरा", + "suffix": "ला" + }, + { + "language": "ta", + "context": "லதாவிடம் அறிக்கை வழங்கப்பட்டது.", + "surface": "லதாவிடம்", + "stem": "லதா", + "suffix": "விடம்", + "split_suffix": "இடம்" + }, + { + "language": "te", + "context": "సీతతో వైద్యుడు మాట్లాడారు.", + "surface": "సీతతో", + "stem": "సీత", + "suffix": "తో" + }, + { + "language": "ml", + "context": "ലതയുടെ റിപ്പോർട്ട് തയ്യാറായി.", + "surface": "ലതയുടെ", + "stem": "ലത", + "suffix": "യുടെ" + }, + { + "language": "kn", + "context": "ಸೀತಾಗೆ ಔಷಧ ನೀಡಲಾಯಿತು.", + "surface": "ಸೀತಾಗೆ", + "stem": "ಸೀತಾ", + "suffix": "ಗೆ" + } + ], + "held_out_names": [ + {"language": "hi", "name": "मेनका"}, + {"language": "mr", "name": "चेतने"}, + {"language": "ta", "name": "பாவை"}, + {"language": "te", "name": "రజని"}, + {"language": "ml", "name": "അനിൽ"}, + {"language": "kn", "name": "ಮಲ್ಲಿಗೆ"} + ] +} diff --git a/tests/unit/core/test_indic_span_refinement.py b/tests/unit/core/test_indic_span_refinement.py new file mode 100644 index 00000000..84aa08f7 --- /dev/null +++ b/tests/unit/core/test_indic_span_refinement.py @@ -0,0 +1,184 @@ +"""Boundary-refinement acceptance gates for Indic inflected names.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openmed.core.decoding.spans import ( + refine_indic_name_span, + refine_privacy_filter_span, +) +from openmed.core.labels import supports_name_boundary_refinement +from openmed.processing.morphology import grapheme_boundaries + +_FIXTURE_PATH = Path(__file__).parents[2] / "fixtures" / "indic_morphology.json" + + +@pytest.fixture(scope="module") +def morphology_fixture() -> dict[str, list[dict[str, str]]]: + """Load the synthetic cross-script morphology fixture.""" + return json.loads(_FIXTURE_PATH.read_text(encoding="utf-8")) + + +def test_refinement_tightens_fixture_spans_and_reconstructs_source_exactly( + morphology_fixture, +) -> None: + for case in morphology_fixture["cases"]: + text = case["context"] + original_start = text.index(case["surface"]) + original_end = original_start + len(case["surface"]) + result = refine_indic_name_span( + "B-NAME", + original_start, + original_end, + text, + enabled=True, + language=case["language"], + confidence=0.99, + allowed_stems={case["stem"]}, + ) + + assert result.applied + assert text[result.start : result.end] == case["stem"] + assert result.start == original_start + assert result.end < original_end + assert result.start in grapheme_boundaries(text) + assert result.end in grapheme_boundaries(text) + assert result.to_source_span(0, result.end - result.start) == ( + result.start, + result.end, + ) + assert ( + "".join(text[start:end] for start, end in result.grapheme_origins) + == (case["stem"]) + ) + + +def test_privacy_filter_hook_is_opt_in_and_disabled_path_is_identical( + morphology_fixture, +) -> None: + baselines: list[tuple[int, int]] = [] + explicit_disabled: list[tuple[int, int]] = [] + for case in morphology_fixture["cases"]: + text = case["context"] + start = text.index(case["surface"]) + end = start + len(case["surface"]) + baselines.append(refine_privacy_filter_span("NAME", start, end, text)) + explicit_disabled.append( + refine_privacy_filter_span( + "NAME", + start, + end, + text, + indic_morphology=False, + language=case["language"], + confidence=0.99, + morphology_allowlist={case["stem"]}, + ) + ) + + regression_samples = [ + ("EMAIL", "alice@hospital.org and another"), + ("PHONE", "+1 212-555-0100"), + ("NAME", " Alice and "), + ] + for label, text in regression_samples: + start, end = 0, len(text) + baselines.append(refine_privacy_filter_span(label, start, end, text)) + explicit_disabled.append( + refine_privacy_filter_span( + label, + start, + end, + text, + indic_morphology=False, + ) + ) + + assert json.dumps(baselines).encode() == json.dumps(explicit_disabled).encode() + + +def test_privacy_filter_hook_refines_when_explicitly_enabled() -> None: + text = "रोगी रामको आज देखा गया।" + start = text.index("रामको") + end = start + len("रामको") + + refined = refine_privacy_filter_span( + "PERSON", + start, + end, + text, + indic_morphology=True, + language="hi", + confidence=0.99, + morphology_allowlist={"राम"}, + ) + + assert text[slice(*refined)] == "राम" + + +def test_non_name_labels_low_confidence_and_missing_allowlist_are_noops() -> None: + text = "रामको" + original = (0, len(text)) + + non_name = refine_indic_name_span( + "LOCATION", + *original, + text, + enabled=True, + language="hi", + confidence=1.0, + allowed_stems={"राम"}, + ) + assert (non_name.start, non_name.end) == original + + for confidence, allowlist in ((0.5, {"राम"}), (1.0, set())): + result = refine_indic_name_span( + "PERSON", + *original, + text, + enabled=True, + language="hi", + confidence=confidence, + allowed_stems=allowlist, + ) + assert not result.applied + assert (result.start, result.end) == original + + +def test_model_span_inside_a_grapheme_is_rejected() -> None: + text = "रामको" + result = refine_indic_name_span( + "PERSON", + 1, + len(text), + text, + enabled=True, + language="hi", + confidence=1.0, + allowed_stems={"ाम"}, + ) + + assert not result.applied + assert result.reason == "unaligned_source_span" + assert (result.start, result.end) == (1, len(text)) + + +@pytest.mark.parametrize( + "label,expected", + [ + ("PERSON", True), + ("B-NAME", True), + ("FIRST_NAME", True), + ("PREFIX", False), + ("USERNAME", False), + ("LOCATION", False), + ], +) +def test_label_allowlist_is_restricted_to_person_names( + label: str, expected: bool +) -> None: + assert supports_name_boundary_refinement(label) is expected diff --git a/tests/unit/processing/test_morphology.py b/tests/unit/processing/test_morphology.py new file mode 100644 index 00000000..45d92057 --- /dev/null +++ b/tests/unit/processing/test_morphology.py @@ -0,0 +1,170 @@ +"""Precision and offset gates for conservative Indic morphology.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openmed.processing.morphology import ( + INDIC_SUFFIX_TABLES, + SUPPORTED_INDIC_MORPHOLOGY_LANGUAGES, + grapheme_boundaries, + split_sandhi, + stem_token, +) + +_FIXTURE_PATH = Path(__file__).parents[2] / "fixtures" / "indic_morphology.json" + + +@pytest.fixture(scope="module") +def morphology_fixture() -> dict[str, list[dict[str, str]]]: + """Load the synthetic cross-script morphology fixture.""" + return json.loads(_FIXTURE_PATH.read_text(encoding="utf-8")) + + +def test_fixture_covers_every_supported_language(morphology_fixture) -> None: + assert {case["language"] for case in morphology_fixture["cases"]} == set( + SUPPORTED_INDIC_MORPHOLOGY_LANGUAGES + ) + + +def test_allowlisted_stems_strip_without_losing_a_grapheme( + morphology_fixture, +) -> None: + for case in morphology_fixture["cases"]: + surface = case["surface"] + stem = case["stem"] + result = stem_token( + surface, + case["language"], + confidence=0.99, + allowed_stems={stem}, + ) + + assert result.applied + assert result.stem == stem + assert result.stem_span == (0, len(stem)) + assert result.stripped_suffixes == (case["suffix"],) + assert result.offset_map.to_source_span(0, len(result.stem)) == ( + 0, + len(stem), + ) + assert surface[slice(*result.stem_span)] == stem + assert len(stem) in grapheme_boundaries(surface) + + +def test_sandhi_split_maps_rendered_parts_back_to_the_joined_source( + morphology_fixture, +) -> None: + for case in morphology_fixture["cases"]: + surface = case["surface"] + stem = case["stem"] + source_suffix = case["suffix"] + split_suffix = case.get("split_suffix", source_suffix) + result = split_sandhi( + surface, + case["language"], + confidence=0.99, + allowed_stems={stem}, + ) + + assert result.applied + assert result.parts == (stem, split_suffix) + assert result.text == f"{stem} {split_suffix}" + assert result.offset_map.to_source_span(0, len(result.text)) == ( + 0, + len(surface), + ) + assert surface[slice(*result.part_spans[0])] == stem + assert surface[slice(*result.part_spans[1])] == source_suffix + output_suffix_start = len(stem) + 1 + assert ( + result.offset_map.to_source_span(output_suffix_start, len(result.text)) + == result.part_spans[1] + ) + if split_suffix != source_suffix: + assert result.offset_map.to_source_span( + output_suffix_start, output_suffix_start + 1 + ) == (len(stem), len(stem) + 2) + + +def test_stemming_and_sandhi_splitting_are_idempotent(morphology_fixture) -> None: + for case in morphology_fixture["cases"]: + language = case["language"] + stem = case["stem"] + first_stem = stem_token( + case["surface"], + language, + confidence=0.99, + allowed_stems={stem}, + ) + second_stem = stem_token( + first_stem.stem, + language, + confidence=0.99, + allowed_stems={stem}, + ) + assert not second_stem.applied + assert second_stem.stem == first_stem.stem + + first_split = split_sandhi( + case["surface"], + language, + confidence=0.99, + allowed_stems={stem}, + ) + second_split = split_sandhi( + first_split.text, + language, + confidence=0.99, + allowed_stems={stem}, + ) + assert second_split.already_split + assert second_split.text == first_split.text + assert second_split.parts == first_split.parts + + +def test_held_out_suffix_like_names_are_never_over_stripped( + morphology_fixture, +) -> None: + all_fixture_stems = {case["stem"] for case in morphology_fixture["cases"]} + for case in morphology_fixture["held_out_names"]: + assert any( + case["name"].endswith(suffix) + for suffix in INDIC_SUFFIX_TABLES[case["language"]] + ) + result = stem_token( + case["name"], + case["language"], + confidence=1.0, + allowed_stems=all_fixture_stems, + ) + assert not result.applied + assert result.stem == case["name"] + assert result.stripped_suffix_spans == () + + +def test_both_confidence_and_allowlist_gates_are_required() -> None: + low_confidence = stem_token( + "रामको", + "hi", + confidence=0.89, + allowed_stems={"राम"}, + ) + missing_allowlist = stem_token( + "रामको", + "hi", + confidence=0.99, + allowed_stems=(), + ) + + assert not low_confidence.applied + assert not missing_allowlist.applied + + +@pytest.mark.parametrize("language", ["bn", "pa", "or"]) +def test_out_of_scope_languages_fail_closed(language: str) -> None: + with pytest.raises(ValueError, match="unsupported morphology language"): + stem_token("example", language, confidence=1.0, allowed_stems={"exam"})