Commit ef665316 authored by Vincent Michel's avatar Vincent Michel

Move the stats logic in a separate module

parent b077bd8e
Pipeline #1047 passed with stages
in 1 minute and 18 seconds
......@@ -3,13 +3,12 @@
from __future__ import absolute_import
import os
from warnings import warn
from collections import namedtuple
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
__all__ = ['init', 'init_handel', 'exit',
......@@ -40,46 +39,6 @@ MAX_STRING_LENGTH = 80
# Helpers
Stats = namedtuple(
'Stats',
'realtime livetime triggers events icr ocr deadtime '
'underflows overflows')
def stats_from_buffer(array):
# Raw statistics
realtime = float(array[0])
livetime = float(array[1])
triggers = int(array[3])
events = int(array[4])
icr = float(array[5])
ocr = float(array[6])
underflows = int(array[7])
overflows = int(array[8])
# Double check the ICR computation
expected_icr = triggers / livetime if livetime != 0 else 0.
if expected_icr != icr:
msg = 'ICR buffer inconsistency: {} != {} (expected)'
warn(msg.format(icr, expected_icr))
# Double check the OCR computation
total = events + underflows + overflows
expected_ocr = total / realtime if realtime != 0 else 0.
if expected_ocr != ocr:
msg = 'OCR buffer inconsistency: {} != {} (expected)'
warn(msg.format(ocr, expected_ocr))
# Deadtime computation
# It's unclear whether icr=ocr=0 should result in a 0.0 or 1.0 deadtime
# Prospect uses 0% so 0. it is.
deadtime = 1 - float(ocr) / icr if icr != 0 else 0.
return Stats(
realtime, livetime, triggers, events, icr, ocr, deadtime,
underflows, overflows)
def to_bytes(arg):
if isinstance(arg, bytes):
return arg
......@@ -216,7 +175,7 @@ def get_module_statistics(module):
code = handel.xiaGetRunData(master, b'module_statistics_2', data)
check_error(code)
# Parse raw data
return {channel: stats_from_buffer(array[index * 9:])
return {channel: stats_from_normal_mode(array[index * 9:])
for index, channel in enumerate(channels)
if channel != -1}
......
"""Parser for the specific XIA INI file format."""
"""Parsers for the specific XIA INI file format and the mapping mode buffer."""
from collections import OrderedDict
from .stats import stats_from_mapping_mode
def parse_xia_ini_file(content):
dct = OrderedDict()
......@@ -129,8 +131,9 @@ def parse_mapping_buffer(raw):
current += size
# Set data
stats = stats_from_mapping_mode(stats_block[index])
spectrums[pixel][channel_id] = spectrum
statistics[pixel][channel_id] = tuple(stats_block[index])
statistics[pixel][channel_id] = stats
# Checks
assert remaining == 0
......
"""Statistics handling."""
from warnings import warn
from collections import namedtuple
CLOCK_TICK = 320e-9
Stats = namedtuple(
'Stats',
'realtime livetime triggers events icr ocr deadtime')
def stats_from_normal_mode(array):
realtime = float(array[0])
livetime = float(array[1])
triggers = int(array[3])
events = int(array[4])
icr = float(array[5])
ocr = float(array[6])
underflows = int(array[7])
overflows = int(array[8])
total_events = events + underflows + overflows
return make_stats(realtime, livetime, triggers, total_events, icr, ocr)
def stats_from_mapping_mode(array):
realtime = array[0] * CLOCK_TICK
livetime = array[1] * CLOCK_TICK
triggers = int(array[2])
events = int(array[3])
return make_stats(realtime, livetime, triggers, events)
def make_stats(realtime, livetime, triggers, events, icr=None, ocr=None):
# Compute ICR
expected_icr = triggers / livetime if livetime != 0 else 0.
if icr is None:
icr = expected_icr
# Double check the ICR computation
elif expected_icr != icr:
msg = 'ICR buffer inconsistency: {} != {} (expected)'
warn(msg.format(icr, expected_icr))
# Compute OCR
expected_ocr = events / realtime if realtime != 0 else 0.
if ocr is None:
ocr = expected_ocr
# Double check the OCR computation
if expected_ocr != ocr:
msg = 'OCR buffer inconsistency: {} != {} (expected)'
warn(msg.format(ocr, expected_ocr))
# Deadtime computation
# It's unclear whether icr=ocr=0 should result in a 0.0 or 1.0 deadtime
# Prospect uses 0% so 0. it is.
deadtime = 1 - float(ocr) / icr if icr != 0 else 0.
return Stats(
realtime, livetime, triggers, events, icr, ocr, deadtime)
......@@ -2,6 +2,8 @@ import pytest
import numpy
import mock
from handel.stats import Stats
@pytest.fixture
def interface():
......@@ -289,16 +291,14 @@ def test_get_module_statistics(interface):
1.,
2.]
expected = {8: interface.Stats(
expected = {8: Stats(
1.00758784,
0.98603936,
3088,
2742,
2745,
3131.720827046904,
2724.3282332585513,
0.1300858589533055,
1,
2)}
0.1300858589533055)}
m.side_effect = side_effect
with mock.patch('handel.interface.get_module_channels') as m2:
......
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