Merge pull request #1354 from PyThaiNLP/copilot/narrow-type-hints-update

Fix runtime AttributeError in wordnet type casts, improve type safety and API contracts

Author: bact

pythainlp/ancient/currency.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 
-def convert_currency(value: float, from_unit: str) -> dict:
+def convert_currency(value: float, from_unit: str) -> dict[str, float]:
     """Convert ancient Thai currency to other units
 
     * เบี้ย (Bia)
@@ -23,7 +23,7 @@ def convert_currency(value: float, from_unit: str) -> dict:
     :param str from_unit: currency unit \
         ('เบี้ย', 'อัฐ', 'ไพ', 'เฟื้อง', 'สลึง', 'บาท', 'ตำลึง', 'ชั่ง')
     :return: Thai currency
-    :rtype: dict
+    :rtype: dict[str, float]
 
     :Example:
     ::
@@ -63,8 +63,8 @@ def convert_currency(value: float, from_unit: str) -> dict:
     # start from 'อัฐ'
     value_in_att = value * conversion_factors_to_att[from_unit]
 
-    # Calculate values ​​in other units
-    results = {}
+    # Calculate values in other units
+    results: dict[str, float] = {}
     for unit, factor in conversion_factors_to_att.items():
         results[unit] = value_in_att / factor
 

pythainlp/augment/word2vec/bpemb_wv.py

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from pythainlp.augment.word2vec.core import Word2VecAug
 
@@ -40,7 +40,7 @@ def tokenizer(self, text: str) -> list[str]:
         """:param str text: Thai text
         :rtype: List[str]
         """
-        return self.bpemb_temp.encode(text)  # type: ignore[no-any-return]
+        return cast(list[str], self.bpemb_temp.encode(text))
 
     def load_w2v(self) -> None:
         """Load BPEmb model"""

pythainlp/augment/word2vec/core.py

@@ -42,7 +42,9 @@ def __init__(
             _gensim_kv_logger.addFilter(_filter)
             try:
                 if type == "file":
-                    self.model = word2vec.KeyedVectors.load_word2vec_format(model)
+                    self.model = word2vec.KeyedVectors.load_word2vec_format(
+                        model
+                    )
                 else:
                     self.model = word2vec.KeyedVectors.load_word2vec_format(
                         model, binary=True, unicode_errors="ignore"

pythainlp/augment/wordnet.py

@@ -122,13 +122,13 @@ class WordNetAug:
     """Text Augment using wordnet"""
 
     synonyms: list[str]
-    list_synsets: list
+    list_synsets: list[Synset]
     p2w_pos: Optional[str]
     synset: Synset
     syn: str
     synonyms_without_duplicates: list[str]
     list_words: list[str]
-    list_synonym: list
+    list_synonym: list[list[str]]
     p_all: int
     list_pos: list[tuple[str, str]]
     temp: list[str]
@@ -152,15 +152,15 @@ def find_synonyms(
         """
         self.synonyms: list[str] = []
         if pos is None:
-            self.list_synsets: list = wordnet.synsets(word)
+            self.list_synsets: list[Synset] = wordnet.synsets(word)
         else:
             self.p2w_pos: Optional[str] = postype2wordnet(pos, postag_corpus)
             if self.p2w_pos != "":
-                self.list_synsets: list = wordnet.synsets(
+                self.list_synsets: list[Synset] = wordnet.synsets(
                     word, pos=self.p2w_pos
                 )
             else:
-                self.list_synsets: list = wordnet.synsets(word)
+                self.list_synsets: list[Synset] = wordnet.synsets(word)
 
         for self.synset in wordnet.synsets(word):
             for self.syn in self.synset.lemma_names(lang="tha"):
@@ -206,7 +206,7 @@ def augment(
         """
         new_sentences = []
         self.list_words: list[str] = tokenize(sentence)
-        self.list_synonym: list = []
+        self.list_synonym: list[list[str]] = []
         self.p_all: int = 1
         if postag:
             self.list_pos: list[tuple[str, str]] = pos_tag(

pythainlp/benchmarks/word_tokenization.py

@@ -10,6 +10,7 @@
 if TYPE_CHECKING:
     import numpy as np
     import pandas as pd
+    from numpy.typing import NDArray
 
 SEPARATOR: str = "|"
 
@@ -43,7 +44,7 @@ def _f1(precision: float, recall: float) -> float:
 
 
 def _flatten_result(
-    my_dict: dict, sep: str = ":"
+    my_dict: dict[str, dict[str, Union[int, str]]], sep: str = ":"
 ) -> dict[str, Union[int, str]]:
     """Flatten two-dimension dictionary.
 
@@ -54,7 +55,8 @@ def _flatten_result(
     { "a:b": 7 }
 
 
-    :param dict my_dict: dictionary containing stats
+    :param dict[str, dict[str, Union[int, str]]] my_dict: dictionary
+        containing stats
     :param str sep: separator between the two keys (default: ":")
 
     :return: a one-dimension dictionary with keys combined
@@ -91,7 +93,7 @@ def benchmark(ref_samples: list[str], samples: list[str]) -> "pd.DataFrame":
                 flat_stats["expected"] = r
                 flat_stats["actual"] = s
                 results.append(flat_stats)
-        except Exception:
+        except Exception as exc:
             reason = """
 [Error]
 Reason: %s
@@ -107,7 +109,7 @@ def benchmark(ref_samples: list[str], samples: list[str]) -> "pd.DataFrame":
                 r,
                 s,
             )
-            raise SystemExit(reason)
+            raise SystemExit(reason) from exc
 
     return pd.DataFrame(results)
 
@@ -209,7 +211,9 @@ def compute_stats(
     }
 
 
-def _binary_representation(txt: str, verbose: bool = False) -> "np.ndarray":
+def _binary_representation(
+    txt: str, verbose: bool = False
+) -> "NDArray[np.int8]":
     """Transform text into {0, 1} sequence.
 
     where (1) indicates that the corresponding character is the beginning of
@@ -219,7 +223,7 @@ def _binary_representation(txt: str, verbose: bool = False) -> "np.ndarray":
     :param bool verbose: for debugging purposes
 
     :return: {0, 1} sequence
-    :rtype: np.ndarray
+    :rtype: numpy.typing.NDArray[numpy.int8]
     """
     import numpy as np
 
@@ -228,7 +232,7 @@ def _binary_representation(txt: str, verbose: bool = False) -> "np.ndarray":
     boundary = np.argwhere(chars == SEPARATOR).reshape(-1)
     boundary = boundary - np.array(range(boundary.shape[0]))
 
-    bin_rept: np.ndarray = np.zeros(len(txt) - boundary.shape[0])
+    bin_rept = np.zeros(len(txt) - boundary.shape[0], dtype=np.int8)
     bin_rept[list(boundary) + [0]] = 1
 
     sample_wo_seps = list(txt.replace(SEPARATOR, ""))
@@ -247,10 +251,13 @@ def _binary_representation(txt: str, verbose: bool = False) -> "np.ndarray":
     return bin_rept
 
 
-def _find_word_boundaries(bin_reps: "np.ndarray") -> list[tuple[int, int]]:
+def _find_word_boundaries(
+    bin_reps: "NDArray[np.int8]",
+) -> list[tuple[int, int]]:
     """Find the starting and ending location of each word.
 
-    :param numpy.ndarray bin_reps: binary representation of a text
+    :param numpy.typing.NDArray[numpy.int8] bin_reps: binary representation
+        of a text
 
     :return: list of tuples (start, end)
     :rtype: list[tuple[int, int]]

pythainlp/chunk/__init__.py

@@ -25,6 +25,7 @@
             print(parser.parse(tokens_pos))
         # output: ['B-NP', 'B-VP', 'I-VP']
 """
+
 from __future__ import annotations
 
 __all__: list[str] = [

pythainlp/chunk/crfchunk.py

@@ -2,10 +2,11 @@
 # SPDX-FileType: SOURCE
 # SPDX-License-Identifier: Apache-2.0
 """CRF-based Thai phrase structure (chunk) parser."""
+
 from __future__ import annotations
 
 from importlib.resources import as_file, files
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, Union, cast
 
 if TYPE_CHECKING:
     import types
@@ -130,7 +131,7 @@ def parse(self, token_pos: list[tuple[str, str]]) -> list[str]:
         :rtype: list[str]
         """
         self.xseq = _extract_features(token_pos)
-        return self.tagger.tag(self.xseq)  # type: ignore[no-any-return]
+        return cast(list[str], self.tagger.tag(self.xseq))
 
     def __enter__(self) -> CRFChunkParser:
         """Context manager entry."""

pythainlp/coref/core.py

@@ -3,24 +3,24 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
-from typing import Any, Optional, Union
+from typing import Any, Union, cast
 
-_MODEL: Optional[Any] = None
+_MODEL_CACHE: dict[tuple[str, str], Any] = {}
 
 
 def coreference_resolution(
     texts: Union[str, list[str]],
     model_name: str = "han-coref-v1.0",
     device: str = "cpu",
-) -> list[dict]:
+) -> list[dict[str, Any]]:
     """Coreference Resolution
 
     :param Union[str, list[str]] texts: list of texts to apply coreference resolution to
     :param str model_name: coreference resolution model
     :param str device: device for running coreference resolution model on\
         ("cpu", "cuda", and others)
     :return: List of texts with coreference resolution
-    :rtype: list[dict]
+    :rtype: list[dict[str, Any]]
 
     :Options for model_name:
         * *han-coref-v1.0* - (default) Han-Coref: Thai coreference resolution\
@@ -43,17 +43,18 @@ def coreference_resolution(
         # 'clusters': [[(0, 10), (50, 52)]]}
         # ]
     """
-    global _MODEL
     if isinstance(texts, str):
         texts = [texts]
 
-    if _MODEL is None and model_name == "han-coref-v1.0":
+    model_key = (model_name, device)
+    if model_key not in _MODEL_CACHE and model_name == "han-coref-v1.0":
         from pythainlp.coref.han_coref import HanCoref
 
-        _MODEL = HanCoref(device=device)
+        _MODEL_CACHE[model_key] = HanCoref(device=device)
 
-    if _MODEL:
-        return _MODEL.predict(texts)  # type: ignore[no-any-return]
+    model = _MODEL_CACHE.get(model_key)
+    if model is not None:
+        return cast(list[dict[str, Any]], model.predict(texts))
 
     return [
         {"text": text, "clusters_string": [], "clusters": []} for text in texts

pythainlp/corpus/common.py

@@ -407,9 +407,12 @@ def thai_synonyms() -> dict[str, Union[list[str], list[list[str]]]]:
             pos = row.get("pos")
             synonym = row.get("synonym")
             if not (
-                word and word.strip()
-                and pos and pos.strip()
-                and synonym and synonym.strip()
+                word
+                and word.strip()
+                and pos
+                and pos.strip()
+                and synonym
+                and synonym.strip()
             ):
                 warnings.warn(
                     f"Skipping thai_synonyms entry with missing or empty field(s): {dict(row)!r}",
@@ -420,7 +423,11 @@ def thai_synonyms() -> dict[str, Union[list[str], list[list[str]]]]:
             words.append(word)
             pos_tags.append(pos)
             synonym_groups.append(synonym.split("|"))
-    _THAI_SYNONYMS = {"word": words, "pos": pos_tags, "synonym": synonym_groups}
+    _THAI_SYNONYMS = {
+        "word": words,
+        "pos": pos_tags,
+        "synonym": synonym_groups,
+    }
     return _THAI_SYNONYMS
 
 

pythainlp/corpus/core.py

@@ -13,7 +13,7 @@
 import zipfile
 from functools import lru_cache
 from importlib.resources import files
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Optional, cast
 
 from pythainlp import __version__
 from pythainlp.corpus import corpus_db_path, corpus_db_url, corpus_path
@@ -26,7 +26,6 @@
 
 if TYPE_CHECKING:
     from http.client import HTTPMessage, HTTPResponse
-    from typing import Any, Optional
 
 _USER_AGENT: str = (
     f"PyThaiNLP/{__version__} "
@@ -50,7 +49,9 @@ def __init__(self, response: HTTPResponse) -> None:
     def json(self) -> dict[str, Any]:
         """Parse JSON content from response."""
         try:
-            return json.loads(self._content.decode("utf-8"))  # type: ignore[no-any-return]
+            return cast(
+                dict[str, Any], json.loads(self._content.decode("utf-8"))
+            )
         except (json.JSONDecodeError, UnicodeDecodeError) as err:
             raise ValueError(f"Failed to parse JSON response: {err}") from err
 
@@ -98,16 +99,15 @@ def get_corpus_db_detail(name: str, version: str = "") -> dict[str, Any]:
     if not version:
         for corpus in local_db["_default"].values():
             if corpus["name"] == name:
-                return corpus  # type: ignore[no-any-return]
+                return cast(dict[str, Any], corpus)
     else:
         for corpus in local_db["_default"].values():
             if corpus["name"] == name and corpus["version"] == version:
-                return corpus  # type: ignore[no-any-return]
+                return cast(dict[str, Any], corpus)
 
     return {}
 
 
-
 @lru_cache(maxsize=None)
 def get_corpus(filename: str, comments: bool = True) -> frozenset[str]:
     """Read corpus data from file and return a frozenset.
@@ -221,7 +221,7 @@ def _load_default_db() -> dict[str, Any]:
     corpus_files = files("pythainlp.corpus")
     default_db_file = corpus_files.joinpath("default_db.json")
     text = default_db_file.read_text(encoding="utf-8-sig")
-    return json.loads(text)  # type: ignore[no-any-return]
+    return cast(dict[str, Any], json.loads(text))
 
 
 def get_corpus_default_db(name: str, version: str = "") -> Optional[str]:
@@ -848,7 +848,6 @@ def remove(name: str) -> bool:
     return False
 
 
-
 def make_safe_directory_name(name: str) -> str:
     """Make safe directory name
 
@@ -921,4 +920,4 @@ def get_hf_hub(repo_id: str, filename: str = "") -> str:
         output_path = snapshot_download(
             repo_id=repo_id, local_dir=root_project
         )
-    return output_path  # type: ignore[no-any-return]
+    return str(output_path)