<commit_before># Filename: daq.py
# pylint: disable=R0903
"""
Pumps for the DAQ data formats.

"""

from collections import namedtuple
from io import BytesIO
import json
import math
import struct
from struct import unpack
import time
import pprint
from urllib.request import urlopen, URLError

import numpy as np

from km3pipe.core import Pump, Module, Blob
from km3pipe.dataclasses import Table
from km3pipe.sys import ignored
from km3pipe.logger import get_logger, get_printer

__author__ = "<NAME>"
__copyright__ = "Copyright 2016, <NAME> and the KM3NeT collaboration."
__credits__ = []
__license__ = "MIT"
__maintainer__ = "<NAME>"
__email__ = "<EMAIL>"
__status__ = "Development"

log = get_logger(__name__)    # pylint: disable=C0103

DATA_TYPES = {
    101: 'DAQSuperFrame',
    201: 'DAQSummaryFrame',
    1001: 'DAQTimeslice',
    2001: 'DAQSummaryslice',
    10001: 'DAQEvent',
}
MINIMAL_RATE_HZ = 2.0e3
MAXIMAL_RATE_HZ = 2.0e6


class TimesliceParser(Module):
    """Preliminary parser for DAQTimeslice"""
    def _get_raw_data(self, blob):
        if 'CHPrefix' in blob:
            if not str(blob['CHPrefix'].tag).startswith('IO_TS'):
                log.info("Not an IO_TS* blob")
                return blob
            return BytesIO(blob['CHData'])
        if 'FileIO' in blob:
            return blob['FileIO']

    def process(self, blob):
        data = self._get_raw_data(blob)
        if data is None:
            return blob
        try:
            tsl_size, datatype = unpack('<ii', data.read(8))
            det_id, run, sqnr = unpack('<iii', data.read(12))
            timestamp, ns_ticks, n_frames = unpack('<iii', data.read(12))

            ts_info = Table.from_template({
                'frame_index': sqnr,
                'slice_id': 0,
                'timestamp': timestamp,
                'nanoseconds': ns_ticks * 16,
                'n_frames': n_frames
            }, 'TimesliceInfo')
            ts_frameinfos = {}

            _dom_ids = []
            _channel_ids = []
            _times = []
            _tots = []
            total_hits = 0
            for _ in range(n_frames):
                frame_size, datatype = unpack('<ii', data.read(8))
                det_id, run, sqnr = unpack('<iii', data.read(12))
                timestamp, ns_ticks, dom_id = unpack('<iii', data.read(12))
                dataqueue_status = unpack('<i', data.read(4))[0]
                dom_status = unpack('<iiii', data.read(4 * 4))
                n_hits = unpack('<i', data.read(4))[0]
                total_hits += n_hits
                ts_frameinfos[dom_id] = Table.from_template({
                    'det_id': det_id,
                    'run_id': run,
                    'sqnr': sqnr,
                    'timestamp': timestamp,
                    'nanoseconds': ns_ticks * 16,
                    'dom_id': dom_id,
                    'dataqueue_status': dataqueue_status,
                    'dom_status': dom_status,
                    'n_hits': n_hits,
                }, 'TimesliceFrameInfo')
                for j in range(n_hits):
                    hit = unpack('!BlB', data.read(6))
                    _dom_ids.append(dom_id)
                    _channel_ids.append(hit[0])
                    _times.append(hit[1])
                    _tots.append(hit[2])

            if total_hits > 0:
                tshits = Table(
                    {
                        'channel_id': np.array(_channel_ids),
                        'dom_id': np.array(_dom_ids),
                        'time': np.array(_times),
                        'tot': np.array(_tots),
                    },
                    'TimesliceHits', h5loc='/timeslice_hits', split_h5=True
                )
                blob['TSHits'] = tshits
            blob['TimesliceInfo'] = ts_info
            blob['TimesliceFrameInfos'] = ts_frameinfos
        except struct.error:
            log.error("Could not parse Timeslice")
            log.error(blob.keys())
        else:
            return blob


class DAQPump(Pump):
    """A pump for binary DAQ files."""
    def configure(self):
        self.filename = self.require('filename')<commit_msg>Always write hits, even if 0 length<commit_after># Filename: daq.py
# pylint: disable=R0903
"""
Pumps for the DAQ data formats.

"""

from collections import namedtuple
from io import BytesIO
import json
import math
import struct
from struct import unpack
import time
import pprint
from urllib.request import urlopen, URLError

import numpy as np

from km3pipe.core import Pump, Module, Blob
from km3pipe.dataclasses import Table
from km3pipe.sys import ignored
from km3pipe.logger import get_logger, get_printer

__author__ = "<NAME>"
__copyright__ = "Copyright 2016, <NAME> and the KM3NeT collaboration."
__credits__ = []
__license__ = "MIT"
__maintainer__ = "<NAME>"
__email__ = "<EMAIL>"
__status__ = "Development"

log = get_logger(__name__)    # pylint: disable=C0103

DATA_TYPES = {
    101: 'DAQSuperFrame',
    201: 'DAQSummaryFrame',
    1001: 'DAQTimeslice',
    2001: 'DAQSummaryslice',
    10001: 'DAQEvent',
}
MINIMAL_RATE_HZ = 2.0e3
MAXIMAL_RATE_HZ = 2.0e6


class TimesliceParser(Module):
    """Preliminary parser for DAQTimeslice"""
    def _get_raw_data(self, blob):
        if 'CHPrefix' in blob:
            if not str(blob['CHPrefix'].tag).startswith('IO_TS'):
                log.info("Not an IO_TS* blob")
                return blob
            return BytesIO(blob['CHData'])
        if 'FileIO' in blob:
            return blob['FileIO']

    def process(self, blob):
        data = self._get_raw_data(blob)
        if data is None:
            return blob
        try:
            tsl_size, datatype = unpack('<ii', data.read(8))
            det_id, run, sqnr = unpack('<iii', data.read(12))
            timestamp, ns_ticks, n_frames = unpack('<iii', data.read(12))

            ts_info = Table.from_template({
                'frame_index': sqnr,
                'slice_id': 0,
                'timestamp': timestamp,
                'nanoseconds': ns_ticks * 16,
                'n_frames': n_frames
            }, 'TimesliceInfo')
            ts_frameinfos = {}

            _dom_ids = []
            _channel_ids = []
            _times = []
            _tots = []
            for _ in range(n_frames):
                frame_size, datatype = unpack('<ii', data.read(8))
                det_id, run, sqnr = unpack('<iii', data.read(12))
                timestamp, ns_ticks, dom_id = unpack('<iii', data.read(12))
                dataqueue_status = unpack('<i', data.read(4))[0]
                dom_status = unpack('<iiii', data.read(4 * 4))
                n_hits = unpack('<i', data.read(4))[0]
                ts_frameinfos[dom_id] = Table.from_template({
                    'det_id': det_id,
                    'run_id': run,
                    'sqnr': sqnr,
                    'timestamp': timestamp,
                    'nanoseconds': ns_ticks * 16,
                    'dom_id': dom_id,
                    'dataqueue_status': dataqueue_status,
                    'dom_status': dom_status,
                    'n_hits': n_hits,
                }, 'TimesliceFrameInfo')
                for j in range(n_hits):
                    hit = unpack('!BlB', data.read(6))
                    _dom_ids.append(dom_id)
                    _channel_ids.append(hit[0])
                    _times.append(hit[1])
                    _tots.append(hit[2])

            tshits = Table(
                {
                    'channel_id': np.array(_channel_ids),
                    'dom_id': np.array(_dom_ids),
                    'time': np.array(_times),
                    'tot': np.array(_tots),
                },
                'TimesliceHits', h5loc='/timeslice_hits', split_h5=True
            )
            blob['TSHits'] = tshits
            blob['TimesliceInfo'] = ts_info
            blob['TimesliceFrameInfos'] = ts_frameinfos
        except struct.error:
            log.error("Could not parse Timeslice")
            log.error(blob.keys())
        else:
            return blob


class DAQPump(Pump):
    """A pump for binary DAQ files."""
    def configure(self):
        self.filename = self.require('filename')
{'fraction_non_alphanumeric': 0.11507115922637148, 'fraction_numerical': 0.017151198151076512, 'mean_word_length': 1.7748903138710765, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '63866', 'n_tokens_mistral': 2665, 'n_tokens_neox': 2464, 'n_words': 562}