Merge pull request #23 from SiLab-Bonn/unit_tests

Unit tests for online analysis

pymosa/m26_configuration.yaml

@@ -2,6 +2,8 @@
 run_number :  # Base run number, will be automatically increased; if none is given, generate filename
 output_folder :  # Output folder for the telescope data; if none is given, the current working directory is used
 m26_configuration_file :  # Configuration file for Mimosa26 sensors, default: 'm26_config/m26_threshold_8.yaml'
+m26_jtag_configuration : True  # Send Mimosa26 configuration via JTAG, default: True
+no_data_timeout : 30  # No data timeout after which the scan will be aborted, in seconds; if 0, the timeout is disabled
 scan_timeout : 0  # Timeout after which the scan will be stopped, in seconds; if 0, the timeout is disabled; use Ctrl-C to stop run
 max_triggers : 0  # Maximum number of triggers; if 0, there is no limit on the number of triggers; use Ctrl-C to stop run
 send_data : 'tcp://127.0.0.1:8500'  # TCP address to which the telescope data is send; to allow incoming connections on all interfaces use 0.0.0.0

pymosa/noise_occupancy_scan.py

@@ -35,7 +35,7 @@ def take_data(self, update_rate=1):
                 try:
                     self.pbar.update(update_rate)
                 except ValueError:
-                        pass
+                    pass
 
             self.pbar.close()
 

pymosa/online.py

@@ -129,6 +129,12 @@ def is_frame_trailer1(word, plane):  # Check if frame trailer1 word for the actu
     return (0x0000ffff & word) == (0xaa50 | plane)
 
 
+# Trigger words
+@njit
+def is_trigger_word(word):  # Check if TLU word (trigger)
+    return (0x80000000 & word) == 0x80000000
+
+
 @njit
 def histogram(raw_data, occ_hist, m26_frame_ids, m26_frame_length, m26_data_loss, m26_word_index, m26_timestamps, last_m26_timestamps, m26_n_words, m26_rows, m26_frame_status, last_completed_m26_frame_ids):
     ''' Raw data to 2D occupancy histogram '''
@@ -230,8 +236,11 @@ def histogram(raw_data, occ_hist, m26_frame_ids, m26_frame_length, m26_data_loss
                                 break
                             # Fill occupancy hist
                             occ_hist[column + k, m26_rows[plane_id], plane_id] += 1
-        else:  # Raw data word is TLU/trigger word
+        elif is_trigger_word(raw_data_word):  # Raw data word is TLU/trigger word
             pass  # Not needed here
+        else:  # Raw data contains unknown word, neither M26 nor TLU word
+            for tmp_plane_index in range(6):
+                m26_data_loss[tmp_plane_index] = True
 
     return m26_frame_ids, m26_frame_length, m26_data_loss, m26_word_index, m26_timestamps, last_m26_timestamps, m26_n_words, m26_rows, m26_frame_status, last_completed_m26_frame_ids
 
@@ -313,14 +322,14 @@ def worker(self, raw_data_queue, shared_array_base, lock, stop):
         # Per frame variables
         m26_frame_ids = np.zeros(shape=(6, ), dtype=np.int64)  # The Mimosa26 frame ID of the actual frame
         m26_frame_length = np.zeros(shape=(6, ), dtype=np.uint32)  # The number of "useful" data words for the actual frame
-        m26_data_loss = np.ones((6, ), dtype=np.bool)  # The data loss status for the actual frame
+        m26_data_loss = np.ones((6, ), dtype=np.bool_)  # The data loss status for the actual frame
         m26_word_index = np.zeros(shape=(6, ), dtype=np.uint32)  # The word index per device of the actual frame
         m26_timestamps = np.zeros(shape=(6, ), dtype=np.int64)  # The timestamp for each plane (in units of 40 MHz)
         last_m26_timestamps = np.zeros(shape=(6, ), dtype=np.int64)
         m26_n_words = np.zeros(shape=(6, ), dtype=np.uint32)  # The number of words containing column / row info
         m26_rows = np.zeros(shape=(6, ), dtype=np.uint32)  # The actual readout row (rolling shutter)
         m26_frame_status = np.zeros(shape=(6, ), dtype=np.uint32)  # The status flags for the actual frames
-        last_completed_m26_frame_ids = np.full(shape=6, dtype=np.int64, fill_value=-1)  # The status if the frame is complete for the actual frame
+        last_completed_m26_frame_ids = -1 * np.ones(shape=6, dtype=np.int64)  # The status if the frame is complete for the actual frame
         while not stop.is_set():
             try:
                 raw_data = raw_data_queue.get(timeout=self._queue_timeout)

pymosa/tests/test_online_analysis.py

@@ -0,0 +1,108 @@
+#
+# ------------------------------------------------------------
+# Copyright (c) All rights reserved
+# SiLab, Institute of Physics, University of Bonn
+# ------------------------------------------------------------
+#
+
+import logging
+import os
+import threading
+import time
+
+from mock import patch
+
+import pytest
+import tables as tb
+import numpy as np
+import matplotlib
+matplotlib.use('Agg')  # noqa: E402 Allow headless plotting
+
+import pymosa  # noqa: E731 E402
+from pymosa import online as oa  # noqa: E402
+from pymosa.tests import utils  # noqa: E40
+
+
+@pytest.fixture()
+def data_folder():
+    pymosa_path = os.path.dirname(pymosa.__file__)
+    print(os.path.abspath(os.path.join(pymosa_path, 'tests')))
+    return os.path.abspath(os.path.join(pymosa_path, 'tests'))
+
+
+@pytest.fixture()
+def ana_log_messages():
+    ana_logger = logging.getLogger('OnlineAnalysis')
+    _ana_log_handler = utils.MockLoggingHandler(level='DEBUG')
+    ana_logger.addHandler(_ana_log_handler)
+    ana_log_messages = _ana_log_handler.messages
+    yield ana_log_messages
+    ana_logger.removeHandler(_ana_log_handler)  # cleanup
+
+
+@pytest.fixture()
+def occ_hist_oa():
+    h = oa.OccupancyHistogramming()
+    yield h
+    h.close()
+    del h
+
+
+def get_raw_data(raw_data_file):
+    ''' Yield data of one readout
+
+        Delay return if replay is too fast
+    '''
+    with tb.open_file(raw_data_file, mode="r") as in_file_h5:
+        meta_data = in_file_h5.root.meta_data[:]
+        raw_data = in_file_h5.root.raw_data
+        n_readouts = meta_data.shape[0]
+
+        for i in range(n_readouts):
+            # Raw data indeces of readout
+            i_start = meta_data['index_start'][i]
+            i_stop = meta_data['index_stop'][i]
+
+            yield raw_data[i_start:i_stop]
+
+
+def test_occupancy_histogramming(data_folder, occ_hist_oa):
+    ''' Test online occupancy histogramming '''
+
+    raw_data_file = os.path.join(data_folder, 'anemone_raw_data.h5')
+    raw_data_file_result = os.path.join(data_folder, 'anemone_raw_data_interpreted_result.h5')
+    for words in get_raw_data(raw_data_file):
+        occ_hist_oa.add(words)
+
+    time.sleep(1.0)
+
+    occ_hist = occ_hist_oa.get()
+
+    with tb.open_file(raw_data_file_result) as in_file:
+        for i in range(6):
+            occ_hist_exptected = in_file.get_node(in_file.root, 'HistOcc_plane%d' % (i + 1))
+            assert(np.array_equal(occ_hist_exptected[:, :], occ_hist[:, :, i]))
+
+
+# def test_occupancy_histogramming_errors(data_folder, occ_hist_oa, ana_log_messages):
+#     # Check error message when requesting histogram before analysis finished
+
+#     def add_data():
+#         for words in get_raw_data(raw_data_file):
+#             for _ in range(100):
+#                 occ_hist_oa.add(words)
+
+#     raw_data_file = os.path.join(data_folder, 'anemone_raw_data.h5')
+#     occ_hist_oa.reset()
+#     thread = threading.Thread(target=add_data)
+#     thread.start()
+#     time.sleep(0.5)  # wait for first data to be added to queue
+#     occ_hist_oa.get(wait=False)
+#     thread.join()
+#     # Check that warning was given
+#     assert('Getting histogram while analyzing data' in ana_log_messages['warning'])
+
+
+if __name__ == '__main__':
+    import pytest
+    pytest.main(['-s', __file__])

pymosa/tests/utils.py

@@ -0,0 +1,31 @@
+import logging
+
+
+class MockLoggingHandler(logging.Handler):
+    """Mock logging handler to check for expected logs.
+
+    Messages are available from an instance's ``messages`` dict, in order, indexed by
+    a lowercase log level string (e.g., 'debug', 'info', etc.).
+
+    https://stackoverflow.com/questions/899067/how-should-i-verify-a-log-message-when-testing-python-code-under-nose
+    """
+
+    def __init__(self, *args, **kwargs):
+        self.messages = {'debug': [], 'info': [], 'notice': [], 'success': [], 'warning': [], 'error': [],
+                         'critical': []}
+        super(MockLoggingHandler, self).__init__(*args, **kwargs)
+
+    def emit(self, record):
+        "Store a message from ``record`` in the instance's ``messages`` dict."
+        try:
+            self.messages[record.levelname.lower()].append(record.getMessage())
+        except Exception:
+            self.handleError(record)
+
+    def reset(self):
+        self.acquire()
+        try:
+            for message_list in self.messages.values():
+                message_list.clear()
+        finally:
+            self.release()