diff --git a/foamlib/_cases.py b/foamlib/_cases.py index 24edc42..08ca64c 100644 --- a/foamlib/_cases.py +++ b/foamlib/_cases.py @@ -20,7 +20,7 @@ import aioshutil -from ._subprocesses import run_process, run_process_async, CalledProcessError +from ._util import run_process, run_process_async, CalledProcessError from ._dictionaries import FoamFile, FoamFieldFile diff --git a/foamlib/_dictionaries/_serialization.py b/foamlib/_dictionaries/_serialization.py index 974f410..1ec5c87 100644 --- a/foamlib/_dictionaries/_serialization.py +++ b/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)}") diff --git a/foamlib/_subprocesses.py b/foamlib/_util.py similarity index 76% rename from foamlib/_subprocesses.py rename to foamlib/_util.py index e104f28..b1be371 100644 --- a/foamlib/_subprocesses.py +++ b/foamlib/_util.py @@ -1,12 +1,28 @@ 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 + +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) + ) + -__all__ = ["run_process", "run_process_async", "CalledProcessError"] +CalledProcessError = subprocess.CalledProcessError def run_process( @@ -16,7 +32,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 +44,7 @@ def run_process( cmd, cwd=cwd, env=env, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, + capture_output=True, shell=shell, check=check, ) @@ -44,7 +59,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,