Pysmurf op data quality checks

socs/agents/pysmurf_controller/agent.py

@@ -24,7 +24,7 @@
 from sodetlib.operations import bias_dets
 
 from socs.agents.pysmurf_controller.smurf_subprocess_util import (
-    RunCfg, RunResult, run_smurf_func)
+    QuantileData, RunCfg, RunResult, run_smurf_func)
 
 
 class PysmurfScriptProtocol(protocol.ProcessProtocol):
@@ -129,7 +129,10 @@ def __init__(self, agent, args):
                 'observatory.{}.feeds.pysmurf_session_data'.format(args.monitor_id),
             )
 
-        self.agent.register_feed('bias_step_quantiles', record=True)
+        self.agent.register_feed('bias_step_results', record=True)
+        self.agent.register_feed('noise_results', record=True)
+        self.agent.register_feed('iv_results', record=True)
+        self.agent.register_feed('bias_wave_results', record=True)
 
     def _on_session_data(self, _data):
         data, feed = _data
@@ -655,6 +658,18 @@ def take_noise(self, session, params):
 
             result = run_smurf_func(cfg)
             set_session_data(session, result)
+            if result.success:
+                block_data = {}
+                for qd in result.return_val['quantiles'].values():
+                    if isinstance(qd, dict):
+                        qd = QuantileData(**qd)
+                    block_data.update(qd.to_block_data())
+                d = {
+                    'timestamp': time.time(),
+                    'block_name': 'noise_results',
+                    'data': block_data
+                }
+                self.agent.publish_to_feed('noise_results', d)
             return result.success, "Finished taking noise"
 
     @ocs_agent.param('kwargs', default=None)
@@ -747,11 +762,10 @@ def take_iv(self, session, params):
 
             >> response.session['data']
             {
-                'bands': Bands number of each resonator
-                'channels': Channel number of each resonator
-                'bgmap': BGMap assignment for each resonator
-                'R_n': Normal resistance for each resonator
                 'filepath': Filepath of saved IVAnalysis object
+                'quantiles': {
+                    'Rn': Rn quantiles
+                }
             }
         """
         if params['kwargs'] is None:
@@ -770,6 +784,18 @@ def take_iv(self, session, params):
             )
             result = run_smurf_func(cfg)
             set_session_data(session, result)
+            if result.success:
+                block_data = {}
+                for qd in result.return_val['quantiles'].values():
+                    if isinstance(qd, dict):
+                        qd = QuantileData(**qd)
+                    block_data.update(qd.to_block_data())
+                d = {
+                    'timestamp': time.time(),
+                    'block_name': 'iv_results',
+                    'data': block_data
+                }
+                self.agent.publish_to_feed('iv_results', d)
             return result.success, "Finished taking IV"
 
     @ocs_agent.param('kwargs', default=None)
@@ -807,14 +833,11 @@ def take_bias_steps(self, session, params):
                 'filepath': Filepath of saved BiasStepAnalysis object
                 'biased_total': Total number of detectors biased into rfrac_range
                 'biased_per_bg': List containing number of biased detectors on each bias line
-                'Rtes_quantiles': {
-                    'Rtes': List of 15%, 25%, 50%, 75%, 85% Rtes quantiles,
-                    'quantiles': List of quantile labels
-                    'count': Total count of the distribution
+                'quantiles': {
+                    'Rtes': Rtes quantiles,
+                    'Rfrac': Rfrac quantiles,
+                    'Si': Si quantiles,
                 }
-                'responsivity_quantiles': Same as above for responsivity
-                'Rfrac_quantiles': Same as above for Rfrac
-
             }
         """
 
@@ -838,15 +861,21 @@ def take_bias_steps(self, session, params):
             result = run_smurf_func(cfg)
             set_session_data(session, result)
             if result.success:  # Publish quantile results
-                for name, d in result.return_val['quantiles'].items():
-                    block = dict(zip(d['labels'], d['values']))
-                    block[f'{name}_count'] = d['count']
-                    pub_data = {
-                        'timestamp': time.time(),
-                        'block_name': f'{name}_quantile',
-                        'data': block
-                    }
-                    self.agent.publish_to_feed('bias_step_quantiles', pub_data)
+                block_data = {
+                    f'biased_bg{bg}': v
+                    for bg, v in enumerate(result.return_val['biased_per_bg'])
+                }
+                block_data['biased_total'] = result.return_val['biased_total']
+                for qd in result.return_val['quantiles'].values():
+                    if isinstance(qd, dict):
+                        qd = QuantileData(**qd)
+                    block_data.update(qd.to_block_data())
+                data = {
+                    'timestamp': time.time(),
+                    'block_name': 'bias_steps_results',
+                    'data': block_data
+                }
+                self.agent.publish_to_feed('bias_step_results', data)
 
             return result.success, "Finished taking bias steps"
 
@@ -883,14 +912,11 @@ def take_bias_waves(self, session, params):
                 'filepath': Filepath of saved BiasWaveAnalysis object
                 'biased_total': Total number of detectors biased into rfrac_range
                 'biased_per_bg': List containing number of biased detectors on each bias line
-                'Rtes_quantiles': {
-                    'Rtes': List of 15%, 25%, 50%, 75%, 85% Rtes quantiles,
-                    'quantiles': List of quantile labels
-                    'count': Total count of the distribution
+                'quantiles': {
+                    'Rtes': Rtes quantiles,
+                    'Rfrac': Rfrac quantiles,
+                    'Si': Si quantiles,
                 }
-                'responsivity_quantiles': Same as above for responsivity
-                'Rfrac_quantiles': Same as above for Rfrac
-
             }
         """
 
@@ -914,15 +940,19 @@ def take_bias_waves(self, session, params):
             result = run_smurf_func(cfg)
             set_session_data(session, result)
             if result.success:  # Publish quantile results
-                for name, d in result.return_val['quantiles'].items():
-                    block = dict(zip(d['labels'], d['values']))
-                    block[f'{name}_count'] = d['count']
-                    pub_data = {
-                        'timestamp': time.time(),
-                        'block_name': f'{name}_quantile',
-                        'data': block
-                    }
-                    self.agent.publish_to_feed('bias_wave_quantiles', pub_data)
+                block_data = {
+                    f'biased_bg{bg}': v
+                    for bg, v in enumerate(result.return_val['biased_per_bg'])
+                }
+                block_data['biased_total'] = result.return_val['biased_total']
+                for qd in result.return_val['quantiles'].values():
+                    block_data.update(QuantileData(**qd).to_block_data())
+                data = {
+                    'timestamp': time.time(),
+                    'block_name': 'bias_wave_results',
+                    'data': block_data
+                }
+                self.agent.publish_to_feed('bias_wave_results', data)
 
             return result.success, "Finished taking bias steps"
 

socs/agents/pysmurf_controller/smurf_subprocess_util.py

@@ -2,7 +2,7 @@
 import os
 import sys
 import traceback
-from dataclasses import asdict, dataclass, field
+from dataclasses import asdict, dataclass, field, is_dataclass
 from typing import Dict, List, Optional
 
 import numpy as np
@@ -23,6 +23,8 @@ def json_safe(data):
         Python basic types.
     - Converting NaN/inf to Num (0 or +/- large number)
     """
+    if is_dataclass(data):
+        return json_safe(asdict(data))
     if isinstance(data, dict):
         return {k: json_safe(v) for k, v in data.items()}
     if isinstance(data, (list, tuple)):
@@ -34,17 +36,50 @@ def json_safe(data):
         return data
     if isinstance(data, float):
         return np.nan_to_num(data)
+    if data is None:
+        return None
+
+    print(f"Unsure how to encode data of type {type(data)}...")
+    print(f"data: {data}")
     # This could still be something weird but json.dumps will
     # probably reject it!
     return data
 
 
+@dataclass
+class QuantileData:
+    name: str
+    quantiles: List[float]
+    values: List[float]
+    total: int
+
+    def to_block_data(self):
+        "Format quantile data for OCS HK block"
+        labels = [f'{self.name}_q{int(q)}' for q in self.quantiles]
+        block_data = dict(zip(labels, self.values))
+        block_data[f'{self.name}_total'] = self.total
+        return block_data
+
+
+def compute_quantiles(name: str, arr: np.ndarray, quantiles: List[float]):
+    """
+    Computes QuantileData object for a given array
+
+    """
+    quantiles = [float(q) for q in quantiles]
+    if np.isnan(arr).all():
+        return QuantileData(name, quantiles, [0 for _ in quantiles], 0)
+    qs = [float(np.nan_to_num(np.nanquantile(arr, q / 100))) for q in quantiles]
+    total = int(np.sum(~np.isnan(arr)))
+    return QuantileData(name, quantiles, qs, total)
+
+
 def encode_dataclass(obj):
     """
     Encodes a data-class into json, replacing any json-unsafe types with
     reasonable alternatives.
     """
-    data = json_safe(asdict(obj))
+    data = json_safe(obj)
     return json.dumps(data).encode()
 
 
@@ -66,7 +101,14 @@ def take_noise(duration, kwargs=None):
     if kwargs is None:
         kwargs = {}
     S, cfg = get_smurf_control()
-    sdl.noise.take_noise(S, cfg, duration, **kwargs)
+    _, res = sdl.noise.take_noise(S, cfg, duration, **kwargs)
+    wls = res['noise_pars'][:, 0]
+    qs = [10, 25, 50, 75, 90]
+    return {
+        'quantiles': {
+            'white_noise_level': compute_quantiles('white_noise_level', wls, qs),
+        }
+    }
 
 
 def take_bgmap(kwargs=None):
@@ -91,7 +133,13 @@ def take_iv(iv_kwargs=None):
     if iv_kwargs is None:
         iv_kwargs = {}
     iva = iv.take_iv(S, cfg, **iv_kwargs)
-    return {'filepath': iva.filepath}
+    quantiles = [10, 25, 50, 75, 90]
+    return {
+        'filepath': iva.filepath,
+        'quantiles': {
+            'Rn': compute_quantiles('Rn', iva.R_n, quantiles)
+        }
+    }
 
 
 def run_uxm_setup(bands=None, kwargs=None):
@@ -112,28 +160,6 @@ def run_uxm_relock(bands=None, kwargs=None):
     return None
 
 
-def _process_quantiles(quantiles: list, arrays: Dict):
-    """
-    Args
-    ----
-    quantiles: list of quantiles to compute (in percent)
-    arrays: Dict of arrays to compute quantiles for
-    """
-    res = {}
-    for name, arr in arrays.items():
-        if np.isnan(arr).all():
-            continue
-        labels = [f'{name}_q{q}' for q in quantiles]
-        qs = [float(np.nan_to_num(np.nanquantile(arr, q / 100))) for q in quantiles]
-        count = int(np.sum(~np.isnan(arr)))
-        res[name] = {
-            'values': qs,
-            'labels': labels,
-            'count': count,
-        }
-    return res
-
-
 def take_bias_steps(kwargs=None, rfrac_range=(0.2, 0.9)):
     """Takes bias steps and computes quantiles for various parameters"""
     if kwargs is None:
@@ -146,18 +172,19 @@ def take_bias_steps(kwargs=None, rfrac_range=(0.2, 0.9)):
         rfrac_range[0] < bsa.Rfrac,
         rfrac_range[1] > bsa.Rfrac
     ])
+    qs = [10, 25, 50, 75, 90]
     data = {
         'filepath': bsa.filepath,
         'biased_total': int(np.sum(biased)),
         'biased_per_bg': [
             int(np.sum(biased[bsa.bgmap == bg])) for bg in range(12)
         ],
+        'quantiles': {
+            'Rfrac': compute_quantiles('Rfrac', bsa.Rfrac, qs),
+            'Si': compute_quantiles('Si', bsa.Si, qs),
+            'Rtes': compute_quantiles('Rtes', bsa.R0, qs),
+        }
     }
-    arrays = {
-        'Rfrac': bsa.Rfrac, 'responsivity': bsa.Si, 'Rtes': bsa.R0,
-    }
-    quantiles = np.array([15, 25, 50, 75, 85])
-    data['quantiles'] = _process_quantiles(quantiles, arrays)
     return data
 
 
@@ -172,18 +199,19 @@ def take_bias_waves(kwargs=None, rfrac_range=(0.2, 0.9)):
         rfrac_range[0] < bwa.Rfrac,
         rfrac_range[1] > bwa.Rfrac
     ])
+    qs = [10, 25, 50, 75, 90]
     data = {
         'filepath': bwa.filepath,
         'biased_total': int(np.sum(biased)),
         'biased_per_bg': [
             int(np.sum(biased[bwa.bgmap == bg])) for bg in range(12)
         ],
+        'quantiles': {
+            'Rfrac': compute_quantiles('Rfrac', bwa.Rfrac, qs),
+            'Si': compute_quantiles('Si', bwa.Si, qs),
+            'Rtes': compute_quantiles('Rtes', bwa.R0, qs),
+        }
     }
-    arrays = {
-        'Rfrac': bwa.Rfrac, 'responsivity': bwa.Si, 'Rtes': bwa.R0,
-    }
-    quantiles = np.array([15, 25, 50, 75, 85])
-    data['quantiles'] = _process_quantiles(quantiles, arrays)
     return data
 
 

tests/agents/test_pysmurf_controller_agent.py

@@ -127,7 +127,7 @@ def mock_take_noise(S, cfg, acq_time, **kwargs):
     **Mock** - Mock take_noise() in sodetlib.
     """
     am = mock.MagicMock()
-    outdict = {'noise_pars': 0,
+    outdict = {'noise_pars': np.zeros((10, 3), dtype=float),
                'bands': 0,
                'channels': 0,
                'band_medians': 0,