use typepigeon to convert values (#125)

* use `typepigeon`

* bump `typepigeon` version

Co-authored-by: zacharyburnett <[email protected]>

coupledmodeldriver/client/check_completion.py

@@ -5,14 +5,16 @@
 from pathlib import Path
 from typing import Any, Iterable, Mapping
 
+from typepigeon import convert_value
+
 from coupledmodeldriver.configure import ModelJSON
 from coupledmodeldriver.generate.adcirc.base import ADCIRCJSON
 from coupledmodeldriver.generate.adcirc.check import (
     check_adcirc_completion,
     CompletionStatus,
     is_adcirc_run_directory,
 )
-from coupledmodeldriver.utilities import convert_value, ProcessPoolExecutorStackTraced
+from coupledmodeldriver.utilities import ProcessPoolExecutorStackTraced
 
 MODELS = {model.name.lower(): model for model in ModelJSON.__subclasses__()}
 

coupledmodeldriver/client/generate_adcirc.py

@@ -1,8 +1,9 @@
 from argparse import ArgumentParser
 from pathlib import Path
 
+from typepigeon import convert_value
+
 from coupledmodeldriver.generate import generate_adcirc_configuration
-from coupledmodeldriver.utilities import convert_value
 
 
 def parse_generate_adcirc_arguments():

coupledmodeldriver/client/initialize_adcirc.py

@@ -7,6 +7,7 @@
 from typing import Any, Mapping
 
 from adcircpy import TidalSource
+from typepigeon import convert_value
 
 from coupledmodeldriver import Platform
 from coupledmodeldriver.configure import (
@@ -26,7 +27,7 @@
 )
 from coupledmodeldriver.generate import ADCIRCRunConfiguration, NEMSADCIRCRunConfiguration
 from coupledmodeldriver.script import EnsembleGenerationJob
-from coupledmodeldriver.utilities import convert_value, get_logger
+from coupledmodeldriver.utilities import get_logger
 
 
 class ForcingConfigurations(Enum):

coupledmodeldriver/configure/base.py

@@ -10,10 +10,11 @@
 from adcircpy.server import SlurmConfig
 from nemspy import ModelingSystem
 from nemspy.model.base import ModelEntry
+from typepigeon import convert_to_json, convert_value
 
 from coupledmodeldriver.platforms import Platform
 from coupledmodeldriver.script import SlurmEmailType
-from coupledmodeldriver.utilities import convert_to_json, convert_value, LOGGER
+from coupledmodeldriver.utilities import LOGGER
 
 
 class ConfigurationJSON(ABC):

coupledmodeldriver/utilities.py

@@ -1,7 +1,4 @@
 from concurrent.futures import ProcessPoolExecutor
-from datetime import datetime, timedelta
-from enum import Enum, EnumMeta
-import json
 import logging
 import os
 from os import PathLike
@@ -10,9 +7,7 @@
 import sys
 import tarfile
 import traceback
-from typing import Any, Collection, Iterable, Mapping, Union
 
-from dateutil.parser import parse as parse_date
 import numpy
 from pyproj import CRS, Geod, Transformer
 from shapely.geometry import Point
@@ -158,127 +153,6 @@ def make_executable(path: PathLike):
     os.chmod(path, mode)
 
 
-def convert_value(value: Any, to_type: type) -> Any:
-    if to_type is None:
-        return value
-    elif isinstance(to_type, str):
-        to_type = eval(to_type)
-    if isinstance(value, Enum):
-        value = value.name
-    if isinstance(to_type, Collection):
-        collection_type = type(to_type)
-        if collection_type is not EnumMeta:
-            if not issubclass(collection_type, Mapping):
-                if value is not None:
-                    to_type = list(to_type)
-                    if not isinstance(value, Iterable) or isinstance(value, str):
-                        value = [value]
-                    if len(to_type) == 1:
-                        to_type = [to_type[0] for _ in value]
-                    elif len(to_type) == len(value):
-                        to_type = to_type[: len(value)]
-                    else:
-                        raise ValueError(
-                            f'unable to convert list of values of length {len(value)} '
-                            f'to list of types of length {len(to_type)}: '
-                            f'{value} -/> {to_type}'
-                        )
-                    value = collection_type(
-                        convert_value(value[index], current_type)
-                        for index, current_type in enumerate(to_type)
-                    )
-                else:
-                    value = collection_type()
-            elif isinstance(value, str):
-                value = json.loads(value)
-            elif isinstance(value, CRS):
-                value = value.to_json_dict()
-        elif value is not None:
-            try:
-                value = to_type[value]
-            except (KeyError, ValueError):
-                try:
-                    value = to_type(value)
-                except (KeyError, ValueError):
-                    raise ValueError(
-                        f'unrecognized entry "{value}"; must be one of {list(to_type)}'
-                    )
-    elif not isinstance(value, to_type) and value is not None:
-        if isinstance(value, timedelta):
-            if issubclass(to_type, str):
-                hours, remainder = divmod(value, timedelta(hours=1))
-                minutes, remainder = divmod(remainder, timedelta(minutes=1))
-                seconds = remainder / timedelta(seconds=1)
-                value = f'{hours:02}:{minutes:02}:{seconds:04.3}'
-            else:
-                value /= timedelta(seconds=1)
-        elif isinstance(value, CRS):
-            if issubclass(to_type, str):
-                value = value.to_wkt()
-            elif issubclass(to_type, dict):
-                value = value.to_json_dict()
-            elif issubclass(to_type, int):
-                value = value.to_epsg()
-        if issubclass(to_type, bool):
-            value = eval(f'{value}')
-        elif issubclass(to_type, datetime):
-            value = parse_date(value)
-        elif issubclass(to_type, timedelta):
-            try:
-                try:
-                    time = datetime.strptime(value, '%H:%M:%S')
-                    value = timedelta(
-                        hours=time.hour, minutes=time.minute, seconds=time.second
-                    )
-                except:
-                    parts = [float(part) for part in value.split(':')]
-                    if len(parts) > 3:
-                        days = parts.pop(0)
-                    else:
-                        days = 0
-                    value = timedelta(
-                        days=days, hours=parts[0], minutes=parts[1], seconds=parts[2]
-                    )
-            except:
-                value = timedelta(seconds=float(value))
-        elif isinstance(value, str):
-            try:
-                value = to_type.from_string(value)
-            except:
-                value = to_type(value)
-        else:
-            value = to_type(value)
-    return value
-
-
-def convert_to_json(value: Any) -> Union[str, float, int, dict, list, bool]:
-    if isinstance(value, Path):
-        value = value.as_posix()
-    elif isinstance(value, Enum):
-        value = value.name
-    if type(value) not in (float, int, bool, str):
-        if isinstance(value, Collection) and not isinstance(value, str):
-            if isinstance(value, Mapping):
-                value = {
-                    convert_to_json(key): convert_to_json(entry)
-                    for key, entry in value.items()
-                }
-            else:
-                value = [convert_to_json(entry) for entry in value]
-        else:
-            try:
-                value = convert_value(value, float)
-            except:
-                try:
-                    value = convert_value(value, int)
-                except:
-                    try:
-                        value = convert_value(value, bool)
-                    except:
-                        value = convert_value(value, str)
-    return value
-
-
 def download_mesh(url: str, directory: PathLike, overwrite: bool = False):
     if not isinstance(directory, Path):
         directory = Path(directory)

setup.py

@@ -15,6 +15,7 @@
     'numpy': [],
     'pyproj': [],
     'requests': [],
+    'typepigeon>=1.0.3': [],
 }
 
 
@@ -41,7 +42,7 @@ def missing_packages(required_packages: {str: [str]}) -> {str: [str]}:
             required_package
             for required_package in required_packages
             if re.split('<|<=|==|>=|>', required_package)[0].lower()
-            not in installed_packages()
+               not in installed_packages()
         ]
 
 
@@ -88,10 +89,10 @@ def missing_packages(required_packages: {str: [str]}) -> {str: [str]}:
             non_conda_packages = [
                 package.replace('-', '').strip()
                 for package in output[
-                    output.index(package_not_found_start) : output.index(
-                        package_not_found_stop
-                    )
-                ].splitlines()[2:]
+                               output.index(package_not_found_start): output.index(
+                                   package_not_found_stop
+                               )
+                               ].splitlines()[2:]
             ]
             conda_packages = [
                 package