Merge pull request #273 from lovit/feature/267

refactor: PMI 함수를 soynlp.word에서 soynlp.utils로 이동 (#267)

Author: lovit

soynlp/utils/__init__.py

@@ -1,6 +1,7 @@
 from soynlp.core.lrgraph import LRGraph
 
 from .math import svd
+from .pmi import pmi
 from .utils import (
     CorpusLoader,
     EojeolCounter,
@@ -23,4 +24,6 @@
     "LRGraph",
     # math
     "svd",
+    # pmi
+    "pmi",
 ]

soynlp/utils/pmi.py

@@ -0,0 +1,76 @@
+from typing import Any
+
+import numpy as np
+from scipy.sparse import csr_matrix, dia_matrix, diags
+
+
+def _as_diag(px: np.ndarray, alpha: float) -> dia_matrix:
+    px_diag: dia_matrix = diags(px.tolist()[0])
+    px_diag.data[0] = np.asarray([0 if v == 0 else 1 / (v + alpha) for v in px_diag.data[0]])
+    return px_diag
+
+
+def _logarithm_and_ppmi(exp_pmi: Any, min_exp_pmi: float) -> csr_matrix:
+    n, m = exp_pmi.shape
+
+    rows, cols = exp_pmi.nonzero()
+    data = exp_pmi.data
+
+    indices = np.where(data >= min_exp_pmi)[0]
+    rows = rows[indices]
+    cols = cols[indices]
+    data = data[indices]
+
+    data = np.log(data)
+    exp_pmi_ = csr_matrix((data, (rows, cols)), shape=(n, m))
+    return exp_pmi_
+
+
+def pmi(
+    X: csr_matrix,
+    py: np.ndarray | None = None,
+    min_pmi: float = 0,
+    alpha: float = 0.0,
+    beta: float = 1,
+) -> tuple[csr_matrix, np.ndarray, np.ndarray]:
+    """Transform `X` to Positive-PMI matrix (CSR sparse matrix)
+
+    Args:
+        X (scipy.sparse.csr_matrix) :
+            shape = (n items, n features)
+        py (numpy.ndarray, optional) :
+            shape = (1, word), probability of context words.
+            If `py` is None, `pmi` function uses normalized row sum of `X`
+        min_pmi (float) :
+            Minimum value of pmi.
+        alpha (float) :
+            Smoothing factor. Default is `0.0`
+        beta (float) :
+            Smoothing factor. Default is `1.0`
+
+    Returns:
+        pmi (scipy.sparse.csr_matrix)
+        px (numpy.ndarray)
+        py (numpy.ndarray)
+    """
+
+    assert 0 < beta <= 1
+
+    px = np.asarray((X.sum(axis=1) / X.sum()).reshape(-1))
+    pxy = X / X.sum()
+    if py is None:
+        py = np.asarray((X.sum(axis=0) / X.sum()).reshape(-1))
+    py_arr: np.ndarray = py
+    if beta < 1:
+        py_arr = py_arr**beta
+        py_arr /= py_arr.sum()
+    assert py_arr.shape[1] == pxy.shape[1]  # type: ignore[index]
+
+    px_diag = _as_diag(px, 0)
+    py_diag = _as_diag(py_arr, alpha)
+    exp_pmi = px_diag.dot(pxy).dot(py_diag)
+
+    min_exp_pmi = 1 if min_pmi == 0 else np.exp(min_pmi)
+    pmi_mat = _logarithm_and_ppmi(exp_pmi, min_exp_pmi)
+
+    return pmi_mat, px, py_arr

soynlp/word/pmi.py

@@ -1,29 +1,14 @@
-from typing import Any
-
-import numpy as np
-from scipy.sparse import csr_matrix, dia_matrix, diags
-
-
-def _as_diag(px: np.ndarray, alpha: float) -> dia_matrix:
-    px_diag: dia_matrix = diags(px.tolist()[0])
-    px_diag.data[0] = np.asarray([0 if v == 0 else 1 / (v + alpha) for v in px_diag.data[0]])
-    return px_diag
+"""Deprecated: soynlp.word.pmi → soynlp.utils.pmi로 이동되었습니다."""
 
+import warnings
 
-def _logarithm_and_ppmi(exp_pmi: Any, min_exp_pmi: float) -> csr_matrix:
-    n, m = exp_pmi.shape
-
-    rows, cols = exp_pmi.nonzero()
-    data = exp_pmi.data
+import numpy as np
+from scipy.sparse import csr_matrix
 
-    indices = np.where(data >= min_exp_pmi)[0]
-    rows = rows[indices]
-    cols = cols[indices]
-    data = data[indices]
+from soynlp.utils.pmi import _as_diag, _logarithm_and_ppmi
+from soynlp.utils.pmi import pmi as _pmi_impl
 
-    data = np.log(data)
-    exp_pmi_ = csr_matrix((data, (rows, cols)), shape=(n, m))
-    return exp_pmi_
+__all__ = ["pmi", "_as_diag", "_logarithm_and_ppmi"]
 
 
 def pmi(
@@ -33,44 +18,10 @@ def pmi(
     alpha: float = 0.0,
     beta: float = 1,
 ) -> tuple[csr_matrix, np.ndarray, np.ndarray]:
-    """Transform `X` to Positive-PMI matrix (CSR sparse matrix)
-
-    Args:
-        X (scipy.sparse.csr_matrix) :
-            shape = (n items, n features)
-        py (numpy.ndarray, optional) :
-            shape = (1, word), probability of context words.
-            If `py` is None, `pmi` function uses normalized row sum of `X`
-        min_pmi (float) :
-            Minimum value of pmi.
-        alpha (float) :
-            Smoothing factor. Default is `0.0`
-        beta (float) :
-            Smoothing factor. Default is `1.0`
-
-    Returns:
-        pmi (scipy.sparse.csr_matrix)
-        px (numpy.ndarray)
-        py (numpy.ndarray)
-    """
-
-    assert 0 < beta <= 1
-
-    px = np.asarray((X.sum(axis=1) / X.sum()).reshape(-1))
-    pxy = X / X.sum()
-    if py is None:
-        py = np.asarray((X.sum(axis=0) / X.sum()).reshape(-1))
-    py_arr: np.ndarray = py
-    if beta < 1:
-        py_arr = py_arr**beta
-        py_arr /= py_arr.sum()
-    assert py_arr.shape[1] == pxy.shape[1]  # type: ignore[index]
-
-    px_diag = _as_diag(px, 0)
-    py_diag = _as_diag(py_arr, alpha)
-    exp_pmi = px_diag.dot(pxy).dot(py_diag)
-
-    min_exp_pmi = 1 if min_pmi == 0 else np.exp(min_pmi)
-    pmi_mat = _logarithm_and_ppmi(exp_pmi, min_exp_pmi)
-
-    return pmi_mat, px, py_arr
+    """.. deprecated:: soynlp.utils.pmi를 사용하세요."""
+    warnings.warn(
+        "soynlp.word.pmi is deprecated. Use soynlp.utils.pmi instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    return _pmi_impl(X, py=py, min_pmi=min_pmi, alpha=alpha, beta=beta)

tests/unit/test_pmi.py

@@ -1,7 +1,9 @@
+import warnings
+
 import numpy as np
 from scipy.sparse import csr_matrix
 
-from soynlp.word import pmi
+from soynlp.utils import pmi
 
 
 def test_pmi():
@@ -19,3 +21,29 @@ def test_pmi():
         print(f"\nPMI: \n{pmi_mat.todense()}")
         print(f"\nPx: {px}")
         print(f"\nPy: {py}")
+
+
+def test_pmi_import_from_utils():
+    """soynlp.utils.pmi가 정상적으로 import되고 동작한다."""
+    from soynlp.utils.pmi import pmi as pmi_direct
+
+    x = csr_matrix(np.ones((3, 4)))
+    pmi_mat, px, py = pmi_direct(x)
+    assert pmi_mat.shape == (3, 4)
+
+
+def test_pmi_deprecated_word_import():
+    """soynlp.word.pmi는 DeprecationWarning을 발생시키며 동일한 결과를 반환한다."""
+    from soynlp.word.pmi import pmi as pmi_word
+
+    x = csr_matrix(np.ones((3, 4)))
+    with warnings.catch_warnings(record=True) as w:
+        warnings.simplefilter("always")
+        result = pmi_word(x)
+        assert len(w) == 1
+        assert issubclass(w[0].category, DeprecationWarning)
+        assert "soynlp.utils.pmi" in str(w[0].message)
+
+    # 결과는 새 경로와 동일해야 함
+    expected = pmi(x)
+    assert (result[0] - expected[0]).nnz == 0