Overview

The Buffer class can be used to open measurement data files that are generated by the Analyzer4D software. It can be used to access the Metadata of those files like process number, channel etc. and the actual measurement data. It can be used together with the BufferMetadataCache to organize large amounts of buffer files.

Example

from qass.tools.analyzer.buffer_parser import Buffer

buffer_file = "path/to/my/buffer_file"
with Buffer(buffer_file) as buff:
    print(buff.process)

qass.tools.analyzer.buffer_parser

Classes:

Name	Description
Buffer
HeaderDtype

Functions:

Name	Description
filter_buffers	This function takes a directory with buffer files and a dictionary with

Attributes:

Name	Type	Description
DataConversion

DataConversion module-attribute

DataConversion = Literal['log', 'delog']

Buffer

Classes:

Name	Description
ADCTYPE
DATAKIND
DATAMODE
DATATYPE

Methods:

Name	Description
__enter__
__exit__
__getstate__	Return path to buffer file, since this is everything needed to reconstruct a buffer object.
__init__
__setstate__	Reconstruct buffer from state, which only contains the path to the buffer file.
block_infos	block_infos iterates through all memory blocks of a buffer file (typically one MB) and fetches the subdata information
db_header	The data block header contains key words and their corresponding
db_header_spec	The method returns the data block header as a dictionary of the data
db_value	Similar to the file header each data block has its own header. This
delog	Method to de-logarithmize a data array. Usually when data are measured
file_size	The file size of the buffer file. The value is not stored in the header
getArray	Wrapper function to 'get_data'.
getRealSpecCount	Wrapper function for the property spec_count.
getSpecDuration	Wrapper Function for property spec_duration.
get_data	This function provides access to the measurement data in the buffer
io_ports	Deprecated method!
io_ports_spec	Deprecated method!
log	Method to logarithmize a data array. Usually when data are measured
spec_to_time_ns	The method calculates the time since measurement start for a given
time_to_spec	The method calculates the index of the nearest spectrum to the time

Attributes:

Name	Type	Description
adc_type		The type of analog/digital converter. The types are defined in class
analyzer_version	Union[str, None]	The analyzer4D version as a dot separated string consisting of major.minor.micro.patch
avg_frq		Property which returns the moving average factor along the
avg_time		Property which returns the moving average factor along the
bit_resolution		The bit resolution after the FFT. Caution: this is not the type of
bytes_per_sample		Number of bytes per sample.
channel		The channel number used.
compression_frq		Property which returns the frequency compression factor.
compression_time		Property which returns the time compression factor.
datakind		The data kind constant is a constant which provides additional
datamode		The data mode is a constant which specifies what kind of data are in
datatype		The data type constant is a constant which specifies different
db_count		Returns the number of data blocks.
db_header_size		Each data block has a header. This property provides the size of the
db_sample_count		Returns the number of samples in a completely filled data block.
db_size		Returns the size of a completely filled data block in bytes.
db_spec_count		The number of spectra within a filled data block. It is calculated by
fft_log_shift		Base of the logarithm applied to the data. Please note that a 1 needs
filepath		Returns the complete path to the file.
frq_bands	int	Each spectrum has a maximum sample number of 512. This is the maximum
frq_end		The property returns the highest represented frequency.
frq_per_band		The property returns the frequency range per band. The maximum number
frq_start		The property returns the lowest represented frequency.
full_blocks		All but the last data block are usually completly filled. In order to
header_hash		A hash value for the buffer's header.
header_size		Each buffer file has a header. This property provides the size of the
last_modification_date_time
last_modification_time		Returns the Posix timestamp of last modification of the file
last_spec
max_amplitude_level		maximum possible amplitude of the buffer, relates to the buffer's dtype
metainfo		The buffer header properties as a dictionary.
partnumber	str	The string representation of the partnumber of the measurement.
preamp_gain		The preamplifier setting in the multiplexer tab of the Analyzer4D software.
process		The process number.
process_date_time		Returns a timestamp of the creation of the file as a string.
process_measure_time_string		Returns the time in ISO format at measure start of the process. If
process_measure_timestamp		Returns the Posix timestamp at measure start of the process in
process_time		Returns the Posix timestamp of creation of the file
project_id
refEnergy		refEnergy provides a normalization factor for compressions.
ref_energy		alias for refEnergy()
sample_count		The number of samples in the buffer. It is calculated by subtracting
spec_count		The number of spectra in the buffer. It is calculated by dividing
spec_duration		The property returns the time for one spectrum in
streamno	Union[int, None]	The index of a stream of an extra channel.

Source code in src/qass/tools/analyzer/buffer_parser.py

class Buffer:
    class DATAMODE(IntEnum):
        DATAMODE_UNDEF = -1
        # Es wird nur ein Zähler übertragen, der im DSP Modul generiert wird
        DATAMODE_COUNTER_UNUSED = 0
        DATAMODE_SIGNAL = (
            auto()
        )  # Es werden die reinen Signaldaten gemessen und übertragen
        # Die FFT wird in der Hardware durchgeführt und das Ergebnis als FFT Daten übertragen
        DATAMODE_FFT = auto()
        DATAMODE_PLOT = auto()  # 2 dimensionale Graph Daten (war INTERLEAVED, das wird nicht mehr genutzt, taucht aber im kernelmodul als Define für DATAMODES noch auf
        # Datenmodus, der nur in importierten oder künstlichen Buffern auftritt
        DATAMODE_OTHER = auto()
        # Datamode is video (This means file is not a normal buffer, but simply a video file)
        DATAMODE_VIDEO = auto()

        DATAMODE_COUNT = auto()

    class DATATYPE(IntEnum):  # Kompressionsmethoden oder auch Buffertypen
        COMP_INVALID = -2
        COMP_UNDEF = -1
        COMP_RAW = (
            0  # Die reinen unkomprimierten Rohdaten, sowie sie aus der Hardware kommen
        )
        # Datenreduktion durch einfaches Downsampling (jedes x-te Sample gelangt in den Buffer)
        COMP_DOWNSAMPLE = auto()
        COMP_MAXIMUM = (
            auto()
        )  # Die Maximalwerte von jeweils x Samples gelangen in den Buffer
        # Die Durchschnittswerte über jeweils x Samples gelangen in den Buffer
        COMP_AVERAGE = auto()
        COMP_STD_DEVIATION = auto()  # Die Standardabweichung
        COMP_ENVELOP = auto()  # NOT USED!! Der Buffer stellt eine Hüllkurve dar
        COMP_MOV_AVERAGE = (
            auto()
        )  # Der gleitende Mittelwert über eine Blockgröße von x Samples
        # Die Bufferdaten wurden aus einer externen Quelle, die nicht mit dem Optimizer aufgezeichnet wurden erstellt
        COMP_EXTERN_DATA = auto()
        # Zur Zeit verwendet, um MinMaxObjekt Energy signature buffer zu taggen
        COMP_ANALYZE_OVERVIEW = auto()
        # Der gleitende Mittelwert über eine Blockgröße von x Samples (optimierte Berechnung)
        COMP_MOV_AVERAGE_OPT = auto()
        COMP_COLLECTION = auto()  # Wild zusammengeworfene Datenmasse
        COMP_IMPORT_SIG = auto()  # Importierte Signaldaten
        COMP_SCOPE_RAW = auto()  # Vom Oszilloskop aufgezeichnet
        # Der gleitende Mittelwert (Zeit und Frequenz) über eine Blockgröße von x Samples
        COMP_MOV_AVERAGE_FRQ = auto()
        COMP_SLOPECHANGE = auto()  # Steigungswechsel der Amplituden über die Zeit
        COMP_OTHER = auto()
        COMP_IO_SIGNAL = auto()  # Aufgezeichnete IO Signale
        COMP_ENERGY = auto()  # Die Energie (in erster Linie für DM=ANALOG)
        COMP_AUDIO_RAW = (
            auto()
        )  # Raw Daten, die mit dem Sound Device aufgezeichnet wurden
        # was COMP_OBJECT before. Now it is used for Datamode PLOT. Type will be displayed as Graphname, if set
        COMP_XY_PLOT = auto()
        COMP_SECOND_FFT = auto()
        # Raw Daten vom Audio Device, bei denen es sich um einen Audiokommentar handelt
        COMP_AUDIO_COMMENT = auto()
        # CrackProperties Energy Signatur, kommt in erster Linie von Energieprofilen, die auf Daten der CrackDefinitionen Berechnet wurden
        COMP_CP_ENERGY_SIG = auto()
        # This is a video stream that has been recorded while measuring
        COMP_VID_MEASURE = auto()
        # This is a screen cast video stream, that has been captured while measuring or session recording
        COMP_VID_SCREEN = auto()
        COMP_VID_EXT_LINK = auto()  # This is an extern linked video file
        COMP_ENVELOPE_UPPER = auto()  # NOT_USED!! Obere Hüllfläche
        COMP_ENVELOPE_LOWER = auto()  # NOT_USED! Untere Hüllfläche
        COMP_PATTERN_REFOBJ = auto()  # NOT USED! A Pattern Recognition reference object
        COMP_SIGNIFICANCE = auto()  # Nur starke Änderungen werden aufgezeichnet
        # this is the signed 32 bit version of a significance buffer
        COMP_SIGNIFICANCE_32 = auto()
        # NOT_USED! A mask buffer for a pattern ref object (likely this is a float buffer)
        COMP_PATTERN_MASK = auto()
        COMP_GRADIENT_FRQ = auto()  # Gradient buffer in frequency direction
        # Not realy a datatype but used to create CSV files from source buffer
        COMP_CSV_EXPORT = auto()
        # Die Anzahl der Einträge in der Datatypesliste, kein wirklicher Datentyp
        COMP_COUNT = auto()

    class DATAKIND(IntEnum):  # Zusätzliche Spezifikation des Buffers
        KIND_UNDEF = -1
        KIND_NONE = 0
        KIND_SENSOR_TEST = auto()  # Sensor Pulse Test Daten
        # Debug Plot Buffer, der die Zeiten für das "freimachen" eines Datenblocks enthält
        KIND_PLOT_FREE_DATABLOCK_TIMING = auto()
        KIND_PATTERN_REF_OBJ_COMPR = auto()
        KIND_PATTERN_REF_OBJ_MASK = auto()
        KIND_PATTERN_REF_OBJ_EXTRA = auto()
        KIND_FREE_6 = auto()
        KIND_FREE_7 = auto()
        DATAKIND_CNT = auto()
        # This datakind is out ouf range. It cannot be stored in bufferId anymore (this is legacy stuff)
        KIND_CAN_NOT_BE_HANDLED = DATAKIND_CNT

        # Werte ab hier zur freien Verwendung (Oft ist das ein Zähler für verschiedene Buffer, die ansonsten den selben FingerPrint hätten)
        KIND_USER = 100

    class ADCTYPE(IntEnum):
        ADC_NOT_USED = 0
        ADC_LEGACY_14BIT = 0
        ADC_16BIT = auto()
        ADC_24BIT = auto()

    def __init__(self, filepath):
        self.__filepath = filepath
        # uint: read with np.uintc and then cast to python int
        # c double is python float
        import numpy as np

        int(np.uintc())
        self.__header_hash = None
        self.__keywords = [
            ("qassdata----", HeaderDtype.INT32),
            ("filevers----", HeaderDtype.INT32),
            ("datavers----", HeaderDtype.INT32),
            ("savefrom----", HeaderDtype.INT32),
            ("datamode----", HeaderDtype.INT32),
            ("datatype----", HeaderDtype.INT32),
            ("datakind----", HeaderDtype.INT32),
            ("framsize----", HeaderDtype.INT32),
            ("smplsize----", HeaderDtype.INT32),
            ("frqbands----", HeaderDtype.INT32),
            ("db_words----", HeaderDtype.INT32),
            ("avgtimba----", HeaderDtype.INT32),
            ("avgfrqba----", HeaderDtype.INT32),
            ("m_u_mask--------", HeaderDtype.UINT64),
            ("b_p_samp----", HeaderDtype.INT32),
            ("s_p_fram----", HeaderDtype.INT32),
            ("db__size----", HeaderDtype.INT32),
            ("max_ampl----", HeaderDtype.INT32),
            ("nul_ampl----", HeaderDtype.INT32),
            ("samplert----", HeaderDtype.INT32),
            ("datarate--------", HeaderDtype.DOUBLE),
            ("samplefr----", HeaderDtype.INT32),
            ("frqshift----", HeaderDtype.INT32),
            ("fftovers----", HeaderDtype.INT32),
            ("fftlogsh----", HeaderDtype.INT32),
            ("fftwinfu----", HeaderDtype.INT32),
            ("dbhdsize----", HeaderDtype.INT32),
            ("comratio----", HeaderDtype.INT32),
            ("tc__real--------", HeaderDtype.DOUBLE),
            ("frqratio----", HeaderDtype.INT32),
            ("fc__real--------", HeaderDtype.DOUBLE),
            ("proj__id--------", HeaderDtype.UINT64),
            ("file__id--------", HeaderDtype.UINT64),
            ("parentid--------", HeaderDtype.INT64),
            ("proc_cnt----", HeaderDtype.INT32),
            ("proc_rng----", HeaderDtype.INT32),
            ("proc_sub----", HeaderDtype.INT32),
            ("poly_cnt----", HeaderDtype.INT32),
            ("polycyid----", HeaderDtype.INT32),
            ("dumpchan----", HeaderDtype.INT32),
            ("del_lock----", HeaderDtype.INT32),
            ("proctime----", HeaderDtype.UINT32),
            ("lmodtime----", HeaderDtype.UINT32),
            ("epoctime--------", HeaderDtype.INT64),
            ("mux_port----", HeaderDtype.INT32),
            ("pampgain----", HeaderDtype.INT32),
            ("dispgain----", HeaderDtype.INT32),
            ("linfgain----", HeaderDtype.INT32),
            ("auxpara0----", HeaderDtype.INT32),
            ("auxpara1----", HeaderDtype.INT32),
            ("auxpara2----", HeaderDtype.INT32),
            ("auxpara3----", HeaderDtype.INT32),
            ("auxpara4----", HeaderDtype.INT32),
            ("auxpara5--------", HeaderDtype.INT64),
            ("skipsamp--------", HeaderDtype.INT64),
            ("skiptime--------", HeaderDtype.INT64),
            ("trunsamp--------", HeaderDtype.INT64),
            ("truntime--------", HeaderDtype.INT64),
            ("skiplfrq----", HeaderDtype.INT32),
            ("trunhfrq----", HeaderDtype.INT32),
            ("startfrq----", HeaderDtype.INT32),
            ("end__frq----", HeaderDtype.INT32),
            ("frqpband--------", HeaderDtype.DOUBLE),
            ("framedur--------", HeaderDtype.DOUBLE),
            ("frameoff----", HeaderDtype.INT32),
            ("p__flags----", HeaderDtype.INT32),
            ("realfrqc----", HeaderDtype.INT32),
            ("sub_data----", HeaderDtype.INT32),
            ("sd_dimen----", HeaderDtype.INT32),
            ("sd_rsize----", HeaderDtype.INT32),
            ("sd_rsizf--------", HeaderDtype.DOUBLE),
            ("sd_dsize----", HeaderDtype.INT32),
            ("frqmasks----", HeaderDtype.UINT32),
            ("frqinmas----", HeaderDtype.UINT32),
            ("eheaderf----", HeaderDtype.UINT32),
            ("adc_type----", HeaderDtype.INT32),
            ("adbitres----", HeaderDtype.INT32),
            ("baserate--------", HeaderDtype.INT64),
            ("interpol----", HeaderDtype.INT32),
            ("dcoffset----", HeaderDtype.INT32),
            ("dispralo----", HeaderDtype.INT32),
            ("disprahi----", HeaderDtype.INT32),
            ("rngtibeg--------", HeaderDtype.INT64),
            ("rngtiend--------", HeaderDtype.INT64),
            ("realsoff--------", HeaderDtype.INT64),
            ("extendid--------", HeaderDtype.UINT64),
            ("sim_mode----", HeaderDtype.INT32),
            ("partnoid----", HeaderDtype.UINT32),
            ("asc_part----", HeaderDtype.UINT32),
            ("asc_desc----", HeaderDtype.UINT32),
            ("comments----", HeaderDtype.UINT32),
            ("sparemem----", HeaderDtype.UINT32),
            ("streamno----", HeaderDtype.UINT32),
            ("an4dvers----", HeaderDtype.HEX_STRING),
            ("headsend", None),
        ]

        self.__db_keywords = [
            ("blochead----", HeaderDtype.INT32),
            ("firstsam--------", HeaderDtype.INT64),
            ("lastsamp--------", HeaderDtype.INT64),
            ("dbfilled----", HeaderDtype.INT32),
            ("mux_port----", HeaderDtype.INT32),
            ("pampgain----", HeaderDtype.INT32),
            ("dispgain----", HeaderDtype.INT32),
            ("io_ports----", HeaderDtype.INT32),
            ("dversion----", HeaderDtype.INT32),
            ("sine_frq----", HeaderDtype.INT32),
            ("sine_amp----", HeaderDtype.INT32),
            ("sd_dimen----", HeaderDtype.INT32),
            ("sd_rsize----", HeaderDtype.INT32),
            ("sd_dsize----", HeaderDtype.INT32),
            ("begin_subdat", HeaderDtype.INT32),
            ("end___subdat", HeaderDtype.INT32),
            (
                "blockend",
                HeaderDtype.INT32,
            ),  # datatypes for last three keywords guessed
        ]
        self.__metainfo = {}
        self.__db_headers = {}

    def __enter__(self):
        self.__last_spectrum = None
        self.file = open(self.__filepath, "rb")
        self._parse_header()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()

    def __getstate__(self):
        """
        Return path to buffer file, since this is everything needed to reconstruct a buffer object.
        This function is needed for pickling buffer objects.

        Returns
        -------
        tuple
            Single-element tuple containing the path to the buffer file.
        """
        return (self.__filepath,)

    def __setstate__(self, state: Tuple):
        """
        Reconstruct buffer from state, which only contains the path to the buffer file.
        This function is needed for unpickling buffer objects.

        Parameters
        ----------
        state: tuple
            Single-element tuple containing the path to the buffer file.
        """
        self.__init__(*state)

    def _get_var_chars(self, content, key, val, idx):
        if key in ["asc_desc", "comments", "sparemem", "asc_part"]:
            d_len = val
            val = content[idx : idx + d_len]
            idx += d_len
        elif key == "begin_subdat":
            end_idx = content.index(b"end___subdat", idx)
            val = content[idx:end_idx]
            idx = end_idx

        return key, val, idx

    def _is_keyword_ascii(self, c):
        return (c >= 95 and c <= 122) or (c >= 48 and c <= 57)

    def _is_possible_keyword(self, testkey):
        """
        Test if given bytearray could be a valid keyword

        Parameters
        ----------
        testkey: bytes

        Returns
        -------
        bool
            returns True, if all characters are lower case ASCII or numbers or _
        """
        for c in testkey:
            if not self._is_keyword_ascii(c):
                return False
        return True

    def _get_header_val(self, idx, content, keyword):
        key: str
        key, data_type = keyword
        key = key.rstrip("-")
        data_len = len(keyword[0]) - len(key)
        data_type = keyword[1]
        curr_content = content[idx : idx + len(key)]

        # TODO: fill the if structur with the correct unpack method from struct.unpack
        def __read_from_bytes(data_type: str, bytes) -> Any:
            """
            Method to translate byte data into correct python datatypes, with the help of given data type
            """
            val: Any = None
            if data_type == HeaderDtype.INT32:
                (val,) = unpack("i", bytes)
            elif data_type == HeaderDtype.INT64:
                (val,) = unpack("q", bytes)
            elif data_type == HeaderDtype.UINT32:
                (val,) = unpack("I", bytes)
            elif data_type == HeaderDtype.UINT64:
                (val,) = unpack("Q", bytes)
            elif data_type == HeaderDtype.FLOAT:
                (val,) = unpack("f", bytes)
            elif data_type == HeaderDtype.DOUBLE:
                (val,) = unpack("d", bytes)
            elif data_type == HeaderDtype.HEX_STRING:
                val = codecs.encode(bytes, "hex")
            elif data_type == None:
                pass
            else:
                val = bytes

            return val

        if curr_content.decode() == key:
            idx += len(key)
            val = None
            if data_len != 0:
                val = __read_from_bytes(data_type, content[idx : idx + data_len])
                # val = int.from_bytes(content[idx: idx + data_len], byteorder='little', signed=True)
            idx += data_len

            return self._get_var_chars(content, key, val, idx)

        return None

    def _read_next_value(self, idx, content, keywords):
        for keyword in keywords:
            res = self._get_header_val(idx, content, keyword)
            if res:
                # key, val, next_idx = res
                return res
        # keyword not found
        current_key = content[idx : idx + 8].decode()
        warnings.warn(f'Error: Key word "{current_key}" not supported by buffer_parser')

    def _parse_header(self):
        self.file.seek(0)
        # this should contain the complete header i sugest...
        header_arr = self.file.read(4096)
        magic_bytes = header_arr[:8]
        if magic_bytes != b"qassdata":
            raise InvalidFileError(f"{self.file.name} is not a valid data stream file")
        _, header_size, _ = self._get_header_val(0, header_arr, self.__keywords[0])

        header_arr = header_arr[:header_size]
        m = hashlib.sha256()
        m.update(header_arr)
        self.__header_hash = m.hexdigest()

        idx = 0
        while idx < header_size:
            try:
                key, val, idx = self._read_next_value(idx, header_arr, self.__keywords)
            except:
                idx += 12  # skip unknown parameter value
                testkey = header_arr[idx : idx + 8]
                # check if current position contains a possible keyword, otherwise
                # add 4 to idx, because most likely the unknown keyword has a 64bit value
                if not self._is_possible_keyword(testkey):
                    idx += 4

            if key:
                self.__metainfo[key] = val

        self.file.seek(0, os.SEEK_END)
        size = self.file.tell()
        self.file_size = size

        self.__header_size = header_size
        self.__db_header_size = self.__metainfo["dbhdsize"]
        self.__bytes_per_sample = (
            self.__metainfo["b_p_samp"] if "b_p_samp" in self.__metainfo else 2
        )
        self.__db_size = self.__metainfo["db__size"]

        self.__frq_bands = None
        for k in ("frqbands", "s_p_fram", "framsize"):
            if k in self.__metainfo:
                self.__frq_bands = self.__metainfo[k]

        if self.__frq_bands is None:
            raise ValueError("Failed to determine the frq_bands of the buffer")
        self.__db_count = math.ceil(
            (self.file_size - self.__header_size)
            / (self.__db_size + self.__db_header_size)
        )

        self._calc_spec_duration()
        self._signalNormalizationFactor()

    def _log(self, data_arr):
        return self.log(data_arr, self.fft_log_shift, self.ad_bit_resolution)

    def _delog(self, data_arr):
        return self.delog(data_arr, self.fft_log_shift, self.ad_bit_resolution)

    def _get_data(
        self,
        specFrom,
        specTo,
        frq_bands,
        conversion: Union[DataConversion, None] = None,
    ):
        pos_start = specFrom * self.__frq_bands * self.__bytes_per_sample
        pos_end = specTo * self.__frq_bands * self.__bytes_per_sample

        db_start = int(pos_start / self.__db_size)
        db_end = math.ceil(pos_end / self.__db_size)

        # The following if clauses check whether the datatype is 2 or 4 bytes long. In case of 4 bytes
        # it checks whether bit 3 is set in p__flags because bit 3 indicates a float buffer.

        if self.__bytes_per_sample == 2:
            dtype = np.uint16
        elif self.__bytes_per_sample == 4:
            if (self.__metainfo["p__flags"] & 8) == 0:
                dtype = np.uint32
            else:
                dtype = np.float32
        else:
            raise ValueError("Unknown value for bytes_per_sample")

        np_arrays = []

        for db in range(db_start, db_end):
            if db == db_start:
                start_in_db = pos_start - db_start * self.__db_size
            else:
                start_in_db = 0

            if db == db_end - 1:
                end_in_db = pos_end - (db_end - 1) * self.__db_size
            else:
                end_in_db = self.__db_size

            block_start_pos_in_file = self.__header_size + db * (
                self.__db_size + self.__db_header_size
            )

            start_pos_in_file = (
                block_start_pos_in_file + start_in_db + self.__db_header_size
            )
            read_len = int((end_in_db - start_in_db) / self.__bytes_per_sample)

            if start_pos_in_file + read_len > self.file_size:
                raise ValueError("The given indices exceed the buffer file's size.")

            # TODO: We should check the length of actual data in this datablock - especially for the last data block
            # but maybe also for intermediate data blocks - since we do not know if the measuring has been paused

            self.file.seek(start_pos_in_file, os.SEEK_SET)  # seek
            arr = np.fromfile(self.file, dtype=dtype, count=read_len)
            if conversion:
                if conversion == "delog":
                    if self.fft_log_shift != 0:
                        arr = self.delog(arr, self.fft_log_shift, self.bit_resolution)
                elif conversion == "log":
                    if self.fft_log_shift == 0:
                        arr = self.log(arr, self.fft_log_shift, self.bit_resolution)
                else:
                    raise InvalidArgumentError("The given conversion is unknown")
            np_arrays.append(arr)

        return np.concatenate(np_arrays).reshape(-1, self.__frq_bands)  # & 0x3fff

    def get_data(self, specFrom=None, specTo=None, conversion: Union[str, None] = None):
        """
        This function provides access to the measurement data in the buffer
        file. The data are retrieved for the range of spectra and stored in a
        numpy array. The data can be logarithmized or not. The datatype of the
        array depends on the type of buffer.

        Parameters
        ----------
        specFrom : int, optional
            First spectrum to be retrieved (default first available spectrum)
        specTo : int, optional
            Last spectrum to be retrieved (default last available spectrum)
        conversion: str, optional
            Conversion ('log' or 'delog') of the data.

        Raises
        ------
        InvalidArgumentError
            The specFrom value is out of range
        InvalidArgumentError
            The specTo value is out of range
        InvalidArgumentError
            specTo or specFrom is not of type int or None
        ValueError
            The given indices exceed the buffer file's size
        InvalidArgumentError
            The given conversion is unknown

        Returns
        -------
        npt.NDArray
            The buffers data.

        Example
        -------
        ```py
        import qass.tools.analyzer.buffer_parser as bp
        proc = range(1,100,1)
        path contains the path to a directory containing buffer files
        buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
        for buffer in buff:
            buff_file = (buffer.filepath)

        with bp.Buffer(buff_file) as buff:
            spec_start = 0
            spec_end = (buff.db_count -1) * buff.db_spec_count
            print('Spec_end: ' + str(spec_end))
            conv = 'delog'
            data = buff.get_data(spec_start, spec_end, conv)
        ```
        """
        specFrom = 0 if specFrom is None else specFrom
        specTo = self.spec_count if specTo is None else specTo

        if not (isinstance(specFrom, int) and isinstance(specTo, int)):
            raise InvalidArgumentError("specFrom/specTo must be int or None!")
        if specFrom > specTo:
            raise InvalidArgumentError(
                f"specFrom {specFrom} cannot be larger than specTo {specTo}!"
            )
        if specTo < 0 or specTo > self.spec_count:
            raise InvalidArgumentError(
                f"specTo {specTo} is out of range (0, {self.spec_count})!"
            )
        if specFrom < 0 or specFrom > self.spec_count:
            raise InvalidArgumentError(
                f"specFrom {specFrom} is out of range (0, {self.spec_count})!"
            )

        if self.datamode == self.DATAMODE.DATAMODE_SIGNAL:
            return self._get_data(specFrom, specTo, 1, conversion).reshape(-1)
        else:
            return self._get_data(specFrom, specTo, self.__frq_bands, conversion)

    def getArray(self, specFrom=None, specTo=None, delog=None):
        """
        Wrapper function to 'get_data'.
        This function provides access to the measurement data in the buffer
        file. The data are retrieved for the range of spectra and stored in a
        numpy array. The data can be logarithmized or not. The datatype of the
        array depends on the type of buffer.

        Parameters
        ----------
        specFrom : int, optional
            First spectrum to be retrieved (default first available spectrum).
        specTo : int, optional
            Last spectrum to be retrieved (default last available spectrum).
        delog : bool, optional
            To de-logarithmize the data (default None).

        Raises
        ------
        InvalidArgumentError
            The specFrom value is out of range.
        InvalidArgumentError
            The specTo value is out of range.

        Returns
        -------
        npt.NDArray
            The buffers data.

        Example
        -------
        ```py
        import qass.tools.analyzer.buffer_parser as bp
        proc = range(1,100,1)
        # path contains the path to a directory containing buffer files
        buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
        for buffer in buff:
            buff_file = (buffer.filepath)

        with bp.Buffer(buff_file) as buff:
            spec_start = 0
            spec_end = (buff.db_count -1) * buff.db_spec_count
            print('Spec_end: ' + str(spec_end))
            conv = 'delog'
            data = buff.getArray(spec_start, spec_end, conv)
        ```
        """
        if delog == True:
            return self.get_data(specFrom, specTo, conversion="delog")
        elif delog == False:
            return self.get_data(specFrom, specTo, conversion="log")
        else:
            return self.get_data(specFrom, specTo)

    def getSpecDuration(self):
        """
        Wrapper Function for property spec_duration.

        See Also
        -------
        spec_duration
        """
        return self.spec_duration

    def _parse_db_header(self, content: bytearray) -> dict:
        db_metainfo = {}

        if content[0] == 0:
            warnings.warn("empty header content")
            return db_metainfo

        idx = 0
        while idx < self.__db_header_size:
            try:
                read_res = self._read_next_value(idx, content, self.__db_keywords)
                if read_res is None:  # key not found
                    warnings.warn("header parse error")
                    idx += 4
                else:
                    key, val, idx = read_res
                    if key:
                        db_metainfo[key] = val
            except Exception:
                raise ValueError("data block header content not parseble")

        return db_metainfo

    def _get_datablock_start_pos(self, db_idx):
        return self.__header_size + db_idx * (self.__db_size + self.__db_header_size)

    def _first_sample_of_datablock(self, db_idx):
        return int(db_idx * self.__db_size / self.__bytes_per_sample)

    def _first_spec_of_datablock(self, db_idx):
        return int(self._first_sample_of_datablock(db_idx) / self.__frq_bands)

    def db_header(self, db_idx):
        """
        The data block header contains key words and their corresponding
        values. This information is retrieved and stored in a dictionary. The
        values can be accessed by providing the corresponding key. Please
        ensure that the key exists in the database prior to its use or
        use the get method to access its content.Otherwise this will
        cause a runtime error.


        Parameters
        ----------
        db_idx : int
            data block index

        Raises
        ------
        ValueError
            The data block index is out of range.

        Returns
        -------
        dict
            a dictionary with the keywords and values

        Example
        -------
        ```py
        # The index of the first data block is 0
        db_idx = 0
        # Function to retrieve the size of the data block
        def db_size(db_idx):
            db_header_dict = db_header(0)
            return(db_header_dict.get('dbfilled', 0))
        ```
        """
        if db_idx > self.__db_count:
            raise ValueError("db_idx is out of bounds")

        if db_idx in self.__db_headers:
            return self.__db_headers[db_idx]

        header_start_pos = self._get_datablock_start_pos(db_idx)

        self.file.seek(header_start_pos, os.SEEK_SET)
        db_header_content = self.file.read(self.__db_header_size)
        db_metainfo = self._parse_db_header(db_header_content)
        self.__db_headers[db_idx] = db_metainfo
        return db_metainfo

    def db_header_spec(self, spec: int):
        """
        The method returns the data block header as a dictionary of the data
        block containing the spectrum whose index is provided.

        Parameters
        ----------
        spec: int
            The index of a spectrum.

        Returns
        -------
        dict
            The data block header with keywords and values.

        Example
        -------
        ```py
        # The index of the spectrum is 50
        spec = 50
        # Function to retrieve the size of the data block
        def db_size(spec):
            db_header_dict = db_header_spec(spec)
            return(db_header_dict.get('dbfilled', 0))
        ```
        """
        db_idx = int(spec * self.__frq_bands / self.db_sample_count)
        return self.db_header(db_idx)

    def db_value(self, db_idx: int, key: str):
        """
        Similar to the file header each data block has its own header. This
        method returns the value of the property specified by key in the
        specified data block. The first data block has the index 0.

        Parameters
        ----------
        db_idx : int
            The index of the data block.
        key: str
            The string containing the key word.

        Returns
        -------
        int
            The data block value for the given key.
        """
        return self.db_header(db_idx)[key]

    def io_ports(self, db_idx: int, byte: int = None, bit: int = None):
        """
        Deprecated method!
        This method provides the state of the io ports at the time the
        dateblock (indicated by its number db_idx) is written. It does not
        provide information wether the state of any io port changes within the
        datablock. This method will be replaced by a method which analyses the
        data within the sub data block, which is part of the data block header
        and provides information regarding the state of the io ports at a
        higher accuracy.

        Parameters
        ----------
        db_idx : int
            Index of the data block
        byte : int
            Number of io port socket [1, 2, 4]
        bit : int, optional
            Number of bit [1, 2, 3, 4, 5, 6, 7, 8]

        Raises
        ------
        ValueError
            The bit argument requires the byte argument.
        ValueError
            The given byte has to be one out of [1, 2, 4].
        ValueError
            The given bit is out of range.

        Returns
        -------
        int
            byte with the appropriate bits set as an int
        """
        io_word = ~(self.db_value(db_idx, "io_ports")) & 0xFFFFFF

        if byte is None and bit is not None:
            raise ValueError("The bit argument requires the byte argument")

        if byte is None:
            return io_word

        if byte == 1:
            io_word = (io_word >> 0) & 0xFF
        elif byte == 2:
            io_word = (io_word >> 8) & 0xFF
        elif byte == 4:
            io_word = (io_word >> 16) & 0xFF
        else:
            raise ValueError("The given byte has to be one out of [1, 2, 4]")

        if bit is None:
            return io_word

        if not 1 <= bit <= 8:
            raise ValueError(f"The given bit is out of range: {bit}")

        return io_word >> (bit - 1) & 1 == 1

    def io_ports_spec(self, spec: int, byte: int, bit: int):
        """
        Deprecated method!
        This method provides the state of the io ports at the time the
        dateblock (indicated by the spec number within the data block) is
        written. It does not provide information wether the state of any io
        port changes within the datablock. This method will be replaced by a
        method which analyses the data within the sub data block, which is
        part of the data block header and provides information regarding the
        state of the io ports at a higher accuracy.

        Parameters
        ----------
        spec : int
            Index of the spectrum
        byte : int
            Number of io port socket [1, 2, 4]
        bit : int
            Number of bit [1, 2, 3, 4, 5, 6, 7, 8]

        Raises
        ------
        ValueError
            The bit argument requires the byte argument.
        ValueError
            The given byte has to be one out of [1, 2, 4].
        ValueError
            The given bit is out of range.

        Returns
        -------
        int
            byte with the appropriate bits set as an int
        """
        db_idx = int(spec * self.__frq_bands / self.db_sample_count)
        return self.io_ports_byte(db_idx, byte, bit)

    def file_size(self):
        """
        The file size of the buffer file. The value is not stored in the header
        but retrieved using the operating system.

        Returns
        -------
        int
            file size in bytes
        """
        return self.file_size

    def _calc_spec_duration(self):
        if "framedur" in self.__metainfo.keys():
            return

        if self.datamode == self.DATAMODE.DATAMODE_FFT:
            needed_keys = ["fftovers", "samplefr", "comratio"]
            if any(key not in self.__metainfo for key in needed_keys):
                raise ValueError(
                    f"Keys are missing to calculate the spec_duration. Missing keys: {[key for key in needed_keys if key not in self.__metainfo]}"
                )

            duration = (
                1e9
                / (
                    (self.__metainfo["samplefr"] * (1 << self.__metainfo["fftovers"]))
                    / 1024
                )
            ) * self.__metainfo["comratio"]
            self.__metainfo["framedur"] = duration
        elif self.datamode == self.DATAMODE.DATAMODE_SIGNAL:
            needed_keys = ["samplefr", "comratio"]
            if any(key not in self.__metainfo for key in needed_keys):
                raise ValueError(
                    f"Keys are missing to calculate the spec_duration. Missing keys: {[key for key in needed_keys if key not in self.__metainfo]}"
                )

            sample_frq = self.__metainfo["samplefr"]
            compression = self.__metainfo["comratio"]
            self.__metainfo["framedur"] = int(1e9 / sample_frq * compression)

    def _signalNormalizationFactor(self, gain=1):
        """_signalNormalizationFactor calculates a ref_energy factor to derive a normalized energy related to time, frequency and amplitude ranges
        of the buffer. A rectangular signal volume, which's base is the rectangle of one spectrums distance and one frequency band's width
        e.g. (100mV * 82�s * 1525Hz) = 0.012505 V. As the voltage is not squared, the result is close to the square-root of an energy.

        Each amplitude could reach a maximum of 1 Volt, the maximum ADC input voltage.
        Therefore we divide the amplitude with the maximum amplitude number to get it's portion in Volt.
        Then we multiply with the time and frequency square, as we assume the amplitude is constant over the complete square (our best guessing).

        When calculating energies of buffer ranges, we simply summarize the contained amplitudes.
        By multiplying the resulting sum with the ref_energy factor, we achieve the above described calculation of the signals volume.

        The resulting energy should be stable, even if compressing time and frequencies, using oversampling and different ADC ranges.

        sum of amplitudes,
        comparable even if sample rate, oversampling etc. is changing
        param gain,
        usually the amplitude ref is 1.0 equal 100% (eliminates the bit resolution of the buffer)
        could be enhanced with the amplification information of the amplifiers

        Parameters
        ----------
        gain : float, optional
            Can reflect changes of the amplification chain, energy is then related where the gain starts (sensor output e.g.), defaults to 1

        Returns
        -------
        float
            a factor to multiply with the sums of amplitudes in time-frequency-buffers
        """

        self.__norm_factor = 1.0
        if gain == 0.0:
            gain = 1.0

        max_amp = self.max_amplitude_level * gain
        # 16 bit=65535, gain=250 => maximum Amp=1V/250=0.004V ~ 4mV
        if max_amp:
            if self.__frq_bands > 1:
                self.__norm_factor = (
                    self.frq_per_band * self.spec_duration / 1e9 / max_amp
                )
            else:
                # normFactor = specDuration() / 1e9 / max_amp;
                # //@todo: calculation may be wrong, check it
                self.__norm_factor = 1.0

        else:
            max_amp = 2 << 16
            max_amp *= gain
            self.__norm_factor = self.frq_per_band * self.spec_duration / 1e9 / max_amp

        return self.__norm_factor

    def block_infos(
        self,
        columns: List[str] = [
            "preamp_gain",
            "mux_port",
            "measure_positions",
            "inputs",
            "outputs",
        ],
        changes_only: bool = False,
        fast_jump: bool = True,
    ):
        """block_infos iterates through all memory blocks of a buffer file (typically one MB) and fetches the subdata information
        each memory block has one set of metadata but e.g. 65 subdata entries for raw files
        or more than 2000 entries for a 32 times compressed file,
        where each entry contains meta information about a small time frame (15 spectrums in raw fft files)

        Parameters
        ----------
        columns : list[str]
            List of columns that should be in the result_array.
            You can define the order of the columns here.
            Possible column names are: ['preamp_gain', 'mux_port', 'measure_positions', 'inputs', 'outputs', 'times', 'index', 'spectrums']
        changes_only : bool, optional
            Flag to enable a summarized output. If True the output does not contain all entries but only entries where the data columns changed.
            In the current implementation the memory consumption does not change here -> in both cases the array is first completely built.
            Defaults to False.
        fast_jump : bool, optional
            If True the function uses seeking in the file instead of parsing every single datablock header.
            The offsets are investigated for the first datablock header and simply applied for all other datablock headers.
            Seeking is usually faster than parsing the datablock headers if they are not already in the cache.
            If the datablock headers are already cached this flag has no effect.
            Defaults to true.

        Returns
        -------
        header_infos : npt.NDArray
            An array containing (spec_index), (index), (times), preamp_gain, mux_port, measure_position, 24bit input, 16bit output, (times), (index), (spec_index)
        """

        data_columns = [
            "preamp_gain",
            "mux_port",
            "measure_positions",
            "inputs",
            "outputs",
        ]
        data_columns_indices = [0, 1, 2, 3, 5]  # positions in subdat block
        index_columns = ["times", "index", "spectrums"]

        allowed_columns = data_columns + index_columns

        for col in columns:
            if col not in allowed_columns:
                raise InvalidArgumentError(
                    f"Column {col} is not a valid column name. Valid columns are: {allowed_columns}"
                )
            if columns.count(col) > 1:
                raise InvalidArgumentError(
                    f"Column {col} is used multiple times. This is not allowed."
                )

        if changes_only:
            if not any(col in index_columns for col in columns):
                raise InvalidArgumentError(
                    "If you use the changes_only option you need to declare at least one index column. Otherwise the results would have no connection."
                )

        # interpreting the entries as int32 makes most sense
        last_sample = 0
        # old header size was 10+32bit entries
        ds_size = 10
        mi = self.db_header(0)
        if "begin_subdat" not in mi:
            raise ValueError(
                f"begin_subdat keyword missing in datablock {0} -> caonnot read IO information"
            )

        subdat = np.frombuffer(mi["begin_subdat"], dtype=np.int32)
        # if it is of extended type, we expect a reasonable value here
        mysize = subdat[10].item()

        if mysize == 80:  # the extended data length, additional sizes may occur
            ds_size = 20  # again the size in 32bit entries

        entries_before = 0

        if fast_jump:
            db_start_pos = self._get_datablock_start_pos(0)
            self.file.seek(db_start_pos, os.SEEK_SET)
            db_header_content = self.file.read(self.__db_header_size)
            start_mark = b"begin_subdat"
            end_mark = b"end___subdat"
            try:
                subdat_offset_start = db_header_content.index(start_mark) + len(
                    start_mark
                )
                subdat_offset_end = db_header_content.index(end_mark)
                subdat_length = subdat_offset_end - subdat_offset_start
                subdat_readlen = int(subdat_length / subdat.itemsize)
            except ValueError as e:
                raise ValueError(
                    f"The datablock header seems not to have a subdat block. Reading block info not possibe. {str(e)}"
                )

        samples_per_entry = mi["sd_rsize"] / self.bytes_per_sample
        specs_per_entry = samples_per_entry / self.frq_bands
        entries_per_spec = 1 / specs_per_entry

        # columns of interest:
        # pgain, mux_port, measure_position, 24bit input (inverted), 16bit output (inverted)
        data_columns_src = []  # column index in the buffers meta info array
        data_columns_dest = []  # column index in the target result_array

        times_col = None
        index_col = None
        specs_col = None

        for idx, col in enumerate(columns):
            if col == "times":
                times_col = idx
            elif col == "index":
                index_col = idx
            elif col == "spectrums":
                specs_col = idx
            elif col in data_columns:
                data_columns_src.append(data_columns_indices[data_columns.index(col)])
                data_columns_dest.append(idx)

        if not data_columns_dest:
            raise InvalidArgumentError("You have to choose at least one data column!")

        # the indices of valid entries in the subdata block are calculated per datablock (checked in Analyzer4D DBgrabber::getSubDataPointer)
        full_datablock_count = self.spec_count // self.db_spec_count
        full_datablock_specs = full_datablock_count * self.db_spec_count
        last_nonfull_datablock_specs = self.spec_count % self.db_spec_count
        entry_count = int(full_datablock_specs * min(1, entries_per_spec)) + int(
            last_nonfull_datablock_specs * min(1, entries_per_spec)
        )

        # 64 bit array if index columns are requested, otherwise 32 bit
        arr_type = (
            np.int64 if any((times_col, index_col, specs_col)) is not None else np.int32
        )
        result_arr = np.empty(
            (entry_count, len(columns)), dtype=arr_type
        )  # allocating the array!

        # loop through the datablocks
        for i in range(self.db_count):
            if fast_jump and i not in self.__db_headers:
                db_start_pos = self._get_datablock_start_pos(i)
                self.file.seek(db_start_pos + subdat_offset_start, os.SEEK_SET)
                entries = np.fromfile(
                    self.file, dtype=np.int32, count=subdat_readlen
                ).reshape(-1, ds_size)
                start_spec = self._first_spec_of_datablock(i)
                end_spec = min(self._first_spec_of_datablock(i + 1), self.spec_count)
            else:
                mi = self.db_header(i)
                if "begin_subdat" not in mi:
                    raise ValueError(
                        f"begin_subdat keyword missing in datablock {i} -> caonnot read IO information"
                    )

                subdat = np.frombuffer(mi["begin_subdat"], dtype=np.int32)
                entries = subdat.reshape(-1, ds_size)
                # f_entries=np.empty((0,ds_size),dtype=np.int32)

                # print("mi['sd_rsize']", mi['sd_rsize'])
                samples_per_entry = mi["sd_rsize"] / self.bytes_per_sample
                specs_per_entry = samples_per_entry / self.frq_bands
                entries_per_spec = 1 / specs_per_entry

                start_spec = int(mi["firstsam"] / self.frq_bands)
                # the end_spec might differ from lastsamp for the last datablock: The real data might be shorter
                end_spec = min(int(mi["lastsamp"] / self.frq_bands), self.spec_count)

            db_spec_count = end_spec - start_spec

            # the number of expected entries might be less than specs for low compressions, but never more than db_spec_count
            entries_expected = int(db_spec_count * min(1, entries_per_spec))

            # max(1, entries_per_spec) because for low compressions we get less than 1 entry per spec -> take every entry
            idxs = np.arange(entries_expected) * max(1, entries_per_spec)
            idxs = idxs.astype(int)

            # filter the entries based on the calculated indexes
            f_entries = entries[idxs]

            assert (
                len(f_entries) == entries_expected
            )  # this must be equal - its critical to have a mismatch here!

            # writing columns_of_interest into the result_arr
            result_arr[
                entries_before : entries_before + len(f_entries), data_columns_dest
            ] = f_entries[:, data_columns_src]

            if times_col is not None:
                start_time = start_spec * self.spec_duration
                end_time = end_spec * self.spec_duration
                times = np.arange(entries_expected, dtype=float)

                # for strong compresssions we get exactly one (and never more than one) entries for each spectrum.
                times *= self.spec_duration / min(1, entries_per_spec)
                times += start_time

                times = times.astype(int)
                result_arr[entries_before : entries_before + len(times), times_col] = (
                    times
                )

            if index_col is not None:
                index = np.arange(
                    entries_before, entries_before + entries_expected, dtype=int
                )
                result_arr[entries_before : entries_before + len(index), index_col] = (
                    index
                )

            if specs_col is not None:
                index = np.arange(entries_expected) * max(1, specs_per_entry)
                index = index.astype(int)
                index += start_spec
                result_arr[entries_before : entries_before + len(index), specs_col] = (
                    index
                )

            entries_before += entries_expected  # len(f_value_list)

        assert (
            entries_before == len(result_arr)
        )  # If this fails we did something wrong when calculating the expected number of entries.

        # Some conversions to ensure readability of the values
        if "preamp_gain" in columns:
            col = columns.index("preamp_gain")
            result_arr[:, col] = result_arr[:, col] >> 16

        if "inputs" in columns:
            col = columns.index("inputs")
            result_arr[:, col] = np.bitwise_and(
                np.bitwise_not(result_arr[:, col]), 0xFFFFFF
            )

        if "outputs" in columns:
            col = columns.index("outputs")
            result_arr[:, col] = np.bitwise_and(
                np.bitwise_not(result_arr[:, col]), 0xFFFF
            )

        if changes_only:
            changes_idx = np.concatenate(
                (
                    (True,),
                    np.any(
                        result_arr[1:, data_columns_dest]
                        != result_arr[:-1, data_columns_dest],
                        axis=1,
                    ),
                )
            )
            return result_arr[changes_idx]
        else:
            return result_arr

    @property
    def header_hash(self):
        """A hash value for the buffer's header.
        The header of the buffer is almost unique for a buffer.
        A buffer should only differ in the length of the contained data.

        Returns
        -------
        int
            The header's hash
        """
        return self.__header_hash

    @property
    def metainfo(self):
        """
        The buffer header properties as a dictionary.

        Returns
        -------
        dict
            The metainfo dictionary

        Example
        -------
        ```py
        key = 'qassdata'
        def keyword(key)
            return buffer.metainfo[key]

        # The above example leads to a runtime error if the keyword is not a
        # key in the dictionary. It would be better to query this beforehand
        # and to provide a default value, thus:

        def keyword(key):
            if key in buffer.metainfo.keys:
                return buffer.metainfo[key]
            else:
                return default_value

        # or shorter:

        def keyword(key):
            return buffer.metainfo.get(key, default_value)
        ```
        """
        return self.__metainfo

    @property
    def filepath(self):
        """
        Returns the complete path to the file.

        Returns
        -------
        str
            Path to the file
        """
        return self.__filepath

    @property
    def header_size(self):
        """
        Each buffer file has a header. This property provides the size of the
        header in bytes (The normal value is 2000 bytes).

        Returns
        -------
        int
            Size in bytes.
        """
        return self.__header_size

    @property
    def project_id(self):
        return self.__metainfo["proj__id"]

    @property
    def process(self):
        """
        The process number.

        Returns
        -------
        int
            The process number.
        """
        return self.__metainfo["proc_cnt"]

    @property
    def channel(self):
        """
        The channel number used.

        Returns
        -------
        int
            The channel number.
        """
        return self.__metainfo["dumpchan"] + 1

    @property
    def streamno(self) -> Union[int, None]:
        """
        The index of a stream of an extra channel.

        Extra channels in the analyzer software can map multiple different data
        streams to the same channel of the software. This makes them only distuinguishable
        by this property.
        """
        return self.__metainfo.get("streamno", None)

    @property
    def datamode(self):
        """
        The data mode is a constant which specifies what kind of data are in
        the buffer. The most important ones are DATAMODE_FFT and
        DATAMODE_SIGNAL. If you want to obtain the name of the constant, use
        `datamode.name`.

        Returns
        -------
        DATAMODE
            The data mode constant
        """
        return self.DATAMODE(self.__metainfo["datamode"])

    @property
    def datakind(self):
        """
        The data kind constant is a constant which provides additional
        buffer specifications. A common value often is
        KIND_UNDEF. If you want to obtain the name of the constant, use
        `datakind.name`.

        Returns
        -------
        DATAKIND
            The data kind constant.
        """
        return self.DATAKIND(self.__metainfo["datakind"])

    @property
    def datatype(self):
        """
        The data type constant is a constant which specifies different
        compression and buffer types. The most important one being
        COMP_RAW. If you want to obtain the name of the constant, use
        `datatype.name`.

        Returns
        -------
        DATATYPE
            The data type constant.
        """
        return self.DATATYPE(self.__metainfo.get("datatype", -1))

    @property
    def process_time(self):
        """
        Returns the Posix timestamp of creation of the file
        in seconds since Thursday, January 1st 1970. Please note that the
        creation time and measure time might be different.

        Returns
        -------
        int
            The timestamp as a number of seconds.
        """
        return self.__metainfo.get("proctime", 0)

    @property
    def process_date_time(self):
        """
        Returns a timestamp of the creation of the file as a string.

        Returns
        -------
        str
            The timestamp in the form 'yyyy-mm-dd hh:mm:ss'
        """
        from datetime import datetime

        return datetime.utcfromtimestamp(int(self.__metainfo["proctime"])).strftime(
            "%Y-%m-%d %H:%M:%S"
        )

    @property
    def process_measure_timestamp(self):
        """
        Returns the Posix timestamp at measure start of the process in
        milliseconds. If epoctime (measure time) is not available due to an
        older buffer format, the creation time of the buffer is returned.
        Please note: The creation time may not be the same as epoc time.
        This is particularly true for compressed buffers which may be created
        much later!

        Returns
        -------
        int
            The timestamp as a number in seconds.
        """
        return self.__metainfo.get("epoctime", self.__metainfo.get("proctime") * 1000)

    @property
    def process_measure_time_string(self):
        """
        Returns the time in ISO format at measure start of the process. If
        epoctime (measure time) is not available due to an older buffer
        format, the creation time of the buffer is returned. Please note:
        The creation time may not be the same as epoc time. This is
        particularly true for compressed buffers which may be created
        much later!

        Returns
        -------
        str
            The timestamp in the form 'yyyy-mm-dd hh:mm:ss'.
        """
        from datetime import datetime

        tsms = self.__metainfo.get("epoctime", self.__metainfo.get("proctime") * 1000)

        datestr = datetime.utcfromtimestamp(tsms // 1000).strftime("%Y-%m-%d %H:%M:%S.")
        datestr += format(tsms % 1000, "03")
        return datestr

    @property
    def last_modification_time(self):
        """
        Returns the Posix timestamp of last modification of the file
        in seconds since Thursday, January 1st 1970.

        Returns
        -------
        int
            The timestamp as a number in seconds.
        """
        return self.__metainfo.get("lmodtime", 0)

    @property
    def last_modification_date_time(self):
        from datetime import datetime

        """
        Returns a timestamp of the last modification as a string.

        Returns
        -------
        str
            The timestamp in the form 'yyyy-mm-dd hh:mm:ss'
        """
        return datetime.utcfromtimestamp(
            int(self.__metainfo.get("lmodtime", 0))
        ).strftime("%Y-%m-%d %H:%M:%S")

    @property
    def db_header_size(self):
        """
        Each data block has a header. This property provides the size of the
        data block header in bytes. The sizes of all data block headers are
        equal.

        Returns
        -------
        int
            Size in bytes
        """
        return self.__db_header_size

    @property
    def bytes_per_sample(self):
        """
        Number of bytes per sample.

        Returns
        -------
        int
            Number of bytes per sample (usually 2 bytes).
        """
        return self.__bytes_per_sample

    @property
    def db_count(self):
        """
        Returns the number of data blocks.

        Returns
        -------
        int
            Number of data blocks.
        """
        return self.__db_count

    @property
    def full_blocks(self):
        """
        All but the last data block are usually completly filled. In order to
        calculate the number of those data blocks which are completely filled
        the header size of the file is substracted from the file size. This
        figure is divided by the sum of the data block header size and data
        block size and rounded down to the nearest integer.

        Returns
        -------
        int
            Number of full data blocks.
        """
        return math.floor(
            (self.file_size - self.__header_size)
            / (self.__db_header_size + self.__db_size)
        )

    @property
    def db_size(self):
        """
        Returns the size of a completely filled data block in bytes.

        Returns
        -------
        int
            Size of a completely filled data block.
        """
        return self.__db_size

    @property
    def db_sample_count(self):
        """
        Returns the number of samples in a completely filled data block.
        It is calculated by dividing the data block size by the number of
        bytes per sample.

        Returns
        -------
        int
            Number of samples.
        """
        return math.floor(self.__db_size / self.__bytes_per_sample)

    @property
    def frq_bands(self) -> int:
        """
        Each spectrum has a maximum sample number of 512. This is the maximum
        number of frequency bands. This figure decreases with compression along
        the frequency axis. The corresponding key word in the metainfo
        dictionary 's_p_fram' (samples per frame).

        Returns
        -------
        int
            Number of frequency bands.
        """
        return self.__frq_bands

    @property
    def db_spec_count(self):
        """
        The number of spectra within a filled data block. It is calculated by
        dividing the number of samples in a data block by the number of
        frequency bands.

        Returns
        -------
        int
            Number of spectra in a filled data block.
        """
        return int(self.db_sample_count / self.__frq_bands)

    @property
    def compression_frq(self):
        """
        Property which returns the frequency compression factor.

        Returns
        -------
        int
            Frequency compression factor.
        """
        return self.__metainfo["frqratio"]

    @property
    def compression_time(self):
        """
        Property which returns the time compression factor.

        Returns
        -------
        int
            Time compression factor.
        """
        return self.__metainfo["comratio"]

    @property
    def avg_time(self):
        """
        Property which returns the moving average factor along the
        time axis.

        Returns
        -------
        int
            Factor of the moving average.
        """
        return self.__metainfo.get("auxpara1", 1)

    @property
    def avg_frq(self):
        """
        Property which returns the moving average factor along the
        frequency axis.

        Returns
        -------
        int
            Factor of the moving average.
        """
        return self.__metainfo.get("auxpara2", 1)

    @property
    def spec_duration(self):
        """
        The property returns the time for one spectrum in
        nanoseconds.

        Returns
        -------
        float
            Time in nanoseconds.
        """
        return self.__metainfo["framedur"]

    @property
    def frq_start(self):
        """
        The property returns the lowest represented frequency.

        Returns
        -------
        int
            Frequency in Hertz (Hz).
        """
        return self.__metainfo["startfrq"] if "startfrq" in self.__metainfo else 0

    @property
    def frq_end(self):
        """
        The property returns the highest represented frequency.

        Returns
        -------
        int
            Frequency in Hertz (Hz).
        """
        return (
            self.__metainfo["end__frq"]
            if "end__frq" in self.__metainfo
            else self.frq_bands * self.frq_per_band
        )

    @property
    def frq_per_band(self):
        """
        The property returns the frequency range per band. The maximum number
        of bands is 512.

        Returns
        -------
        float
            Frequency range in Hertz (Hz) for one band.
        """
        return self.__metainfo["frqpband"]

    @property
    def sample_count(self):
        """
        The number of samples in the buffer. It is calculated by subtracting
        the header size and the data block size times the number of datablocks
        from the file size. This figure is divided by the bytes per sample to
        obtain the number of samples. The number is cast to an int as the
        division usually results in a float due to an incompletely filled
        final data block.

        Returns
        -------
        int
            Number of samples
        """
        without_header = self.file_size - self.__header_size
        return int(
            (without_header - self.__db_count * self.__db_header_size)
            / self.__bytes_per_sample
        )

    def getRealSpecCount(self):
        """
        Wrapper function for the property spec_count.
        The number of spectra within a filled data block. It is calculated by
        dividing the number of samples in a data block by the number of
        frequency bands.

        Returns
        -------
        int
            Number of spectra in a filled data block.
        """
        return self.spec_count

    @property
    def spec_count(self):
        """
        The number of spectra in the buffer. It is calculated by dividing
        the number of samples by the number of frequency bands.

        Returns
        -------
        int
            number of spectra
        """
        return int(self.sample_count / self.__frq_bands)

    @property
    def last_spec(self):
        if self.__last_spectrum is None:
            last_sample = 0

            last_db = self.db_count - 1
            mi = self.db_header(last_db)
            if "lastsamp" not in mi.keys():
                raise ValueError(
                    'The datablock header does not contain a key "lastsamp"'
                )

            last_sample = mi["lastsamp"]
            self.__last_spectrum = int(last_sample / self.frq_bands)

        return self.__last_spectrum

    @property
    def adc_type(self):
        """
        The type of analog/digital converter. The types are defined in class
        ADC_TYPE with the most important being ADC_16BIT and ADC_24BIT. If
        you want to obtain the name of the constant, use `adc_type.name`.

        Returns
        -------
        ADCTYPE
            The ADCTYPE constant
        """
        return self.ADCTYPE(self.__metainfo["adc_type"])

    @property
    def bit_resolution(self):
        """
        The bit resolution after the FFT. Caution: this is not the type of
        analog digital converter used. The usual value is 16. If no logarithm
        is applied to the data the value is 32.

        Returns
        -------
        int
            The bit resolution.
        """
        return self.__metainfo["adbitres"]

    @property
    def fft_log_shift(self):
        """
        Base of the logarithm applied to the data. Please note that a 1 needs
        to be added to the value.

        Returns
        -------
        int
            Base of the logarithm.
        """
        return self.__metainfo["fftlogsh"]

    @property
    def max_amplitude_level(self):
        """
        maximum possible amplitude of the buffer, relates to the buffer's dtype

        Returns
        -------
        int, float
        """
        return self.__metainfo["max_ampl"]

    @property
    def refEnergy(self):
        """
        refEnergy provides a normalization factor for compressions.
        This makes sums of amplitudes of different compressions comparable to each other.
        See _signalNormalizationFactor for a detailed description.

        Returns
        -------
        float
        """
        return self.__norm_factor

    @property
    def ref_energy(self):
        """
        alias for refEnergy()

        Returns
        -------
        float
        """
        return self.refEnergy

    @property
    def preamp_gain(self):
        """
        The preamplifier setting in the multiplexer tab of the Analyzer4D software.

        Returns
        -------
        int
        """
        return self.__metainfo["pampgain"]

    @property
    def analyzer_version(self) -> Union[str, None]:
        """
        The analyzer4D version as a dot separated string consisting of major.minor.micro.patch

        Returns
        -------
        str, None
        """
        version = self.__metainfo.get("an4dvers", None)
        if version is None:
            return None
        version = version.decode("utf-8")
        version_tuple = tuple(version[i : i + 2] for i in range(0, len(version), 2))
        return ".".join(version_tuple[::-1])

    @property
    def partnumber(self) -> str:
        """
        The string representation of the partnumber of the measurement.

        Returns
        -------
        str
        """
        asc_part_string = self.__metainfo.get("asc_part", "")
        if isinstance(asc_part_string, str):
            return asc_part_string
        return asc_part_string.decode("utf-8").replace("\x00", "")

    def spec_to_time_ns(self, spec):
        """
        The method calculates the time since measurement start for a given
        index of a spectrum.

        Parameters
        ----------
        spec : int
            The index of a spectrum

        Returns
        -------
        float
            time in nanoseconds
        """
        return spec * self.spec_duration

    def time_to_spec(self, time_ns):
        """
        The method calculates the index of the nearest spectrum to the time
        given since the start of the measurement.

        Parameters
        ----------
        time_ns : float
            Time elapsed since measurement start in ns.

        Returns
        -------
        int
            index of the spectrum
        """
        return int(time_ns / self.spec_duration)

    @staticmethod
    def delog(data_arr, fft_log_shift, ad_bit_resolution):
        """
        Method to de-logarithmize a data array. Usually when data are measured
        with the Optimizer the data are logarithmized. The problem with
        logarithms is that adding them up constitutes a multiplication of the
        non logarithmized data. So in order to carry out calculations such as
        calculating sums etc. it is crucial to use de-logarithmized data.

        Parameters
        ----------
        data_arr : npt.NDArray
            Array with the logarithmized data.
        fft_log_shift : int
            Base of the logarithm.
        ad_bit_resolution : int
            The bit resolution (usually 16).

        Returns
        -------
        npt.NDArray[np.floating]
            Array with de-logarithmized data.
        """
        shift = fft_log_shift - 1
        bits = ad_bit_resolution if ad_bit_resolution >= 1 else 16
        bit_res_idx = [14, 16, 24].index(bits)

        # remember negative values:
        negative = data_arr < 0
        np.abs(data_arr, out=data_arr)

        # ensure data_arr to be a floating point array
        data_arr = data_arr.astype(np.double, copy=False)

        shift_offset = [23, 23, 31][bit_res_idx]
        max_amp = 1 << bits

        data_arr = data_arr * (shift_offset - shift) / max_amp + shift

        data_arr = (2**data_arr) - (1 << shift)
        data_arr[negative] *= -1

        return data_arr

    @staticmethod
    def log(data_arr, fft_log_shift, ad_bit_resolution):
        """
        Method to logarithmize a data array. Usually when data are measured
        with the Optimizer the data are logarithmized. The problem with
        logarithms is that adding them up constitutes a multiplication of the
        non logarithmized data. So in order to carry out calculations such as
        calculating sums etc. it is crucial to use de-logarithmized data.
        Sometimes it is necessary to logarithmize data which were artificially
        created or measured without a logarithm.

        Parameters
        ----------
        data_arr : npt.NDArray[np.floating]
            Array with non-logarithmized data.
        fft_log_shift : int
            Base of the logarithm.
        ad_bit_resolution : int
            The bit resolution (usually 16).

        Returns
        -------
        npt.NDArray[np.floating]
        """
        shift = fft_log_shift - 1
        bits = ad_bit_resolution if ad_bit_resolution >= 1 else 16
        bit_res_idx = [14, 16, 24].index(bits)

        # remember negative values:
        negative = data_arr < 0
        np.abs(data_arr, out=data_arr)

        # ensure data_arr to be a floating point array
        data_arr = data_arr.astype(np.double, copy=False)

        shift_offset = [23, 23, 31][bit_res_idx]
        max_amp = 1 << bits

        data_arr = np.log2(data_arr + (1 << shift)) - shift
        data_arr = data_arr * max_amp / (shift_offset - shift)

        data_arr[negative] *= -1

        return data_arr

__db_headers instance-attribute

__db_headers = {}

__db_keywords instance-attribute

__db_keywords = [('blochead----', INT32), ('firstsam--------', INT64), ('lastsamp--------', INT64), ('dbfilled----', INT32), ('mux_port----', INT32), ('pampgain----', INT32), ('dispgain----', INT32), ('io_ports----', INT32), ('dversion----', INT32), ('sine_frq----', INT32), ('sine_amp----', INT32), ('sd_dimen----', INT32), ('sd_rsize----', INT32), ('sd_dsize----', INT32), ('begin_subdat', INT32), ('end___subdat', INT32), ('blockend', INT32)]

__filepath instance-attribute

__filepath = filepath

__header_hash instance-attribute

__header_hash = None

__keywords instance-attribute

__keywords = [('qassdata----', INT32), ('filevers----', INT32), ('datavers----', INT32), ('savefrom----', INT32), ('datamode----', INT32), ('datatype----', INT32), ('datakind----', INT32), ('framsize----', INT32), ('smplsize----', INT32), ('frqbands----', INT32), ('db_words----', INT32), ('avgtimba----', INT32), ('avgfrqba----', INT32), ('m_u_mask--------', UINT64), ('b_p_samp----', INT32), ('s_p_fram----', INT32), ('db__size----', INT32), ('max_ampl----', INT32), ('nul_ampl----', INT32), ('samplert----', INT32), ('datarate--------', DOUBLE), ('samplefr----', INT32), ('frqshift----', INT32), ('fftovers----', INT32), ('fftlogsh----', INT32), ('fftwinfu----', INT32), ('dbhdsize----', INT32), ('comratio----', INT32), ('tc__real--------', DOUBLE), ('frqratio----', INT32), ('fc__real--------', DOUBLE), ('proj__id--------', UINT64), ('file__id--------', UINT64), ('parentid--------', INT64), ('proc_cnt----', INT32), ('proc_rng----', INT32), ('proc_sub----', INT32), ('poly_cnt----', INT32), ('polycyid----', INT32), ('dumpchan----', INT32), ('del_lock----', INT32), ('proctime----', UINT32), ('lmodtime----', UINT32), ('epoctime--------', INT64), ('mux_port----', INT32), ('pampgain----', INT32), ('dispgain----', INT32), ('linfgain----', INT32), ('auxpara0----', INT32), ('auxpara1----', INT32), ('auxpara2----', INT32), ('auxpara3----', INT32), ('auxpara4----', INT32), ('auxpara5--------', INT64), ('skipsamp--------', INT64), ('skiptime--------', INT64), ('trunsamp--------', INT64), ('truntime--------', INT64), ('skiplfrq----', INT32), ('trunhfrq----', INT32), ('startfrq----', INT32), ('end__frq----', INT32), ('frqpband--------', DOUBLE), ('framedur--------', DOUBLE), ('frameoff----', INT32), ('p__flags----', INT32), ('realfrqc----', INT32), ('sub_data----', INT32), ('sd_dimen----', INT32), ('sd_rsize----', INT32), ('sd_rsizf--------', DOUBLE), ('sd_dsize----', INT32), ('frqmasks----', UINT32), ('frqinmas----', UINT32), ('eheaderf----', UINT32), ('adc_type----', INT32), ('adbitres----', INT32), ('baserate--------', INT64), ('interpol----', INT32), ('dcoffset----', INT32), ('dispralo----', INT32), ('disprahi----', INT32), ('rngtibeg--------', INT64), ('rngtiend--------', INT64), ('realsoff--------', INT64), ('extendid--------', UINT64), ('sim_mode----', INT32), ('partnoid----', UINT32), ('asc_part----', UINT32), ('asc_desc----', UINT32), ('comments----', UINT32), ('sparemem----', UINT32), ('streamno----', UINT32), ('an4dvers----', HEX_STRING), ('headsend', None)]

__metainfo instance-attribute

__metainfo = {}

adc_type property

adc_type

The type of analog/digital converter. The types are defined in class ADC_TYPE with the most important being ADC_16BIT and ADC_24BIT. If you want to obtain the name of the constant, use adc_type.name.

Returns:

Type	Description
ADCTYPE	The ADCTYPE constant

analyzer_version property

analyzer_version: Union[str, None]

The analyzer4D version as a dot separated string consisting of major.minor.micro.patch

Returns:

Type	Description
(str, None)

avg_frq property

avg_frq

Property which returns the moving average factor along the frequency axis.

Returns:

Type	Description
int	Factor of the moving average.

avg_time property

avg_time

Property which returns the moving average factor along the time axis.

Returns:

Type	Description
int	Factor of the moving average.

bit_resolution property

bit_resolution

The bit resolution after the FFT. Caution: this is not the type of analog digital converter used. The usual value is 16. If no logarithm is applied to the data the value is 32.

Returns:

Type	Description
int	The bit resolution.

bytes_per_sample property

bytes_per_sample

Number of bytes per sample.

Returns:

Type	Description
int	Number of bytes per sample (usually 2 bytes).

channel property

channel

The channel number used.

Returns:

Type	Description
int	The channel number.

compression_frq property

compression_frq

Property which returns the frequency compression factor.

Returns:

Type	Description
int	Frequency compression factor.

compression_time property

compression_time

Property which returns the time compression factor.

Returns:

Type	Description
int	Time compression factor.

datakind property

datakind

The data kind constant is a constant which provides additional buffer specifications. A common value often is KIND_UNDEF. If you want to obtain the name of the constant, use datakind.name.

Returns:

Type	Description
DATAKIND	The data kind constant.

datamode property

datamode

The data mode is a constant which specifies what kind of data are in the buffer. The most important ones are DATAMODE_FFT and DATAMODE_SIGNAL. If you want to obtain the name of the constant, use datamode.name.

Returns:

Type	Description
DATAMODE	The data mode constant

datatype property

datatype

The data type constant is a constant which specifies different compression and buffer types. The most important one being COMP_RAW. If you want to obtain the name of the constant, use datatype.name.

Returns:

Type	Description
DATATYPE	The data type constant.

db_count property

db_count

Returns the number of data blocks.

Returns:

Type	Description
int	Number of data blocks.

db_header_size property

db_header_size

Each data block has a header. This property provides the size of the data block header in bytes. The sizes of all data block headers are equal.

Returns:

Type	Description
int	Size in bytes

db_sample_count property

db_sample_count

Returns the number of samples in a completely filled data block. It is calculated by dividing the data block size by the number of bytes per sample.

Returns:

Type	Description
int	Number of samples.

db_size property

db_size

Returns the size of a completely filled data block in bytes.

Returns:

Type	Description
int	Size of a completely filled data block.

db_spec_count property

db_spec_count

The number of spectra within a filled data block. It is calculated by dividing the number of samples in a data block by the number of frequency bands.

Returns:

Type	Description
int	Number of spectra in a filled data block.

fft_log_shift property

fft_log_shift

Base of the logarithm applied to the data. Please note that a 1 needs to be added to the value.

Returns:

Type	Description
int	Base of the logarithm.

filepath property

filepath

Returns the complete path to the file.

Returns:

Type	Description
str	Path to the file

frq_bands property

frq_bands: int

Each spectrum has a maximum sample number of 512. This is the maximum number of frequency bands. This figure decreases with compression along the frequency axis. The corresponding key word in the metainfo dictionary 's_p_fram' (samples per frame).

Returns:

Type	Description
int	Number of frequency bands.

frq_end property

frq_end

The property returns the highest represented frequency.

Returns:

Type	Description
int	Frequency in Hertz (Hz).

frq_per_band property

frq_per_band

The property returns the frequency range per band. The maximum number of bands is 512.

Returns:

Type	Description
float	Frequency range in Hertz (Hz) for one band.

frq_start property

frq_start

The property returns the lowest represented frequency.

Returns:

Type	Description
int	Frequency in Hertz (Hz).

full_blocks property

full_blocks

All but the last data block are usually completly filled. In order to calculate the number of those data blocks which are completely filled the header size of the file is substracted from the file size. This figure is divided by the sum of the data block header size and data block size and rounded down to the nearest integer.

Returns:

Type	Description
int	Number of full data blocks.

header_hash property

header_hash

A hash value for the buffer's header. The header of the buffer is almost unique for a buffer. A buffer should only differ in the length of the contained data.

Returns:

Type	Description
int	The header's hash

header_size property

header_size

Each buffer file has a header. This property provides the size of the header in bytes (The normal value is 2000 bytes).

Returns:

Type	Description
int	Size in bytes.

last_modification_date_time property

last_modification_date_time

last_modification_time property

last_modification_time

Returns the Posix timestamp of last modification of the file in seconds since Thursday, January 1st 1970.

Returns:

Type	Description
int	The timestamp as a number in seconds.

last_spec property

last_spec

max_amplitude_level property

max_amplitude_level

maximum possible amplitude of the buffer, relates to the buffer's dtype

Returns:

Type	Description
(int, float)

metainfo property

metainfo

The buffer header properties as a dictionary.

Returns:

Type	Description
dict	The metainfo dictionary

Example

key = 'qassdata'
def keyword(key)
    return buffer.metainfo[key]

# The above example leads to a runtime error if the keyword is not a
# key in the dictionary. It would be better to query this beforehand
# and to provide a default value, thus:

def keyword(key):
    if key in buffer.metainfo.keys:
        return buffer.metainfo[key]
    else:
        return default_value

# or shorter:

def keyword(key):
    return buffer.metainfo.get(key, default_value)

partnumber property

partnumber: str

The string representation of the partnumber of the measurement.

Returns:

Type	Description
str

preamp_gain property

preamp_gain

The preamplifier setting in the multiplexer tab of the Analyzer4D software.

Returns:

Type	Description
int

process property

process

The process number.

Returns:

Type	Description
int	The process number.

process_date_time property

process_date_time

Returns a timestamp of the creation of the file as a string.

Returns:

Type	Description
str	The timestamp in the form 'yyyy-mm-dd hhss'

process_measure_time_string property

process_measure_time_string

Returns the time in ISO format at measure start of the process. If epoctime (measure time) is not available due to an older buffer format, the creation time of the buffer is returned. Please note: The creation time may not be the same as epoc time. This is particularly true for compressed buffers which may be created much later!

Returns:

Type	Description
str	The timestamp in the form 'yyyy-mm-dd hhss'.

process_measure_timestamp property

process_measure_timestamp

Returns the Posix timestamp at measure start of the process in milliseconds. If epoctime (measure time) is not available due to an older buffer format, the creation time of the buffer is returned. Please note: The creation time may not be the same as epoc time. This is particularly true for compressed buffers which may be created much later!

Returns:

Type	Description
int	The timestamp as a number in seconds.

process_time property

process_time

Returns the Posix timestamp of creation of the file in seconds since Thursday, January 1st 1970. Please note that the creation time and measure time might be different.

Returns:

Type	Description
int	The timestamp as a number of seconds.

project_id property

project_id

refEnergy property

refEnergy

refEnergy provides a normalization factor for compressions. This makes sums of amplitudes of different compressions comparable to each other. See _signalNormalizationFactor for a detailed description.

Returns:

Type	Description
float

ref_energy property

ref_energy

alias for refEnergy()

Returns:

Type	Description
float

sample_count property

sample_count

The number of samples in the buffer. It is calculated by subtracting the header size and the data block size times the number of datablocks from the file size. This figure is divided by the bytes per sample to obtain the number of samples. The number is cast to an int as the division usually results in a float due to an incompletely filled final data block.

Returns:

Type	Description
int	Number of samples

spec_count property

spec_count

The number of spectra in the buffer. It is calculated by dividing the number of samples by the number of frequency bands.

Returns:

Type	Description
int	number of spectra

spec_duration property

spec_duration

The property returns the time for one spectrum in nanoseconds.

Returns:

Type	Description
float	Time in nanoseconds.

streamno property

streamno: Union[int, None]

The index of a stream of an extra channel.

Extra channels in the analyzer software can map multiple different data streams to the same channel of the software. This makes them only distuinguishable by this property.

ADCTYPE

Bases: IntEnum

Attributes:

Name	Type	Description
ADC_16BIT
ADC_24BIT
ADC_LEGACY_14BIT
ADC_NOT_USED

Source code in src/qass/tools/analyzer/buffer_parser.py

class ADCTYPE(IntEnum):
    ADC_NOT_USED = 0
    ADC_LEGACY_14BIT = 0
    ADC_16BIT = auto()
    ADC_24BIT = auto()

ADC_16BIT class-attribute instance-attribute

ADC_16BIT = auto()

ADC_24BIT class-attribute instance-attribute

ADC_24BIT = auto()

ADC_LEGACY_14BIT class-attribute instance-attribute

ADC_LEGACY_14BIT = 0

ADC_NOT_USED class-attribute instance-attribute

ADC_NOT_USED = 0

DATAKIND

Bases: IntEnum

Attributes:

Name	Type	Description
DATAKIND_CNT
KIND_CAN_NOT_BE_HANDLED
KIND_FREE_6
KIND_FREE_7
KIND_NONE
KIND_PATTERN_REF_OBJ_COMPR
KIND_PATTERN_REF_OBJ_EXTRA
KIND_PATTERN_REF_OBJ_MASK
KIND_PLOT_FREE_DATABLOCK_TIMING
KIND_SENSOR_TEST
KIND_UNDEF
KIND_USER

Source code in src/qass/tools/analyzer/buffer_parser.py

class DATAKIND(IntEnum):  # Zusätzliche Spezifikation des Buffers
    KIND_UNDEF = -1
    KIND_NONE = 0
    KIND_SENSOR_TEST = auto()  # Sensor Pulse Test Daten
    # Debug Plot Buffer, der die Zeiten für das "freimachen" eines Datenblocks enthält
    KIND_PLOT_FREE_DATABLOCK_TIMING = auto()
    KIND_PATTERN_REF_OBJ_COMPR = auto()
    KIND_PATTERN_REF_OBJ_MASK = auto()
    KIND_PATTERN_REF_OBJ_EXTRA = auto()
    KIND_FREE_6 = auto()
    KIND_FREE_7 = auto()
    DATAKIND_CNT = auto()
    # This datakind is out ouf range. It cannot be stored in bufferId anymore (this is legacy stuff)
    KIND_CAN_NOT_BE_HANDLED = DATAKIND_CNT

    # Werte ab hier zur freien Verwendung (Oft ist das ein Zähler für verschiedene Buffer, die ansonsten den selben FingerPrint hätten)
    KIND_USER = 100

DATAKIND_CNT class-attribute instance-attribute

DATAKIND_CNT = auto()

KIND_CAN_NOT_BE_HANDLED class-attribute instance-attribute

KIND_CAN_NOT_BE_HANDLED = DATAKIND_CNT

KIND_FREE_6 class-attribute instance-attribute

KIND_FREE_6 = auto()

KIND_FREE_7 class-attribute instance-attribute

KIND_FREE_7 = auto()

KIND_NONE class-attribute instance-attribute

KIND_NONE = 0

KIND_PATTERN_REF_OBJ_COMPR class-attribute instance-attribute

KIND_PATTERN_REF_OBJ_COMPR = auto()

KIND_PATTERN_REF_OBJ_EXTRA class-attribute instance-attribute

KIND_PATTERN_REF_OBJ_EXTRA = auto()

KIND_PATTERN_REF_OBJ_MASK class-attribute instance-attribute

KIND_PATTERN_REF_OBJ_MASK = auto()

KIND_PLOT_FREE_DATABLOCK_TIMING class-attribute instance-attribute

KIND_PLOT_FREE_DATABLOCK_TIMING = auto()

KIND_SENSOR_TEST class-attribute instance-attribute

KIND_SENSOR_TEST = auto()

KIND_UNDEF class-attribute instance-attribute

KIND_UNDEF = -1

KIND_USER class-attribute instance-attribute

KIND_USER = 100

DATAMODE

Bases: IntEnum

Attributes:

Name	Type	Description
DATAMODE_COUNT
DATAMODE_COUNTER_UNUSED
DATAMODE_FFT
DATAMODE_OTHER
DATAMODE_PLOT
DATAMODE_SIGNAL
DATAMODE_UNDEF
DATAMODE_VIDEO

Source code in src/qass/tools/analyzer/buffer_parser.py

class DATAMODE(IntEnum):
    DATAMODE_UNDEF = -1
    # Es wird nur ein Zähler übertragen, der im DSP Modul generiert wird
    DATAMODE_COUNTER_UNUSED = 0
    DATAMODE_SIGNAL = (
        auto()
    )  # Es werden die reinen Signaldaten gemessen und übertragen
    # Die FFT wird in der Hardware durchgeführt und das Ergebnis als FFT Daten übertragen
    DATAMODE_FFT = auto()
    DATAMODE_PLOT = auto()  # 2 dimensionale Graph Daten (war INTERLEAVED, das wird nicht mehr genutzt, taucht aber im kernelmodul als Define für DATAMODES noch auf
    # Datenmodus, der nur in importierten oder künstlichen Buffern auftritt
    DATAMODE_OTHER = auto()
    # Datamode is video (This means file is not a normal buffer, but simply a video file)
    DATAMODE_VIDEO = auto()

    DATAMODE_COUNT = auto()

DATAMODE_COUNT class-attribute instance-attribute

DATAMODE_COUNT = auto()

DATAMODE_COUNTER_UNUSED class-attribute instance-attribute

DATAMODE_COUNTER_UNUSED = 0

DATAMODE_FFT class-attribute instance-attribute

DATAMODE_FFT = auto()

DATAMODE_OTHER class-attribute instance-attribute

DATAMODE_OTHER = auto()

DATAMODE_PLOT class-attribute instance-attribute

DATAMODE_PLOT = auto()

DATAMODE_SIGNAL class-attribute instance-attribute

DATAMODE_SIGNAL = auto()

DATAMODE_UNDEF class-attribute instance-attribute

DATAMODE_UNDEF = -1

DATAMODE_VIDEO class-attribute instance-attribute

DATAMODE_VIDEO = auto()

DATATYPE

Bases: IntEnum

Attributes:

Name	Type	Description
COMP_ANALYZE_OVERVIEW
COMP_AUDIO_COMMENT
COMP_AUDIO_RAW
COMP_AVERAGE
COMP_COLLECTION
COMP_COUNT
COMP_CP_ENERGY_SIG
COMP_CSV_EXPORT
COMP_DOWNSAMPLE
COMP_ENERGY
COMP_ENVELOP
COMP_ENVELOPE_LOWER
COMP_ENVELOPE_UPPER
COMP_EXTERN_DATA
COMP_GRADIENT_FRQ
COMP_IMPORT_SIG
COMP_INVALID
COMP_IO_SIGNAL
COMP_MAXIMUM
COMP_MOV_AVERAGE
COMP_MOV_AVERAGE_FRQ
COMP_MOV_AVERAGE_OPT
COMP_OTHER
COMP_PATTERN_MASK
COMP_PATTERN_REFOBJ
COMP_RAW
COMP_SCOPE_RAW
COMP_SECOND_FFT
COMP_SIGNIFICANCE
COMP_SIGNIFICANCE_32
COMP_SLOPECHANGE
COMP_STD_DEVIATION
COMP_UNDEF
COMP_VID_EXT_LINK
COMP_VID_MEASURE
COMP_VID_SCREEN
COMP_XY_PLOT

Source code in src/qass/tools/analyzer/buffer_parser.py

class DATATYPE(IntEnum):  # Kompressionsmethoden oder auch Buffertypen
    COMP_INVALID = -2
    COMP_UNDEF = -1
    COMP_RAW = (
        0  # Die reinen unkomprimierten Rohdaten, sowie sie aus der Hardware kommen
    )
    # Datenreduktion durch einfaches Downsampling (jedes x-te Sample gelangt in den Buffer)
    COMP_DOWNSAMPLE = auto()
    COMP_MAXIMUM = (
        auto()
    )  # Die Maximalwerte von jeweils x Samples gelangen in den Buffer
    # Die Durchschnittswerte über jeweils x Samples gelangen in den Buffer
    COMP_AVERAGE = auto()
    COMP_STD_DEVIATION = auto()  # Die Standardabweichung
    COMP_ENVELOP = auto()  # NOT USED!! Der Buffer stellt eine Hüllkurve dar
    COMP_MOV_AVERAGE = (
        auto()
    )  # Der gleitende Mittelwert über eine Blockgröße von x Samples
    # Die Bufferdaten wurden aus einer externen Quelle, die nicht mit dem Optimizer aufgezeichnet wurden erstellt
    COMP_EXTERN_DATA = auto()
    # Zur Zeit verwendet, um MinMaxObjekt Energy signature buffer zu taggen
    COMP_ANALYZE_OVERVIEW = auto()
    # Der gleitende Mittelwert über eine Blockgröße von x Samples (optimierte Berechnung)
    COMP_MOV_AVERAGE_OPT = auto()
    COMP_COLLECTION = auto()  # Wild zusammengeworfene Datenmasse
    COMP_IMPORT_SIG = auto()  # Importierte Signaldaten
    COMP_SCOPE_RAW = auto()  # Vom Oszilloskop aufgezeichnet
    # Der gleitende Mittelwert (Zeit und Frequenz) über eine Blockgröße von x Samples
    COMP_MOV_AVERAGE_FRQ = auto()
    COMP_SLOPECHANGE = auto()  # Steigungswechsel der Amplituden über die Zeit
    COMP_OTHER = auto()
    COMP_IO_SIGNAL = auto()  # Aufgezeichnete IO Signale
    COMP_ENERGY = auto()  # Die Energie (in erster Linie für DM=ANALOG)
    COMP_AUDIO_RAW = (
        auto()
    )  # Raw Daten, die mit dem Sound Device aufgezeichnet wurden
    # was COMP_OBJECT before. Now it is used for Datamode PLOT. Type will be displayed as Graphname, if set
    COMP_XY_PLOT = auto()
    COMP_SECOND_FFT = auto()
    # Raw Daten vom Audio Device, bei denen es sich um einen Audiokommentar handelt
    COMP_AUDIO_COMMENT = auto()
    # CrackProperties Energy Signatur, kommt in erster Linie von Energieprofilen, die auf Daten der CrackDefinitionen Berechnet wurden
    COMP_CP_ENERGY_SIG = auto()
    # This is a video stream that has been recorded while measuring
    COMP_VID_MEASURE = auto()
    # This is a screen cast video stream, that has been captured while measuring or session recording
    COMP_VID_SCREEN = auto()
    COMP_VID_EXT_LINK = auto()  # This is an extern linked video file
    COMP_ENVELOPE_UPPER = auto()  # NOT_USED!! Obere Hüllfläche
    COMP_ENVELOPE_LOWER = auto()  # NOT_USED! Untere Hüllfläche
    COMP_PATTERN_REFOBJ = auto()  # NOT USED! A Pattern Recognition reference object
    COMP_SIGNIFICANCE = auto()  # Nur starke Änderungen werden aufgezeichnet
    # this is the signed 32 bit version of a significance buffer
    COMP_SIGNIFICANCE_32 = auto()
    # NOT_USED! A mask buffer for a pattern ref object (likely this is a float buffer)
    COMP_PATTERN_MASK = auto()
    COMP_GRADIENT_FRQ = auto()  # Gradient buffer in frequency direction
    # Not realy a datatype but used to create CSV files from source buffer
    COMP_CSV_EXPORT = auto()
    # Die Anzahl der Einträge in der Datatypesliste, kein wirklicher Datentyp
    COMP_COUNT = auto()

COMP_ANALYZE_OVERVIEW class-attribute instance-attribute

COMP_ANALYZE_OVERVIEW = auto()

COMP_AUDIO_COMMENT class-attribute instance-attribute

COMP_AUDIO_COMMENT = auto()

COMP_AUDIO_RAW class-attribute instance-attribute

COMP_AUDIO_RAW = auto()

COMP_AVERAGE class-attribute instance-attribute

COMP_AVERAGE = auto()

COMP_COLLECTION class-attribute instance-attribute

COMP_COLLECTION = auto()

COMP_COUNT class-attribute instance-attribute

COMP_COUNT = auto()

COMP_CP_ENERGY_SIG class-attribute instance-attribute

COMP_CP_ENERGY_SIG = auto()

COMP_CSV_EXPORT class-attribute instance-attribute

COMP_CSV_EXPORT = auto()

COMP_DOWNSAMPLE class-attribute instance-attribute

COMP_DOWNSAMPLE = auto()

COMP_ENERGY class-attribute instance-attribute

COMP_ENERGY = auto()

COMP_ENVELOP class-attribute instance-attribute

COMP_ENVELOP = auto()

COMP_ENVELOPE_LOWER class-attribute instance-attribute

COMP_ENVELOPE_LOWER = auto()

COMP_ENVELOPE_UPPER class-attribute instance-attribute

COMP_ENVELOPE_UPPER = auto()

COMP_EXTERN_DATA class-attribute instance-attribute

COMP_EXTERN_DATA = auto()

COMP_GRADIENT_FRQ class-attribute instance-attribute

COMP_GRADIENT_FRQ = auto()

COMP_IMPORT_SIG class-attribute instance-attribute

COMP_IMPORT_SIG = auto()

COMP_INVALID class-attribute instance-attribute

COMP_INVALID = -2

COMP_IO_SIGNAL class-attribute instance-attribute

COMP_IO_SIGNAL = auto()

COMP_MAXIMUM class-attribute instance-attribute

COMP_MAXIMUM = auto()

COMP_MOV_AVERAGE class-attribute instance-attribute

COMP_MOV_AVERAGE = auto()

COMP_MOV_AVERAGE_FRQ class-attribute instance-attribute

COMP_MOV_AVERAGE_FRQ = auto()

COMP_MOV_AVERAGE_OPT class-attribute instance-attribute

COMP_MOV_AVERAGE_OPT = auto()

COMP_OTHER class-attribute instance-attribute

COMP_OTHER = auto()

COMP_PATTERN_MASK class-attribute instance-attribute

COMP_PATTERN_MASK = auto()

COMP_PATTERN_REFOBJ class-attribute instance-attribute

COMP_PATTERN_REFOBJ = auto()

COMP_RAW class-attribute instance-attribute

COMP_RAW = 0

COMP_SCOPE_RAW class-attribute instance-attribute

COMP_SCOPE_RAW = auto()

COMP_SECOND_FFT class-attribute instance-attribute

COMP_SECOND_FFT = auto()

COMP_SIGNIFICANCE class-attribute instance-attribute

COMP_SIGNIFICANCE = auto()

COMP_SIGNIFICANCE_32 class-attribute instance-attribute

COMP_SIGNIFICANCE_32 = auto()

COMP_SLOPECHANGE class-attribute instance-attribute

COMP_SLOPECHANGE = auto()

COMP_STD_DEVIATION class-attribute instance-attribute

COMP_STD_DEVIATION = auto()

COMP_UNDEF class-attribute instance-attribute

COMP_UNDEF = -1

COMP_VID_EXT_LINK class-attribute instance-attribute

COMP_VID_EXT_LINK = auto()

COMP_VID_MEASURE class-attribute instance-attribute

COMP_VID_MEASURE = auto()

COMP_VID_SCREEN class-attribute instance-attribute

COMP_VID_SCREEN = auto()

COMP_XY_PLOT class-attribute instance-attribute

COMP_XY_PLOT = auto()

__enter__

__enter__()

Source code in src/qass/tools/analyzer/buffer_parser.py

def __enter__(self):
    self.__last_spectrum = None
    self.file = open(self.__filepath, "rb")
    self._parse_header()
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)

Source code in src/qass/tools/analyzer/buffer_parser.py

def __exit__(self, exc_type, exc_val, exc_tb):
    self.file.close()

__getstate__

__getstate__()

Return path to buffer file, since this is everything needed to reconstruct a buffer object. This function is needed for pickling buffer objects.

Returns:

Type	Description
tuple	Single-element tuple containing the path to the buffer file.

Source code in src/qass/tools/analyzer/buffer_parser.py

def __getstate__(self):
    """
    Return path to buffer file, since this is everything needed to reconstruct a buffer object.
    This function is needed for pickling buffer objects.

    Returns
    -------
    tuple
        Single-element tuple containing the path to the buffer file.
    """
    return (self.__filepath,)

__init__

__init__(filepath)

Source code in src/qass/tools/analyzer/buffer_parser.py

def __init__(self, filepath):
    self.__filepath = filepath
    # uint: read with np.uintc and then cast to python int
    # c double is python float
    import numpy as np

    int(np.uintc())
    self.__header_hash = None
    self.__keywords = [
        ("qassdata----", HeaderDtype.INT32),
        ("filevers----", HeaderDtype.INT32),
        ("datavers----", HeaderDtype.INT32),
        ("savefrom----", HeaderDtype.INT32),
        ("datamode----", HeaderDtype.INT32),
        ("datatype----", HeaderDtype.INT32),
        ("datakind----", HeaderDtype.INT32),
        ("framsize----", HeaderDtype.INT32),
        ("smplsize----", HeaderDtype.INT32),
        ("frqbands----", HeaderDtype.INT32),
        ("db_words----", HeaderDtype.INT32),
        ("avgtimba----", HeaderDtype.INT32),
        ("avgfrqba----", HeaderDtype.INT32),
        ("m_u_mask--------", HeaderDtype.UINT64),
        ("b_p_samp----", HeaderDtype.INT32),
        ("s_p_fram----", HeaderDtype.INT32),
        ("db__size----", HeaderDtype.INT32),
        ("max_ampl----", HeaderDtype.INT32),
        ("nul_ampl----", HeaderDtype.INT32),
        ("samplert----", HeaderDtype.INT32),
        ("datarate--------", HeaderDtype.DOUBLE),
        ("samplefr----", HeaderDtype.INT32),
        ("frqshift----", HeaderDtype.INT32),
        ("fftovers----", HeaderDtype.INT32),
        ("fftlogsh----", HeaderDtype.INT32),
        ("fftwinfu----", HeaderDtype.INT32),
        ("dbhdsize----", HeaderDtype.INT32),
        ("comratio----", HeaderDtype.INT32),
        ("tc__real--------", HeaderDtype.DOUBLE),
        ("frqratio----", HeaderDtype.INT32),
        ("fc__real--------", HeaderDtype.DOUBLE),
        ("proj__id--------", HeaderDtype.UINT64),
        ("file__id--------", HeaderDtype.UINT64),
        ("parentid--------", HeaderDtype.INT64),
        ("proc_cnt----", HeaderDtype.INT32),
        ("proc_rng----", HeaderDtype.INT32),
        ("proc_sub----", HeaderDtype.INT32),
        ("poly_cnt----", HeaderDtype.INT32),
        ("polycyid----", HeaderDtype.INT32),
        ("dumpchan----", HeaderDtype.INT32),
        ("del_lock----", HeaderDtype.INT32),
        ("proctime----", HeaderDtype.UINT32),
        ("lmodtime----", HeaderDtype.UINT32),
        ("epoctime--------", HeaderDtype.INT64),
        ("mux_port----", HeaderDtype.INT32),
        ("pampgain----", HeaderDtype.INT32),
        ("dispgain----", HeaderDtype.INT32),
        ("linfgain----", HeaderDtype.INT32),
        ("auxpara0----", HeaderDtype.INT32),
        ("auxpara1----", HeaderDtype.INT32),
        ("auxpara2----", HeaderDtype.INT32),
        ("auxpara3----", HeaderDtype.INT32),
        ("auxpara4----", HeaderDtype.INT32),
        ("auxpara5--------", HeaderDtype.INT64),
        ("skipsamp--------", HeaderDtype.INT64),
        ("skiptime--------", HeaderDtype.INT64),
        ("trunsamp--------", HeaderDtype.INT64),
        ("truntime--------", HeaderDtype.INT64),
        ("skiplfrq----", HeaderDtype.INT32),
        ("trunhfrq----", HeaderDtype.INT32),
        ("startfrq----", HeaderDtype.INT32),
        ("end__frq----", HeaderDtype.INT32),
        ("frqpband--------", HeaderDtype.DOUBLE),
        ("framedur--------", HeaderDtype.DOUBLE),
        ("frameoff----", HeaderDtype.INT32),
        ("p__flags----", HeaderDtype.INT32),
        ("realfrqc----", HeaderDtype.INT32),
        ("sub_data----", HeaderDtype.INT32),
        ("sd_dimen----", HeaderDtype.INT32),
        ("sd_rsize----", HeaderDtype.INT32),
        ("sd_rsizf--------", HeaderDtype.DOUBLE),
        ("sd_dsize----", HeaderDtype.INT32),
        ("frqmasks----", HeaderDtype.UINT32),
        ("frqinmas----", HeaderDtype.UINT32),
        ("eheaderf----", HeaderDtype.UINT32),
        ("adc_type----", HeaderDtype.INT32),
        ("adbitres----", HeaderDtype.INT32),
        ("baserate--------", HeaderDtype.INT64),
        ("interpol----", HeaderDtype.INT32),
        ("dcoffset----", HeaderDtype.INT32),
        ("dispralo----", HeaderDtype.INT32),
        ("disprahi----", HeaderDtype.INT32),
        ("rngtibeg--------", HeaderDtype.INT64),
        ("rngtiend--------", HeaderDtype.INT64),
        ("realsoff--------", HeaderDtype.INT64),
        ("extendid--------", HeaderDtype.UINT64),
        ("sim_mode----", HeaderDtype.INT32),
        ("partnoid----", HeaderDtype.UINT32),
        ("asc_part----", HeaderDtype.UINT32),
        ("asc_desc----", HeaderDtype.UINT32),
        ("comments----", HeaderDtype.UINT32),
        ("sparemem----", HeaderDtype.UINT32),
        ("streamno----", HeaderDtype.UINT32),
        ("an4dvers----", HeaderDtype.HEX_STRING),
        ("headsend", None),
    ]

    self.__db_keywords = [
        ("blochead----", HeaderDtype.INT32),
        ("firstsam--------", HeaderDtype.INT64),
        ("lastsamp--------", HeaderDtype.INT64),
        ("dbfilled----", HeaderDtype.INT32),
        ("mux_port----", HeaderDtype.INT32),
        ("pampgain----", HeaderDtype.INT32),
        ("dispgain----", HeaderDtype.INT32),
        ("io_ports----", HeaderDtype.INT32),
        ("dversion----", HeaderDtype.INT32),
        ("sine_frq----", HeaderDtype.INT32),
        ("sine_amp----", HeaderDtype.INT32),
        ("sd_dimen----", HeaderDtype.INT32),
        ("sd_rsize----", HeaderDtype.INT32),
        ("sd_dsize----", HeaderDtype.INT32),
        ("begin_subdat", HeaderDtype.INT32),
        ("end___subdat", HeaderDtype.INT32),
        (
            "blockend",
            HeaderDtype.INT32,
        ),  # datatypes for last three keywords guessed
    ]
    self.__metainfo = {}
    self.__db_headers = {}

__setstate__

__setstate__(state: Tuple)

Reconstruct buffer from state, which only contains the path to the buffer file. This function is needed for unpickling buffer objects.

Parameters:

Name	Type	Description	Default
state	Tuple	Single-element tuple containing the path to the buffer file.	required

Source code in src/qass/tools/analyzer/buffer_parser.py

def __setstate__(self, state: Tuple):
    """
    Reconstruct buffer from state, which only contains the path to the buffer file.
    This function is needed for unpickling buffer objects.

    Parameters
    ----------
    state: tuple
        Single-element tuple containing the path to the buffer file.
    """
    self.__init__(*state)

block_infos

block_infos(columns: List[str] = ['preamp_gain', 'mux_port', 'measure_positions', 'inputs', 'outputs'], changes_only: bool = False, fast_jump: bool = True)

block_infos iterates through all memory blocks of a buffer file (typically one MB) and fetches the subdata information each memory block has one set of metadata but e.g. 65 subdata entries for raw files or more than 2000 entries for a 32 times compressed file, where each entry contains meta information about a small time frame (15 spectrums in raw fft files)

Parameters:

Name	Type	Description	Default
columns	list[str]	List of columns that should be in the result_array. You can define the order of the columns here. Possible column names are: ['preamp_gain', 'mux_port', 'measure_positions', 'inputs', 'outputs', 'times', 'index', 'spectrums']	['preamp_gain', 'mux_port', 'measure_positions', 'inputs', 'outputs']
changes_only	bool	Flag to enable a summarized output. If True the output does not contain all entries but only entries where the data columns changed. In the current implementation the memory consumption does not change here -> in both cases the array is first completely built. Defaults to False.	False
fast_jump	bool	If True the function uses seeking in the file instead of parsing every single datablock header. The offsets are investigated for the first datablock header and simply applied for all other datablock headers. Seeking is usually faster than parsing the datablock headers if they are not already in the cache. If the datablock headers are already cached this flag has no effect. Defaults to true.	True

Returns:

Name	Type	Description
header_infos	NDArray	An array containing (spec_index), (index), (times), preamp_gain, mux_port, measure_position, 24bit input, 16bit output, (times), (index), (spec_index)

Source code in src/qass/tools/analyzer/buffer_parser.py

def block_infos(
    self,
    columns: List[str] = [
        "preamp_gain",
        "mux_port",
        "measure_positions",
        "inputs",
        "outputs",
    ],
    changes_only: bool = False,
    fast_jump: bool = True,
):
    """block_infos iterates through all memory blocks of a buffer file (typically one MB) and fetches the subdata information
    each memory block has one set of metadata but e.g. 65 subdata entries for raw files
    or more than 2000 entries for a 32 times compressed file,
    where each entry contains meta information about a small time frame (15 spectrums in raw fft files)

    Parameters
    ----------
    columns : list[str]
        List of columns that should be in the result_array.
        You can define the order of the columns here.
        Possible column names are: ['preamp_gain', 'mux_port', 'measure_positions', 'inputs', 'outputs', 'times', 'index', 'spectrums']
    changes_only : bool, optional
        Flag to enable a summarized output. If True the output does not contain all entries but only entries where the data columns changed.
        In the current implementation the memory consumption does not change here -> in both cases the array is first completely built.
        Defaults to False.
    fast_jump : bool, optional
        If True the function uses seeking in the file instead of parsing every single datablock header.
        The offsets are investigated for the first datablock header and simply applied for all other datablock headers.
        Seeking is usually faster than parsing the datablock headers if they are not already in the cache.
        If the datablock headers are already cached this flag has no effect.
        Defaults to true.

    Returns
    -------
    header_infos : npt.NDArray
        An array containing (spec_index), (index), (times), preamp_gain, mux_port, measure_position, 24bit input, 16bit output, (times), (index), (spec_index)
    """

    data_columns = [
        "preamp_gain",
        "mux_port",
        "measure_positions",
        "inputs",
        "outputs",
    ]
    data_columns_indices = [0, 1, 2, 3, 5]  # positions in subdat block
    index_columns = ["times", "index", "spectrums"]

    allowed_columns = data_columns + index_columns

    for col in columns:
        if col not in allowed_columns:
            raise InvalidArgumentError(
                f"Column {col} is not a valid column name. Valid columns are: {allowed_columns}"
            )
        if columns.count(col) > 1:
            raise InvalidArgumentError(
                f"Column {col} is used multiple times. This is not allowed."
            )

    if changes_only:
        if not any(col in index_columns for col in columns):
            raise InvalidArgumentError(
                "If you use the changes_only option you need to declare at least one index column. Otherwise the results would have no connection."
            )

    # interpreting the entries as int32 makes most sense
    last_sample = 0
    # old header size was 10+32bit entries
    ds_size = 10
    mi = self.db_header(0)
    if "begin_subdat" not in mi:
        raise ValueError(
            f"begin_subdat keyword missing in datablock {0} -> caonnot read IO information"
        )

    subdat = np.frombuffer(mi["begin_subdat"], dtype=np.int32)
    # if it is of extended type, we expect a reasonable value here
    mysize = subdat[10].item()

    if mysize == 80:  # the extended data length, additional sizes may occur
        ds_size = 20  # again the size in 32bit entries

    entries_before = 0

    if fast_jump:
        db_start_pos = self._get_datablock_start_pos(0)
        self.file.seek(db_start_pos, os.SEEK_SET)
        db_header_content = self.file.read(self.__db_header_size)
        start_mark = b"begin_subdat"
        end_mark = b"end___subdat"
        try:
            subdat_offset_start = db_header_content.index(start_mark) + len(
                start_mark
            )
            subdat_offset_end = db_header_content.index(end_mark)
            subdat_length = subdat_offset_end - subdat_offset_start
            subdat_readlen = int(subdat_length / subdat.itemsize)
        except ValueError as e:
            raise ValueError(
                f"The datablock header seems not to have a subdat block. Reading block info not possibe. {str(e)}"
            )

    samples_per_entry = mi["sd_rsize"] / self.bytes_per_sample
    specs_per_entry = samples_per_entry / self.frq_bands
    entries_per_spec = 1 / specs_per_entry

    # columns of interest:
    # pgain, mux_port, measure_position, 24bit input (inverted), 16bit output (inverted)
    data_columns_src = []  # column index in the buffers meta info array
    data_columns_dest = []  # column index in the target result_array

    times_col = None
    index_col = None
    specs_col = None

    for idx, col in enumerate(columns):
        if col == "times":
            times_col = idx
        elif col == "index":
            index_col = idx
        elif col == "spectrums":
            specs_col = idx
        elif col in data_columns:
            data_columns_src.append(data_columns_indices[data_columns.index(col)])
            data_columns_dest.append(idx)

    if not data_columns_dest:
        raise InvalidArgumentError("You have to choose at least one data column!")

    # the indices of valid entries in the subdata block are calculated per datablock (checked in Analyzer4D DBgrabber::getSubDataPointer)
    full_datablock_count = self.spec_count // self.db_spec_count
    full_datablock_specs = full_datablock_count * self.db_spec_count
    last_nonfull_datablock_specs = self.spec_count % self.db_spec_count
    entry_count = int(full_datablock_specs * min(1, entries_per_spec)) + int(
        last_nonfull_datablock_specs * min(1, entries_per_spec)
    )

    # 64 bit array if index columns are requested, otherwise 32 bit
    arr_type = (
        np.int64 if any((times_col, index_col, specs_col)) is not None else np.int32
    )
    result_arr = np.empty(
        (entry_count, len(columns)), dtype=arr_type
    )  # allocating the array!

    # loop through the datablocks
    for i in range(self.db_count):
        if fast_jump and i not in self.__db_headers:
            db_start_pos = self._get_datablock_start_pos(i)
            self.file.seek(db_start_pos + subdat_offset_start, os.SEEK_SET)
            entries = np.fromfile(
                self.file, dtype=np.int32, count=subdat_readlen
            ).reshape(-1, ds_size)
            start_spec = self._first_spec_of_datablock(i)
            end_spec = min(self._first_spec_of_datablock(i + 1), self.spec_count)
        else:
            mi = self.db_header(i)
            if "begin_subdat" not in mi:
                raise ValueError(
                    f"begin_subdat keyword missing in datablock {i} -> caonnot read IO information"
                )

            subdat = np.frombuffer(mi["begin_subdat"], dtype=np.int32)
            entries = subdat.reshape(-1, ds_size)
            # f_entries=np.empty((0,ds_size),dtype=np.int32)

            # print("mi['sd_rsize']", mi['sd_rsize'])
            samples_per_entry = mi["sd_rsize"] / self.bytes_per_sample
            specs_per_entry = samples_per_entry / self.frq_bands
            entries_per_spec = 1 / specs_per_entry

            start_spec = int(mi["firstsam"] / self.frq_bands)
            # the end_spec might differ from lastsamp for the last datablock: The real data might be shorter
            end_spec = min(int(mi["lastsamp"] / self.frq_bands), self.spec_count)

        db_spec_count = end_spec - start_spec

        # the number of expected entries might be less than specs for low compressions, but never more than db_spec_count
        entries_expected = int(db_spec_count * min(1, entries_per_spec))

        # max(1, entries_per_spec) because for low compressions we get less than 1 entry per spec -> take every entry
        idxs = np.arange(entries_expected) * max(1, entries_per_spec)
        idxs = idxs.astype(int)

        # filter the entries based on the calculated indexes
        f_entries = entries[idxs]

        assert (
            len(f_entries) == entries_expected
        )  # this must be equal - its critical to have a mismatch here!

        # writing columns_of_interest into the result_arr
        result_arr[
            entries_before : entries_before + len(f_entries), data_columns_dest
        ] = f_entries[:, data_columns_src]

        if times_col is not None:
            start_time = start_spec * self.spec_duration
            end_time = end_spec * self.spec_duration
            times = np.arange(entries_expected, dtype=float)

            # for strong compresssions we get exactly one (and never more than one) entries for each spectrum.
            times *= self.spec_duration / min(1, entries_per_spec)
            times += start_time

            times = times.astype(int)
            result_arr[entries_before : entries_before + len(times), times_col] = (
                times
            )

        if index_col is not None:
            index = np.arange(
                entries_before, entries_before + entries_expected, dtype=int
            )
            result_arr[entries_before : entries_before + len(index), index_col] = (
                index
            )

        if specs_col is not None:
            index = np.arange(entries_expected) * max(1, specs_per_entry)
            index = index.astype(int)
            index += start_spec
            result_arr[entries_before : entries_before + len(index), specs_col] = (
                index
            )

        entries_before += entries_expected  # len(f_value_list)

    assert (
        entries_before == len(result_arr)
    )  # If this fails we did something wrong when calculating the expected number of entries.

    # Some conversions to ensure readability of the values
    if "preamp_gain" in columns:
        col = columns.index("preamp_gain")
        result_arr[:, col] = result_arr[:, col] >> 16

    if "inputs" in columns:
        col = columns.index("inputs")
        result_arr[:, col] = np.bitwise_and(
            np.bitwise_not(result_arr[:, col]), 0xFFFFFF
        )

    if "outputs" in columns:
        col = columns.index("outputs")
        result_arr[:, col] = np.bitwise_and(
            np.bitwise_not(result_arr[:, col]), 0xFFFF
        )

    if changes_only:
        changes_idx = np.concatenate(
            (
                (True,),
                np.any(
                    result_arr[1:, data_columns_dest]
                    != result_arr[:-1, data_columns_dest],
                    axis=1,
                ),
            )
        )
        return result_arr[changes_idx]
    else:
        return result_arr

db_header

db_header(db_idx)

The data block header contains key words and their corresponding values. This information is retrieved and stored in a dictionary. The values can be accessed by providing the corresponding key. Please ensure that the key exists in the database prior to its use or use the get method to access its content.Otherwise this will cause a runtime error.

Parameters:

Name	Type	Description	Default
db_idx	int	data block index	required

Raises:

Type	Description
ValueError	The data block index is out of range.

Returns:

Type	Description
dict	a dictionary with the keywords and values

Example

# The index of the first data block is 0
db_idx = 0
# Function to retrieve the size of the data block
def db_size(db_idx):
    db_header_dict = db_header(0)
    return(db_header_dict.get('dbfilled', 0))

Source code in src/qass/tools/analyzer/buffer_parser.py

def db_header(self, db_idx):
    """
    The data block header contains key words and their corresponding
    values. This information is retrieved and stored in a dictionary. The
    values can be accessed by providing the corresponding key. Please
    ensure that the key exists in the database prior to its use or
    use the get method to access its content.Otherwise this will
    cause a runtime error.


    Parameters
    ----------
    db_idx : int
        data block index

    Raises
    ------
    ValueError
        The data block index is out of range.

    Returns
    -------
    dict
        a dictionary with the keywords and values

    Example
    -------
    ```py
    # The index of the first data block is 0
    db_idx = 0
    # Function to retrieve the size of the data block
    def db_size(db_idx):
        db_header_dict = db_header(0)
        return(db_header_dict.get('dbfilled', 0))
    ```
    """
    if db_idx > self.__db_count:
        raise ValueError("db_idx is out of bounds")

    if db_idx in self.__db_headers:
        return self.__db_headers[db_idx]

    header_start_pos = self._get_datablock_start_pos(db_idx)

    self.file.seek(header_start_pos, os.SEEK_SET)
    db_header_content = self.file.read(self.__db_header_size)
    db_metainfo = self._parse_db_header(db_header_content)
    self.__db_headers[db_idx] = db_metainfo
    return db_metainfo

db_header_spec

db_header_spec(spec: int)

The method returns the data block header as a dictionary of the data block containing the spectrum whose index is provided.

Parameters:

Name	Type	Description	Default
spec	int	The index of a spectrum.	required

Returns:

Type	Description
dict	The data block header with keywords and values.

Example

# The index of the spectrum is 50
spec = 50
# Function to retrieve the size of the data block
def db_size(spec):
    db_header_dict = db_header_spec(spec)
    return(db_header_dict.get('dbfilled', 0))

Source code in src/qass/tools/analyzer/buffer_parser.py

def db_header_spec(self, spec: int):
    """
    The method returns the data block header as a dictionary of the data
    block containing the spectrum whose index is provided.

    Parameters
    ----------
    spec: int
        The index of a spectrum.

    Returns
    -------
    dict
        The data block header with keywords and values.

    Example
    -------
    ```py
    # The index of the spectrum is 50
    spec = 50
    # Function to retrieve the size of the data block
    def db_size(spec):
        db_header_dict = db_header_spec(spec)
        return(db_header_dict.get('dbfilled', 0))
    ```
    """
    db_idx = int(spec * self.__frq_bands / self.db_sample_count)
    return self.db_header(db_idx)

db_value

db_value(db_idx: int, key: str)

Similar to the file header each data block has its own header. This method returns the value of the property specified by key in the specified data block. The first data block has the index 0.

Parameters:

Name	Type	Description	Default
db_idx	int	The index of the data block.	required
key	str	The string containing the key word.	required

Returns:

Type	Description
int	The data block value for the given key.

Source code in src/qass/tools/analyzer/buffer_parser.py

def db_value(self, db_idx: int, key: str):
    """
    Similar to the file header each data block has its own header. This
    method returns the value of the property specified by key in the
    specified data block. The first data block has the index 0.

    Parameters
    ----------
    db_idx : int
        The index of the data block.
    key: str
        The string containing the key word.

    Returns
    -------
    int
        The data block value for the given key.
    """
    return self.db_header(db_idx)[key]

delog staticmethod

delog(data_arr, fft_log_shift, ad_bit_resolution)

Method to de-logarithmize a data array. Usually when data are measured with the Optimizer the data are logarithmized. The problem with logarithms is that adding them up constitutes a multiplication of the non logarithmized data. So in order to carry out calculations such as calculating sums etc. it is crucial to use de-logarithmized data.

Parameters:

Name	Type	Description	Default
data_arr	NDArray	Array with the logarithmized data.	required
fft_log_shift	int	Base of the logarithm.	required
ad_bit_resolution	int	The bit resolution (usually 16).	required

Returns:

Type	Description
NDArray[floating]	Array with de-logarithmized data.

Source code in src/qass/tools/analyzer/buffer_parser.py

@staticmethod
def delog(data_arr, fft_log_shift, ad_bit_resolution):
    """
    Method to de-logarithmize a data array. Usually when data are measured
    with the Optimizer the data are logarithmized. The problem with
    logarithms is that adding them up constitutes a multiplication of the
    non logarithmized data. So in order to carry out calculations such as
    calculating sums etc. it is crucial to use de-logarithmized data.

    Parameters
    ----------
    data_arr : npt.NDArray
        Array with the logarithmized data.
    fft_log_shift : int
        Base of the logarithm.
    ad_bit_resolution : int
        The bit resolution (usually 16).

    Returns
    -------
    npt.NDArray[np.floating]
        Array with de-logarithmized data.
    """
    shift = fft_log_shift - 1
    bits = ad_bit_resolution if ad_bit_resolution >= 1 else 16
    bit_res_idx = [14, 16, 24].index(bits)

    # remember negative values:
    negative = data_arr < 0
    np.abs(data_arr, out=data_arr)

    # ensure data_arr to be a floating point array
    data_arr = data_arr.astype(np.double, copy=False)

    shift_offset = [23, 23, 31][bit_res_idx]
    max_amp = 1 << bits

    data_arr = data_arr * (shift_offset - shift) / max_amp + shift

    data_arr = (2**data_arr) - (1 << shift)
    data_arr[negative] *= -1

    return data_arr

file_size

file_size()

The file size of the buffer file. The value is not stored in the header but retrieved using the operating system.

Returns:

Type	Description
int	file size in bytes

Source code in src/qass/tools/analyzer/buffer_parser.py

def file_size(self):
    """
    The file size of the buffer file. The value is not stored in the header
    but retrieved using the operating system.

    Returns
    -------
    int
        file size in bytes
    """
    return self.file_size

getArray

getArray(specFrom=None, specTo=None, delog=None)

Wrapper function to 'get_data'. This function provides access to the measurement data in the buffer file. The data are retrieved for the range of spectra and stored in a numpy array. The data can be logarithmized or not. The datatype of the array depends on the type of buffer.

Parameters:

Name	Type	Description	Default
specFrom	int	First spectrum to be retrieved (default first available spectrum).	None
specTo	int	Last spectrum to be retrieved (default last available spectrum).	None
delog	bool	To de-logarithmize the data (default None).	None

Raises:

Type	Description
InvalidArgumentError	The specFrom value is out of range.
InvalidArgumentError	The specTo value is out of range.

Returns:

Type	Description
NDArray	The buffers data.

Example

import qass.tools.analyzer.buffer_parser as bp
proc = range(1,100,1)
# path contains the path to a directory containing buffer files
buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
for buffer in buff:
    buff_file = (buffer.filepath)

with bp.Buffer(buff_file) as buff:
    spec_start = 0
    spec_end = (buff.db_count -1) * buff.db_spec_count
    print('Spec_end: ' + str(spec_end))
    conv = 'delog'
    data = buff.getArray(spec_start, spec_end, conv)

Source code in src/qass/tools/analyzer/buffer_parser.py

def getArray(self, specFrom=None, specTo=None, delog=None):
    """
    Wrapper function to 'get_data'.
    This function provides access to the measurement data in the buffer
    file. The data are retrieved for the range of spectra and stored in a
    numpy array. The data can be logarithmized or not. The datatype of the
    array depends on the type of buffer.

    Parameters
    ----------
    specFrom : int, optional
        First spectrum to be retrieved (default first available spectrum).
    specTo : int, optional
        Last spectrum to be retrieved (default last available spectrum).
    delog : bool, optional
        To de-logarithmize the data (default None).

    Raises
    ------
    InvalidArgumentError
        The specFrom value is out of range.
    InvalidArgumentError
        The specTo value is out of range.

    Returns
    -------
    npt.NDArray
        The buffers data.

    Example
    -------
    ```py
    import qass.tools.analyzer.buffer_parser as bp
    proc = range(1,100,1)
    # path contains the path to a directory containing buffer files
    buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
    for buffer in buff:
        buff_file = (buffer.filepath)

    with bp.Buffer(buff_file) as buff:
        spec_start = 0
        spec_end = (buff.db_count -1) * buff.db_spec_count
        print('Spec_end: ' + str(spec_end))
        conv = 'delog'
        data = buff.getArray(spec_start, spec_end, conv)
    ```
    """
    if delog == True:
        return self.get_data(specFrom, specTo, conversion="delog")
    elif delog == False:
        return self.get_data(specFrom, specTo, conversion="log")
    else:
        return self.get_data(specFrom, specTo)

getRealSpecCount

getRealSpecCount()

Wrapper function for the property spec_count. The number of spectra within a filled data block. It is calculated by dividing the number of samples in a data block by the number of frequency bands.

Returns:

Type	Description
int	Number of spectra in a filled data block.

Source code in src/qass/tools/analyzer/buffer_parser.py

def getRealSpecCount(self):
    """
    Wrapper function for the property spec_count.
    The number of spectra within a filled data block. It is calculated by
    dividing the number of samples in a data block by the number of
    frequency bands.

    Returns
    -------
    int
        Number of spectra in a filled data block.
    """
    return self.spec_count

getSpecDuration

getSpecDuration()

Wrapper Function for property spec_duration.

See Also

spec_duration

Source code in src/qass/tools/analyzer/buffer_parser.py

def getSpecDuration(self):
    """
    Wrapper Function for property spec_duration.

    See Also
    -------
    spec_duration
    """
    return self.spec_duration

get_data

get_data(specFrom=None, specTo=None, conversion: Union[str, None] = None)

This function provides access to the measurement data in the buffer file. The data are retrieved for the range of spectra and stored in a numpy array. The data can be logarithmized or not. The datatype of the array depends on the type of buffer.

Parameters:

Name	Type	Description	Default
specFrom	int	First spectrum to be retrieved (default first available spectrum)	None
specTo	int	Last spectrum to be retrieved (default last available spectrum)	None
conversion	Union[str, None]	Conversion ('log' or 'delog') of the data.	None

Raises:

Type	Description
InvalidArgumentError	The specFrom value is out of range
InvalidArgumentError	The specTo value is out of range
InvalidArgumentError	specTo or specFrom is not of type int or None
ValueError	The given indices exceed the buffer file's size
InvalidArgumentError	The given conversion is unknown

Returns:

Type	Description
NDArray	The buffers data.

Example

import qass.tools.analyzer.buffer_parser as bp
proc = range(1,100,1)
path contains the path to a directory containing buffer files
buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
for buffer in buff:
    buff_file = (buffer.filepath)

with bp.Buffer(buff_file) as buff:
    spec_start = 0
    spec_end = (buff.db_count -1) * buff.db_spec_count
    print('Spec_end: ' + str(spec_end))
    conv = 'delog'
    data = buff.get_data(spec_start, spec_end, conv)

Source code in src/qass/tools/analyzer/buffer_parser.py

def get_data(self, specFrom=None, specTo=None, conversion: Union[str, None] = None):
    """
    This function provides access to the measurement data in the buffer
    file. The data are retrieved for the range of spectra and stored in a
    numpy array. The data can be logarithmized or not. The datatype of the
    array depends on the type of buffer.

    Parameters
    ----------
    specFrom : int, optional
        First spectrum to be retrieved (default first available spectrum)
    specTo : int, optional
        Last spectrum to be retrieved (default last available spectrum)
    conversion: str, optional
        Conversion ('log' or 'delog') of the data.

    Raises
    ------
    InvalidArgumentError
        The specFrom value is out of range
    InvalidArgumentError
        The specTo value is out of range
    InvalidArgumentError
        specTo or specFrom is not of type int or None
    ValueError
        The given indices exceed the buffer file's size
    InvalidArgumentError
        The given conversion is unknown

    Returns
    -------
    npt.NDArray
        The buffers data.

    Example
    -------
    ```py
    import qass.tools.analyzer.buffer_parser as bp
    proc = range(1,100,1)
    path contains the path to a directory containing buffer files
    buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
    for buffer in buff:
        buff_file = (buffer.filepath)

    with bp.Buffer(buff_file) as buff:
        spec_start = 0
        spec_end = (buff.db_count -1) * buff.db_spec_count
        print('Spec_end: ' + str(spec_end))
        conv = 'delog'
        data = buff.get_data(spec_start, spec_end, conv)
    ```
    """
    specFrom = 0 if specFrom is None else specFrom
    specTo = self.spec_count if specTo is None else specTo

    if not (isinstance(specFrom, int) and isinstance(specTo, int)):
        raise InvalidArgumentError("specFrom/specTo must be int or None!")
    if specFrom > specTo:
        raise InvalidArgumentError(
            f"specFrom {specFrom} cannot be larger than specTo {specTo}!"
        )
    if specTo < 0 or specTo > self.spec_count:
        raise InvalidArgumentError(
            f"specTo {specTo} is out of range (0, {self.spec_count})!"
        )
    if specFrom < 0 or specFrom > self.spec_count:
        raise InvalidArgumentError(
            f"specFrom {specFrom} is out of range (0, {self.spec_count})!"
        )

    if self.datamode == self.DATAMODE.DATAMODE_SIGNAL:
        return self._get_data(specFrom, specTo, 1, conversion).reshape(-1)
    else:
        return self._get_data(specFrom, specTo, self.__frq_bands, conversion)

io_ports

io_ports(db_idx: int, byte: int = None, bit: int = None)

Deprecated method! This method provides the state of the io ports at the time the dateblock (indicated by its number db_idx) is written. It does not provide information wether the state of any io port changes within the datablock. This method will be replaced by a method which analyses the data within the sub data block, which is part of the data block header and provides information regarding the state of the io ports at a higher accuracy.

Parameters:

Name	Type	Description	Default
db_idx	int	Index of the data block	required
byte	int	Number of io port socket [1, 2, 4]	None
bit	int	Number of bit [1, 2, 3, 4, 5, 6, 7, 8]	None

Raises:

Type	Description
ValueError	The bit argument requires the byte argument.
ValueError	The given byte has to be one out of [1, 2, 4].
ValueError	The given bit is out of range.

Returns:

Type	Description
int	byte with the appropriate bits set as an int

Source code in src/qass/tools/analyzer/buffer_parser.py

def io_ports(self, db_idx: int, byte: int = None, bit: int = None):
    """
    Deprecated method!
    This method provides the state of the io ports at the time the
    dateblock (indicated by its number db_idx) is written. It does not
    provide information wether the state of any io port changes within the
    datablock. This method will be replaced by a method which analyses the
    data within the sub data block, which is part of the data block header
    and provides information regarding the state of the io ports at a
    higher accuracy.

    Parameters
    ----------
    db_idx : int
        Index of the data block
    byte : int
        Number of io port socket [1, 2, 4]
    bit : int, optional
        Number of bit [1, 2, 3, 4, 5, 6, 7, 8]

    Raises
    ------
    ValueError
        The bit argument requires the byte argument.
    ValueError
        The given byte has to be one out of [1, 2, 4].
    ValueError
        The given bit is out of range.

    Returns
    -------
    int
        byte with the appropriate bits set as an int
    """
    io_word = ~(self.db_value(db_idx, "io_ports")) & 0xFFFFFF

    if byte is None and bit is not None:
        raise ValueError("The bit argument requires the byte argument")

    if byte is None:
        return io_word

    if byte == 1:
        io_word = (io_word >> 0) & 0xFF
    elif byte == 2:
        io_word = (io_word >> 8) & 0xFF
    elif byte == 4:
        io_word = (io_word >> 16) & 0xFF
    else:
        raise ValueError("The given byte has to be one out of [1, 2, 4]")

    if bit is None:
        return io_word

    if not 1 <= bit <= 8:
        raise ValueError(f"The given bit is out of range: {bit}")

    return io_word >> (bit - 1) & 1 == 1

io_ports_spec

io_ports_spec(spec: int, byte: int, bit: int)

Deprecated method! This method provides the state of the io ports at the time the dateblock (indicated by the spec number within the data block) is written. It does not provide information wether the state of any io port changes within the datablock. This method will be replaced by a method which analyses the data within the sub data block, which is part of the data block header and provides information regarding the state of the io ports at a higher accuracy.

Parameters:

Name	Type	Description	Default
spec	int	Index of the spectrum	required
byte	int	Number of io port socket [1, 2, 4]	required
bit	int	Number of bit [1, 2, 3, 4, 5, 6, 7, 8]	required

Raises:

Type	Description
ValueError	The bit argument requires the byte argument.
ValueError	The given byte has to be one out of [1, 2, 4].
ValueError	The given bit is out of range.

Returns:

Type	Description
int	byte with the appropriate bits set as an int

Source code in src/qass/tools/analyzer/buffer_parser.py

def io_ports_spec(self, spec: int, byte: int, bit: int):
    """
    Deprecated method!
    This method provides the state of the io ports at the time the
    dateblock (indicated by the spec number within the data block) is
    written. It does not provide information wether the state of any io
    port changes within the datablock. This method will be replaced by a
    method which analyses the data within the sub data block, which is
    part of the data block header and provides information regarding the
    state of the io ports at a higher accuracy.

    Parameters
    ----------
    spec : int
        Index of the spectrum
    byte : int
        Number of io port socket [1, 2, 4]
    bit : int
        Number of bit [1, 2, 3, 4, 5, 6, 7, 8]

    Raises
    ------
    ValueError
        The bit argument requires the byte argument.
    ValueError
        The given byte has to be one out of [1, 2, 4].
    ValueError
        The given bit is out of range.

    Returns
    -------
    int
        byte with the appropriate bits set as an int
    """
    db_idx = int(spec * self.__frq_bands / self.db_sample_count)
    return self.io_ports_byte(db_idx, byte, bit)

log staticmethod

log(data_arr, fft_log_shift, ad_bit_resolution)

Method to logarithmize a data array. Usually when data are measured with the Optimizer the data are logarithmized. The problem with logarithms is that adding them up constitutes a multiplication of the non logarithmized data. So in order to carry out calculations such as calculating sums etc. it is crucial to use de-logarithmized data. Sometimes it is necessary to logarithmize data which were artificially created or measured without a logarithm.

Parameters:

Name	Type	Description	Default
data_arr	NDArray[floating]	Array with non-logarithmized data.	required
fft_log_shift	int	Base of the logarithm.	required
ad_bit_resolution	int	The bit resolution (usually 16).	required

Returns:

Type	Description
NDArray[floating]

Source code in src/qass/tools/analyzer/buffer_parser.py

@staticmethod
def log(data_arr, fft_log_shift, ad_bit_resolution):
    """
    Method to logarithmize a data array. Usually when data are measured
    with the Optimizer the data are logarithmized. The problem with
    logarithms is that adding them up constitutes a multiplication of the
    non logarithmized data. So in order to carry out calculations such as
    calculating sums etc. it is crucial to use de-logarithmized data.
    Sometimes it is necessary to logarithmize data which were artificially
    created or measured without a logarithm.

    Parameters
    ----------
    data_arr : npt.NDArray[np.floating]
        Array with non-logarithmized data.
    fft_log_shift : int
        Base of the logarithm.
    ad_bit_resolution : int
        The bit resolution (usually 16).

    Returns
    -------
    npt.NDArray[np.floating]
    """
    shift = fft_log_shift - 1
    bits = ad_bit_resolution if ad_bit_resolution >= 1 else 16
    bit_res_idx = [14, 16, 24].index(bits)

    # remember negative values:
    negative = data_arr < 0
    np.abs(data_arr, out=data_arr)

    # ensure data_arr to be a floating point array
    data_arr = data_arr.astype(np.double, copy=False)

    shift_offset = [23, 23, 31][bit_res_idx]
    max_amp = 1 << bits

    data_arr = np.log2(data_arr + (1 << shift)) - shift
    data_arr = data_arr * max_amp / (shift_offset - shift)

    data_arr[negative] *= -1

    return data_arr

spec_to_time_ns

spec_to_time_ns(spec)

The method calculates the time since measurement start for a given index of a spectrum.

Parameters:

Name	Type	Description	Default
spec	int	The index of a spectrum	required

Returns:

Type	Description
float	time in nanoseconds

Source code in src/qass/tools/analyzer/buffer_parser.py

def spec_to_time_ns(self, spec):
    """
    The method calculates the time since measurement start for a given
    index of a spectrum.

    Parameters
    ----------
    spec : int
        The index of a spectrum

    Returns
    -------
    float
        time in nanoseconds
    """
    return spec * self.spec_duration

time_to_spec

time_to_spec(time_ns)

The method calculates the index of the nearest spectrum to the time given since the start of the measurement.

Parameters:

Name	Type	Description	Default
time_ns	float	Time elapsed since measurement start in ns.	required

Returns:

Type	Description
int	index of the spectrum

Source code in src/qass/tools/analyzer/buffer_parser.py

def time_to_spec(self, time_ns):
    """
    The method calculates the index of the nearest spectrum to the time
    given since the start of the measurement.

    Parameters
    ----------
    time_ns : float
        Time elapsed since measurement start in ns.

    Returns
    -------
    int
        index of the spectrum
    """
    return int(time_ns / self.spec_duration)

HeaderDtype

Bases: IntEnum

Attributes:

Name	Type	Description
DOUBLE
FLOAT
HEX_STRING
INT32
INT64
UINT32
UINT64

Source code in src/qass/tools/analyzer/buffer_parser.py

class HeaderDtype(IntEnum):
    # Enum helper class to mark data types
    INT32 = auto()
    INT64 = auto()
    UINT32 = auto()
    UINT64 = auto()
    FLOAT = auto()
    DOUBLE = auto()
    HEX_STRING = auto()

DOUBLE class-attribute instance-attribute

DOUBLE = auto()

FLOAT class-attribute instance-attribute

FLOAT = auto()

HEX_STRING class-attribute instance-attribute

HEX_STRING = auto()

INT32 class-attribute instance-attribute

INT32 = auto()

INT64 class-attribute instance-attribute

INT64 = auto()

UINT32 class-attribute instance-attribute

UINT32 = auto()

UINT64 class-attribute instance-attribute

UINT64 = auto()

filter_buffers

filter_buffers(directory, filters)

This function takes a directory with buffer files and a dictionary with fiter criteria and returns a list of buffer objects from this dictionary which fulfill the filter criteria.

Parameters:

Name	Type	Description	Default
directory	str	Path to a directory with buffer files	required
filters	dict	Dictionary with filters	required

Returns:

Type	Description
list	A list of buffer objects

Example

from qass.tools.analyzer import buffer_parser as bp
proc = range(1, 100, 1)
# path contains the path to a directory containing buffer files
buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})

Source code in src/qass/tools/analyzer/buffer_parser.py

def filter_buffers(directory, filters):
    """
    This function takes a directory with buffer files and a dictionary with
    fiter criteria and returns a list of buffer objects from this
    dictionary which fulfill the filter criteria.

    Parameters
    ----------
    directory : str
        Path to a directory with buffer files
    filters : dict
        Dictionary with filters

    Returns
    -------
    list
        A list of buffer objects

    Example
    -------
    ```py
    from qass.tools.analyzer import buffer_parser as bp
    proc = range(1, 100, 1)
    # path contains the path to a directory containing buffer files
    buff = bp.filter_buffers(path, {'wanted_process': proc , 'datamode': bp.Buffer.DATAMODE.DATAMODE_FFT})
    ```
    """
    from pathlib import Path

    pattern = "*p*"
    if "process" in filters:
        pattern += str(filters["process"])
    pattern += "c"

    if "channel" in filters:
        pattern += str(filters["channel"] - 1)
    else:
        pattern += "*"
    pattern += "b*"

    buffers = []
    for file in Path(directory).rglob(pattern):
        try:
            with Buffer(file) as buff:
                if "process" in filters and buff.process != filters["process"]:
                    continue
                if (
                    "unwanted_process" in filters
                    and buff.process in filters["unwanted_process"]
                ):
                    continue
                if (
                    "wanted_process" in filters
                    and buff.process not in filters["wanted_process"]
                ):
                    continue
                if "channel" in filters and filters["channel"] != buff.channel:
                    continue
                if "datamode" in filters and filters["datamode"] != buff.datamode:
                    continue
                if "datakind" in filters and filters["datakind"] != buff.datakind:
                    continue
                if "datatype" in filters and filters["datatype"] != buff.datatype:
                    continue
                if (
                    "compression_time" in filters
                    and filters["compression_time"] != buff.compression_time
                ):
                    continue
                if (
                    "compression_frq" in filters
                    and filters["compression_frq"] != buff.compression_frq
                ):
                    continue
                if "avg_time" in filters and filters["avg_time"] != buff.avg_time:
                    continue
                if "avg_frq" in filters and filters["avg_frq"] != buff.avg_frq:
                    continue

                buffers.append(buff)
        except Exception:
            warnings.warn(f"filter_buffers could not parse: {file.name}")
            pass

    return buffers