Commit 45679f35 authored by Vincent Michel's avatar Vincent Michel

Fix spectrum parsing for falconX

parent 7dbc1e7f
Pipeline #1337 passed with stages
in 1 minute and 22 seconds
......@@ -255,10 +255,7 @@ def get_raw_buffer(master, buffer_id):
raise RuntimeError(
"The buffer {} associated with channel {} is empty"
.format(str(buffer_id), master))
# XMAP/Mercury
if array[1] == 0:
return array[::2]
# FalconX
# Return array
return array
......
......@@ -18,6 +18,14 @@ def parse_mapping_buffer(raw):
"""
spectrums, statistics = {}, {}
# XMAP/Mercury parsing
if raw[1] == 0:
raw = raw[::2]
spectrum_type = 'uint16'
# FalconX parsing
else:
spectrum_type = 'uint32'
# Header advance
header = raw[0:256]
current = 256
......@@ -78,6 +86,7 @@ def parse_mapping_buffer(raw):
# Sectrum Advance
spectrum = raw[current:current+size]
spectrum.dtype = spectrum_type
current += size
# Set data
......
import pytest
import numpy
import mock
import numpy
import pytest
from handel.stats import Stats
from handel.error import HandelError
......@@ -417,19 +417,7 @@ def test_get_raw_buffer(interface):
buff = range(1, 6)
m.side_effect = side_effect
expected = numpy.array(range(1, 6), dtype='uint32')
diff = interface.get_raw_buffer(1, 'a') == expected
assert diff.all()
m.assert_called()
arg = m.call_args[0][2]
m.assert_called_with(1, b'buffer_a', arg)
# Make sure errors have been checked
interface.check_error.assert_called_with(0)
# Second test
buff = [2*x + ((2*x+1) << 16) for x in range(1, 6)]
expected = numpy.array(range(2, 12), dtype='uint32')
diff = interface.get_raw_buffer(1, 'a') == expected
assert diff.all()
assert numpy.array_equal(interface.get_raw_buffer(1, 'a')[::2], expected)
m.assert_called()
arg = m.call_args[0][2]
m.assert_called_with(1, b'buffer_a', arg)
......
"""Test module for mapping mode helpers."""
import numpy
from handel.stats import Stats, stats_from_mapping_mode
from handel.mapping import parse_mapping_buffer
......@@ -16,8 +18,9 @@ def test_stats_from_mapping_mode():
assert stats_from_mapping_mode([10e4, 5e4, 100, 50]) == expected
def test_parse_mapping_buffer():
def test_parse_xmap_mapping_buffer():
raw = [0]*0x300
raw = numpy.array(raw, dtype='uint32')
# Header
raw[0] = 0x55aa # Token
raw[1] = 0xaa55 # Token
......@@ -38,7 +41,42 @@ def test_parse_mapping_buffer():
raw[0x120+0:0x120+8:2] = [1, 2, 3, 4]
# Spectrum
raw[0x200:0x300] = range(256)
# Convert
raw.dtype = 'uint16'
# Test
spectrums, stats = parse_mapping_buffer(raw)
assert numpy.array_equal(spectrums[1][2], list(range(256)))
assert stats == {1: {2: stats_from_mapping_mode([1, 2, 3, 4])}}
def test_parse_falconx_mapping_buffer():
raw = [0]*0x300
raw = numpy.array(raw, dtype='uint16')
# 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
s = numpy.array(range(128), dtype='uint32')
s.dtype = 'uint16'
raw[0x200:0x300] = s
# Convert
raw.dtype = 'uint16'
# Test
spectrums, stats = parse_mapping_buffer(raw)
assert spectrums == {1: {2: list(range(256))}}
assert numpy.array_equal(spectrums[1][2], list(range(128)))
assert stats == {1: {2: stats_from_mapping_mode([1, 2, 3, 4])}}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment