separated parallel routines into parallel.py, fixed readiness bug in …

…subsampling in readiness

descqa/configs/srv_ngals.yaml

@@ -1,28 +1,26 @@
 subclass_name: srv_ngals.CheckNgals
 description: 'Check flags'
 included_by_default: true
-
 ra: 'ra'
 dec: 'dec'
-nside: 64
 
 flags_to_check:
   - quantities: 'psFlux_flag_u'
     kind: 'bool'
-    flag_val: True
+    flag_val: False
 
   - quantities: 'good'
     kind: 'bool'
-    flag_val: True
+    flag_val: False
 
   - quantities: 'I_flag'
     kind: 'bool'
-    flag_val: True
+    flag_val: False
 
   - quantities: 'cModelFlux_flag_r'
     kind: 'bool'
-    flag_val: True
+    flag_val: False
 
 catalog_filters:
 
-  - filters: ['mag_r < 28', 'mag_i < 28']
+  - filters: ['extendedness==1']

descqa/configs/srv_readiness.yaml

@@ -2,7 +2,7 @@ subclass_name: readiness_test_synchronous.CheckQuantities
 description: 'Plot histograms of listed quantities and perform range, finiteness, mean and standard deviation checks.'
 included_by_default: true
 
-percentile_max: 100000
+subset_size: 100000
 
 quantities_to_check:
   - quantities: ['ra', 'dec']
@@ -12,13 +12,6 @@ quantities_to_check:
     max: [2.49, 2.51]
     f_inf: 0
 
-  - quantities: ['blendedness']
-    kind: 'double'
-    label: 'blendedness'
-    min: [-0.05, 0.05]
-    max: [0.95, 1.05]
-    f_inf: 0
-
   - quantities: 'cModelFlux_i'
     kind: 'double'
     label: 'cModelFlux'

descqa/parallel.py

@@ -0,0 +1,103 @@
+"""
+MPI-related utility functions for descqa
+"""
+from __future__ import unicode_literals, division, print_function, absolute_import
+import numpy as np
+from mpi4py import MPI
+
+comm = MPI.COMM_WORLD
+size = comm.Get_size()
+rank = comm.Get_rank()
+
+
+__all__ = [
+    'send_to_master',
+    'get_ra_dec',
+]
+
+
+def send_to_master(value, kind):
+    """
+    Parameters
+    ----------
+    value : ndarray
+        rank-local array of values to communicate
+    kind : str
+        type of variable. Currently implemented options are double, bool 
+
+    Returns
+    -------
+    recvbuf : ndarray
+        array of all rank values for rank 0 
+        None value otherwise
+    """
+    count = len(value)
+    tot_num = comm.reduce(count)
+    counts = comm.allgather(count)
+
+    if rank==0:
+        if kind=='double':
+            recvbuf = np.zeros(tot_num)
+        elif kind=='bool':
+            recvbuf = np.zeros(tot_num)!=0.0
+        else:
+            raise NotImplementedError
+    else:
+        recvbuf = None
+
+    displs = np.array([sum(counts[:p]) for p in range(size)])
+    if kind=='double':
+        comm.Gatherv([value,MPI.DOUBLE], [recvbuf,counts,displs,MPI.DOUBLE],root=0)
+    elif kind=='float':
+        comm.Gatherv([value,MPI.FLOAT], [recvbuf,counts,displs,MPI.FLOAT],root=0)
+    elif kind=='int':
+        comm.Gatherv([value,MPI.INT], [recvbuf,counts,displs,MPI.INT],root=0)
+    elif kind=='bool':
+        comm.Gatherv([value,MPI.BOOL], [recvbuf,counts,displs,MPI.BOOL],root=0)
+    elif kind=='int64':
+        comm.Gatherv([value,MPI.INT64_T], [recvbuf, counts, displs, MPI.INT64_T],root=0)
+    else:
+        raise NotImplementedError
+
+    return recvbuf
+
+
+def get_ra_dec(ra,dec,catalog_data):
+    """
+    Parameters
+    ----------
+    ra : str
+        variable name representing right ascension 
+    dec: str
+        variable name representing declination
+    catalog_data : GCRCatalog object
+        GCRCatalog object holding catalog data 
+        
+    Returns
+    -------
+    recvbuf_ra : ndarray
+        rank 0 outputs array of all ra values
+        other ranks output a None value 
+    recvbuf_dec : ndarray 
+        rank 0 outputs array of all dec values 
+        other ranks output a None value
+
+    """
+    data_rank={}
+    recvbuf={}
+    for quantity in [ra,dec]:
+        data_rank[quantity] = catalog_data[quantity]
+        count = len(data_rank[quantity])
+        tot_num = comm.reduce(count)
+        counts = comm.allgather(count)
+        if rank==0:
+            recvbuf[quantity] = np.zeros(tot_num)
+        else:
+            recvbuf[quantity] = None
+        displs = np.array([sum(counts[:p]) for p in range(size)])
+        comm.Gatherv([data_rank[quantity],MPI.DOUBLE], [recvbuf[quantity], counts, displs, MPI.DOUBLE],root=0)
+    recvbuf_ra = recvbuf[ra]
+    recvbuf_dec = recvbuf[dec]
+
+    return recvbuf_ra, recvbuf_dec
+