pr review changes

sup3r/models/abstract.py

@@ -910,7 +910,8 @@ class AbstractWindInterface(ABC):
     for Sup3r wind model subclasses
     """
     # pylint: disable=E0211
-    def set_model_params(self, **kwargs):
+    @staticmethod
+    def set_model_params(**kwargs):
         """Set parameters used for training the model
 
         Parameters

sup3r/models/wind.py

@@ -69,7 +69,7 @@ def set_model_params(self, **kwargs):
             Keyword arguments including 'training_features', 'output_features',
             'smoothed_features', 's_enhance', 't_enhance', 'smoothing'
         """
-        AbstractWindInterface.set_model_params(self, **kwargs)
+        AbstractWindInterface.set_model_params(**kwargs)
         Sup3rGan.set_model_params(self, **kwargs)
 
     @tf.function

sup3r/models/wind_conditional_moments.py

@@ -33,7 +33,7 @@ def set_model_params(self, **kwargs):
             Keyword arguments including 'training_features', 'output_features',
             'smoothed_features', 's_enhance', 't_enhance', 'smoothing'
         """
-        AbstractWindInterface.set_model_params(self, **kwargs)
+        AbstractWindInterface.set_model_params(**kwargs)
         Sup3rCondMom.set_model_params(self, **kwargs)
 
     @tf.function

sup3r/pipeline/forward_pass.py

@@ -24,10 +24,10 @@
 from sup3r.postprocessing.file_handling import (OutputHandlerH5,
                                                 OutputHandlerNC,
                                                 OutputHandler)
-from sup3r.utilities.utilities import (DistributedProcess,
-                                       get_chunk_slices,
+from sup3r.utilities.utilities import (get_chunk_slices,
                                        get_source_type,
                                        get_input_handler_class)
+from sup3r.utilities.execution import DistributedProcess
 from sup3r.utilities import ModuleName
 from sup3r.utilities.cli import BaseCLI
 

sup3r/utilities/execution.py

@@ -0,0 +1,135 @@
+# -*- coding: utf-8 -*-
+"""Execution methods for running some cli routines
+
+@author: bbenton
+"""
+
+import numpy as np
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+
+class DistributedProcess:
+    """High-level class with commonly used functionality for processes
+    distributed across multiple nodes"""
+
+    def __init__(self, max_nodes=1, n_chunks=None, max_chunks=None,
+                 incremental=False):
+        """
+        Parameters
+        ----------
+        max_nodes : int, optional
+            Max number of nodes to distribute processes across
+        n_chunks : int, optional
+            Number of chunks to split all processes into. These process
+            chunks will be distributed across nodes.
+        max_chunks : int, optional
+            Max number of chunks processes can be split into.
+        incremental : bool
+            Whether to skip previously run process chunks or to overwrite.
+        """
+        msg = ('For a distributed process either max_chunks or '
+               'max_chunks + n_chunks must be specified. Received '
+               f'max_chunks={max_chunks}, n_chunks={n_chunks}.')
+        assert max_chunks is not None, msg
+        self._node_chunks = None
+        self._n_chunks = n_chunks
+        self._max_nodes = max_nodes
+        self._max_chunks = max_chunks
+        self._out_files = None
+        self._failed_chunks = False
+        self.incremental = incremental
+
+    def __len__(self):
+        """Get total number of process chunks"""
+        return self.chunks
+
+    def node_finished(self, node_index):
+        """Check if all out files for a given node have been saved
+
+        Parameters
+        ----------
+        node_index : int
+            Index of node to check for completed processes
+
+        Returns
+        -------
+        bool
+            Whether all processes for the given node have finished
+        """
+        return all(self.chunk_finished(i)
+                   for i in self.node_chunks[node_index])
+
+    # pylint: disable=E1136
+    def chunk_finished(self, chunk_index):
+        """Check if process for given chunk_index has already been run.
+
+        Parameters
+        ----------
+        chunk_index : int
+            Index of the process chunk to check for completion. Considered
+            finished if there is already an output file and incremental is
+            False.
+
+        Returns
+        -------
+        bool
+            Whether the process for the given chunk has finished
+        """
+        out_file = self.out_files[chunk_index]
+        if os.path.exists(out_file) and self.incremental:
+            logger.info('Not running chunk index {}, output file '
+                        'exists: {}'.format(chunk_index, out_file))
+            return True
+        return False
+
+    @property
+    def all_finished(self):
+        """Check if all out files have been saved"""
+        return all(self.node_finished(i) for i in range(self.nodes))
+
+    @property
+    def out_files(self):
+        """Get list of out files to write process output to"""
+        return self._out_files
+
+    @property
+    def max_nodes(self):
+        """Get uncapped max number of nodes to distribute processes across"""
+        return self._max_nodes
+
+    @property
+    def chunks(self):
+        """Get the number of process chunks for this distributed routine."""
+        if self._n_chunks is None:
+            return self._max_chunks
+        else:
+            return min(self._n_chunks, self._max_chunks)
+
+    @property
+    def nodes(self):
+        """Get the max number of nodes to distribute chunks across, limited by
+        the number of process chunks"""
+        return len(self.node_chunks)
+
+    @property
+    def node_chunks(self):
+        """Get the chunk indices for different nodes"""
+        if self._node_chunks is None:
+            n_chunks = min(self.max_nodes, self.chunks)
+            self._node_chunks = np.array_split(np.arange(self.chunks),
+                                               n_chunks)
+        return self._node_chunks
+
+    @property
+    def failed_chunks(self):
+        """Check whether any processes have failed."""
+        return self._failed_chunks
+
+    @failed_chunks.setter
+    def failed_chunks(self, failed):
+        """Set failed_chunks value. Should be set to True if there is a failed
+        chunk"""
+        self._failed_chunks = failed

sup3r/utilities/regridder.py

@@ -15,7 +15,7 @@
 
 from sup3r.postprocessing.file_handling import OutputMixIn, RexOutputs
 from sup3r.utilities import ModuleName
-from sup3r.utilities.utilities import DistributedProcess
+from sup3r.utilities.execution import DistributedProcess
 from sup3r.utilities.cli import BaseCLI
 
 logger = logging.getLogger(__name__)

sup3r/utilities/utilities.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
-"""Utilities module for preparing
-training data
+"""Utilities module for preparing training data
 
 @author: bbenton
 """
@@ -26,130 +25,6 @@
 logger = logging.getLogger(__name__)
 
 
-class DistributedProcess:
-    """High-level class with commonly used functionality for processes
-    distributed across multiple nodes"""
-
-    def __init__(self, max_nodes=1, n_chunks=None, max_chunks=None,
-                 incremental=False):
-        """
-        Parameters
-        ----------
-        max_nodes : int, optional
-            Max number of nodes to distribute processes across
-        n_chunks : int, optional
-            Number of chunks to split all processes into. These process
-            chunks will be distributed across nodes.
-        max_chunks : int, optional
-            Max number of chunks processes can be split into.
-        incremental : bool
-            Whether to skip previously run process chunks or to overwrite.
-        """
-        msg = ('For a distributed process either max_chunks or '
-               'max_chunks + n_chunks must be specified. Received '
-               f'max_chunks={max_chunks}, n_chunks={n_chunks}.')
-        assert max_chunks is not None, msg
-        self._node_chunks = None
-        self._n_chunks = n_chunks
-        self._max_nodes = max_nodes
-        self._max_chunks = max_chunks
-        self._out_files = None
-        self._failed_chunks = False
-        self.incremental = incremental
-
-    def __len__(self):
-        """Get total number of process chunks"""
-        return self.chunks
-
-    def node_finished(self, node_index):
-        """Check if all out files for a given node have been saved
-
-        Parameters
-        ----------
-        node_index : int
-            Index of node to check for completed processes
-
-        Returns
-        -------
-        bool
-            Whether all processes for the given node have finished
-        """
-        return all(self.chunk_finished(i)
-                   for i in self.node_chunks[node_index])
-
-    # pylint: disable=E1136
-    def chunk_finished(self, chunk_index):
-        """Check if process for given chunk_index has already been run.
-
-        Parameters
-        ----------
-        chunk_index : int
-            Index of the process chunk to check for completion. Considered
-            finished if there is already an output file and incremental is
-            False.
-
-        Returns
-        -------
-        bool
-            Whether the process for the given chunk has finished
-        """
-        out_file = self.out_files[chunk_index]
-        if os.path.exists(out_file) and self.incremental:
-            logger.info('Not running chunk index {}, output file '
-                        'exists: {}'.format(chunk_index, out_file))
-            return True
-        return False
-
-    @property
-    def all_finished(self):
-        """Check if all out files have been saved"""
-        return all(self.node_finished(i) for i in range(self.nodes))
-
-    @property
-    def out_files(self):
-        """Get list of out files to write process output to"""
-        return self._out_files
-
-    @property
-    def max_nodes(self):
-        """Get uncapped max number of nodes to distribute processes across"""
-        return self._max_nodes
-
-    @property
-    def chunks(self):
-        """Get the number of processes chunks for this distributed routine."""
-        if self._n_chunks is None:
-            return self._max_chunks
-        else:
-            return min(self._n_chunks, self._max_chunks)
-
-    @property
-    def nodes(self):
-        """Get the max number of nodes to distribute chunks across, limited by
-        the number of process chunks"""
-        return len(self.node_chunks)
-
-    @property
-    def node_chunks(self):
-        """Get the chunk indices for different nodes"""
-        if self._node_chunks is None:
-            n_chunks = min(self.max_nodes, self.chunks)
-            self._node_chunks = np.array_split(np.arange(self.chunks),
-                                               n_chunks)
-        return self._node_chunks
-
-    @property
-    def failed_chunks(self):
-        """Check whether any processes have failed."""
-        return self._failed_chunks
-
-    @failed_chunks.setter
-    def failed_chunks(self, failed):
-        """Set failed_chunks value. Should be set to True if there is a failed
-        chunk"""
-        self._failed_chunks = failed
-
-
 def correct_path(path):
     """If running on windows we need to replace backslashes with double
     backslashes so paths can be parsed correctly with safe_open_json"""