Pipeline

burmesenlp.pipeline.BurmeseNLP

BurmeseNLP(dictionary_path: Optional[str] = None, *, lexicon: Optional[Lexicon] = None, split_on_final_particles: bool = True, gazetteer: bool = True, gazetteer_manager: Optional[GazetteerManager] = None)

Myanmar (Burmese) NLP preprocessing pipeline.

All character positions returned by this class refer to the normalized form of the input (zero-width characters stripped, NFC applied); normalize() is exposed so callers can reproduce it.

Zawgyi is not auto-converted: call zg2uni / to_unicode before process when the encoding is Zawgyi or unknown.

Loading a dictionary that does not exist or is malformed raises LexiconError instead of silently falling back. Custom files (.json or .txt) are merged on top of the bundled default lexicon with per-word tag union; pass lexicon=... for full control.

Source code in src/burmesenlp/pipeline/__init__.py

def __init__(
    self,
    dictionary_path: Optional[str] = None,
    *,
    lexicon: Optional[Lexicon] = None,
    split_on_final_particles: bool = True,
    gazetteer: bool = True,
    gazetteer_manager: Optional[GazetteerManager] = None,
):
    if lexicon is not None:
        self.lexicon = lexicon
    elif dictionary_path is not None:
        self.lexicon = Lexicon.from_file(dictionary_path, merge_default=True)
    else:
        self.lexicon = Lexicon.default()

    self._segmenter = WordSegmenter(self.lexicon)
    self._sentencer = SentenceSegmenter(
        split_on_final_particles=split_on_final_particles
    )
    self._mwe = BMWEEngine(lexicon=self.lexicon)
    self._tagger = POSTagger(self.lexicon)
    self._chunker = PhraseChunker()
    self._clause_parser = ClauseParser()
    self._gazetteer: Optional[GazetteerManager] = None
    # Gazetteer NER after POS on post-BMWE words → Document.entities
    if gazetteer_manager is not None:
        self._gazetteer_enabled = True
        self._gazetteer = gazetteer_manager
    else:
        self._gazetteer_enabled = gazetteer
        self._gazetteer = None

syllable_segment

syllable_segment(text: str) -> List[str]

Segment text into syllables (foundation for all other steps).

Source code in src/burmesenlp/pipeline/__init__.py

def syllable_segment(self, text: str) -> List[str]:
    """Segment text into syllables (foundation for all other steps)."""
    return [t.text for t in tokenize(normalize(text))]

syllable_tokens

syllable_tokens(text: str) -> List[Token]

Syllable tokens with offsets into the normalized text.

Source code in src/burmesenlp/pipeline/__init__.py

def syllable_tokens(self, text: str) -> List[Token]:
    """Syllable tokens with offsets into the normalized text."""
    return tokenize(normalize(text))

word_segment

word_segment(text: str) -> List[str]

Segment text into words.

Source code in src/burmesenlp/pipeline/__init__.py

def word_segment(self, text: str) -> List[str]:
    """Segment text into words."""
    return [t.text for t in self._word_tokens(normalize(text))]

word_tokens

word_tokens(text: str) -> List[Token]

Word tokens with offsets into the normalized text.

Source code in src/burmesenlp/pipeline/__init__.py

def word_tokens(self, text: str) -> List[Token]:
    """Word tokens with offsets into the normalized text."""
    return self._word_tokens(normalize(text))

sentence_segment

sentence_segment(text: str) -> List[str]

Segment text into sentences (grammar-aware: POS + chunks).

Source code in src/burmesenlp/pipeline/__init__.py

def sentence_segment(self, text: str) -> List[str]:
    """Segment text into sentences (grammar-aware: POS + chunks)."""
    norm = normalize(text)
    if not norm:
        return []
    *_, sentences, _syntax, _entities = self._analyze(norm)
    return [s.text for s in sentences]

sentence_segment_with_positions

sentence_segment_with_positions(text: str) -> List[Tuple[str, int, int]]

Sentences with (start, end) offsets into the normalized text.

Guaranteed: normalize(text)[start:end] == sentence.

Source code in src/burmesenlp/pipeline/__init__.py

def sentence_segment_with_positions(self, text: str) -> List[Tuple[str, int, int]]:
    """Sentences with (start, end) offsets into the *normalized* text.

    Guaranteed: ``normalize(text)[start:end] == sentence``.
    """
    norm = normalize(text)
    if not norm:
        return []
    *_, sentences, _syntax, _entities = self._analyze(norm)
    return [(s.text, s.start, s.end) for s in sentences]

pos_tag

pos_tag(words: Sequence[str], *, mwe: Optional[Sequence[MWEToken]] = None) -> List[Tuple[str, str]]

Tag an already-segmented word list (optionally MWE-aware).

Source code in src/burmesenlp/pipeline/__init__.py

def pos_tag(
    self,
    words: Sequence[str],
    *,
    mwe: Optional[Sequence[MWEToken]] = None,
) -> List[Tuple[str, str]]:
    """Tag an already-segmented word list (optionally MWE-aware)."""
    return self._tagger.tag(words, mwe=mwe)

chunk_from_tokens

chunk_from_tokens(words: Sequence[str], pos_tags: PosInput) -> List[Chunk]

Chunk from words + POS tags (does not re-tag).

Source code in src/burmesenlp/pipeline/__init__.py

def chunk_from_tokens(
    self,
    words: Sequence[str],
    pos_tags: PosInput,
) -> List[Chunk]:
    """Chunk from words + POS tags (does not re-tag)."""
    return self._chunker.chunk(words, pos_tags)

chunk

chunk(text: str) -> List[Chunk]

Segment, MWE-merge, POS-tag, then chunk text.

Source code in src/burmesenlp/pipeline/__init__.py

def chunk(self, text: str) -> List[Chunk]:
    """Segment, MWE-merge, POS-tag, then chunk *text*."""
    pre = self.word_segment(text)
    words, mwe_spans = self._mwe.process_detailed(pre)
    return self.chunk_from_tokens(words, self.pos_tag(words, mwe=mwe_spans))

load_mwe

load_mwe(path: str, *, category: Optional[str] = None, priority: int = 0) -> int

Load an additional MWE resource (JSON/TXT) into the engine.

Source code in src/burmesenlp/pipeline/__init__.py

def load_mwe(
    self,
    path: str,
    *,
    category: Optional[str] = None,
    priority: int = 0,
) -> int:
    """Load an additional MWE resource (JSON/TXT) into the engine."""
    return self._mwe.load(path, category=category, priority=priority)

process

process(text: str) -> Document

Run the full pipeline once, with all outputs mutually consistent.

Flow: normalize → syllables → words → BMWE → POS → gazetteer NER → phrase chunk (entity spans locked as NP) → sentences → ClauseParser.

doc.entities is the semantic gazetteer layer; matching spans also appear as NP chunks with features["entity"]. Pass gazetteer=False to skip NER.

Source code in src/burmesenlp/pipeline/__init__.py

def process(self, text: str) -> Document:
    """Run the full pipeline once, with all outputs mutually consistent.

    Flow: normalize → syllables → words → BMWE → POS → gazetteer NER
    → phrase chunk (entity spans locked as NP) → sentences → ClauseParser.

    ``doc.entities`` is the semantic gazetteer layer; matching spans also
    appear as NP chunks with ``features["entity"]``. Pass ``gazetteer=False``
    to skip NER.
    """
    norm = normalize(text)
    syllable_tokens = tokenize(norm)
    if not norm:
        return Document(
            raw_text=norm,
            syllables=[],
            words=[],
            sentences=[],
            pos_tags=[],
            sentence_word_tags=[],
            chunks=[],
            mwe=[],
            entities=[],
            sentence_trees=[],
        )

    (
        _word_tokens,
        words,
        mwe_spans,
        pos_tags,
        chunks,
        sentences,
        syntax,
        entities,
    ) = self._analyze(norm)

    sentence_word_tags: List[List[Tuple[str, str]]] = [
        pos_tags[s.word_start : s.word_end] for s in sentences
    ]

    return Document(
        raw_text=norm,
        syllables=[t.text for t in syllable_tokens],
        words=words,
        sentences=[s.text for s in sentences],
        pos_tags=pos_tags,
        sentence_word_tags=sentence_word_tags,
        chunks=chunks,
        mwe=mwe_spans,
        entities=entities,
        sentence_trees=list(syntax),
    )

add_to_dictionary

add_to_dictionary(word: str, tags: Iterable[str]) -> None

Add a word with POS tags (validated; raises LexiconError).

Source code in src/burmesenlp/pipeline/__init__.py

def add_to_dictionary(self, word: str, tags: Iterable[str]) -> None:
    """Add a word with POS tags (validated; raises LexiconError)."""
    self.lexicon.add(word, tags)

save_dictionary

save_dictionary(path: str) -> None

Atomically save the current dictionary as UTF-8 JSON.

Source code in src/burmesenlp/pipeline/__init__.py

def save_dictionary(self, path: str) -> None:
    """Atomically save the current dictionary as UTF-8 JSON."""
    self.lexicon.save(path)

get_stats

get_stats(text: str) -> Dict

Basic statistics about the text.

Source code in src/burmesenlp/pipeline/__init__.py

def get_stats(self, text: str) -> Dict:
    """Basic statistics about the text."""
    result = self.process(text)
    dist: Dict[str, int] = defaultdict(int)
    for _, tag in result["pos_tags"]:
        dist[tag] += 1
    return {
        "char_count": len(result["raw_text"]),
        "syllable_count": len(result["syllables"]),
        "word_count": len(result["words"]),
        "sentence_count": len(result["sentences"]),
        "avg_words_per_sentence": (
            len(result["words"]) / max(len(result["sentences"]), 1)
        ),
        "avg_syllables_per_word": (
            len(result["syllables"]) / max(len(result["words"]), 1)
        ),
        "pos_distribution": dict(dist),
    }

extract_features_for_crf

extract_features_for_crf(text: str) -> List[Dict]

Per-syllable feature dicts for training CRF / BiLSTM-CRF models.

Source code in src/burmesenlp/pipeline/__init__.py

def extract_features_for_crf(self, text: str) -> List[Dict]:
    """Per-syllable feature dicts for training CRF / BiLSTM-CRF models."""
    tokens = tokenize(normalize(text))
    features: List[Dict] = []

    for i, tok in enumerate(tokens):
        syl = tok.text
        feat: Dict = {
            "syllable": syl,
            "len": len(syl),
            "first_char": syl[0],
            "last_char": syl[-1],
            "has_medial": any(c in MEDIALS for c in syl),
            "has_vowel_sign": any(c in VOWEL_SIGNS for c in syl),
            "has_tone": any(c in TONES for c in syl),
            "has_asat": ASAT in syl,
            "has_anusvara": ANUSVARA in syl,
            "has_stacked": STACK_VIRAMA in syl,
            "is_digit": all(c in MY_DIGITS for c in syl),
            "is_punctuation": syl in (SECTION, FULL_STOP),
        }

        if i > 0:
            feat["prev_syllable"] = tokens[i - 1].text
            feat["prev_len"] = len(tokens[i - 1].text)
        else:
            feat["BOS"] = True

        if i < len(tokens) - 1:
            feat["next_syllable"] = tokens[i + 1].text
            feat["next_len"] = len(tokens[i + 1].text)
        else:
            feat["EOS"] = True

        if len(syl) >= 2:
            feat["bigram_0_1"] = syl[:2]
            feat["bigram_-2_-1"] = syl[-2:]
        if len(syl) >= 3:
            feat["trigram_0_2"] = syl[:3]

        features.append(feat)

    return features

burmesenlp.pipeline.process

process(text: str, **kwargs) -> Document

One-shot pipeline: normalize → words → MWE → POS → gazetteer → phrases → sentences → clauses.

Does not auto-convert Zawgyi; use zg2uni / to_unicode first if needed.

Source code in src/burmesenlp/pipeline/__init__.py

def process(text: str, **kwargs) -> Document:
    """One-shot pipeline: normalize → words → MWE → POS → gazetteer → phrases → sentences → clauses.

    Does not auto-convert Zawgyi; use ``zg2uni`` / ``to_unicode`` first if needed.
    """
    return BurmeseNLP(**kwargs).process(text)

burmesenlp.pipeline.document.Document dataclass

Document(raw_text: str, syllables: List[str], words: List[str], sentences: List[str], pos_tags: List[Tuple[str, str]], sentence_word_tags: List[List[Tuple[str, str]]], chunks: List[Chunk] = list(), mwe: List[MWEToken] = list(), entities: List[GazetteerHit] = list(), sentence_trees: List[SyntaxSentence] = list())

Full-pipeline output with attribute and mapping access.

Layers stay separate::

entities  — semantic gazetteer NER (PERSON / TOWN / …)
chunks    — syntactic phrases (NP / VP / PP / …)
sentence_trees / clauses — clause syntax

For json.dump, use doc.to_dict().

clauses property

clauses: List[Clause]

Flat list of clauses from sentence_trees (syntactic layer).

to_dict

to_dict() -> Dict[str, Any]

Plain dict suitable for json.dump / json.dumps.

Source code in src/burmesenlp/pipeline/document.py

def to_dict(self) -> Dict[str, Any]:
    """Plain dict suitable for ``json.dump`` / ``json.dumps``."""
    return {
        "raw_text": self.raw_text,
        "syllables": list(self.syllables),
        "words": list(self.words),
        "sentences": list(self.sentences),
        "pos_tags": [list(pair) for pair in self.pos_tags],
        "sentence_word_tags": [
            [list(pair) for pair in sent] for sent in self.sentence_word_tags
        ],
        "mwe": [
            {
                "text": m.text,
                "tokens": list(m.tokens),
                "category": m.category,
                "start": m.start,
                "end": m.end,
                "priority": m.priority,
                "pos": m.resolved_pos(),
                "index": m.index,
            }
            for m in self.mwe
        ],
        "entities": [
            {
                "text": e.text,
                "type": e.entity_type.value,
                "start": e.start,
                "end": e.end,
                "tokens": list(e.tokens),
                "attributes": dict(e.attributes),
            }
            for e in self.entities
        ],
        "chunks": [
            {
                "type": c.type.value,
                "text": c.text,
                "tokens": list(c.tokens),
                "pos_tags": list(c.pos_tags),
                "start": c.start,
                "end": c.end,
                "features": dict(c.features),
            }
            for c in self.chunks
        ],
        "sentence_trees": [s.to_dict() for s in self.sentence_trees],
        "clauses": [c.to_dict() for c in self.clauses],
    }