Reduce cognitive complexity in five scikit-learn modules through helper method extraction

sklearn/datasets/_svmlight_format_io.py

@@ -448,6 +448,31 @@ def _dump_svmlight(X, y, f, multilabel, one_based, comment, query_id):
     )
 
 
+def _validate_comment(comment):
+    """Validate and encode the comment for SVMlight format."""
+    if comment is None:
+        return comment
+    # Convert comment string to list of lines in UTF-8.
+    # If a byte string is passed, then check whether it's ASCII;
+    # if a user wants to get fancy, they'll have to decode themselves.
+    if isinstance(comment, bytes):
+        comment.decode("ascii")  # just for the exception
+    else:
+        comment = comment.encode("utf-8")
+    if b"\0" in comment:
+        raise ValueError("comment string contains NUL byte")
+    return comment
+
+
+def _sort_sparse_indices(val, original):
+    """Sort sparse matrix indices, returning the sorted result."""
+    if val is original and hasattr(val, "sorted_indices"):
+        return val.sorted_indices()
+    if hasattr(val, "sort_indices"):
+        val.sort_indices()
+    return val
+
+
 @validate_params(
     {
         "X": ["array-like", "sparse matrix"],
@@ -524,16 +549,7 @@ def dump_svmlight_file(
     >>> output_file = "my_dataset.svmlight"
     >>> dump_svmlight_file(X, y, output_file)  # doctest: +SKIP
     """
-    if comment is not None:
-        # Convert comment string to list of lines in UTF-8.
-        # If a byte string is passed, then check whether it's ASCII;
-        # if a user wants to get fancy, they'll have to decode themselves.
-        if isinstance(comment, bytes):
-            comment.decode("ascii")  # just for the exception
-        else:
-            comment = comment.encode("utf-8")
-        if b"\0" in comment:
-            raise ValueError("comment string contains NUL byte")
+    comment = _validate_comment(comment)
 
     yval = check_array(y, accept_sparse="csr", ensure_2d=False)
     if sp.issparse(yval):
@@ -555,19 +571,8 @@ def dump_svmlight_file(
     # We had some issues with CSR matrices with unsorted indices (e.g. #1501),
     # so sort them here, but first make sure we don't modify the user's X.
     # TODO We can do this cheaper; sorted_indices copies the whole matrix.
-    if yval is y and hasattr(yval, "sorted_indices"):
-        y = yval.sorted_indices()
-    else:
-        y = yval
-        if hasattr(y, "sort_indices"):
-            y.sort_indices()
-
-    if x_val is X and hasattr(x_val, "sorted_indices"):
-        X = x_val.sorted_indices()
-    else:
-        X = x_val
-        if hasattr(X, "sort_indices"):
-            X.sort_indices()
+    y = _sort_sparse_indices(yval, y)
+    X = _sort_sparse_indices(x_val, X)
 
     if query_id is None:
         # NOTE: query_id is passed to Cython functions using a fused type on query_id.

sklearn/decomposition/_dict_learning.py

@@ -545,28 +545,8 @@ def _update_dict(
         print(f"{n_unused} unused atoms resampled.")
 
 
-def _dict_learning(
-    X,
-    n_components,
-    *,
-    alpha,
-    max_iter,
-    tol,
-    method_params,
-    n_jobs,
-    init,
-    callback,
-    verbose,
-    random_state,
-    return_n_iter,
-    positive,
-):
-    """Main dictionary learning algorithm"""
-    positive_dict, positive_code = positive
-    dict_init, code_init = init
-    method, method_max_iter = method_params
-
-    t0 = time.time()
+def _initialize_code_and_dictionary(X, n_components, code_init, dict_init):
+    """Initialize code and dictionary for dictionary learning."""
     # Init the code and the dictionary with SVD of Y
     if code_init is not None and dict_init is not None:
         code = np.array(code_init, order="F")
@@ -590,6 +570,49 @@ def _dict_learning(
     # Fortran-order dict better suited for the sparse coding which is the
     # bottleneck of this algorithm.
     dictionary = np.asfortranarray(dictionary)
+    return code, dictionary
+
+
+def _check_convergence(ii, errors, tol, verbose):
+    """Check convergence and log if converged. Returns True if converged."""
+    if ii > 0:
+        d_e = errors[-2] - errors[-1]
+
+        if d_e < tol * errors[-1]:
+            if verbose == 1:
+                # A line return
+                print("")
+            elif verbose:
+                print("--- Convergence reached after %d iterations" % ii)
+            return True
+    return False
+
+
+def _dict_learning(
+    X,
+    n_components,
+    *,
+    alpha,
+    max_iter,
+    tol,
+    method_params,
+    n_jobs,
+    init,
+    callback,
+    verbose,
+    random_state,
+    return_n_iter,
+    positive,
+):
+    """Main dictionary learning algorithm"""
+    positive_dict, positive_code = positive
+    dict_init, code_init = init
+    method, method_max_iter = method_params
+
+    t0 = time.time()
+    code, dictionary = _initialize_code_and_dictionary(
+        X, n_components, code_init, dict_init
+    )
 
     errors = []
     current_cost = np.nan
@@ -640,16 +663,8 @@ def _dict_learning(
         )
         errors.append(current_cost)
 
-        if ii > 0:
-            d_e = errors[-2] - errors[-1]
-
-            if d_e < tol * errors[-1]:
-                if verbose == 1:
-                    # A line return
-                    print("")
-                elif verbose:
-                    print("--- Convergence reached after %d iterations" % ii)
-                break
+        if _check_convergence(ii, errors, tol, verbose):
+            break
         if ii % 5 == 0 and callback is not None:
             callback(locals())
 

sklearn/decomposition/_lda.py

@@ -633,6 +633,34 @@ def partial_fit(self, X, y=None):
 
         return self
 
+    def _check_perplexity_and_convergence(
+        self, X, i, evaluate_every, max_iter, last_bound, parallel
+    ):
+        """Evaluate perplexity and check for convergence during fitting.
+
+        Returns (last_bound, should_break).
+        """
+        if evaluate_every > 0 and (i + 1) % evaluate_every == 0:
+            doc_topics_distr, _ = self._e_step(
+                X, cal_sstats=False, random_init=False, parallel=parallel
+            )
+            bound = self._perplexity_precomp_distr(
+                X, doc_topics_distr, sub_sampling=False
+            )
+            if self.verbose:
+                print(
+                    "iteration: %d of max_iter: %d, perplexity: %.4f"
+                    % (i + 1, max_iter, bound)
+                )
+
+            if last_bound and abs(last_bound - bound) < self.perp_tol:
+                return bound, True
+            return bound, False
+
+        elif self.verbose:
+            print("iteration: %d of max_iter: %d" % (i + 1, max_iter))
+        return last_bound, False
+
     @_fit_context(prefer_skip_nested_validation=True)
     def fit(self, X, y=None):
         """Learn model for the data X with variational Bayes method.
@@ -685,25 +713,11 @@ def fit(self, X, y=None):
                     )
 
                 # check perplexity
-                if evaluate_every > 0 and (i + 1) % evaluate_every == 0:
-                    doc_topics_distr, _ = self._e_step(
-                        X, cal_sstats=False, random_init=False, parallel=parallel
-                    )
-                    bound = self._perplexity_precomp_distr(
-                        X, doc_topics_distr, sub_sampling=False
-                    )
-                    if self.verbose:
-                        print(
-                            "iteration: %d of max_iter: %d, perplexity: %.4f"
-                            % (i + 1, max_iter, bound)
-                        )
-
-                    if last_bound and abs(last_bound - bound) < self.perp_tol:
-                        break
-                    last_bound = bound
-
-                elif self.verbose:
-                    print("iteration: %d of max_iter: %d" % (i + 1, max_iter))
+                last_bound, converged = self._check_perplexity_and_convergence(
+                    X, i, evaluate_every, max_iter, last_bound, parallel
+                )
+                if converged:
+                    break
                 self.n_iter_ += 1
 
         # calculate final perplexity value on train set

sklearn/discriminant_analysis.py

@@ -1050,6 +1050,30 @@ def _solve_svd(self, X):
 
         return scaling, rotation, cov
 
+    def _check_rank(self, scaling_class, n_features, x_class, class_label):
+        """Raise an error if the covariance matrix is not full rank."""
+        rank = np.sum(scaling_class > self.tol)
+        if rank < n_features:
+            n_samples_class = x_class.shape[0]
+            if self.solver == "svd" and n_samples_class <= n_features:
+                raise linalg.LinAlgError(
+                    f"The covariance matrix of class {class_label} is not full "
+                    f"rank. When using `solver='svd'` the number of samples in "
+                    f"each class should be more than the number of features, but "
+                    f"class {class_label} has {n_samples_class} samples and "
+                    f"{n_features} features. Try using `solver='eigen'` and "
+                    f"setting the parameter `shrinkage` for regularization."
+                )
+            else:
+                msg_param = (
+                    "shrinkage" if self.solver == "eigen" else "reg_param"
+                )
+                raise linalg.LinAlgError(
+                    f"The covariance matrix of class {class_label} is not full "
+                    f"rank. Increase the value of `{msg_param}` to reduce the "
+                    f"collinearity.",
+                )
+
     @_fit_context(prefer_skip_nested_validation=True)
     def fit(self, X, y):
         """Fit the model according to the given training data and parameters.
@@ -1122,25 +1146,7 @@ def fit(self, X, y):
 
             scaling_class, rotation_class, cov_class = specific_solver(x_class)
 
-            rank = np.sum(scaling_class > self.tol)
-            if rank < n_features:
-                n_samples_class = x_class.shape[0]
-                if self.solver == "svd" and n_samples_class <= n_features:
-                    raise linalg.LinAlgError(
-                        f"The covariance matrix of class {class_label} is not full "
-                        f"rank. When using `solver='svd'` the number of samples in "
-                        f"each class should be more than the number of features, but "
-                        f"class {class_label} has {n_samples_class} samples and "
-                        f"{n_features} features. Try using `solver='eigen'` and "
-                        f"setting the parameter `shrinkage` for regularization."
-                    )
-                else:
-                    msg_param = "shrinkage" if self.solver == "eigen" else "reg_param"
-                    raise linalg.LinAlgError(
-                        f"The covariance matrix of class {class_label} is not full "
-                        f"rank. Increase the value of `{msg_param}` to reduce the "
-                        f"collinearity.",
-                    )
+            self._check_rank(scaling_class, n_features, x_class, class_label)
 
             cov.append(cov_class)
             scalings.append(scaling_class)