50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
"""Italian → English vocabulary normalisation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from pipeline.errors import PipelineError
|
|
|
|
|
|
class Vocabulary:
|
|
"""Lookup table loaded from a vocab YAML file."""
|
|
|
|
def __init__(self, path: Path) -> None:
|
|
with path.open(encoding="utf-8") as fh:
|
|
raw = yaml.safe_load(fh) or {}
|
|
self.canonical_values: set[str] = set(raw.get("canonical_values", []) or [])
|
|
normalize = raw.get("normalize", {}) or {}
|
|
self._normalize: dict[str, str] = {
|
|
str(key).strip().lower(): str(value).strip() for key, value in normalize.items()
|
|
}
|
|
|
|
def to_english(self, italian_name: str, *, kind: str) -> str:
|
|
"""Map an Italian (or already-English) term to its canonical English form."""
|
|
key = italian_name.strip().lower()
|
|
if not key:
|
|
raise PipelineError(f"Empty {kind} name cannot be normalised.")
|
|
|
|
if key in self._normalize:
|
|
return self._normalize[key]
|
|
|
|
# Already a canonical English value (case-insensitive match).
|
|
for value in self.canonical_values:
|
|
if value.lower() == key:
|
|
return value
|
|
|
|
raise PipelineError(
|
|
f"No English mapping found for {kind} '{italian_name}'. "
|
|
f"Add it to the vocabulary file."
|
|
)
|
|
|
|
|
|
def load_crop_vocab(path: Path) -> Vocabulary:
|
|
return Vocabulary(path)
|
|
|
|
|
|
def load_disease_vocab(path: Path) -> Vocabulary:
|
|
return Vocabulary(path)
|