Merge pull request #826 from stan-dev/mypy-tests

Fix typing issues in tests

Author: WardBrian

.github/workflows/main.yml

@@ -68,11 +68,11 @@ jobs:
         run: python -m pip freeze
 
       - name: Run flake8, pylint, mypy
-        if: matrix.python-version == '3.11'
+        if: matrix.python-version == '3.14'
         run: |
           flake8 cmdstanpy test
           pylint -v cmdstanpy test
-          mypy cmdstanpy
+          mypy cmdstanpy test
 
       - name: CmdStan installation cacheing
         id: cache-cmdstan

.pre-commit-config.yaml

@@ -23,7 +23,6 @@ repos:
     rev: v1.5.0
     hooks:
       - id: mypy
-        exclude: ^test/
         additional_dependencies: [ numpy >= 1.22]
   # local uses the user-installed pylint, this allows dependency checking
   - repo: local

cmdstanpy/cmdstan_args.py

@@ -643,7 +643,7 @@ def __init__(
             PathfinderArgs,
         ],
         data: Union[Mapping[str, Any], str, None] = None,
-        seed: Union[int, list[int], None] = None,
+        seed: Union[int, np.integer, list[int], list[np.integer], None] = None,
         inits: Union[int, float, str, list[str], None] = None,
         output_dir: OptionalPath = None,
         sig_figs: Optional[int] = None,

cmdstanpy/install_cmdstan.py

@@ -29,9 +29,10 @@
 import urllib.error
 import urllib.request
 from collections import OrderedDict
+from functools import cached_property
 from pathlib import Path
 from time import sleep
-from typing import TYPE_CHECKING, Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, Union
 
 from tqdm.auto import tqdm
 
@@ -47,18 +48,6 @@
 
 from . import progress as progbar
 
-if sys.version_info >= (3, 8) or TYPE_CHECKING:
-    # mypy only knows about the new built-in cached_property
-    from functools import cached_property
-else:
-    # on older Python versions, this is the recommended
-    # way to get the same effect
-    from functools import lru_cache
-
-    def cached_property(fun):
-        return property(lru_cache(maxsize=None)(fun))
-
-
 try:
     # on MacOS and Linux, importing this
     # improves the UX of the input() function

cmdstanpy/model.py

@@ -13,7 +13,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from io import StringIO
 from multiprocessing import cpu_count
-from typing import Any, Callable, Mapping, Optional, TypeVar, Union
+from typing import Any, Callable, Mapping, Optional, Sequence, TypeVar, Union
 
 import numpy as np
 import pandas as pd
@@ -461,8 +461,7 @@ def sample(
             Mapping[str, Any],
             float,
             str,
-            list[str],
-            list[Mapping[str, Any]],
+            Sequence[Union[str, Mapping[str, Any]]],
             None,
         ] = None,
         iter_warmup: Optional[int] = None,
@@ -493,7 +492,7 @@ def sample(
             str,
             np.ndarray,
             Mapping[str, Any],
-            list[Union[str, np.ndarray, Mapping[str, Any]]],
+            Sequence[Union[str, np.ndarray, Mapping[str, Any]]],
             None,
         ] = None,
     ) -> CmdStanMCMC:
@@ -1360,7 +1359,13 @@ def pathfinder(
         calculate_lp: bool = True,
         # arguments standard to all methods
         seed: Optional[int] = None,
-        inits: Union[dict[str, float], float, str, os.PathLike, None] = None,
+        inits: Union[
+            Mapping[str, Any],
+            float,
+            str,
+            Sequence[Union[str, Mapping[str, Any]]],
+            None,
+        ] = None,
         output_dir: OptionalPath = None,
         sig_figs: Optional[int] = None,
         save_profile: bool = False,

cmdstanpy/stanfit/__init__.py

@@ -248,7 +248,7 @@ def from_csv(
             mode: CmdStanMLE = from_csv(
                 config_dict['mode'],  # type: ignore
                 method='optimize',
-            )  # type: ignore
+            )
             return CmdStanLaplace(runset, mode=mode)
         elif config_dict['method'] == 'pathfinder':
             pathfinder_args = PathfinderArgs(

cmdstanpy/stanfit/mcmc.py

@@ -348,7 +348,7 @@ def _validate_csv_files(self) -> dict[str, Any]:
                     save_warmup=self._save_warmup,
                     thin=self._thin,
                 )
-                self._chain_time.append(dzero['time'])  # type: ignore
+                self._chain_time.append(dzero['time'])
                 if not self._is_fixed_param:
                     self._divergences[i] = dzero['ct_divergences']
                     self._max_treedepths[i] = dzero['ct_max_treedepth']
@@ -360,7 +360,7 @@ def _validate_csv_files(self) -> dict[str, Any]:
                     save_warmup=self._save_warmup,
                     thin=self._thin,
                 )
-                self._chain_time.append(drest['time'])  # type: ignore
+                self._chain_time.append(drest['time'])
                 for key in dzero:
                     # check args that matter for parsing, plus name, version
                     if (

cmdstanpy/stanfit/pathfinder.py

@@ -208,7 +208,7 @@ def is_resampled(self) -> bool:
         Returns True if the draws were resampled from several Pathfinder
         approximations, False otherwise.
         """
-        return (  # type: ignore
+        return (
             self._metadata.cmdstan_config.get("num_paths", 4) > 1
             and self._metadata.cmdstan_config.get('psis_resample', 1)
             in (1, 'true')

cmdstanpy/utils/__init__.py

@@ -80,7 +80,7 @@ def show_versions(output: bool = True) -> str:
             deps_info.append((module, None))
         else:
             try:
-                ver = mod.__version__  # type: ignore
+                ver = mod.__version__
                 deps_info.append((module, ver))
             # pylint: disable=broad-except
             except Exception:

cmdstanpy/utils/filesystem.py

@@ -8,7 +8,7 @@
 import re
 import shutil
 import tempfile
-from typing import Any, Iterator, Mapping, Optional, Union
+from typing import Any, Iterator, Mapping, Optional, Sequence, Union
 
 import numpy as np
 
@@ -131,10 +131,12 @@ def _temp_single_json(
 
 
 def _temp_multiinput(
-    input: Union[str, os.PathLike, Mapping[str, Any], list[Any], None],
+    input: Union[str, os.PathLike, Mapping[str, Any], Sequence[Any], None],
     base: int = 1,
 ) -> Iterator[Optional[str]]:
-    if isinstance(input, list):
+    if isinstance(input, Sequence) and not isinstance(
+        input, (str, os.PathLike)
+    ):
         # most complicated case: list of inits
         # for multiple chains, we need to create multiple files
         # which look like somename_{i}.json and then pass somename.json
@@ -170,7 +172,7 @@ def _temp_multiinput(
 @contextlib.contextmanager
 def temp_metrics(
     metrics: Union[
-        str, os.PathLike, Mapping[str, Any], np.ndarray, list[Any], None
+        str, os.PathLike, Mapping[str, Any], np.ndarray, Sequence[Any], None
     ],
     *,
     id: int = 1,
@@ -200,7 +202,7 @@ def temp_metrics(
 @contextlib.contextmanager
 def temp_inits(
     inits: Union[
-        str, os.PathLike, Mapping[str, Any], float, int, list[Any], None
+        str, os.PathLike, Mapping[str, Any], float, int, Sequence[Any], None
     ],
     *,
     allow_multiple: bool = True,
@@ -212,7 +214,9 @@ def temp_inits(
     if allow_multiple:
         yield from _temp_multiinput(inits, base=id)
     else:
-        if isinstance(inits, list):
+        if isinstance(inits, Sequence) and not isinstance(
+            inits, (str, os.PathLike)
+        ):
             raise ValueError('Expected single initialization, got list')
         yield from _temp_single_json(inits)