Update internal utility functions

foamlib/_cases.py

@@ -20,7 +20,7 @@
 
 import aioshutil
 
-from ._subprocesses import run_process, run_process_async, CalledProcessError
+from ._util import is_sequence, run_process, run_process_async, CalledProcessError
 from ._dictionaries import FoamFile, FoamFieldFile
 
 
@@ -215,7 +215,7 @@ def _env(self) -> Mapping[str, str]:
     def _parallel_cmd(
         self, cmd: Union[Sequence[Union[str, Path]], str, Path]
     ) -> Union[Sequence[Union[str, Path]], str]:
-        if isinstance(cmd, str) or not isinstance(cmd, Sequence):
+        if not is_sequence(cmd):
             return f"mpiexec -np {self._nprocessors} {cmd} -parallel"
         else:
             return [

foamlib/_dictionaries/_serialization.py

@@ -1,23 +1,8 @@
 from contextlib import suppress
-from typing import Any, Mapping, Sequence
+from typing import Any, Mapping
 
 from ._base import FoamDictionaryBase
-
-try:
-    import numpy as np
-except ModuleNotFoundError:
-    numpy = False
-else:
-    numpy = True
-
-
-def _is_sequence(value: Any) -> bool:
-    return (
-        isinstance(value, Sequence)
-        and not isinstance(value, str)
-        or numpy
-        and isinstance(value, np.ndarray)
-    )
+from .._util import is_sequence
 
 
 def _serialize_bool(value: Any) -> str:
@@ -30,14 +15,14 @@ def _serialize_bool(value: Any) -> str:
 
 
 def _serialize_list(value: Any) -> str:
-    if _is_sequence(value):
+    if is_sequence(value):
         return f"({' '.join(_serialize_value(v) for v in value)})"
     else:
         raise TypeError(f"Not a valid sequence: {type(value)}")
 
 
 def _serialize_field(value: Any) -> str:
-    if _is_sequence(value):
+    if is_sequence(value):
         try:
             s = _serialize_list(value)
         except TypeError:
@@ -64,7 +49,7 @@ def _serialize_field(value: Any) -> str:
 
 
 def _serialize_dimensions(value: Any) -> str:
-    if _is_sequence(value) and len(value) == 7:
+    if is_sequence(value) and len(value) == 7:
         return f"[{' '.join(str(v) for v in value)}]"
     else:
         raise TypeError(f"Not a valid dimension set: {type(value)}")

foamlib/_subprocesses.py → foamlib/_util.py

@@ -1,12 +1,32 @@
 import asyncio
 import sys
+import subprocess
 
 from pathlib import Path
-from typing import Union, Sequence, Mapping
-import subprocess
-from subprocess import CalledProcessError
+from typing import Any, Mapping, Sequence, Union
+
+from typing_extensions import TypeGuard
+
+try:
+    import numpy as np
+except ModuleNotFoundError:
+    numpy = False
+else:
+    numpy = True
+
+
+def is_sequence(
+    value: Any,
+) -> TypeGuard[Union["Sequence[Any]", "np.ndarray[Any, Any]"]]:
+    return (
+        isinstance(value, Sequence)
+        and not isinstance(value, str)
+        or numpy
+        and isinstance(value, np.ndarray)
+    )
+
 
-__all__ = ["run_process", "run_process_async", "CalledProcessError"]
+CalledProcessError = subprocess.CalledProcessError
 
 
 def run_process(
@@ -16,7 +36,7 @@ def run_process(
     cwd: Union[None, str, Path] = None,
     env: Union[None, Mapping[str, str]] = None,
 ) -> "subprocess.CompletedProcess[bytes]":
-    shell = isinstance(cmd, str) or not isinstance(cmd, Sequence)
+    shell = not is_sequence(cmd)
 
     if sys.version_info < (3, 8):
         if shell:
@@ -28,8 +48,7 @@ def run_process(
         cmd,
         cwd=cwd,
         env=env,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
+        capture_output=True,
         shell=shell,
         check=check,
     )
@@ -44,7 +63,7 @@ async def run_process_async(
     cwd: Union[None, str, Path] = None,
     env: Union[None, Mapping[str, str]] = None,
 ) -> "subprocess.CompletedProcess[bytes]":
-    if isinstance(cmd, str) or not isinstance(cmd, Sequence):
+    if not is_sequence(cmd):
         proc = await asyncio.create_subprocess_shell(
             str(cmd),
             cwd=cwd,