Move mapping mode to its own module and add tests

handel/interface.py

@@ -9,7 +9,8 @@ import numpy
 from .error import check_error
 from ._cffi import handel, ffi
 from .stats import stats_from_normal_mode
-from .parser import parse_xia_ini_file, parse_mapping_buffer
+from .parser import parse_xia_ini_file
+from .mapping import parse_mapping_buffer
 
 __all__ = ['init', 'init_handel', 'exit',
            'new_detector', 'get_num_detectors', 'get_detectors',

handel/mapping.py

+"""Helpers specific to the mapping mode."""
+
+from .stats import stats_from_mapping_mode
+
+
+def dword(array, index, bitsize=16):
+    """Extract a dword value from two words starting at the given index."""
+    return array[index] | array[index+1] << bitsize
+
+
+def parse_mapping_buffer(raw):
+    """Parse the given mapping buffer and return a (spectrums, statistics) tuple.
+
+    Both results are dictionaries of dictionaries, first indexed by pixel,
+    then channels:
+    - spectrums: <pixel: <channel: spectrum (1D array)>
+    - statistics: <pixel: <channel: stats (Stats object)>
+    """
+    spectrums, statistics = {}, {}
+
+    # Header advance
+    header = raw[0:256]
+    current = 256
+
+    # Header parsing
+    assert header[0] == 0x55aa
+    assert header[1] == 0xaa55
+    assert header[2] == 0x100
+    mapping_mode = header[3]
+    buffer_index = dword(header, 4)
+    buffer_id = header[7]
+    pixel_number = header[8]
+    starting_pixel = dword(header, 9)
+    module_serial_number = header[11]
+    channel_ids = header[12:20:2]
+    channel_detector_ids = header[13:20:2]
+
+    # Checks
+    assert mapping_mode == 1  # MCA mapping mode
+    assert starting_pixel == dword(raw, 256+4)
+    assert buffer_id in (0, 1)
+
+    # Unused information, should we do something with it?
+    buffer_index, module_serial_number, channel_detector_ids
+
+    # Iterate over pixels
+    for _ in range(pixel_number):
+
+        # Pixel header advance
+        pixel_header = raw[current:current+256]
+        current += 256
+
+        # Pixel header parsing
+        assert pixel_header[0] == 0x33cc
+        assert pixel_header[1] == 0xcc33
+        assert pixel_header[2] == 0x100
+        assert pixel_header[3] == mapping_mode
+        pixel = dword(pixel_header, 4)
+        total_size = dword(pixel_header, 6)
+        sizes = pixel_header[8:12]
+
+        # Statistics block
+        stats_block = [
+            [dword(pixel_header, 32 + 8 * i + 2 * j) for j in range(4)]
+            for i in range(4)]
+
+        # Iterate over channels
+        spectrums[pixel] = {}
+        statistics[pixel] = {}
+        remaining = total_size - 256
+        for index, channel_id, size in zip(range(4), channel_ids, sizes):
+
+            # Update remaining size
+            assert remaining >= 0
+            if remaining == 0:
+                break
+            remaining -= size
+
+            # Sectrum Advance
+            spectrum = raw[current:current+size]
+            current += size
+
+            # Set data
+            stats = stats_from_mapping_mode(stats_block[index])
+            spectrums[pixel][channel_id] = spectrum
+            statistics[pixel][channel_id] = stats
+
+        # Checks
+        assert remaining == 0
+
+    # Return results
+    return spectrums, statistics

handel/parser.py

-"""Parsers for the specific XIA INI file format and the mapping mode buffer."""
+"""Parsers for the specific XIA INI file format."""
 
 from collections import OrderedDict
 
-from .stats import stats_from_mapping_mode
-
 
 def parse_xia_ini_file(content):
+    """Parse the content of a XIA INI file.
+
+    The return result is an OrderedDict of <section name: list>,
+    where the list items are OrderedDict of <key: value>.
+    """
     dct = OrderedDict()
     section, item = None, None
     # Loop over striped lines
@@ -59,84 +62,3 @@ def parse_xia_ini_file(content):
             raise ValueError('Line not recognized: {!r}'.format(line))
     # Return result
     return dct
-
-
-def dword(array, index, bitsize=16):
-    return array[index] | array[index+1] << bitsize
-
-
-def parse_mapping_buffer(raw):
-    spectrums, statistics = {}, {}
-
-    # Header advance
-    header = raw[0:256]
-    current = 256
-
-    # Header parsing
-    assert header[0] == 0x55aa
-    assert header[1] == 0xaa55
-    assert header[2] == 0x100
-    mapping_mode = header[3]
-    buffer_index = dword(header, 4)
-    buffer_id = header[7]
-    pixel_number = header[8]
-    starting_pixel = dword(header, 9)
-    module_serial_number = header[11]
-    channel_ids = header[12:20:2]
-    channel_detector_ids = header[13:20:2]
-
-    # Checks
-    assert mapping_mode == 1  # MCA mapping mode
-    assert starting_pixel == dword(raw, 256+4)
-    assert buffer_id in (0, 1)
-
-    # Unused information, should we do something with it?
-    buffer_index, module_serial_number, channel_detector_ids
-
-    # Iterate over pixels
-    for _ in range(pixel_number):
-
-        # Pixel header advance
-        pixel_header = raw[current:current+256]
-        current += 256
-
-        # Pixel header parsing
-        assert pixel_header[0] == 0x33cc
-        assert pixel_header[1] == 0xcc33
-        assert pixel_header[2] == 0x100
-        assert pixel_header[3] == mapping_mode
-        pixel = dword(pixel_header, 4)
-        total_size = dword(pixel_header, 6)
-        sizes = pixel_header[8:12]
-
-        # Statistics block
-        stats_block = [
-            [dword(pixel_header, 32 + 8 * i + 2 * j) for j in range(4)]
-            for i in range(4)]
-
-        # Iterate over channels
-        spectrums[pixel] = {}
-        statistics[pixel] = {}
-        remaining = total_size - 256
-        for index, channel_id, size in zip(range(4), channel_ids, sizes):
-
-            # Update remaining size
-            assert remaining >= 0
-            if remaining == 0:
-                break
-            remaining -= size
-
-            # Sectrum Advance
-            spectrum = raw[current:current+size]
-            current += size
-
-            # Set data
-            stats = stats_from_mapping_mode(stats_block[index])
-            spectrums[pixel][channel_id] = spectrum
-            statistics[pixel][channel_id] = stats
-
-        # Checks
-        assert remaining == 0
-
-    # Return results
-    return spectrums, statistics

tests/test_mapping.py

+"""Test module for mapping mode helpers."""
+
+from handel.stats import Stats, stats_from_mapping_mode
+from handel.mapping import parse_mapping_buffer
+
+
+def test_stats_from_mapping_mode():
+    expected = Stats(
+        realtime=0.032,
+        livetime=0.016,
+        triggers=100,
+        events=50,
+        icr=6250.0,
+        ocr=1562.5,
+        deadtime=0.75)
+    assert stats_from_mapping_mode([10e4, 5e4, 100, 50]) == expected
+
+
+def test_parse_mapping_buffer():
+    raw = [0]*0x300
+    # Header
+    raw[0] = 0x55aa  # Token
+    raw[1] = 0xaa55  # Token
+    raw[2] = 0x100   # Header size
+    raw[3] = 0x1     # Mapping mode
+    raw[8] = 0x1     # Number of pixel
+    raw[9] = 0x1     # Starting pixel
+    raw[12] = 0x2    # First channel ID
+    # Pixel header
+    raw[0x100+0] = 0x33cc  # Token
+    raw[0x100+1] = 0xcc33  # Token
+    raw[0x100+2] = 0x100   # Pixel header size
+    raw[0x100+3] = 0x1     # Mapping mode
+    raw[0x100+4] = 0x1     # Pixel id
+    raw[0x100+6] = 0x200   # Total spectrum size
+    raw[0x100+8] = 0x100   # First channel spectrum size
+    # Statistics
+    raw[0x120+0:0x120+8:2] = [1, 2, 3, 4]
+    # Spectrum
+    raw[0x200:0x300] = range(256)
+    # Test
+    spectrums, stats = parse_mapping_buffer(raw)
+    assert spectrums == {1: {2: list(range(256))}}
+    assert stats == {1: {2: stats_from_mapping_mode([1, 2, 3, 4])}}