Added PBS backend and make float32 default for Marchenko class

Author: mrava87

README.md

@@ -2,7 +2,7 @@
 
 [![PyPI version](https://badge.fury.io/py/pylops-distributed.svg)](https://badge.fury.io/py/pylops-distributed)
 [![Build Status](https://travis-ci.org/equinor/pylops-distributed.svg?branch=master)](https://travis-ci.org/equinor/pylops-distributed)
-[![AzureDevOps Status](https://dev.azure.com/MRAVA/PyLops/_apis/build/status/equinor.pylops-distributed?branchName=master)](https://dev.azure.com/MRAVA/PyLops/_build/latest?definitionId=3&branchName=master)
+[![AzureDevOps Status](https://dev.azure.com/matteoravasi/PyLops/_apis/build/status/equinor.pylops-distributed?branchName=master)](https://dev.azure.com/matteoravasi/PyLops/_build/latest?definitionId=4&branchName=master)
 [![Documentation Status](https://readthedocs.org/projects/pylops-distributed/badge/?version=latest)](https://pylops-distributed.readthedocs.io/en/latest/?badge=latest)
 [![OS-support](https://img.shields.io/badge/OS-linux,osx-850A8B.svg)](https://github.com/equinor/pylops)
 [![Slack Status](https://img.shields.io/badge/chat-slack-green.svg)](https://pylops.slack.com)

environment-dev.yml

@@ -16,6 +16,7 @@ dependencies:
   - numba
   - icc_rt
   - dask>=2.0.0
+  - dask-jobqueue
   - zarr
   - pylops
   - pip:

environment.yml

@@ -10,4 +10,5 @@ dependencies:
   - numba
   - icc_rt
   - dask>=2.0.0
+  - dask-jobqueue
   - pylops

pylops_distributed/__init__.py

@@ -28,4 +28,8 @@
 try:
     from .version import version as __version__
 except ImportError:
-    __version__ = '0.0.0'
+    # If it was not installed, then we don't know the version. We could throw a
+    # warning here, but this case *should* be rare. pylops should be installed
+    # properly!
+    from datetime import datetime
+    __version__ = 'unknown-'+datetime.today().strftime('%Y%m%d')

pylops_distributed/utils/backend.py

@@ -1,8 +1,15 @@
 from dask.distributed import Client, LocalCluster
 
 
+try:
+    from dask_jobqueue import PBSCluster
+    jobqueue = True
+except:
+    jobqueue = False
+
+
 def dask(hardware='single', client=None, processes=False, n_workers=1,
-         threads_per_worker=1):
+         threads_per_worker=1, **kwargscluster):
     r"""Dask backend initialization.
 
     Create connection to drive computations using Dask distributed.
@@ -11,30 +18,45 @@ def dask(hardware='single', client=None, processes=False, n_workers=1,
     ----------
     hardware : :obj:`str`, optional
         Hardware used to run Dask distributed. Currently available options
-        are ``single`` for single-machine distribution.
+        are ``single`` for single-machine distribution, ``ssh`` for
+        SSH-bases multi-machine distribution and ``pbs`` for
+        PBS-bases multi-machine distribution
     client : :obj:`str`, optional
-        Name of scheduler (use ``None`` for ``hardware=single``)n.
+        Name of scheduler (use ``None`` for ``hardware=single``).
     processes : :obj:`str`, optional
         Whether to use processes (``True``) or threads (``False``).
     n_workers : :obj:`int`, optional
         Number of workers
     threads_per_worker : :obj:`int`, optional
         Number of threads per each worker
+    kwargscluster:
+        Additional parameters to be passed to the cluster creation routine
 
     Returns
     -------
     client : :obj:`dask.distributed.client.Client`
+        Client
+    cluster :
+        Cluster
 
     Raises
     ------
-    ValueError
-        If ``hardware`` is not ``single``
+    NotImplementedError
+        If ``hardware`` is not ``single``, ``ssh``, or ``pbs``
 
     """
     if hardware == 'single':
         cluster = LocalCluster(processes=processes, n_workers=n_workers,
                                threads_per_worker=threads_per_worker)
-    else:
+    elif hardware == 'ssh':
         cluster = client
+    elif hardware == 'pbs':
+        if jobqueue == False:
+            raise ModuleNotFoundError('dask-jobqueue not installed. ' \
+                                      'Run "pip install dask-jobqueue".')
+        cluster = PBSCluster(**kwargscluster)
+        cluster.scale(jobs=n_workers)
+    else:
+        raise NotImplementedError('hardware must be single, ssh, or pbs')
     client = Client(cluster)
-    return client
+    return client, cluster

pylops_distributed/waveeqprocessing/marchenko.py

@@ -79,7 +79,7 @@ class Marchenko():
 
     """
     def __init__(self, R, nt, dt=0.004, dr=1., wav=None, toff=0.0,
-                 nsmooth=10, saveRt=True, prescaled=False, dtype='float64'):
+                 nsmooth=10, saveRt=True, prescaled=False, dtype='float32'):
         # Save inputs into class
         self.nt = nt
         self.dt = dt
@@ -179,8 +179,8 @@ def apply_onepoint(self, trav, dist=None, G0=None, nfft=None,
         Rollop = Roll(self.nt2 * self.ns,
                       dims=(self.nt2, self.ns),
                       dir=0, shift=-1, dtype=self.dtype)
-        Wop = Diagonal(da.from_array(w.T.flatten()))
-        Iop = Identity(self.nr * self.nt2)
+        Wop = Diagonal(da.from_array(w.T.flatten()), dtype=self.dtype)
+        Iop = Identity(self.nr * self.nt2, dtype=self.dtype)
         Mop = Block([[Iop, -1 * Wop * Rop],
                      [-1 * Wop * Rollop * R1op, Iop]]) * BlockDiag([Wop, Wop])
         Gop = Block([[Iop, -1 * Rop],
@@ -233,8 +233,7 @@ def apply_onepoint(self, trav, dist=None, G0=None, nfft=None,
         f1_inv = cgls(Mop, d.flatten(), **kwargs_cgls)[0]
         f1_inv = f1_inv.reshape(2 * self.nt2, self.nr)
         f1_inv_tot = f1_inv + da.concatenate((da.zeros((self.nt2, self.nr),
-                                                       dtype=self.dtype),
-                                              fd_plus))
+                                                       dtype=self.dtype), fd_plus))
         # Create Green's functions
         if greens:
             g_inv = Gop * f1_inv_tot.flatten()
@@ -351,8 +350,9 @@ def apply_multiplepoints(self, trav, dist=None, G0=None, nfft=None,
         Rollop = Roll(self.ns * nvs * self.nt2,
                       dims=(self.nt2, self.ns, nvs),
                       dir=0, shift=-1, dtype=self.dtype)
-        Wop = Diagonal(da.from_array(w.transpose(2, 0, 1).flatten()))
-        Iop = Identity(self.nr * nvs * self.nt2)
+        Wop = Diagonal(da.from_array(w.transpose(2, 0, 1).flatten()),
+                       dtype=self.dtype)
+        Iop = Identity(self.nr * nvs * self.nt2, dtype=self.dtype)
         Mop = Block([[Iop, -1 * Wop * Rop],
                      [-1 * Wop * Rollop * R1op, Iop]]) * BlockDiag([Wop, Wop])
         Gop = Block([[Iop, -1 * Rop],

pylops_distributed/waveeqprocessing/mdd.py

@@ -34,8 +34,6 @@ def MDC(G, nt, nv, dt=1., dr=1., twosided=True,
     twosided : :obj:`bool`, optional
         MDC operator has both negative and positive time (``True``) or
         only positive (``False``)
-    dtype : :obj:`str`, optional
-        *Deprecated*, will be removed in v2.0.0
     saveGt : :obj:`bool`, optional
         Save ``G`` and ``G^H`` to speed up the computation of adjoint of
         :class:`pylops_distributed.signalprocessing.Fredholm1` (``True``) or create

requirements-dev.txt

@@ -2,6 +2,7 @@ numpy>=1.15.0
 scipy>=1.1.0
 numba
 dask[complete]>=2.0.0
+dask-jobqueue
 zarr
 pylops[advanced]
 matplotlib

requirements.txt

@@ -2,4 +2,5 @@ numpy>=1.15.0
 scipy>=1.1.0
 numba
 dask[complete]>=2.0.0
+dask-jobqueue
 pylops