Merge pull request #420 from ExaWorks/parse_walltime_function

Move the walltime parsing routine in the serialiation module to

Author: hategan

src/psij/job_attributes.py

@@ -1,9 +1,14 @@
+import re
 from datetime import timedelta
 from typing import Optional, Dict
 
 from typeguard import check_argument_types
 
 
+_WALLTIME_FMT_ERROR = 'Unknown walltime format: %s. Accepted formats are hh:mm:ss, ' \
+                      'hh:mm, mm, or n\\s*[h|m|s].'
+
+
 class JobAttributes(object):
     """A class containing ancillary job information that describes how a job is to be run."""
 
@@ -92,3 +97,48 @@ def __eq__(self, o: object) -> bool:
                 return False
 
         return True
+
+    @staticmethod
+    def parse_walltime(walltime: str) -> timedelta:
+        r"""
+        Parses a walltime string into a :class:`~datetime.timedelta`.
+
+        The accepted walltime strings formats are:
+        * hh:mm:ss
+        * hh:mm
+        * mm
+        * n\s*[y|M|d|h|m\s]
+
+        Parameters
+        ----------
+        walltime
+            A string in one of the above formats representing a time duration
+
+        Returns
+        -------
+        A :class:`~datetime.timedelta` representing the same time duration as the ``walltime``
+        parameter.
+        """
+        if ':' in walltime:
+            parts = walltime.split(':')
+            seconds = 0
+            if len(parts) == 3:
+                seconds = int(parts[2])
+            if len(parts) <= 3:
+                return timedelta(hours=int(parts[0]), minutes=int(parts[1]), seconds=seconds)
+            else:
+                raise ValueError(_WALLTIME_FMT_ERROR % walltime)
+        if walltime.isdigit():
+            return timedelta(minutes=int(walltime))
+        m = re.search(r'(\d+)\s*([hms])', walltime)
+        if m:
+            digits = m.group(1)
+            unit = m.group(2)
+            val = int(digits)
+            if unit == 'h':
+                return timedelta(hours=val)
+            elif unit == 'm':
+                return timedelta(minutes=val)
+            elif unit == 's':
+                return timedelta(seconds=val)
+        raise ValueError(_WALLTIME_FMT_ERROR % walltime)

src/psij/serialize.py

@@ -1,6 +1,5 @@
 import inspect
 import json
-import re
 import typing
 from abc import ABC, abstractmethod
 from datetime import timedelta
@@ -245,7 +244,7 @@ def _to_object(self, s: object, t: object) -> object:
                 return s
             if issubclass(t, timedelta):
                 assert isinstance(s, str)
-                return self._to_timedelta(s)
+                return JobAttributes.parse_walltime(s)
         else:
             if t == Union[str, Path] or t == Optional[Union[str, Path]]:
                 assert isinstance(s, str)
@@ -273,40 +272,6 @@ def _to_dict(self, d: Dict[str, object]) -> Dict[str, object]:
     def _to_list(self, lst: List[object]) -> List[object]:
         return [self._to_object(v, type(v)) for v in lst]
 
-    _TIMEDELTA_FMT_ERROR = 'Unknown time interval format: %s. Accepted formats are hh:mm:ss, ' \
-                           'hh:mm, mm, or n\\s*[h|m|s].'
-
-    def _to_timedelta(self, s: str) -> timedelta:
-        # we accept multiple formats here:
-        #   hh:mm:ss
-        #   hh:mm
-        #   mm
-        #   n\s*[y|M|d|h|m\s]
-
-        if ':' in s:
-            parts = s.split(':')
-            seconds = 0
-            if len(parts) == 3:
-                seconds = int(parts[2])
-            if len(parts) <= 3:
-                return timedelta(hours=int(parts[0]), minutes=int(parts[1]), seconds=seconds)
-            else:
-                raise ValueError(Serializer._TIMEDELTA_FMT_ERROR % s)
-        if s.isdigit():
-            return timedelta(minutes=int(s))
-        m = re.search(r'(\d+)\s*([hms])', s)
-        if m:
-            digits = m.group(1)
-            unit = m.group(2)
-            val = int(digits)
-            if unit == 'h':
-                return timedelta(hours=val)
-            elif unit == 'm':
-                return timedelta(minutes=val)
-            elif unit == 's':
-                return timedelta(seconds=val)
-        raise ValueError(Serializer._TIMEDELTA_FMT_ERROR % s)
-
 
 class JSONSerializer(Serializer):
     """A JSON serializer."""