Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions soynlp/core/lrgraph.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import copy
import os
from collections import defaultdict
from collections.abc import Iterable, Sized
from collections.abc import Iterable, Iterator, Sized


class LRGraph:
Expand Down Expand Up @@ -178,7 +178,7 @@ def get_original_r(self, word: str) -> dict[str, int]:
"""Return the original R-frequency dict for `word` (before any compound extraction edits)."""
return dict(self._lr_origin.get(word, {}))

def iter_original_lr(self):
def iter_original_lr(self) -> Iterator[tuple[str, dict[str, int]]]:
"""Yield (L, R_freq_dict) pairs from the original (frozen) L-R graph."""
yield from self._lr_origin.items()

Expand Down
4 changes: 2 additions & 2 deletions soynlp/predicator/eomi.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ def __init__(
min_num_of_features: int = 5,
verbose: bool = True,
logpath: str | None = None,
):
) -> None:
self.lrgraph = lrgraph
self._stems = stems
self._nouns = nouns
Expand Down Expand Up @@ -128,7 +128,7 @@ def _has_stem_at_last(self, l: str) -> bool:
return True
return False

def _refine_features(self, features: list, r: str) -> list:
def _refine_features(self, features: list[tuple[str, int]], r: str) -> list[tuple[str, int]]:
return [(l, count) for l, count in features if (l in self._stem_surfaces) and (not self._exist_longer_pos(l, r))]

def _candidates_from_stem_surfaces(self, condition: str | None = None) -> dict[str, int]:
Expand Down
34 changes: 18 additions & 16 deletions soynlp/predicator/predicator.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ def __init__(
extract_stem: bool = False,
verbose: bool = True,
ensure_normalized: bool = False,
):
) -> None:
if not josas:
josas = self._load_default_josa()
if (adjectives is None) or (verbs is None):
Expand Down Expand Up @@ -149,7 +149,7 @@ def train_extract(
min_stem_score: float = 0.7,
min_stem_frequency: int = 100,
n_workers: int = 1,
):
) -> tuple[dict, dict]:
self.train(
inputs,
min_eojeol_frequency,
Expand Down Expand Up @@ -180,7 +180,7 @@ def train(
min_stem_score: float = 0.7,
min_stem_frequency: int = 100,
n_workers: int = 1,
):
) -> None:
if isinstance(inputs, LRGraph):
self._train_with_eojeol_counter(inputs.to_EojeolCounter(), min_eojeol_frequency) # type: ignore[union-attr]
elif isinstance(inputs, EojeolCounter):
Expand Down Expand Up @@ -210,7 +210,7 @@ def train(

def _train_with_sentences(
self, sentences, min_eojeol_frequency: int = 2, filtering_checkpoint: int = 100000, n_workers: int = 1
):
) -> None:
logger.info("counting eojeols ...")

preprocess = (lambda x: x) if self.ensure_normalized else normalize_sent_for_lrgraph
Expand All @@ -224,7 +224,7 @@ def _train_with_sentences(
)
self._train_with_eojeol_counter(eojeol_counter)

def _train_with_eojeol_counter(self, eojeol_counter: EojeolCounter, min_eojeol_frequency: int = 2):
def _train_with_eojeol_counter(self, eojeol_counter: EojeolCounter, min_eojeol_frequency: int = 2) -> None:
eojeol_counter._counter = {
eojeol: count for eojeol, count in eojeol_counter._counter.items() if count >= min_eojeol_frequency
}
Expand All @@ -238,14 +238,14 @@ def _train_with_eojeol_counter(self, eojeol_counter: EojeolCounter, min_eojeol_f

logger.info("#eojeols=%d, mem=%.3f Gb", self._num_of_eojeols, get_process_memory())

def extract(self, candidates=None, min_predicator_frequency: int = 1):
def extract(self, candidates=None, min_predicator_frequency: int = 1) -> tuple[dict, dict]:
"""Extract predicators. candidates is EojeolCounter or dict format."""
self._num_of_covered_eojeols = 0
predicators = self._extract_predicator(candidates, min_predicator_frequency)
adjectives, verbs = self._separate_adjective_verb(predicators)
return adjectives, verbs

def _prepare_predicator_lrgraph(self):
def _prepare_predicator_lrgraph(self) -> LRGraph:
def contains_noun(eojeol: str) -> bool:
n = len(eojeol)
for e in range(2, n + 1):
Expand All @@ -259,11 +259,11 @@ def contains_noun(eojeol: str) -> bool:

def _extract_eomi(
self,
lrgraph,
lrgraph: LRGraph,
min_num_of_features: int = 5,
min_eomi_score: float = 0.3,
min_eomi_frequency: int = 1,
):
) -> None:
eomi_extractor = EomiExtractor(
lrgraph=lrgraph,
stems=self._stems,
Expand All @@ -287,13 +287,13 @@ def _extract_eomi(

def _extract_stem(
self,
lrgraph,
lrgraph: LRGraph,
min_num_of_unique_R_char: int = 10,
min_entropy_of_R_char: float = 0.5,
min_entropy_of_R: float = 1.5,
min_stem_score: float = 0.7,
min_stem_frequency: int = 100,
):
) -> None:
stem_extractor = StemExtractor(
lrgraph=lrgraph,
stems=self._stems,
Expand All @@ -315,7 +315,7 @@ def _extract_stem(

logger.info("stems: %d -> %d", n_before, n_after)

def _extract_predicator(self, eojeol_counter=None, min_frequency: int = 1) -> dict:
def _extract_predicator(self, eojeol_counter=None, min_frequency: int = 1) -> dict[str, Predicator]:
def all_characters_are_complete_korean(s: str) -> bool:
return all(character_is_complete_korean(c) for c in s)

Expand All @@ -332,7 +332,7 @@ def all_characters_are_complete_korean(s: str) -> bool:

return lemmas

def _as_lemma_candidates(self, eojeol_counter=None) -> dict:
def _as_lemma_candidates(self, eojeol_counter=None) -> dict[str, Predicator]:
def is_noun_josa(eojeol: str) -> bool:
for i in range(1, len(eojeol)):
if (eojeol[:i] in self._nouns) and (eojeol[i:] in self._josas):
Expand Down Expand Up @@ -378,8 +378,8 @@ def is_noun_josa(eojeol: str) -> bool:

return lemmas

def _remove_wrong_eomis(self, lemmas: dict, eomi_to_word_count: dict) -> dict:
def noun_proportion(word_count: list) -> tuple[float, float]:
def _remove_wrong_eomis(self, lemmas: dict[str, Predicator], eomi_to_word_count: dict[str, list]) -> dict[str, Predicator]:
def noun_proportion(word_count: list[tuple[str, int]]) -> tuple[float, float]:
sum_ = sum(1 for w, v in word_count if len(w) == 2)
prop = sum(1 for w, v in word_count if (w in self._nouns) and (len(w) == 2))
prop_len2 = 0.0
Expand Down Expand Up @@ -426,7 +426,9 @@ def noun_proportion(word_count: list) -> tuple[float, float]:

return lemmas

def _separate_adjective_verb(self, predicators: dict) -> tuple[dict, dict]:
def _separate_adjective_verb(
self, predicators: dict[str, Predicator]
) -> tuple[dict[str, Predicator], dict[str, Predicator]]:
adjectives: dict = {}
verbs: dict = {}

Expand Down
17 changes: 9 additions & 8 deletions soynlp/predicator/stem.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
import math
from typing import cast

from soynlp.lemmatizer import conjugate, lemma_candidate

Expand All @@ -16,7 +17,7 @@ def __init__(
min_entropy_of_R_char: float = 0.5,
min_entropy_of_R: float = 1.5,
verbose: bool = True,
):
) -> None:
self.lrgraph = lrgraph
self.stems = stems
self.eomis = eomis
Expand Down Expand Up @@ -62,7 +63,7 @@ def extract(
L_ignore: set[str] | None = None,
min_stem_score: float = 0.7,
min_stem_frequency: int = 100,
) -> dict:
) -> dict[str, tuple[float, float]]:
if L_ignore is None:
L_ignore = set()

Expand Down Expand Up @@ -95,7 +96,7 @@ def _batch_prediction(
candidates: dict[str, int],
min_stem_score: float,
min_frequency: int,
) -> dict[str, tuple[float, int] | None]:
) -> dict[str, tuple[float, int]]:
extracted: dict[str, tuple[float, int] | None] = {l: None for l in self.L}

for l in sorted(candidates, key=lambda x: -len(x)):
Expand All @@ -109,7 +110,7 @@ def _batch_prediction(

extracted[l] = (score, freq)

return {l: score for l, score in extracted.items() if l not in self.L}
return cast(dict[str, tuple[float, int]], {l: score for l, score in extracted.items() if l not in self.L})

def predict(self, l: str, min_stem_score: float = 0.7, min_frequency: int = 1, debug: bool = False) -> tuple[float, int]:
features = self.lrgraph.get_r(l, -1)
Expand Down Expand Up @@ -191,7 +192,7 @@ def _exist_longer_eomi(self, l: str, r: str) -> bool:
return True
return False

def _post_processing(self, extracted: dict) -> tuple[dict, set[str]]:
def _post_processing(self, extracted: dict[str, tuple[float, int]]) -> tuple[dict[str, tuple[float, int]], set[str]]:
def is_stem_and_eomi(l: str) -> bool:
n = len(l)
for i in range(1, n):
Expand All @@ -215,11 +216,11 @@ def exist_subword(l: str) -> bool:
extracted = {l: score for l, score in extracted.items() if l not in removals}
return extracted, removals

def _to_stem(self, surfaces: dict) -> dict:
def merge_score(freq0: int, score0: float, freq1: int, score1: float) -> tuple[int, float]:
def _to_stem(self, surfaces: dict[str, tuple[float, int]]) -> dict[str, tuple[float, float]]:
def merge_score(freq0: float, score0: float, freq1: float, score1: float) -> tuple[float, float]:
return (freq0 + freq1, (score0 * freq0 + score1 * freq1) / (freq0 + freq1))

stems: dict[str, tuple[int, float]] = {}
stems: dict[str, tuple[float, float]] = {}
for l, (freq0, score0) in surfaces.items():
for r, count in self.lrgraph.get_r(l, -1):
try:
Expand Down
Loading