Source code for pycantonese.parsing

import collections
import concurrent.futures as cf
import functools
import multiprocessing as mp
import re
import uuid
from string import ascii_uppercase

from import _File, Utterance

from pycantonese.corpus import CHATReader, Token
from pycantonese.jyutping.characters import characters_to_jyutping
from pycantonese.pos_tagging.tagger import pos_tag

# Punctuation marks for utterance segmentation.
_UTTERANCE_PUNCT_MARKS = frozenset(("。", "!", "?"))
_ASCII_UPPERCASE = frozenset(ascii_uppercase)


_CPU_COUNT = mp.cpu_count()

def _parse_text(text: str, segment_kwargs, pos_tag_kwargs):
    chars_jps = characters_to_jyutping(text, **(segment_kwargs or {}))
    segmented, jyutping = [], []
    for chars, jps in chars_jps:
    tags = [pos for _, pos in pos_tag(segmented, **(pos_tag_kwargs or {}))]
    return segmented, tags, jyutping

def _get_utterance(
    unparsed_sent, segment_kwargs, pos_tag_kwargs, participant
) -> Utterance:
    if participant is not None:
    elif isinstance(unparsed_sent, str):
        participant = _UNKNOWN_PARTICIPANT
    elif isinstance(unparsed_sent, tuple):
        participant, unparsed_sent, *_ = unparsed_sent
        raise TypeError(
            "Utterance must be either a string or "
            f"a tuple of (participant, utterance): {unparsed_sent}"
    participant = str(participant)

    if not unparsed_sent:
        return Utterance(
            participant=participant, tokens=[], time_marks=None, tiers={participant: ""}
    words, tags, jps = _parse_text(unparsed_sent, segment_kwargs, pos_tag_kwargs)

    tokens = [
        Token(word, pos, jp, None, None, None)
        for word, pos, jp in zip(words, tags, jps)

    return Utterance(
            # TODO or question: Convert full-width punct to CHAT-styled punct?
            participant: " ".join(words),
            "%mor": " ".join(
                if pos == "PUNCT" or pos[0].upper() not in _ASCII_UPPERCASE
                else f"{pos}|{jp or ''}"
                for word, pos, jp in zip(words, tags, jps)

[docs]def parse_text( data, *, segment_kwargs=None, pos_tag_kwargs=None, participant: str = None, parallel: bool = True, ) -> CHATReader: """Parse raw Cantonese text. Parameters ---------- data : str or Iterable[str] or Iterable[Tuple[str, str]] Raw Cantonese text data, in one of the following formats: - A single string, e.g., ``"廣東話好難學?都唔係吖!"`` (which would be two utterances). Basic utterance segmentation (i.e., splitting by the end-of-line character ``\\n`` or one of the Chinese full-width punctuation marks from {"。", "!", "?"}) will be applied to this string, and each segmented utterance will be an utterance in the resulting CHAT reader. - An iterable of strings, e.g., ``["廣東話好難學?", "都唔係吖!"]``. No utterance segmentation will be done. Use this option to pass in data that's utterance-segmented to your liking. - An iterable of tuples, where each tuple has two strings, one for the participant and the other for the utterance, e.g., ``[("小芬", "你食咗飯未呀?"), ("小明", "我食咗喇。")]``. segment_kwargs : dict, optional To customize word segmentation, provide a dictionary here which would then be passed as keyword arguments to :func:`~pycantonese.segment`. pos_tag_kwargs : dict, optional To customize part-of-speech tagging, provide a dictionary here which would then be passed as keyword arguments to :func:`~pycantonese.pos_tag`. participant : str, optional If provided, this will be the participant in the output CHAT-formatted data (and will override all the particpants if your input to ``data`` is an iterable of tuples). If not provided, a default dummy participant ``"X"`` is used when your ``data`` is either a single string or an iterable of strings. parallel : bool, optional If ``True`` (the default), this function attempts to parallelize parsing for speed-up. (In case the data volume is very small, the parsing is not parallelized even if you pass in ``True``.) Under certain circumstances (e.g., your application is already parallelized and further parallelization from within this function might be undesirable), you may like to consider setting this parameter to ``False``. Returns ------- :class:`~pycantonese.CHATReader` """ if isinstance(data, str): # Perform basic sentence segmentation. for punct in _UTTERANCE_PUNCT_MARKS: data = data.replace(punct, f"{punct}\n") data = data.replace("\r\n", "\n") data = re.sub(r"\n{2,}", "\n", data) data = data.strip().split("\n") # Internally, word segmentation is actually going to be done by # `characters_to_jyutping` instead of `segment`. # `characters_to_jyutping`'s segmenter kwarg is called `segmenter`, # while that of `segment` is called `cls`. segment_kwargs = segment_kwargs or {} if "cls" in segment_kwargs: segment_kwargs["segmenter"] = segment_kwargs["cls"] del segment_kwargs["cls"] # If there's not much data, don't bother with parallelization. if parallel and len(data) < (_CPU_COUNT * _CHUNK_SIZE): parallel = False if parallel: func = functools.partial( _get_utterance, segment_kwargs=segment_kwargs, pos_tag_kwargs=pos_tag_kwargs, participant=participant, ) with cf.ProcessPoolExecutor() as executor: utterances = list(, data, chunksize=_CHUNK_SIZE)) else: utterances = [ _get_utterance(sent, segment_kwargs, pos_tag_kwargs, participant) for sent in data ] f = _File( file_path=str(uuid.uuid4()), header={}, utterances=utterances, ) reader = CHATReader() reader._files = collections.deque([f]) return reader