MAX_UINT64 = (1 << 64) - 1
DATA_TYPE_UNSIGNED_INTEL = 0
DATA_TYPE_UNSIGNED_MOTOROLA = 1
DATA_TYPE_SIGNED_INTEL = 2
DATA_TYPE_SIGNED_MOTOROLA = 3
DATA_TYPE_REAL_INTEL = 4
DATA_TYPE_REAL_MOTOROLA = 5
DATA_TYPE_STRING_LATIN_1 = 6
DATA_TYPE_STRING_UTF_8 = 7
DATA_TYPE_STRING_UTF_16_LE = 8
DATA_TYPE_STRING_UTF_16_BE = 9
DATA_TYPE_BYTEARRAY = 10
DATA_TYPE_MIME_SAMPLE = 11
DATA_TYPE_MIME_STREAM = 12
DATA_TYPE_CANOPEN_DATE = 13
DATA_TYPE_CANOPEN_TIME = 14
DATA_TYPE_COMPLEX_INTEL = 15
DATA_TYPE_COMPLEX_MOTOROLA = 16
DATA_TYPE_TO_STRING = dict(
    enumerate(
        [
            "UNSIGNED_INTEL",
            "UNSIGNED_MOTOROLA",
            "SIGNED_INTEL",
            "SIGNED_MOTOROLA",
            "REAL_INTEL",
            "REAL_MOTOROLA",
            "STRING_LATIN_1",
            "STRING_UTF_8",
            "STRING_UTF_16_LE",
            "STRING_UTF_16_BE",
            "BYTEARRAY",
            "MIME_SAMPLE",
            "MIME_STREAM",
            "CANOPEN_DATE",
            "CANOPEN_TIME",
            "COMPLEX_INTEL",
            "COMPLEX_MOTOROLA",
        ]
    )
)
NON_SCALAR_TYPES = {
    DATA_TYPE_BYTEARRAY,
    DATA_TYPE_MIME_SAMPLE,
    DATA_TYPE_MIME_STREAM,
    DATA_TYPE_STRING_UTF_8,
    DATA_TYPE_STRING_LATIN_1,
    DATA_TYPE_STRING_UTF_16_BE,
    DATA_TYPE_STRING_UTF_16_LE,
    DATA_TYPE_CANOPEN_DATE,
    DATA_TYPE_CANOPEN_TIME,
}
STRING_TYPES = {
    DATA_TYPE_STRING_UTF_8,
    DATA_TYPE_STRING_LATIN_1,
    DATA_TYPE_STRING_UTF_16_BE,
    DATA_TYPE_STRING_UTF_16_LE,
}
SIGNAL_TYPE_SCALAR = 0
SIGNAL_TYPE_STRING = 1
SIGNAL_TYPE_CANOPEN = 2
SIGNAL_TYPE_STRUCTURE_COMPOSITION = 3
SIGNAL_TYPE_ARRAY = 4
SIGNAL_TYPE_BYTEARRAY = 5
SIGNED_INT = {DATA_TYPE_SIGNED_INTEL, DATA_TYPE_SIGNED_MOTOROLA}
FLOATS = {DATA_TYPE_REAL_INTEL, DATA_TYPE_REAL_MOTOROLA}
STANDARD_INT_SIZES = {8, 16, 32, 64}
INT_TYPES = {
    DATA_TYPE_UNSIGNED_INTEL,
    DATA_TYPE_UNSIGNED_MOTOROLA,
    DATA_TYPE_SIGNED_INTEL,
    DATA_TYPE_SIGNED_MOTOROLA,
}
CHANNEL_TYPE_VALUE = 0
CHANNEL_TYPE_VLSD = 1
CHANNEL_TYPE_MASTER = 2
CHANNEL_TYPE_VIRTUAL_MASTER = 3
CHANNEL_TYPE_SYNC = 4
CHANNEL_TYPE_MLSD = 5
CHANNEL_TYPE_VIRTUAL = 6
SYNC_TYPE_NONE = 0
SYNC_TYPE_TIME = 1
SYNC_TYPE_ANGLE = 2
SYNC_TYPE_DISTANCE = 3
SYNC_TYPE_INDEX = 4
SYNC_TYPE_TO_STRING = dict(enumerate(["NONE", "TIME", "ANGLE", "DISTANCE", "INDEX"]))
CHANNEL_TYPE_TO_STRING = CHANNEL_TYPE_TO_DESCRIPTION = {
    CHANNEL_TYPE_VALUE: "VALUE",
    CHANNEL_TYPE_VLSD: "VLSD",
    CHANNEL_TYPE_MASTER: "MASTER",
    CHANNEL_TYPE_VIRTUAL_MASTER: "VIRTUAL MASTER",
    CHANNEL_TYPE_SYNC: "SYNC",
    CHANNEL_TYPE_MLSD: "MLSD",
    CHANNEL_TYPE_VIRTUAL: "VIRTUAL",
}
MASTER_TYPES = {CHANNEL_TYPE_MASTER, CHANNEL_TYPE_VIRTUAL_MASTER}
VIRTUAL_TYPES = {CHANNEL_TYPE_VIRTUAL, CHANNEL_TYPE_VIRTUAL_MASTER}
CONVERSION_TYPE_NON = 0
CONVERSION_TYPE_LIN = 1
CONVERSION_TYPE_RAT = 2
CONVERSION_TYPE_ALG = 3
CONVERSION_TYPE_TABI = 4
CONVERSION_TYPE_TAB = 5
CONVERSION_TYPE_RTAB = 6
CONVERSION_TYPE_TABX = 7
CONVERSION_TYPE_RTABX = 8
CONVERSION_TYPE_TTAB = 9
CONVERSION_TYPE_TRANS = 10
CONVERSION_TYPE_BITFIELD = 11
CONVERSION_TYPE_TO_STRING = dict(
    enumerate(
        [
            "NON",
            "LIN",
            "RAT",
            "ALG",
            "TABI",
            "TAB",
            "RTAB",
            "TABX",
            "RTABX",
            "TTAB",
            "TRANS",
            "BITFIELD",
        ]
    )
)
CONV_RAT_TEXT = "(P1 * X**2 + P2 * X + P3) / (P4 * X**2 + P5 * X + P6)"
TABULAR_CONVERSIONS = {
    CONVERSION_TYPE_TABX,
    CONVERSION_TYPE_RTABX,
    CONVERSION_TYPE_TTAB,
    CONVERSION_TYPE_BITFIELD,
}
CONVERSIONS_WITH_TEXTS = {
    CONVERSION_TYPE_ALG,
    CONVERSION_TYPE_RTABX,
    CONVERSION_TYPE_TABX,
    CONVERSION_TYPE_TRANS,
    CONVERSION_TYPE_TTAB,
    CONVERSION_TYPE_BITFIELD,
}
CONVERSION_GROUP_1 = {CONVERSION_TYPE_NON, CONVERSION_TYPE_TRANS, CONVERSION_TYPE_TTAB}
CONVERSION_GROUP_2 = {
    CONVERSION_TYPE_LIN,
    CONVERSION_TYPE_RAT,
    CONVERSION_TYPE_ALG,
    CONVERSION_TYPE_TABI,
    CONVERSION_TYPE_TAB,
    CONVERSION_TYPE_RTAB,
}
CA_TYPE_ARRAY = 0
CA_TYPE_SCALE_AXIS = 1
CA_TYPE_LOOKUP = 2
CA_STORAGE_TYPE_CN_TEMPLATE = 0
CA_STORAGE_TYPE_CG_TEMPLATE = 1
CA_STORAGE_TYPE_DG_TEMPLATE = 2
SOURCE_OTHER = 0
SOURCE_ECU = 1
SOURCE_BUS = 2
SOURCE_IO = 3
SOURCE_TOOL = 4
SOURCE_USER = 5
SOURCE_TYPE_TO_STRING = {
    SOURCE_OTHER: "OTHER",
    SOURCE_ECU: "ECU",
    SOURCE_BUS: "BUS",
    SOURCE_IO: "IO",
    SOURCE_TOOL: "TOOL",
    SOURCE_USER: "USER",
}
BUS_TYPE_NONE = 0
BUS_TYPE_OTHER = 1
BUS_TYPE_CAN = 2
BUS_TYPE_LIN = 3
BUS_TYPE_MOST = 4
BUS_TYPE_FLEXRAY = 5
BUS_TYPE_K_LINE = 6
BUS_TYPE_ETHERNET = 7
BUS_TYPE_USB = 8
BUS_TYPE_TO_STRING = {
    BUS_TYPE_NONE: "NONE",
    BUS_TYPE_OTHER: "OTHER",
    BUS_TYPE_CAN: "CAN",
    BUS_TYPE_LIN: "LIN",
    BUS_TYPE_MOST: "MOST",
    BUS_TYPE_FLEXRAY: "FLEXRAY",
    BUS_TYPE_K_LINE: "K_LINE",
    BUS_TYPE_ETHERNET: "ETHERNET",
    BUS_TYPE_USB: "USB",
}
EVENT_TYPE_RECORDING = 0
EVENT_TYPE_RECORDING_INTERRUPT = 1
EVENT_TYPE_ACQUISITION_INTERRUPT = 2
EVENT_TYPE_START_RECORDING_TRIGGER = 3
EVENT_TYPE_STOP_RECORDING_TRIGGER = 4
EVENT_TYPE_TRIGGER = 5
EVENT_TYPE_MARKER = 6
EVENT_TYPE_TO_STRING = {
    EVENT_TYPE_RECORDING: "Recording",
    EVENT_TYPE_RECORDING_INTERRUPT: "REcording interrupt",
    EVENT_TYPE_ACQUISITION_INTERRUPT: "Acquisition interrupt",
    EVENT_TYPE_START_RECORDING_TRIGGER: "Start recording trigger",
    EVENT_TYPE_STOP_RECORDING_TRIGGER: "Stop recording trigger",
    EVENT_TYPE_TRIGGER: "Trigger",
    EVENT_TYPE_MARKER: "Marker",
}
EVENT_SYNC_TYPE_S = 1
EVENT_SYNC_TYPE_RAD = 2
EVENT_SYNC_TYPE_M = 3
EVENT_SYNC_TYPE_INDEX = 4
EVENT_RANGE_TYPE_POINT = 0
EVENT_RANGE_TYPE_BEGINNING = 1
EVENT_RANGE_TYPE_END = 2
EVENT_CAUSE_OTHER = 0
EVENT_CAUSE_ERROR = 1
EVENT_CAUSE_TOOL = 2
EVENT_CAUSE_SCRIPT = 3
EVENT_CAUSE_USER = 4
SEEK_START = 0
SEEK_REL = 1
SEEK_END = 2
TIME_CH_SIZE = 8
SI_BLOCK_SIZE = 56
FH_BLOCK_SIZE = 56
DG_BLOCK_SIZE = 64
HD_BLOCK_SIZE = 104
CN_BLOCK_SIZE = 160
CN_SINGLE_ATTACHMENT_BLOCK_SIZE = 168
CG_BLOCK_SIZE = 104
CG_RM_BLOCK_SIZE = 112
COMMON_SIZE = 24
COMMON_SHORT_SIZE = 16
CC_NONE_BLOCK_SIZE = 80
CC_ALG_BLOCK_SIZE = 88
CC_LIN_BLOCK_SIZE = 96
AT_COMMON_SIZE = 96
DZ_COMMON_SIZE = 48
DZ_COMMON_INFO_SIZE = 22
CC_COMMON_BLOCK_SIZE = 80
HL_BLOCK_SIZE = 40
IDENTIFICATION_BLOCK_SIZE = 64
HEADER_BLOCK_SIZE = 104
SR_BLOCK_SIZE = 64
FLAG_ALL_SAMPLES_VALID = 1
FLAG_INVALIDATION_BIT_VALID = 2
FLAG_PRECISION = 1
FLAG_PHY_RANGE_OK = 1 << 4
FLAG_VAL_RANGE_OK = 1 << 3
FLAG_AT_EMBEDDED = 1
FLAG_AT_COMPRESSED_EMBEDDED = 2
FLAG_AT_MD5_VALID = 4
FLAG_DZ_DEFLATE = 0
FLAG_DZ_TRANPOSED_DEFLATE = 1
FLAG_CA_DYNAMIC_AXIS = 1
FLAG_CA_INPUT_QUANTITY = 1 << 1
FLAG_CA_OUTPUT_QUANTITY = 1 << 2
FLAG_CA_COMPARISON_QUANTITY = 1 << 3
FLAG_CA_AXIS = 1 << 4
FLAG_CA_FIXED_AXIS = 1 << 5
FLAG_CA_INVERSE_LAYOUT = 1 << 6
FLAG_CA_LEFT_OPENED_INTERVAL = 1 << 7
FLAG_CA_STANDARD_AXIS = 1 << 8
FLAG_DL_EQUAL_LENGHT = 1
FLAG_DL_TIME_VALUES = 1 << 1
FLAG_DL_ANGLE_VALUES = 1 << 2
FLAG_DL_DISTANCE_VALUES = 1 << 3
FLAG_LD_EQUAL_LENGHT = 1
FLAG_LD_TIME_VALUES = 1 << 1
FLAG_LD_ANGLE_VALUES = 1 << 2
FLAG_LD_DISTANCE_VALUES = 1 << 3
FLAG_LD_INVALIDATION_PRESENT = 1 << 31
FLAG_EV_POST_PROCESSING = 1
FLAG_EV_GROUP_NAME = 1 << 1
FLAG_CG_VLSD = 1
FLAG_CG_BUS_EVENT = 1 << 1
FLAG_CG_PLAIN_BUS_EVENT = 1 << 2
FLAG_CG_REMOTE_MASTER = 1 << 3
FLAG_CG_EVENT_GROUP = 1 << 4
FLAG_UNFIN_UPDATE_CG_COUNTER = 0x1
FLAG_UNFIN_UPDATE_SR_COUNTER = 0x2
FLAG_UNFIN_UPDATE_LAST_DT_LENGTH = 0x4
FLAG_UNFIN_UPDATE_LAST_RD_LENGTH = 0x8
FLAG_UNFIN_UPDATE_LAST_DL = 0x10
FLAG_UNFIN_UPDATE_VLSD_BYTES = 0x20
FLAG_UNFIN_UPDATE_VLSD_OFFSETS = 0x40
FLAG_CG_TO_STRING = {
    FLAG_CG_VLSD: "VLSD",
    FLAG_CG_BUS_EVENT: "BUS_EVENT",
    FLAG_CG_PLAIN_BUS_EVENT: "PLAIN_BUS_EVENT",
    FLAG_CG_REMOTE_MASTER: "REMOTE_MASTER",
    FLAG_CG_EVENT_GROUP: "EVENT_GROUP",
}
FLAG_AT_TO_STRING = {
    FLAG_AT_EMBEDDED: "Embedded",
    FLAG_AT_COMPRESSED_EMBEDDED: "Compressed",
    FLAG_AT_MD5_VALID: "MD5 sum valid",
}
FLAG_CN_ALL_INVALID = 1
FLAG_CN_INVALIDATION_PRESENT = 1 << 1
FLAG_CN_PRECISION = 1 << 2
FLAG_CN_VALUE_RANGE = 1 << 3
FLAG_CN_LIMIT_RANGE = 1 << 4
FLAG_CN_EXTENDED_LIMIT_RANGE = 1 << 5
FLAG_CN_DISCRETE = 1 << 6
FLAG_CN_CALIBRATION = 1 << 7
FLAG_CN_CALCULATED = 1 << 8
FLAG_CN_VIRTUAL = 1 << 9
FLAG_CN_BUS_EVENT = 1 << 10
FLAG_CN_MONOTONOUS = 1 << 11
FLAG_CN_DEFAULT_X = 1 << 12
FLAG_CN_EVENT_SIGNAL = 1 << 13
FLAG_CN_VLSD = 1 << 14
FLAG_CN_TO_STRING = {
    FLAG_CN_ALL_INVALID: "ALL_INVALID",
    FLAG_CN_INVALIDATION_PRESENT: "INVALIDATION_PRESENT",
    FLAG_CN_PRECISION: "PRECISION",
    FLAG_CN_VALUE_RANGE: "VALUE_RANGE",
    FLAG_CN_LIMIT_RANGE: "LIMIT_RANGE",
    FLAG_CN_EXTENDED_LIMIT_RANGE: "EXTENDED_LIMIT_RANGE",
    FLAG_CN_DISCRETE: "DISCRETE",
    FLAG_CN_CALIBRATION: "CALIBRATION",
    FLAG_CN_CALCULATED: "CALCULATED",
    FLAG_CN_VIRTUAL: "VIRTUAL",
    FLAG_CN_BUS_EVENT: "BUS_EVENT",
    FLAG_CN_MONOTONOUS: "MONOTONOUS",
    FLAG_CN_DEFAULT_X: "DEFAULT_X",
    FLAG_CN_EVENT_SIGNAL: "EVENT_SIGNAL",
    FLAG_CN_VLSD: "VLSD",
}
FLAG_CC_PRECISION = 1
FLAG_CC_RANGE = 1 << 1
FLAG_HD_LOCAL_TIME = 1
FLAG_HD_TIME_OFFSET_VALID = 2
LOCATION_ORIGINAL_FILE = 0
LOCATION_TEMPORARY_FILE = 1
LOCATION_MEMORY = 2
DT_BLOCK = 0
DZ_BLOCK_DEFLATE = 1
DZ_BLOCK_TRANSPOSED = 2
DZ_BLOCK_LZ = 3
FMT_CHANNEL = "<4sI2Q{}Q4B4I2BH6d"
FMT_CHANNEL_PARAMS = "<4B4I2BH6d"
FMT_CHANNEL_FILTER = "<4s12x4Q32xQ"
CHANNEL_FILTER_SIZE = 88
FMT_SIMPLE_CHANNEL = "<4sI10Q4B4I2BH6d"
KEYS_SIMPLE_CHANNEL = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_ch_addr",
    "component_addr",
    "name_addr",
    "source_addr",
    "conversion_addr",
    "data_block_addr",
    "unit_addr",
    "comment_addr",
    "channel_type",
    "sync_type",
    "data_type",
    "bit_offset",
    "byte_offset",
    "bit_count",
    "flags",
    "pos_invalidation_bit",
    "precision",
    "reserved1",
    "attachment_nr",
    "min_raw_value",
    "max_raw_value",
    "lower_limit",
    "upper_limit",
    "lower_ext_limit",
    "upper_ext_limit",
)
FMT_SIMPLE_CHANNEL_PARAMS = "<8Q4B4I2BH6d"
SIMPLE_CHANNEL_PACK = struct.Struct(FMT_SIMPLE_CHANNEL).pack
FMT_SINGLE_ATTACHMENT_CHANNEL = "<4sI11Q4B4I2BH6d"
SINGLE_ATTACHMENT_CHANNEL_PACK = struct.Struct(FMT_SINGLE_ATTACHMENT_CHANNEL).pack
FMT_SINGLE_ATTACHMENT_CHANNEL_PARAMS = "<9Q4B4I2BH6d"
FMT_TEXT_BLOCK = "<4sIQQ{}s"
KEYS_TEXT_BLOCK = ("id", "reserved0", "block_len", "links_nr", "text")
FMT_SOURCE_INFORMATION = "<4sI5Q3B5s"
KEYS_SOURCE_INFORMATION = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "name_addr",
    "path_addr",
    "comment_addr",
    "source_type",
    "bus_type",
    "flags",
    "reserved1",
)
SOURCE_INFORMATION_PACK = struct.Struct(FMT_SOURCE_INFORMATION).pack
FMT_CHANNEL_GROUP = "<4sI10Q2H3I"
KEYS_CHANNEL_GROUP = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_cg_addr",
    "first_ch_addr",
    "acq_name_addr",
    "acq_source_addr",
    "first_sample_reduction_addr",
    "comment_addr",
    "record_id",
    "cycles_nr",
    "flags",
    "path_separator",
    "reserved1",
    "samples_byte_nr",
    "invalidation_bytes_nr",
)
FMT_CHANNEL_GROUP_SHORT = "<8Q2H3I"
KEYS_CHANNEL_GROUP_SHORT = (
    "next_cg_addr",
    "first_ch_addr",
    "acq_name_addr",
    "acq_source_addr",
    "first_sample_reduction_addr",
    "comment_addr",
    "record_id",
    "cycles_nr",
    "flags",
    "path_separator",
    "reserved1",
    "samples_byte_nr",
    "invalidation_bytes_nr",
)
FMT_CHANNEL_GROUP_RM = "<4sI11Q2H3I"
KEYS_CHANNEL_GROUP_RM = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_cg_addr",
    "first_ch_addr",
    "acq_name_addr",
    "acq_source_addr",
    "first_sample_reduction_addr",
    "comment_addr",
    "cg_master_addr",
    "record_id",
    "cycles_nr",
    "flags",
    "path_separator",
    "reserved1",
    "samples_byte_nr",
    "invalidation_bytes_nr",
)
FMT_CHANNEL_GROUP_RM_SHORT = "<9Q2H3I"
KEYS_CHANNEL_GROUP_RM_SHORT = (
    "next_cg_addr",
    "first_ch_addr",
    "acq_name_addr",
    "acq_source_addr",
    "first_sample_reduction_addr",
    "comment_addr",
    "cg_master_addr",
    "record_id",
    "cycles_nr",
    "flags",
    "path_separator",
    "reserved1",
    "samples_byte_nr",
    "invalidation_bytes_nr",
)
FMT_DATA_BLOCK = "<4sI2Q{}s"
KEYS_DATA_BLOCK = ("id", "reserved0", "block_len", "links_nr", "data")
FMT_COMMON = "<4sI2Q"
FMT_COMMON_SHORT = "<4s4xQ"
FMT_FILE_HISTORY = "<4sI5Q2hB3s"
KEYS_FILE_HISTORY = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_fh_addr",
    "comment_addr",
    "abs_time",
    "tz_offset",
    "daylight_save_time",
    "time_flags",
    "reserved1",
)
FMT_DATA_GROUP = "<4sI6QB7s"
KEYS_DATA_GROUP = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_dg_addr",
    "first_cg_addr",
    "data_block_addr",
    "comment_addr",
    "record_id_len",
    "reserved1",
)
FMT_DATA_LIST = "<4sI2Q{}QB3sIQ"
FMT_CONVERSION_NONE = "<4sI6Q2B3H2d"
KEYS_CONVERSION_NONE = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "name_addr",
    "unit_addr",
    "comment_addr",
    "inv_conv_addr",
    "conversion_type",
    "precision",
    "flags",
    "ref_param_nr",
    "val_param_nr",
    "min_phy_value",
    "max_phy_value",
)
CONVERSION_NONE_PACK = struct.Struct(FMT_CONVERSION_NONE).pack
FMT_CONVERSION_NONE_INIT = "<4Q2B3H2d"
FMT_CONVERSION_LINEAR = FMT_CONVERSION_NONE + "2d"
CONVERSION_LINEAR_PACK = struct.Struct(FMT_CONVERSION_LINEAR).pack
KEYS_CONVERSION_LINEAR = KEYS_CONVERSION_NONE + ("b", "a")
FMT_CONVERSION_LINEAR_INIT = "<4Q2B3H4d"
FMT_CONVERSION_ALGEBRAIC = "<4sI7Q2B3H2d"
CONVERSION_ALGEBRAIC_PACK = struct.Struct(FMT_CONVERSION_ALGEBRAIC).pack
KEYS_CONVERSION_ALGEBRAIC = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "name_addr",
    "unit_addr",
    "comment_addr",
    "inv_conv_addr",
    "formula_addr",
    "conversion_type",
    "precision",
    "flags",
    "ref_param_nr",
    "val_param_nr",
    "min_phy_value",
    "max_phy_value",
)
FMT_CONVERSION_ALGEBRAIC_INIT = "<5Q2B3H2d"
FMT_CONVERSION_RAT = FMT_CONVERSION_NONE + "6d"
CONVERSION_RAT_PACK = struct.Struct(FMT_CONVERSION_RAT).pack
KEYS_CONVERSION_RAT = KEYS_CONVERSION_NONE + ("P1", "P2", "P3", "P4", "P5", "P6")
FMT_CONVERSION_RAT_INIT = "<4Q2B3H8d"
FMT_CONVERSION_RAT_INIT = "<4Q2B3H8d"
FMT_HEADER_BLOCK = "<4sI9Q2h4B2Q"
FMT_IDENTIFICATION_BLOCK = "<8s8s8s4sH30s2H"
KEYS_HEADER_BLOCK = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "first_dg_addr",
    "file_history_addr",
    "channel_tree_addr",
    "first_attachment_addr",
    "first_event_addr",
    "comment_addr",
    "abs_time",
    "tz_offset",
    "daylight_save_time",
    "time_flags",
    "time_quality",
    "flags",
    "reserved1",
    "start_angle",
    "start_distance",
)
KEYS_IDENTIFICATION_BLOCK = (
    "file_identification",
    "version_str",
    "program_identification",
    "reserved0",
    "mdf_version",
    "reserved1",
    "unfinalized_standard_flags",
    "unfinalized_custom_flags",
)
FMT_AT_COMMON = "<4sI6Q2HI16s2Q"
KEYS_AT_BLOCK = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_at_addr",
    "file_name_addr",
    "mime_addr",
    "comment_addr",
    "flags",
    "creator_index",
    "reserved1",
    "md5_sum",
    "original_size",
    "embedded_size",
    "embedded_data",
)
FMT_DZ_COMMON = "<4sI2Q2s2BI2Q"
KEYS_DZ_BLOCK = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "original_type",
    "zip_type",
    "reserved1",
    "param",
    "original_size",
    "zip_size",
    "data",
)
FMT_DZ_INFO_COMMON = "<B1xI2Q"
DZ_INFO_COMMON_OFFSET = 26
FMT_HL_BLOCK = "<4sI3QHB5s"
KEYS_HL_BLOCK = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "first_dl_addr",
    "flags",
    "zip_type",
    "reserved1",
)
FMT_EVENT_PARAMS = "<5B3sI2Hqd"
FMT_EVENT = "<4sI2Q{}Q5B3sI2Hqd"
FMT_SR_BLOCK = "<4sI5Qd2B6s"
KEYS_SR_BLOCK = (
    "id",
    "reserved0",
    "block_len",
    "links_nr",
    "next_sr_addr",
    "data_block_addr",
    "cycles_nr",
    "interval",
    "sync_type",
    "flags",
    "reserved1",
)
ASAM_XML_NAMESPACE = "{http://www.asam.net/mdf/v4}"
CAN_ID_PATTERN = re.compile(r"(?i:can)(?P<id>\d+)")
CAN_DATA_FRAME_PATTERN = re.compile(r"(?i:CAN_DataFrame)_(?P<id>\d+)")
LIN_DATA_FRAME_PATTERN = re.compile(r"(?i:LIN_Frame)_(?P<id>\d+)")
CN_COMMENT_TEMPLATE = """<CNcomment>
<TX>{}</TX>
<names>
{}
</names>
</CNcomment>"""
HD_COMMENT_TEMPLATE = """<HDcomment>
<TX>{}</TX>
<common_properties>
    <e name="author">{}</e>
    <e name="department">{}</e>
    <e name="project">{}</e>
    <e name="subject">{}</e>
</common_properties>
</HDcomment>"""
CANOPEN_TIME_FIELDS = ("ms", "days")
CANOPEN_DATE_FIELDS = (
    "ms",
    "min",
    "hour",
    "day",
    "month",
    "year",
    "summer_time",
    "day_of_week",
)
CAN_ERROR_TYPES = {
    0: "Other/unknown error ",
    1: "Bit Error",
    2: "Form Error",
    3: "Bit Stuffing Error",
    4: "CRC Error",
    5: "Acknowledgment Error",
}
 def from_dict(conversion: dict[str, Any]) -> v4b.ChannelConversion:
    if not conversion:
        conversion = None

    elif "a" in conversion:
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_LIN
        conversion = v4b.ChannelConversion(**conversion)

    elif "formula" in conversion:
        conversion["formula"] = conversion["formula"]
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_ALG
        conversion = v4b.ChannelConversion(**conversion)

    elif all(key in conversion for key in [f"P{i}" for i in range(1, 7)]):
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_RAT
        conversion = v4b.ChannelConversion(**conversion)

    elif "raw_0" in conversion and "phys_0" in conversion:
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_TAB
        nr = 0
        while f"phys_{nr}" in conversion:
            nr += 1
        conversion["val_param_nr"] = nr * 2
        if "interpolation" in conversion:
            conversion["conversion_type"] = v4c.CONVERSION_TYPE_TABI
        else:
            conversion["conversion_type"] = v4c.CONVERSION_TYPE_TAB
        conversion = v4b.ChannelConversion(**conversion)

    elif "mask_0" in conversion and "text_0" in conversion:
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_BITFIELD
        nr = 0
        while f"text_{nr}" in conversion:
            val = conversion[f"text_{nr}"]
            if isinstance(val, (bytes, str)):
                partial_conversion = {
                    "conversion_type": v4c.CONVERSION_TYPE_RTABX,
                    "upper_0": conversion[f"upper_{nr}"],
                    "lower_0": conversion[f"lower_{nr}"],
                    "text_0": conversion[f"text_{nr}"]
                    if isinstance(conversion[f"text_{nr}"], bytes)
                    else conversion[f"text_{nr}"].encode("utf-8"),
                    "default": b"",
                }
                conversion[f"text_{nr}"] = from_dict(partial_conversion)

            nr += 1

        conversion["ref_param_nr"] = nr
        conversion["val_param_nr"] = nr
        conversion = v4b.ChannelConversion(**conversion)

    elif "upper_0" in conversion and "phys_0" in conversion:
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_RTAB
        nr = 0
        while f"phys_{nr}" in conversion:
            nr += 1
        conversion["val_param_nr"] = nr * 3 + 1
        conversion = v4b.ChannelConversion(**conversion)

    elif "val_0" in conversion and "text_0" in conversion:
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_TABX
        nr = 0
        while f"text_{nr}" in conversion:
            val = conversion[f"text_{nr}"]
            if isinstance(val, str):
                conversion[f"text_{nr}"] = val.encode("utf-8")
            elif isinstance(val, dict):
                conversion[f"text_{nr}"] = from_dict(val)
            nr += 1

        val = conversion.get("default_addr", b"")
        if isinstance(val, str):
            conversion[f"default_addr"] = val.encode("utf-8")
        elif isinstance(val, dict):
            conversion[f"default_addr"] = from_dict(val)

        conversion["ref_param_nr"] = nr + 1
        conversion = v4b.ChannelConversion(**conversion)

    elif "upper_0" in conversion and "text_0" in conversion:
        conversion["conversion_type"] = v4c.CONVERSION_TYPE_RTABX
        nr = 0
        while f"text_{nr}" in conversion:
            val = conversion[f"text_{nr}"]
            if isinstance(val, str):
                conversion[f"text_{nr}"] = val.encode("utf-8")
            elif isinstance(val, dict):
                conversion[f"text_{nr}"] = from_dict(val)
            nr += 1

        conversion["ref_param_nr"] = nr + 1

        val = conversion.get("default_addr", b"")
        if isinstance(val, str):
            conversion[f"default_addr"] = val.encode("utf-8")
        elif isinstance(val, dict):
            conversion[f"default_addr"] = from_dict(val)
        conversion = v4b.ChannelConversion(**conversion)

    else:
        conversion = v4b.ChannelConversion(conversion_type=v4c.CONVERSION_TYPE_NON)

    return conversion def csv_bytearray2hex(val, size: int | None = None) -> str:
    """format CAN payload as hex strings

    b'\xa2\xc3\x08' -> A2 C3 08

    """
    if size is not None:
        val = val.tobytes()[:size].hex(" ", 1).upper()
    else:
        val = val.tobytes().hex(" ", 1).upper()

    return val
 def extract_cncomment_xml(comment: str) -> str:
    """extract *TX* tag or otherwise the *common_properties* from a xml comment

    Parameters
    ----------
    comment : str
        xml string comment

    Returns
    -------
    comment : str
        extracted string

    """

    comment = comment.replace(' xmlns="http://www.asam.net/mdf/v4"', "")
    try:
        comment = ET.fromstring(comment)
        match = comment.find(".//TX")
        if match is None:
            common_properties = comment.find(".//common_properties")
            if common_properties is not None:
                comment = []
                for e in common_properties:
                    field = f'{e.get("name")}: {e.text}'
                    comment.append(field)
                comment = "\n".join(field)
            else:
                comment = ""
        else:
            comment = match.text or ""
    except ET.ParseError:
        pass

    return comment
 def load_can_database(
    path: StrPathType, contents: bytes | str | None = None, **kwargs
) -> CanMatrix | None:
    path = Path(path)
    import_type = path.suffix.lstrip(".").lower()
    if contents is None:
        func = canmatrix.formats.loadp
        arg = path
    else:
        func = canmatrix.formats.loads
        arg = contents

    try:
        dbs = func(arg, import_type=import_type, key="db", **kwargs)
    except UnicodeDecodeError:
        if contents is None:
            contents = path.read_bytes()

        encoding = detect(contents)["encoding"]

        try:
            dbs = func(
                arg,
                import_type=import_type,
                key="db",
                encoding=encoding,
                **kwargs,
            )
        except:
            dbs = None

    if dbs:
        first_bus = list(dbs)[0]
        can_matrix = dbs[first_bus]
    else:
        can_matrix = None

    return can_matrix
 class MdfException(Exception):
    """MDF Exception class"""

    pass
 class MDF:
    """Unified access to MDF v3 and v4 files. Underlying _mdf's attributes and
    methods are linked to the `MDF` object via *setattr*. This is done to expose
    them to the user code and for performance considerations.

    Parameters
    ----------
    name : string | BytesIO | zipfile.ZipFile | bz2.BZ2File | gzip.GzipFile
        mdf file name (if provided it must be a real file name), file-like object or
        compressed file opened as Python object

        .. versionchanged:: 6.2.0

            added support for zipfile.ZipFile, bz2.BZ2File and gzip.GzipFile

    version : string
        mdf file version from ('2.00', '2.10', '2.14', '3.00', '3.10', '3.20',
        '3.30', '4.00', '4.10', '4.11', '4.20'); default '4.10'. This argument is
        only used for MDF objects created from scratch; for MDF objects created
        from a file the version is set to file version

    channels (None) : iterable
        channel names that will used for selective loading. This can dramatically
        improve the file loading time. Default None -> load all channels

        .. versionadded:: 6.1.0

        .. versionchanged:: 6.3.0 make the default None


    callback (\*\*kwargs) : function
        keyword only argument: function to call to update the progress; the
        function must accept two arguments (the current progress and maximum
        progress value)
    use_display_names (\*\*kwargs) : bool
        keyword only argument: for MDF4 files parse the XML channel comment to
        search for the display name; XML parsing is quite expensive so setting
        this to *False* can decrease the loading times very much; default
        *False*
    remove_source_from_channel_names (\*\*kwargs) : bool
        remove source from channel names ("Speed\XCP3" -> "Speed")
    copy_on_get (\*\*kwargs) : bool
        copy arrays in the get method; default *True*
    expand_zippedfile (\*\*kwargs) : bool
        only for bz2.BZ2File and gzip.GzipFile, load the file content into a
        BytesIO before parsing (avoids the huge performance penalty of doing
        random reads from the zipped file); default *True*
    raise_on_multiple_occurrences (\*\*kwargs) : bool
        raise exception when there are multiple channel occurrences in the file and
        the `get` call is ambiguous; default True

        .. versionadded:: 7.0.0

    temporary_folder (\*\*kwargs) : str | pathlib.Path
        folder to use for temporary files

        .. versionadded:: 7.0.0

    Examples
    --------
    >>> mdf = MDF(version='3.30') # new MDF object with version 3.30
    >>> mdf = MDF('path/to/file.mf4') # MDF loaded from file
    >>> mdf = MDF(BytesIO(data)) # MDF from file contents
    >>> mdf = MDF(zipfile.ZipFile('data.zip')) # MDF creating using the first valid MDF from archive
    >>> mdf = MDF(bz2.BZ2File('path/to/data.bz2', 'rb')) # MDF from bz2 object
    >>> mdf = MDF(gzip.GzipFile('path/to/data.gzip', 'rb')) # MDF from gzip object

    """

    _terminate = False

    def __init__(
        self,
        name: InputType | None = None,
        version: str = "4.10",
        channels: list[str] | None = None,
        **kwargs,
    ) -> None:
        self._mdf = None

        temporary_folder = kwargs.get("temporary_folder", None)
        if temporary_folder:
            try:
                os.makedirs(temporary_folder, exist_ok=True)
            except:
                kwargs["temporary_folder"] = None

        if name:
            if is_file_like(name):

                if isinstance(name, BytesIO):
                    original_name = None
                    file_stream = name
                    do_close = False

                elif isinstance(name, bz2.BZ2File):
                    original_name = Path(name._fp.name)
                    name = get_temporary_filename(original_name, dir=temporary_folder)
                    name.write_bytes(name.read())
                    file_stream = open(name, "rb")
                    do_close = True
                elif isinstance(name, gzip.GzipFile):

                    original_name = Path(name.name)
                    name = get_temporary_filename(original_name, dir=temporary_folder)
                    name.write_bytes(name.read())
                    file_stream = open(name, "rb")
                    do_close = True

            else:
                name = original_name = Path(name)
                if not name.is_file() or not name.exists():
                    raise MdfException(f'File "{name}" does not exist')

                if original_name.suffix.lower() in (".mf4z", ".zip"):
                    name = get_temporary_filename(original_name, dir=temporary_folder)
                    with zipfile.ZipFile(original_name, allowZip64=True) as archive:
                        files = archive.namelist()
                        if len(files) != 1:
                            raise Exception(
                                "invalid zipped MF4: must contain a single file"
                            )
                        fname = files[0]

                        if Path(fname).suffix.lower() not in (".mdf", ".dat", ".mf4"):
                            raise Exception(
                                "invalid zipped MF4: must contain a single MDF file"
                            )

                        tmpdir = mkdtemp()
                        output = archive.extract(fname, tmpdir)

                        move(output, name)

                file_stream = open(name, "rb")
                do_close = True

            file_stream.seek(0)
            magic_header = file_stream.read(8)

            if magic_header.strip() not in (b"MDF", b"UnFinMF"):
                raise MdfException(
                    f'"{name}" is not a valid ASAM MDF file: magic header is {magic_header}'
                )

            file_stream.seek(8)
            version = file_stream.read(4).decode("ascii").strip(" \0")
            if not version:
                _, version = get_measurement_timestamp_and_version(file_stream)

            if do_close:
                file_stream.close()

            kwargs["original_name"] = original_name

            if version in MDF3_VERSIONS:
                self._mdf = MDF3(name, channels=channels, **kwargs)
            elif version in MDF4_VERSIONS:
                self._mdf = MDF4(name, channels=channels, **kwargs)
            elif version in MDF2_VERSIONS:
                self._mdf = MDF2(name, channels=channels, **kwargs)
            else:
                message = f'"{name}" is not a supported MDF file; "{version}" file version was found'
                raise MdfException(message)

        else:
            kwargs["original_name"] = None
            version = validate_version_argument(version)
            if version in MDF2_VERSIONS:
                self._mdf = MDF3(version=version, **kwargs)
            elif version in MDF3_VERSIONS:
                self._mdf = MDF3(version=version, **kwargs)
            elif version in MDF4_VERSIONS:
                self._mdf = MDF4(version=version, **kwargs)
            else:
                message = (
                    f'"{version}" is not a supported MDF file version; '
                    f"Supported versions are {SUPPORTED_VERSIONS}"
                )
                raise MdfException(message)

        # we need a backreference to the MDF object to avoid it being garbage
        # collected in code like this:
        # MDF(filename).convert('4.10')
        self._mdf._parent = self

    def __setattr__(self, item: str, value: Any) -> None:
        if item == "_mdf":
            super().__setattr__(item, value)
        else:
            setattr(self._mdf, item, value)

    def __getattr__(self, item: str) -> Any:
        return getattr(self._mdf, item)

    def __dir__(self) -> list[str]:
        return sorted(set(super().__dir__()) | set(dir(self._mdf)))

    def __enter__(self) -> MDF:
        return self

    def __exit__(
        self,
        exc_type: Type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> bool | None:
        if self._mdf is not None:
            try:
                self.close()
            except:
                pass
        self._mdf = None

    def __del__(self) -> None:
        if self._mdf is not None:
            try:
                self.close()
            except:
                pass
        self._mdf = None

    def __lt__(self, other: MDF) -> bool:
        if self.header.start_time < other.header.start_time:
            return True
        elif self.header.start_time > other.header.start_time:
            return False
        else:

            t_min = []
            for i, group in enumerate(self.groups):
                cycles_nr = group.channel_group.cycles_nr
                if cycles_nr and i in self.masters_db:
                    master_min = self.get_master(i, record_offset=0, record_count=1)
                    if len(master_min):
                        t_min.append(master_min[0])

            other_t_min = []
            for i, group in enumerate(other.groups):
                cycles_nr = group.channel_group.cycles_nr
                if cycles_nr and i in other.masters_db:
                    master_min = other.get_master(i, record_offset=0, record_count=1)
                    if len(master_min):
                        other_t_min.append(master_min[0])

            if not t_min or not other_t_min:
                return True
            else:
                return min(t_min) < min(other_t_min)

    def _transfer_events(self, other: MDF) -> None:
        def get_scopes(event, events):
            if event.scopes:
                return event.scopes
            else:
                if event.parent is not None:
                    return get_scopes(events[event.parent], events)
                elif event.range_start is not None:
                    return get_scopes(events[event.range_start], events)
                else:
                    return event.scopes

        if other.version >= "4.00":
            for event in other.events:
                if self.version >= "4.00":
                    new_event = deepcopy(event)
                    event_valid = True
                    for i, ref in enumerate(new_event.scopes):
                        try:
                            dg_cntr, ch_cntr = ref
                            try:
                                (self.groups[dg_cntr].channels[ch_cntr])
                            except:
                                event_valid = False
                        except TypeError:
                            dg_cntr = ref
                            try:
                                (self.groups[dg_cntr].channel_group)
                            except:
                                event_valid = False
                    # ignore attachments for now
                    for i in range(new_event.attachment_nr):
                        key = f"attachment_{i}_addr"
                        event[key] = 0
                    if event_valid:
                        self.events.append(new_event)
                else:
                    ev_type = event.event_type
                    ev_range = event.range_type
                    ev_base = event.sync_base
                    ev_factor = event.sync_factor

                    timestamp = ev_base * ev_factor

                    try:
                        comment = ET.fromstring(
                            event.comment.replace(
                                ' xmlns="http://www.asam.net/mdf/v4"', ""
                            )
                        )
                        pre = comment.find(".//pre_trigger_interval")
                        if pre is not None:
                            pre = float(pre.text)
                        else:
                            pre = 0.0
                        post = comment.find(".//post_trigger_interval")
                        if post is not None:
                            post = float(post.text)
                        else:
                            post = 0.0
                        comment = comment.find(".//TX")
                        if comment is not None:
                            comment = comment.text
                        else:
                            comment = ""

                    except:
                        pre = 0.0
                        post = 0.0
                        comment = event.comment

                    if comment:
                        comment += ": "

                    if ev_range == v4c.EVENT_RANGE_TYPE_BEGINNING:
                        comment += "Begin of "
                    elif ev_range == v4c.EVENT_RANGE_TYPE_END:
                        comment += "End of "
                    else:
                        comment += "Single point "

                    if ev_type == v4c.EVENT_TYPE_RECORDING:
                        comment += "recording"
                    elif ev_type == v4c.EVENT_TYPE_RECORDING_INTERRUPT:
                        comment += "recording interrupt"
                    elif ev_type == v4c.EVENT_TYPE_ACQUISITION_INTERRUPT:
                        comment += "acquisition interrupt"
                    elif ev_type == v4c.EVENT_TYPE_START_RECORDING_TRIGGER:
                        comment += "measurement start trigger"
                    elif ev_type == v4c.EVENT_TYPE_STOP_RECORDING_TRIGGER:
                        comment += "measurement stop trigger"
                    elif ev_type == v4c.EVENT_TYPE_TRIGGER:
                        comment += "trigger"
                    else:
                        comment += "marker"

                    scopes = get_scopes(event, other.events)
                    if scopes:
                        for i, ref in enumerate(scopes):
                            event_valid = True
                            try:
                                dg_cntr, ch_cntr = ref
                                try:
                                    (self.groups[dg_cntr])
                                except:
                                    event_valid = False
                            except TypeError:
                                dg_cntr = ref
                                try:
                                    (self.groups[dg_cntr])
                                except:
                                    event_valid = False
                            if event_valid:

                                self.add_trigger(
                                    dg_cntr,
                                    timestamp,
                                    pre_time=pre,
                                    post_time=post,
                                    comment=comment,
                                )
                    else:
                        for i, _ in enumerate(self.groups):
                            self.add_trigger(
                                i,
                                timestamp,
                                pre_time=pre,
                                post_time=post,
                                comment=comment,
                            )

        else:
            for trigger_info in other.iter_get_triggers():
                comment = trigger_info["comment"]
                timestamp = trigger_info["time"]
                group = trigger_info["group"]

                if self.version < "4.00":
                    self.add_trigger(
                        group,
                        timestamp,
                        pre_time=trigger_info["pre_time"],
                        post_time=trigger_info["post_time"],
                        comment=comment,
                    )
                else:
                    if timestamp:
                        ev_type = v4c.EVENT_TYPE_TRIGGER
                    else:
                        ev_type = v4c.EVENT_TYPE_START_RECORDING_TRIGGER
                    event = EventBlock(
                        event_type=ev_type,
                        sync_base=int(timestamp * 10**9),
                        sync_factor=10**-9,
                        scope_0_addr=0,
                    )
                    event.comment = comment
                    event.scopes.append(group)
                    self.events.append(event)

    def _transfer_header_data(self, other: MDF, message: str = "") -> None:
        self.header.author = other.header.author
        self.header.department = other.header.department
        self.header.project = other.header.project
        self.header.subject = other.header.subject
        self.header.comment = other.header.comment
        if self.version >= "4.00" and message:
            fh = FileHistory()
            fh.comment = f"""<FHcomment>
    <TX>{message}</TX>
    <tool_id>asammdf</tool_id>
    <tool_vendor>asammdf</tool_vendor>
    <tool_version>{__version__}</tool_version>
</FHcomment>"""

            self.file_history = [fh]

    @staticmethod
    def _transfer_channel_group_data(
        sgroup: ChannelGroupType, ogroup: ChannelGroupType
    ) -> None:
        if not hasattr(sgroup, "acq_name") or not hasattr(ogroup, "acq_name"):
            sgroup.comment = ogroup.comment
        else:

            sgroup.flags = ogroup.flags
            sgroup.path_separator = ogroup.path_separator
            sgroup.comment = ogroup.comment
            sgroup.acq_name = ogroup.acq_name
            acq_source = ogroup.acq_source
            if acq_source:
                sgroup.acq_source = acq_source.copy()

    def _transfer_metadata(self, other: MDF, message: str = "") -> None:
        self._transfer_events(other)
        self._transfer_header_data(other, message)

    def __contains__(self, channel: str) -> bool:
        """if *'channel name'* in *'mdf file'*"""
        return channel in self.channels_db

    def __iter__(self) -> Iterator[Signal]:
        """iterate over all the channels found in the file; master channels
        are skipped from iteration

        """
        yield from self.iter_channels()

    def configure(
        self,
        *,
        from_other: MDF_v2_v3_v4 | None = None,
        read_fragment_size: int | None = None,
        write_fragment_size: int | None = None,
        use_display_names: bool | None = None,
        single_bit_uint_as_bool: bool | None = None,
        integer_interpolation: IntInterpolationModeType
        | IntegerInterpolation
        | None = None,
        copy_on_get: bool | None = None,
        float_interpolation: FloatInterpolationModeType
        | FloatInterpolation
        | None = None,
        raise_on_multiple_occurrences: bool | None = None,
        temporary_folder: str | None = None,
    ) -> None:
        """configure MDF parameters

        The default values for the options are the following:
        * read_fragment_size = 0
        * write_fragment_size = 4MB
        * use_display_names = False
        * single_bit_uint_as_bool = False
        * integer_interpolation = 0 (fill - use previous sample)
        * float_interpolation = 1 (linear interpolation)
        * copy_on_get = False
        * raise_on_multiple_occurrences = True
        * temporary_folder = ""

        Parameters
        ----------
        read_fragment_size : int
            size hint of split data blocks, default 8MB; if the initial size is
            smaller, then no data list is used. The actual split size depends on
            the data groups' records size
        write_fragment_size : int
            size hint of split data blocks, default 4MB; if the initial size is
            smaller, then no data list is used. The actual split size depends on
            the data groups' records size. Maximum size is 4MB to ensure
            compatibility with CANape
        use_display_names : bool
            search for display name in the Channel XML comment
        single_bit_uint_as_bool : bool
            return single bit channels are np.bool arrays
        integer_interpolation : int
            interpolation mode for integer channels:

                * 0 - repeat previous sample
                * 1 - use linear interpolation
                * 2 - hybrid interpolation: channels with integer data type (raw values) that have a
                  conversion that outputs float values will use linear interpolation, otherwise
                  the previous sample is used

                .. versionchanged:: 6.2.0
                    added hybrid mode interpolation

        copy_on_get : bool
            copy arrays in the get method

        float_interpolation : int
            interpolation mode for float channels:

                * 0 - repeat previous sample
                * 1 - use linear interpolation

                .. versionadded:: 6.2.0

        raise_on_multiple_occurrences : bool
            raise exception when there are multiple channel occurrences in the file and
            the `get` call is ambiguous; default True

            .. versionadded:: 6.2.0

        from_other : MDF
            copy configuration options from other MDF

            .. versionadded:: 6.2.0

        temporary_folder : str
            default folder for temporary files

            .. versionadded:: 7.0.0

        """

        if from_other is not None:
            self._read_fragment_size = from_other._read_fragment_size
            self._write_fragment_size = from_other._write_fragment_size
            self._use_display_names = from_other._use_display_names
            self._single_bit_uint_as_bool = from_other._single_bit_uint_as_bool
            self._integer_interpolation = from_other._integer_interpolation
            self.copy_on_get = from_other.copy_on_get
            self._float_interpolation = from_other._float_interpolation
            self._raise_on_multiple_occurrences = (
                from_other._raise_on_multiple_occurrences
            )

        if read_fragment_size is not None:
            self._read_fragment_size = int(read_fragment_size)

        if write_fragment_size is not None:
            self._write_fragment_size = min(int(write_fragment_size), 4 * 2**20)

        if use_display_names is not None:
            self._use_display_names = bool(use_display_names)

        if single_bit_uint_as_bool is not None:
            self._single_bit_uint_as_bool = bool(single_bit_uint_as_bool)

        if integer_interpolation is not None:
            self._integer_interpolation = IntegerInterpolation(integer_interpolation)

        if copy_on_get is not None:
            self.copy_on_get = copy_on_get

        if float_interpolation is not None:
            self._float_interpolation = FloatInterpolation(float_interpolation)

        if temporary_folder is not None:
            try:
                os.makedirs(temporary_folder, exist_ok=True)
                self.temporary_folder = temporary_folder
            except:
                self.temporary_folder = None

        if raise_on_multiple_occurrences is not None:
            self._raise_on_multiple_occurrences = bool(raise_on_multiple_occurrences)

    def convert(self, version: str) -> MDF:
        """convert *MDF* to other version

        Parameters
        ----------
        version : str
            new mdf file version from ('2.00', '2.10', '2.14', '3.00', '3.10',
            '3.20', '3.30', '4.00', '4.10', '4.11', '4.20'); default '4.10'

        Returns
        -------
        out : MDF
            new *MDF* object

        """
        version = validate_version_argument(version)

        out = MDF(version=version, **self._kwargs)

        out.configure(from_other=self)

        out.header.start_time = self.header.start_time

        groups_nr = len(self.virtual_groups)

        if self._callback:
            self._callback(0, groups_nr)

        cg_nr = None

        self.configure(copy_on_get=False)

        # walk through all groups and get all channels
        for i, virtual_group in enumerate(self.virtual_groups):

            for idx, sigs in enumerate(
                self._yield_selected_signals(virtual_group, version=version)
            ):
                if idx == 0:
                    if sigs:
                        cg = self.groups[virtual_group].channel_group
                        cg_nr = out.append(
                            sigs,
                            common_timebase=True,
                            comment=cg.comment,
                        )
                        MDF._transfer_channel_group_data(
                            out.groups[cg_nr].channel_group, cg
                        )
                    else:
                        break
                else:
                    out.extend(cg_nr, sigs)

            if self._callback:
                self._callback(i + 1, groups_nr)

            if self._terminate:
                return

        out._transfer_metadata(self, message=f"Converted from <{self.name}>")
        self.configure(copy_on_get=True)
        if self._callback:
            out._callback = out._mdf._callback = self._callback

        return out

    def cut(
        self,
        start: float | None = None,
        stop: float | None = None,
        whence: int = 0,
        version: str | None = None,
        include_ends: bool = True,
        time_from_zero: bool = False,
    ) -> MDF:
        """cut *MDF* file. *start* and *stop* limits are absolute values
        or values relative to the first timestamp depending on the *whence*
        argument.

        Parameters
        ----------
        start : float
            start time, default *None*. If *None* then the start of measurement
            is used
        stop : float
            stop time, default *None*. If *None* then the end of measurement is
            used
        whence : int
            how to search for the start and stop values

            * 0 : absolute
            * 1 : relative to first timestamp

        version : str
            new mdf file version from ('2.00', '2.10', '2.14', '3.00', '3.10',
            '3.20', '3.30', '4.00', '4.10', '4.11', 4.20'); default *None* and in this
            case the original file version is used
        include_ends : bool
            include the *start* and *stop* timestamps after cutting the signal.
            If *start* and *stop* are found in the original timestamps, then
            the new samples will be computed using interpolation. Default *True*
        time_from_zero : bool
            start time stamps from 0s in the cut measurement

        Returns
        -------
        out : MDF
            new MDF object

        """

        if version is None:
            version = self.version
        else:
            version = validate_version_argument(version)

        out = MDF(
            version=version,
            **self._kwargs,
        )

        integer_interpolation_mode = self._integer_interpolation
        float_interpolation_mode = self._float_interpolation
        out.configure(from_other=self)

        self.configure(copy_on_get=False)

        if whence == 1:
            timestamps = []
            for group in self.virtual_groups:
                master = self.get_master(group, record_offset=0, record_count=1)
                if master.size:
                    timestamps.append(master[0])

            if timestamps:
                first_timestamp = np.amin(timestamps)
            else:
                first_timestamp = 0

            if start is not None:
                start += first_timestamp
            if stop is not None:
                stop += first_timestamp

        if time_from_zero:
            delta = start
            t_epoch = self.header.start_time.timestamp() + delta
            out.header.start_time = datetime.fromtimestamp(t_epoch)
        else:
            delta = 0
            out.header.start_time = self.header.start_time

        groups_nr = len(self.virtual_groups)

        if self._callback:
            self._callback(0, groups_nr)

        # walk through all groups and get all channels
        for i, (group_index, virtual_group) in enumerate(self.virtual_groups.items()):

            included_channels = self.included_channels(group_index)[group_index]
            if not included_channels:
                continue

            idx = 0
            signals = []
            for j, sigs in enumerate(
                self._yield_selected_signals(group_index, groups=included_channels)
            ):
                if not sigs:
                    break
                if j == 0:
                    master = sigs[0].timestamps
                    signals = sigs
                else:
                    master = sigs[0][0]

                if not len(master):
                    continue

                needs_cutting = True

                # check if this fragement is within the cut interval or
                # if the cut interval has ended
                if start is None and stop is None:
                    fragment_start = None
                    fragment_stop = None
                    start_index = 0
                    stop_index = len(master)
                    needs_cutting = False
                elif start is None:
                    fragment_start = None
                    start_index = 0
                    if master[0] > stop:
                        break
                    else:
                        fragment_stop = min(stop, master[-1])
                        stop_index = np.searchsorted(
                            master, fragment_stop, side="right"
                        )
                        if stop_index == len(master):
                            needs_cutting = False

                elif stop is None:
                    fragment_stop = None
                    if master[-1] < start:
                        continue
                    else:
                        fragment_start = max(start, master[0])
                        start_index = np.searchsorted(
                            master, fragment_start, side="left"
                        )
                        stop_index = len(master)
                        if start_index == 0:
                            needs_cutting = False
                else:
                    if master[0] > stop:
                        break
                    elif master[-1] < start:
                        continue
                    else:
                        fragment_start = max(start, master[0])
                        start_index = np.searchsorted(
                            master, fragment_start, side="left"
                        )
                        fragment_stop = min(stop, master[-1])
                        stop_index = np.searchsorted(
                            master, fragment_stop, side="right"
                        )
                        if start_index == 0 and stop_index == len(master):
                            needs_cutting = False

                # update the signal is this is not the first yield
                if j:
                    for signal, (samples, invalidation) in zip(signals, sigs[1:]):
                        signal.samples = samples
                        signal.timestamps = master
                        signal.invalidation_bits = invalidation

                if needs_cutting:
                    master = (
                        Signal(master, master, name="_")
                        .cut(
                            fragment_start,
                            fragment_stop,
                            include_ends,
                            integer_interpolation_mode=integer_interpolation_mode,
                            float_interpolation_mode=float_interpolation_mode,
                        )
                        .timestamps
                    )

                    if not len(master):
                        continue

                    signals = [
                        sig.cut(
                            master[0],
                            master[-1],
                            include_ends=include_ends,
                            integer_interpolation_mode=integer_interpolation_mode,
                            float_interpolation_mode=float_interpolation_mode,
                        )
                        for sig in signals
                    ]

                else:
                    for sig in signals:
                        native = sig.samples.dtype.newbyteorder("=")
                        if sig.samples.dtype != native:
                            sig.samples = sig.samples.astype(native)

                if time_from_zero:
                    master = master - delta
                    for sig in signals:
                        sig.timestamps = master

                if idx == 0:

                    if start:
                        start_ = f"{start}s"
                    else:
                        start_ = "start of measurement"
                    if stop:
                        stop_ = f"{stop}s"
                    else:
                        stop_ = "end of measurement"
                    cg = self.groups[group_index].channel_group
                    cg_nr = out.append(
                        signals,
                        common_timebase=True,
                        comment=cg.comment,
                    )
                    MDF._transfer_channel_group_data(
                        out.groups[cg_nr].channel_group, cg
                    )

                else:
                    sigs = [(sig.samples, sig.invalidation_bits) for sig in signals]
                    sigs.insert(0, (master, None))
                    out.extend(cg_nr, sigs)

                idx += 1

            # if the cut interval is not found in the measurement
            # then append a data group with 0 cycles
            if idx == 0 and signals:
                for sig in signals:
                    sig.samples = sig.samples[:0]
                    sig.timestamps = sig.timestamps[:0]
                    if sig.invalidation_bits is not None:
                        sig.invalidation_bits = sig.invalidation_bits[:0]

                if start:
                    start_ = f"{start}s"
                else:
                    start_ = "start of measurement"
                if stop:
                    stop_ = f"{stop}s"
                else:
                    stop_ = "end of measurement"
                cg = self.groups[group_index].channel_group
                cg_nr = out.append(
                    signals,
                    common_timebase=True,
                )
                MDF._transfer_channel_group_data(out.groups[cg_nr].channel_group, cg)

            if self._callback:
                self._callback(i + 1, groups_nr)

            if self._terminate:
                return

        self.configure(copy_on_get=True)

        out._transfer_metadata(self, message=f"Cut from {start_} to {stop_}")
        if self._callback:
            out._callback = out._mdf._callback = self._callback
        return out

    def export(
        self,
        fmt: Literal["csv", "hdf5", "mat", "parquet"],
        filename: StrPathType | None = None,
        **kwargs,
    ) -> None:
        r"""export *MDF* to other formats. The *MDF* file name is used is
        available, else the *filename* argument must be provided.

        The *pandas* export option was removed. you should use the method
        *to_dataframe* instead.

        Parameters
        ----------
        fmt : string
            can be one of the following:

            * `csv` : CSV export that uses the "," delimiter. This option
              will generate a new csv file for each data group
              (<MDFNAME>_DataGroup_<cntr>.csv)

            * `hdf5` : HDF5 file output; each *MDF* data group is mapped to
              a *HDF5* group with the name 'DataGroup_<cntr>'
              (where <cntr> is the index)

            * `mat` : Matlab .mat version 4, 5 or 7.3 export. If
              *single_time_base==False* the channels will be renamed in the mat
              file to 'D<cntr>_<channel name>'. The channel group
              master will be renamed to 'DM<cntr>_<channel name>'
              ( *<cntr>* is the data group index starting from 0)

            * `parquet` : export to Apache parquet format

        filename : string | pathlib.Path
            export file name

        \*\*kwargs

            * `single_time_base`: resample all channels to common time base,
              default *False*
            * `raster`: float time raster for resampling. Valid if
              *single_time_base* is *True*
            * `time_from_zero`: adjust time channel to start from 0
            * `use_display_names`: use display name instead of standard channel
              name, if available.
            * `empty_channels`: behaviour for channels without samples; the
              options are *skip* or *zeros*; default is *skip*
            * `format`: only valid for *mat* export; can be '4', '5' or '7.3',
              default is '5'
            * `oned_as`: only valid for *mat* export; can be 'row' or 'column'
            * `keep_arrays` : keep arrays and structure channels as well as the
              component channels. If *True* this can be very slow. If *False*
              only the component channels are saved, and their names will be
              prefixed with the parent channel.
            * `reduce_memory_usage` : bool
              reduce memory usage by converting all float columns to float32 and
              searching for minimum dtype that can reprezent the values found
              in integer columns; default *False*
            * `compression` : str
              compression to be used

              * for ``parquet`` : "GZIP" or "SANPPY"
              * for ``hfd5`` : "gzip", "lzf" or "szip"
              * for ``mat`` : bool

            * `time_as_date` (False) : bool
              export time as local timezone datetimee; only valid for CSV export

              .. versionadded:: 5.8.0

            * `ignore_value2text_conversions` (False) : bool
              valid only for the channels that have value to text conversions and
              if *raw=False*. If this is True then the raw numeric values will be
              used, and the conversion will not be applied.

              .. versionadded:: 5.8.0

            * raw (False) : bool
              export all channels using the raw values

              .. versionadded:: 6.0.0

            * delimiter (',') : str
              only valid for CSV: see cpython documentation for csv.Dialect.delimiter

              .. versionadded:: 6.2.0

            * doublequote (True) : bool
              only valid for CSV: see cpython documentation for csv.Dialect.doublequote

              .. versionadded:: 6.2.0

            * escapechar (None) : str
              only valid for CSV: see cpython documentation for csv.Dialect.escapechar

              .. versionadded:: 6.2.0

            * lineterminator ("\\r\\n") : str
              only valid for CSV: see cpython documentation for csv.Dialect.lineterminator

              .. versionadded:: 6.2.0

            * quotechar ('"') : str
              only valid for CSV: see cpython documentation for csv.Dialect.quotechar

              .. versionadded:: 6.2.0

            * quoting ("MINIMAL") : str
              only valid for CSV: see cpython documentation for csv.Dialect.quoting. Use the
              last part of the quoting constant name

              .. versionadded:: 6.2.0


        """

        header_items = (
            "date",
            "time",
            "author_field",
            "department_field",
            "project_field",
            "subject_field",
        )

        if fmt != "pandas" and filename is None and self.name is None:
            message = (
                "Must specify filename for export"
                "if MDF was created without a file name"
            )
            logger.warning(message)
            return

        single_time_base = kwargs.get("single_time_base", False)
        raster = kwargs.get("raster", None)
        time_from_zero = kwargs.get("time_from_zero", True)
        use_display_names = kwargs.get("use_display_names", True)
        empty_channels = kwargs.get("empty_channels", "skip")
        format = kwargs.get("format", "5")
        oned_as = kwargs.get("oned_as", "row")
        reduce_memory_usage = kwargs.get("reduce_memory_usage", False)
        compression = kwargs.get("compression", "")
        time_as_date = kwargs.get("time_as_date", False)
        ignore_value2text_conversions = kwargs.get(
            "ignore_value2text_conversions", False
        )
        raw = bool(kwargs.get("raw", False))

        if compression == "SNAPPY":
            try:
                import snappy
            except ImportError:
                logger.warning(
                    "snappy compressor is not installed; compression will be set to GZIP"
                )
                compression = "GZIP"

        filename = Path(filename) if filename else self.name

        if fmt == "parquet":
            try:
                from fastparquet import write as write_parquet
            except ImportError:
                logger.warning(
                    "fastparquet not found; export to parquet is unavailable"
                )
                return

        elif fmt == "hdf5":
            try:
                from h5py import File as HDF5
            except ImportError:
                logger.warning("h5py not found; export to HDF5 is unavailable")
                return

        elif fmt == "mat":
            if format == "7.3":
                try:
                    from hdf5storage import savemat
                except ImportError:
                    logger.warning(
                        "hdf5storage not found; export to mat v7.3 is unavailable"
                    )
                    return
            else:
                try:
                    from scipy.io import savemat
                except ImportError:
                    logger.warning("scipy not found; export to mat is unavailable")
                    return

        elif fmt not in ("csv",):
            raise MdfException(f"Export to {fmt} is not implemented")

        name = ""

        if self._callback:
            self._callback(0, 100)

        if single_time_base or fmt == "parquet":
            df = self.to_dataframe(
                raster=raster,
                time_from_zero=time_from_zero,
                use_display_names=use_display_names,
                empty_channels=empty_channels,
                reduce_memory_usage=reduce_memory_usage,
                ignore_value2text_conversions=ignore_value2text_conversions,
                raw=raw,
                numeric_1D_only=fmt == "parquet",
            )
            units = OrderedDict()
            comments = OrderedDict()
            used_names = UniqueDB()

            dropped = {}

            groups_nr = len(self.groups)
            for i, grp in enumerate(self.groups):
                if self._terminate:
                    return

                for ch in grp.channels:

                    if use_display_names:
                        channel_name = (
                            list(ch.display_names)[0] if ch.display_names else ch.name
                        )
                    else:
                        channel_name = ch.name

                    channel_name = used_names.get_unique_name(channel_name)

                    if hasattr(ch, "unit"):
                        unit = ch.unit
                        if ch.conversion:
                            unit = unit or ch.conversion.unit
                    else:
                        unit = ""
                    comment = ch.comment

                    units[channel_name] = unit
                    comments[channel_name] = comment

                if self._callback:
                    self._callback(i + 1, groups_nr * 2)

        if fmt == "hdf5":
            filename = filename.with_suffix(".hdf")

            if single_time_base:

                with HDF5(str(filename), "w") as hdf:
                    # header information
                    group = hdf.create_group(str(filename))

                    if self.version in MDF2_VERSIONS + MDF3_VERSIONS:
                        for item in header_items:
                            group.attrs[item] = self.header[item].replace(b"\0", b"")

                    # save each data group in a HDF5 group called
                    # "DataGroup_<cntr>" with the index starting from 1
                    # each HDF5 group will have a string attribute "master"
                    # that will hold the name of the master channel

                    count = len(df.columns)

                    for i, channel in enumerate(df):
                        samples = df[channel]
                        unit = units.get(channel, "")
                        comment = comments.get(channel, "")

                        if samples.dtype.kind == "O":
                            if isinstance(samples[0], np.ndarray):
                                samples = np.vstack(samples)
                            else:
                                continue

                        if compression:
                            dataset = group.create_dataset(
                                channel, data=samples, compression=compression
                            )
                        else:
                            dataset = group.create_dataset(channel, data=samples)
                        unit = unit.replace("\0", "")
                        if unit:
                            dataset.attrs["unit"] = unit
                        comment = comment.replace("\0", "")
                        if comment:
                            dataset.attrs["comment"] = comment

                        if self._callback:
                            self._callback(i + 1 + count, count * 2)

            else:
                with HDF5(str(filename), "w") as hdf:
                    # header information
                    group = hdf.create_group(str(filename))

                    if self.version in MDF2_VERSIONS + MDF3_VERSIONS:
                        for item in header_items:
                            group.attrs[item] = self.header[item].replace(b"\0", b"")

                    # save each data group in a HDF5 group called
                    # "DataGroup_<cntr>" with the index starting from 1
                    # each HDF5 group will have a string attribute "master"
                    # that will hold the name of the master channel

                    groups_nr = len(self.virtual_groups)
                    for i, (group_index, virtual_group) in enumerate(
                        self.virtual_groups.items()
                    ):
                        channels = self.included_channels(group_index)[group_index]

                        if not channels:
                            continue

                        names = UniqueDB()
                        if self._terminate:
                            return

                        if len(virtual_group.groups) == 1:
                            comment = self.groups[
                                virtual_group.groups[0]
                            ].channel_group.comment
                        else:
                            comment = "Virtual group i"

                        group_name = r"/" + f"ChannelGroup_{i}"
                        group = hdf.create_group(group_name)

                        group.attrs["comment"] = comment

                        master_index = self.masters_db.get(group_index, -1)

                        if master_index >= 0:
                            group.attrs["master"] = (
                                self.groups[group_index].channels[master_index].name
                            )

                        channels = [
                            (None, gp_index, ch_index)
                            for gp_index, channel_indexes in channels.items()
                            for ch_index in channel_indexes
                        ]

                        if not channels:
                            continue

                        channels = self.select(channels, raw=raw)

                        for j, sig in enumerate(channels):
                            if use_display_names:
                                name = (
                                    list(sig.display_names)[0]
                                    if sig.display_names
                                    else sig.name
                                )
                            else:
                                name = sig.name
                            name = name.replace("\\", "_").replace("/", "_")
                            name = names.get_unique_name(name)
                            if reduce_memory_usage:
                                sig.samples = downcast(sig.samples)
                            if compression:
                                dataset = group.create_dataset(
                                    name, data=sig.samples, compression=compression
                                )
                            else:
                                dataset = group.create_dataset(
                                    name, data=sig.samples, dtype=sig.samples.dtype
                                )
                            unit = sig.unit.replace("\0", "")
                            if unit:
                                dataset.attrs["unit"] = unit
                            comment = sig.comment.replace("\0", "")
                            if comment:
                                dataset.attrs["comment"] = comment

                        if self._callback:
                            self._callback(i + 1, groups_nr)

        elif fmt == "csv":
            fmtparams = {
                "delimiter": kwargs.get("delimiter", ",")[0],
                "doublequote": kwargs.get("doublequote", True),
                "lineterminator": kwargs.get("lineterminator", "\r\n"),
                "quotechar": kwargs.get("quotechar", '"')[0],
            }

            quoting = kwargs.get("quoting", "MINIMAL").upper()
            quoting = getattr(csv, f"QUOTE_{quoting}")

            fmtparams["quoting"] = quoting

            escapechar = kwargs.get("escapechar", None)
            if escapechar is not None:
                escapechar = escapechar[0]

            fmtparams["escapechar"] = escapechar

            if single_time_base:
                filename = filename.with_suffix(".csv")
                message = f'Writing csv export to file "{filename}"'
                logger.info(message)

                if time_as_date:
                    index = (
                        pd.to_datetime(
                            df.index + self.header.start_time.timestamp(), unit="s"
                        )
                        .tz_localize("UTC")
                        .tz_convert(LOCAL_TIMEZONE)
                        .astype(str)
                    )
                    df.index = index
                    df.index.name = "timestamps"

                if hasattr(self, "can_logging_db") and self.can_logging_db:

                    dropped = {}

                    for name_ in df.columns:
                        if name_.endswith("CAN_DataFrame.ID"):
                            dropped[name_] = pd.Series(
                                csv_int2hex(df[name_].astype("<u4") & 0x1FFFFFFF),
                                index=df.index,
                            )

                        elif name_.endswith("CAN_DataFrame.DataBytes"):
                            dropped[name_] = pd.Series(
                                csv_bytearray2hex(df[name_]), index=df.index
                            )

                    df = df.drop(columns=list(dropped))
                    for name, s in dropped.items():
                        df[name] = s

                with open(filename, "w", newline="") as csvfile:

                    writer = csv.writer(csvfile, **fmtparams)

                    names_row = [df.index.name, *df.columns]
                    writer.writerow(names_row)

                    if reduce_memory_usage:
                        vals = [df.index, *(df[name] for name in df)]
                    else:
                        vals = [
                            df.index.to_list(),
                            *(df[name].to_list() for name in df),
                        ]
                    count = len(df.index)

                    if self._terminate:
                        return

                    for i, row in enumerate(zip(*vals)):
                        writer.writerow(row)

                        if self._callback:
                            self._callback(i + 1 + count, count * 2)

            else:

                filename = filename.with_suffix(".csv")

                gp_count = len(self.virtual_groups)
                for i, (group_index, virtual_group) in enumerate(
                    self.virtual_groups.items()
                ):

                    if self._terminate:
                        return

                    message = f"Exporting group {i+1} of {gp_count}"
                    logger.info(message)

                    if len(virtual_group.groups) == 1:
                        comment = self.groups[
                            virtual_group.groups[0]
                        ].channel_group.comment
                    else:
                        comment = ""

                    if comment:
                        for char in f'\n\t\r\b <>\/:"?*|':
                            comment = comment.replace(char, "_")
                        group_csv_name = (
                            filename.parent
                            / f"{filename.stem}.ChannelGroup_{i}_{comment}.csv"
                        )
                    else:
                        group_csv_name = (
                            filename.parent / f"{filename.stem}.ChannelGroup_{i}.csv"
                        )

                    df = self.get_group(
                        group_index,
                        raster=raster,
                        time_from_zero=time_from_zero,
                        use_display_names=use_display_names,
                        reduce_memory_usage=reduce_memory_usage,
                        ignore_value2text_conversions=ignore_value2text_conversions,
                        raw=raw,
                    )

                    if time_as_date:
                        index = (
                            pd.to_datetime(
                                df.index + self.header.start_time.timestamp(), unit="s"
                            )
                            .tz_localize("UTC")
                            .tz_convert(LOCAL_TIMEZONE)
                            .astype(str)
                        )
                        df.index = index
                        df.index.name = "timestamps"

                    with open(group_csv_name, "w", newline="") as csvfile:
                        writer = csv.writer(csvfile, **fmtparams)

                        if hasattr(self, "can_logging_db") and self.can_logging_db:

                            dropped = {}

                            for name_ in df.columns:
                                if name_.endswith("CAN_DataFrame.ID"):
                                    dropped[name_] = pd.Series(
                                        csv_int2hex(df[name_] & 0x1FFFFFFF),
                                        index=df.index,
                                    )

                                elif name_.endswith("CAN_DataFrame.DataBytes"):
                                    dropped[name_] = pd.Series(
                                        csv_bytearray2hex(df[name_]), index=df.index
                                    )

                            df = df.drop(columns=list(dropped))
                            for name_, s in dropped.items():
                                df[name_] = s

                        names_row = [df.index.name, *df.columns]
                        writer.writerow(names_row)

                        if reduce_memory_usage:
                            vals = [df.index, *(df[name] for name in df)]
                        else:
                            vals = [
                                df.index.to_list(),
                                *(df[name].to_list() for name in df),
                            ]

                        for i, row in enumerate(zip(*vals)):
                            writer.writerow(row)

                    if self._callback:
                        self._callback(i + 1, gp_count)

        elif fmt == "mat":

            filename = filename.with_suffix(".mat")

            if not single_time_base:
                mdict = {}

                master_name_template = "DGM{}_{}"
                channel_name_template = "DG{}_{}"
                used_names = UniqueDB()

                groups_nr = len(self.virtual_groups)

                for i, (group_index, virtual_group) in enumerate(
                    self.virtual_groups.items()
                ):
                    if self._terminate:
                        return

                    channels = self.included_channels(group_index)[group_index]

                    if not channels:
                        continue

                    channels = [
                        (None, gp_index, ch_index)
                        for gp_index, channel_indexes in channels.items()
                        for ch_index in channel_indexes
                    ]

                    if not channels:
                        continue

                    channels = self.select(
                        channels,
                        ignore_value2text_conversions=ignore_value2text_conversions,
                        raw=raw,
                    )

                    master = channels[0].copy()
                    master.samples = master.timestamps

                    channels.insert(0, master)

                    for j, sig in enumerate(channels):
                        if j == 0:
                            channel_name = master_name_template.format(i, "timestamps")
                        else:

                            if use_display_names:
                                channel_name = (
                                    list(sig.display_names)[0]
                                    if sig.display_names
                                    else sig.name
                                )
                            else:
                                channel_name = sig.name
                            channel_name = channel_name_template.format(i, channel_name)

                        channel_name = matlab_compatible(channel_name)
                        channel_name = used_names.get_unique_name(channel_name)

                        if sig.samples.dtype.names:
                            sig.samples.dtype.names = [
                                matlab_compatible(name)
                                for name in sig.samples.dtype.names
                            ]

                        mdict[channel_name] = sig.samples

                    if self._callback:
                        self._callback(i + 1, groups_nr + 1)

            else:
                used_names = UniqueDB()
                mdict = {}

                count = len(df.columns)

                for i, name in enumerate(df.columns):
                    channel_name = matlab_compatible(name)
                    channel_name = used_names.get_unique_name(channel_name)

                    mdict[channel_name] = df[name].values

                    if hasattr(mdict[channel_name].dtype, "categories"):
                        mdict[channel_name] = np.array(mdict[channel_name], dtype="S")

                    if self._callback:
                        self._callback(i + 1 + count, count * 2)

                mdict["timestamps"] = df.index.values

            if self._callback:
                self._callback(80, 100)
            if format == "7.3":

                savemat(
                    str(filename),
                    mdict,
                    long_field_names=True,
                    format="7.3",
                    delete_unused_variables=False,
                    oned_as=oned_as,
                    structured_numpy_ndarray_as_struct=True,
                )
            else:
                savemat(
                    str(filename),
                    mdict,
                    long_field_names=True,
                    oned_as=oned_as,
                    do_compression=bool(compression),
                )
            if self._callback:
                self._callback(100, 100)

        elif fmt == "parquet":
            filename = filename.with_suffix(".parquet")
            if compression:
                write_parquet(filename, df, compression=compression)
            else:
                write_parquet(filename, df)

        else:
            message = (
                'Unsopported export type "{}". '
                'Please select "csv", "excel", "hdf5", "mat" or "pandas"'
            )
            message.format(fmt)
            logger.warning(message)

    def filter(self, channels: ChannelsType, version: str | None = None) -> MDF:
        """return new *MDF* object that contains only the channels listed in
        *channels* argument

        Parameters
        ----------
        channels : list
            list of items to be filtered; each item can be :

                * a channel name string
                * (channel name, group index, channel index) list or tuple
                * (channel name, group index) list or tuple
                * (None, group index, channel index) list or tuple

        version : str
            new mdf file version from ('2.00', '2.10', '2.14', '3.00', '3.10',
            '3.20', '3.30', '4.00', '4.10', '4.11', '4.20'); default *None* and in this
            case the original file version is used

        Returns
        -------
        mdf : MDF
            new *MDF* file

        Examples
        --------
        >>> from asammdf import MDF, Signal
        >>> import numpy as np
        >>> t = np.arange(5)
        >>> s = np.ones(5)
        >>> mdf = MDF()
        >>> for i in range(4):
        ...     sigs = [Signal(s*(i*10+j), t, name='SIG') for j in range(1,4)]
        ...     mdf.append(sigs)
        ...
        >>> filtered = mdf.filter(['SIG', ('SIG', 3, 1), ['SIG', 2], (None, 1, 2)])
        >>> for gp_nr, ch_nr in filtered.channels_db['SIG']:
        ...     print(filtered.get(group=gp_nr, index=ch_nr))
        ...
        <Signal SIG:
                samples=[ 1.  1.  1.  1.  1.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        <Signal SIG:
                samples=[ 31.  31.  31.  31.  31.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        <Signal SIG:
                samples=[ 21.  21.  21.  21.  21.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        <Signal SIG:
                samples=[ 12.  12.  12.  12.  12.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">

        """
        if version is None:
            version = self.version
        else:
            version = validate_version_argument(version)

        # group channels by group index
        gps = self.included_channels(channels=channels)

        mdf = MDF(
            version=version,
            **self._kwargs,
        )

        integer_interpolation_mode = self._integer_interpolation
        float_interpolation_mode = self._float_interpolation
        mdf.configure(from_other=self)
        mdf.header.start_time = self.header.start_time

        self.configure(copy_on_get=False)

        if self.name:
            origin = self.name.name
        else:
            origin = "New MDF"

        groups_nr = len(gps)

        if self._callback:
            self._callback(0, groups_nr)

        for i, (group_index, groups) in enumerate(gps.items()):

            for idx, sigs in enumerate(
                self._yield_selected_signals(
                    group_index, groups=groups, version=version
                )
            ):
                if not sigs:
                    break

                if idx == 0:

                    if sigs:
                        cg = self.groups[group_index].channel_group
                        cg_nr = mdf.append(
                            sigs,
                            common_timebase=True,
                            comment=cg.comment,
                        )
                        MDF._transfer_channel_group_data(
                            mdf.groups[cg_nr].channel_group, cg
                        )
                    else:
                        break

                else:
                    mdf.extend(cg_nr, sigs)

            if self._callback:
                self._callback(i + 1, groups_nr)

            if self._terminate:
                return

        self.configure(copy_on_get=True)

        mdf._transfer_metadata(self, message=f"Filtered from {self.name}")
        if self._callback:
            mdf._callback = mdf._mdf._callback = self._callback
        return mdf

    @overload
    def iter_get(
        self,
        name: str | None = ...,
        group: int | None = ...,
        index: int | None = ...,
        raster: float | None = ...,
        samples_only: Literal[False] = ...,
        raw: bool = ...,
    ) -> Iterator[Signal]:
        ...

    @overload
    def iter_get(
        self,
        name: str | None = ...,
        group: int | None = ...,
        index: int | None = ...,
        raster: float | None = ...,
        samples_only: Literal[True] = ...,
        raw: bool = ...,
    ) -> Iterator[tuple[NDArray[Any], NDArray[Any] | None]]:
        ...

    def iter_get(
        self,
        name: str | None = None,
        group: int | None = None,
        index: int | None = None,
        raster: float | None = None,
        samples_only: bool = False,
        raw: bool = False,
    ) -> Iterator[Signal] | Iterator[tuple[NDArray[Any], NDArray[Any] | None]]:
        """iterator over a channel

        This is usefull in case of large files with a small number of channels.

        If the *raster* keyword argument is not *None* the output is
        interpolated accordingly

        Parameters
        ----------
        name : string
            name of channel
        group : int
            0-based group index
        index : int
            0-based channel index
        raster : float
            time raster in seconds
        samples_only : bool
            if *True* return only the channel samples as numpy array; if
                *False* return a *Signal* object
        raw : bool
            return channel samples without appling the conversion rule; default
            `False`

        """

        gp_nr, ch_nr = self._validate_channel_selection(name, group, index)

        grp = self.groups[gp_nr]

        data = self._load_data(grp)

        for fragment in data:
            yield self.get(
                group=gp_nr,
                index=ch_nr,
                raster=raster,
                samples_only=samples_only,
                data=fragment,
                raw=raw,
            )

    @staticmethod
    def concatenate(
        files: Sequence[MDF | InputType],
        version: str = "4.10",
        sync: bool = True,
        add_samples_origin: bool = False,
        direct_timestamp_continuation: bool = False,
        **kwargs,
    ) -> MDF:
        """concatenates several files. The files
        must have the same internal structure (same number of groups, and same
        channels in each group).

        The order of the input files is always preserved, only the samples timestamps are influenced
        by the ``sync`` argument.

        Parameters
        ----------
        files : list | tuple
            list of *MDF* file names or *MDF*, zipfile.ZipFile, bz2.BZ2File or gzip.GzipFile
            instances

            ..versionchanged:: 6.2.0

                added support for zipfile.ZipFile, bz2.BZ2File and gzip.GzipFile

        version : str
            merged file version
        sync : bool
            sync the files based on the start of measurement, default *True*. The order of the
            input files is preserved, only the samples timestamps are influenced by this
            argument
        add_samples_origin : bool
            option to create a new "__samples_origin" channel that will hold
            the index of the measurement from where each timestamp originated
        direct_timestamp_continuation (False) : bool
            the time stamps from the next file will be added right after the last
            time stamp from the previous file; default False

            ..versionadded:: 6.0.0

        kwargs :

            use_display_names (False) : bool

        Examples
        --------
        >>> conc = MDF.concatenate(
            [
                'path/to/file.mf4',
                MDF(BytesIO(data)),
                MDF(zipfile.ZipFile('data.zip')),
                MDF(bz2.BZ2File('path/to/data.bz2', 'rb')),
                MDF(gzip.GzipFile('path/to/data.gzip', 'rb')),
            ],
            version='4.00',
            sync=False,
        )

        Returns
        -------
        concatenate : MDF
            new *MDF* object with concatenated channels

        Raises
        ------
        MdfException : if there are inconsistencies between the files

        """

        if not files:
            raise MdfException("No files given for merge")

        callback = kwargs.get("callback", None)
        if callback:
            callback(0, 100)

        mdf_nr = len(files)

        input_types = [isinstance(mdf, MDF) for mdf in files]
        use_display_names = kwargs.get("use_display_names", False)

        versions = []
        if sync:
            timestamps = []
            for file in files:
                if isinstance(file, MDF):
                    timestamps.append(file.header.start_time)
                    versions.append(file.version)
                else:
                    if is_file_like(file):
                        ts, version = get_measurement_timestamp_and_version(file)
                        timestamps.append(ts)
                        versions.append(version)
                    else:
                        with open(file, "rb") as mdf:
                            ts, version = get_measurement_timestamp_and_version(mdf)
                            timestamps.append(ts)
                            versions.append(version)

            try:
                oldest = min(timestamps)
            except TypeError:
                timestamps = [
                    timestamp.astimezone(timezone.utc) for timestamp in timestamps
                ]
                oldest = min(timestamps)

            offsets = [(timestamp - oldest).total_seconds() for timestamp in timestamps]
            offsets = [offset if offset > 0 else 0 for offset in offsets]

        else:
            file = files[0]
            if isinstance(file, MDF):
                oldest = file.header.start_time
                versions.append(file.version)
            else:
                if is_file_like(file):
                    ts, version = get_measurement_timestamp_and_version(file)
                    versions.append(version)
                else:
                    with open(file, "rb") as mdf:
                        ts, version = get_measurement_timestamp_and_version(mdf)
                        versions.append(version)

                oldest = ts

            offsets = [0 for _ in files]

        included_channel_names = []
        cg_map = {}

        if add_samples_origin:
            origin_conversion = {}
            for i, mdf in enumerate(files):
                origin_conversion[f"val_{i}"] = i
                if isinstance(mdf, MDF):
                    origin_conversion[f"text_{i}"] = str(mdf.name)
                else:
                    origin_conversion[f"text_{i}"] = str(mdf)
            origin_conversion = from_dict(origin_conversion)

        for mdf_index, (offset, mdf) in enumerate(zip(offsets, files)):
            if not isinstance(mdf, MDF):
                mdf = MDF(
                    mdf,
                    use_display_names=use_display_names,
                )

            if mdf_index == 0:
                version = validate_version_argument(version)

                kwargs = dict(mdf._kwargs)
                kwargs.pop("callback", None)

                merged = MDF(
                    version=version,
                    callback=callback,
                    **kwargs,
                )

                merged.configure(from_other=mdf)

                merged.header.start_time = oldest

            mdf.configure(copy_on_get=False)

            if mdf_index == 0:
                last_timestamps = [None for gp in mdf.virtual_groups]
                groups_nr = len(last_timestamps)

            else:
                if len(mdf.virtual_groups) != groups_nr:
                    raise MdfException(
                        f"internal structure of file <{mdf.name}> is different; different channel groups count"
                    )

            for i, group_index in enumerate(mdf.virtual_groups):
                included_channels = mdf.included_channels(group_index)[group_index]

                if mdf_index == 0:
                    included_channel_names.append(
                        [
                            mdf.groups[gp_index].channels[ch_index].name
                            for gp_index, channels in included_channels.items()
                            for ch_index in channels
                        ]
                    )
                    different_channel_order = False
                else:
                    names = [
                        mdf.groups[gp_index].channels[ch_index].name
                        for gp_index, channels in included_channels.items()
                        for ch_index in channels
                    ]
                    different_channel_order = False
                    if names != included_channel_names[i]:
                        if sorted(names) != sorted(included_channel_names[i]):
                            raise MdfException(
                                f"internal structure of file {mdf_index} is different; different channels"
                            )
                        else:
                            original_names = included_channel_names[i]
                            different_channel_order = True
                            remap = [original_names.index(name) for name in names]

                if not included_channels:
                    continue

                last_timestamp = last_timestamps[i]
                first_timestamp = None
                original_first_timestamp = None

                for idx, signals in enumerate(
                    mdf._yield_selected_signals(group_index, groups=included_channels)
                ):
                    if not signals:
                        break
                    if mdf_index == 0 and idx == 0:

                        first_signal = signals[0]
                        if len(first_signal):
                            if offset > 0:
                                timestamps = first_signal.timestamps + offset
                                for sig in signals:
                                    sig.timestamps = timestamps
                            last_timestamp = first_signal.timestamps[-1]
                            first_timestamp = first_signal.timestamps[0]
                            original_first_timestamp = first_timestamp

                        if add_samples_origin:
                            signals.append(
                                Signal(
                                    samples=np.ones(len(first_signal), dtype="<u2")
                                    * mdf_index,
                                    timestamps=first_signal.timestamps,
                                    conversion=origin_conversion,
                                    name="__samples_origin",
                                )
                            )

                        cg = mdf.groups[group_index].channel_group
                        cg_nr = merged.append(
                            signals,
                            common_timebase=True,
                        )
                        MDF._transfer_channel_group_data(
                            merged.groups[cg_nr].channel_group, cg
                        )
                        cg_map[group_index] = cg_nr

                    else:
                        if different_channel_order:
                            new_signals = [None for _ in signals]
                            if idx == 0:
                                for new_index, sig in zip(remap, signals):
                                    new_signals[new_index] = sig
                            else:
                                for new_index, sig in zip(remap, signals[1:]):
                                    new_signals[new_index + 1] = sig
                                new_signals[0] = signals[0]

                            signals = new_signals

                        if idx == 0:
                            signals = [(signals[0].timestamps, None)] + [
                                (sig.samples, sig.invalidation_bits) for sig in signals
                            ]

                        master = signals[0][0]
                        _copied = False

                        if len(master):
                            if original_first_timestamp is None:
                                original_first_timestamp = master[0]
                            if offset > 0:
                                master = master + offset
                                _copied = True
                            if last_timestamp is None:
                                last_timestamp = master[-1]
                            else:
                                if (
                                    last_timestamp >= master[0]
                                    or direct_timestamp_continuation
                                ):
                                    if len(master) >= 2:
                                        delta = master[1] - master[0]
                                    else:
                                        delta = 0.001
                                    if _copied:
                                        master -= master[0]
                                    else:
                                        master = master - master[0]
                                        _copied = True
                                    master += last_timestamp + delta
                                last_timestamp = master[-1]

                            signals[0] = master, None

                            if add_samples_origin:
                                signals.append(
                                    (
                                        np.ones(len(master), dtype="<u2") * mdf_index,
                                        None,
                                    )
                                )
                            cg_nr = cg_map[group_index]
                            merged.extend(cg_nr, signals)

                            if first_timestamp is None:
                                first_timestamp = master[0]

                last_timestamps[i] = last_timestamp

            mdf.configure(copy_on_get=True)

            if mdf_index == 0:
                merged._transfer_metadata(mdf)

            if not input_types[mdf_index]:
                mdf.close()

            if callback:
                callback(i + 1 + mdf_index * groups_nr, groups_nr * mdf_nr)

            if MDF._terminate:
                return

        try:
            merged._process_bus_logging()
        except:
            pass

        return merged

    @staticmethod
    def stack(
        files: Sequence[MDF | InputType],
        version: str = "4.10",
        sync: bool = True,
        **kwargs,
    ) -> MDF:
        """stack several files and return the stacked *MDF* object

        Parameters
        ----------
        files : list | tuple
            list of *MDF* file names or *MDF*, zipfile.ZipFile, bz2.BZ2File or gzip.GzipFile
            instances

            ..versionchanged:: 6.2.0

                added support for zipfile.ZipFile, bz2.BZ2File and gzip.GzipFile
        version : str
            merged file version
        sync : bool
            sync the files based on the start of measurement, default *True*

        kwargs :

            use_display_names (False) : bool

        Examples
        --------
        >>> stacked = MDF.stack(
            [
                'path/to/file.mf4',
                MDF(BytesIO(data)),
                MDF(zipfile.ZipFile('data.zip')),
                MDF(bz2.BZ2File('path/to/data.bz2', 'rb')),
                MDF(gzip.GzipFile('path/to/data.gzip', 'rb')),
            ],
            version='4.00',
            sync=False,
        )

        Returns
        -------
        stacked : MDF
            new *MDF* object with stacked channels

        """
        if not files:
            raise MdfException("No files given for stack")

        version = validate_version_argument(version)

        callback = kwargs.get("callback", None)
        use_display_names = kwargs.get("use_display_names", False)

        files_nr = len(files)

        input_types = [isinstance(mdf, MDF) for mdf in files]

        if callback:
            callback(0, files_nr)

        if sync:
            timestamps = []
            for file in files:
                if isinstance(file, MDF):
                    timestamps.append(file.header.start_time)
                else:
                    if is_file_like(file):
                        ts, version = get_measurement_timestamp_and_version(file)
                        timestamps.append(ts)
                    else:
                        with open(file, "rb") as mdf:
                            ts, version = get_measurement_timestamp_and_version(mdf)
                            timestamps.append(ts)

            try:
                oldest = min(timestamps)
            except TypeError:
                timestamps = [
                    timestamp.astimezone(timezone.utc) for timestamp in timestamps
                ]
                oldest = min(timestamps)

            offsets = [(timestamp - oldest).total_seconds() for timestamp in timestamps]

        else:
            offsets = [0 for file in files]

        for mdf_index, (offset, mdf) in enumerate(zip(offsets, files)):
            if not isinstance(mdf, MDF):
                mdf = MDF(mdf, use_display_names=use_display_names)

            if mdf_index == 0:
                version = validate_version_argument(version)

                kwargs = dict(mdf._kwargs)
                kwargs.pop("callback", None)

                stacked = MDF(
                    version=version,
                    callback=callback,
                    **kwargs,
                )

                stacked.configure(from_other=mdf)

                if sync:
                    stacked.header.start_time = oldest
                else:
                    stacked.header.start_time = mdf.header.start_time

            mdf.configure(copy_on_get=False)

            for i, group in enumerate(mdf.virtual_groups):
                dg_cntr = None
                included_channels = mdf.included_channels(group)[group]
                if not included_channels:
                    continue

                for idx, signals in enumerate(
                    mdf._yield_selected_signals(
                        group, groups=included_channels, version=version
                    )
                ):
                    if not signals:
                        break
                    if idx == 0:
                        if sync:
                            timestamps = signals[0].timestamps + offset
                            for sig in signals:
                                sig.timestamps = timestamps
                        cg = mdf.groups[group].channel_group
                        dg_cntr = stacked.append(
                            signals,
                            common_timebase=True,
                        )
                        MDF._transfer_channel_group_data(
                            stacked.groups[dg_cntr].channel_group, cg
                        )
                    else:
                        master = signals[0][0]
                        if sync:
                            master = master + offset
                            signals[0] = master, None

                        stacked.extend(dg_cntr, signals)

                if dg_cntr is not None:
                    for index in range(dg_cntr, len(stacked.groups)):

                        stacked.groups[
                            index
                        ].channel_group.comment = (
                            f'stacked from channel group {i} of "{mdf.name.parent}"'
                        )

            if callback:
                callback(mdf_index, files_nr)

            mdf.configure(copy_on_get=True)

            if mdf_index == 0:
                stacked._transfer_metadata(mdf)

            if not input_types[mdf_index]:
                mdf.close()

            if MDF._terminate:
                return

        try:
            stacked._process_bus_logging()
        except:
            pass

        return stacked

    def iter_channels(
        self, skip_master: bool = True, copy_master: bool = True
    ) -> Iterator[Signal]:
        """generator that yields a *Signal* for each non-master channel

        Parameters
        ----------
        skip_master : bool
            do not yield master channels; default *True*
        copy_master : bool
            copy master for each yielded channel

        """

        for index in self.virtual_groups:

            channels = [
                (None, gp_index, ch_index)
                for gp_index, channel_indexes in self.included_channels(index)[
                    index
                ].items()
                for ch_index in channel_indexes
            ]

            channels = self.select(channels, copy_master=copy_master)

            yield from channels

    def iter_groups(
        self,
        raster: RasterType | None = None,
        time_from_zero: bool = True,
        empty_channels: EmptyChannelsType = "skip",
        keep_arrays: bool = False,
        use_display_names: bool = False,
        time_as_date: bool = False,
        reduce_memory_usage: bool = False,
        raw: bool = False,
        ignore_value2text_conversions: bool = False,
        only_basenames: bool = False,
    ) -> Iterator[pd.DataFrame]:
        """generator that yields channel groups as pandas DataFrames. If there
        are multiple occurrences for the same channel name inside a channel
        group, then a counter will be used to make the names unique
        (<original_name>_<counter>)


        Parameters
        ----------
        use_display_names : bool
            use display name instead of standard channel name, if available.

            .. versionadded:: 5.21.0

        reduce_memory_usage : bool
            reduce memory usage by converting all float columns to float32 and
            searching for minimum dtype that can reprezent the values found
            in integer columns; default *False*

            .. versionadded:: 5.21.0

        raw (False) : bool
            the dataframe will contain the raw channel values

            .. versionadded:: 5.21.0

        ignore_value2text_conversions (False) : bool
            valid only for the channels that have value to text conversions and
            if *raw=False*. If this is True then the raw numeric values will be
            used, and the conversion will not be applied.

            .. versionadded:: 5.21.0

        keep_arrays (False) : bool
            keep arrays and structure channels as well as the
            component channels. If *True* this can be very slow. If *False*
            only the component channels are saved, and their names will be
            prefixed with the parent channel.

            .. versionadded:: 5.21.0

        empty_channels ("skip") : str
            behaviour for channels without samples; the options are *skip* or
            *zeros*; default is *skip*

            .. versionadded:: 5.21.0

        only_basenames (False) : bool
            use just the field names, without prefix, for structures and channel
            arrays

            .. versionadded:: 5.21.0

        raster : float | np.array | str
            new raster that can be

            * a float step value
            * a channel name who's timestamps will be used as raster (starting with asammdf 5.5.0)
            * an array (starting with asammdf 5.5.0)

            see `resample` for examples of using this argument

            .. versionadded:: 5.21.0

        """

        for i in self.virtual_groups:
            yield self.get_group(
                i,
                raster=None,
                time_from_zero=time_from_zero,
                empty_channels=empty_channels,
                keep_arrays=keep_arrays,
                use_display_names=use_display_names,
                time_as_date=time_as_date,
                reduce_memory_usage=reduce_memory_usage,
                raw=raw,
                ignore_value2text_conversions=ignore_value2text_conversions,
                only_basenames=only_basenames,
            )

    def resample(
        self,
        raster: RasterType,
        version: str | None = None,
        time_from_zero: bool = False,
    ) -> MDF:
        """resample all channels using the given raster. See *configure* to select
        the interpolation method for interger channels

        Parameters
        ----------
        raster : float | np.array | str
            new raster that can be

            * a float step value
            * a channel name who's timestamps will be used as raster (starting with asammdf 5.5.0)
            * an array (starting with asammdf 5.5.0)

        version : str
            new mdf file version from ('2.00', '2.10', '2.14', '3.00', '3.10',
            '3.20', '3.30', '4.00', '4.10', '4.11', '4.20'); default *None* and
            in this case the original file version is used

        time_from_zero : bool
            start time stamps from 0s in the cut measurement

        Returns
        -------
        mdf : MDF
            new *MDF* with resampled channels

        Examples
        --------
        >>> from asammdf import MDF, Signal
        >>> import numpy as np
        >>> mdf = MDF()
        >>> sig = Signal(name='S1', samples=[1,2,3,4], timestamps=[1,2,3,4])
        >>> mdf.append(sig)
        >>> sig = Signal(name='S2', samples=[1,2,3,4], timestamps=[1.1, 3.5, 3.7, 3.9])
        >>> mdf.append(sig)
        >>> resampled = mdf.resample(raster=0.1)
        >>> resampled.select(['S1', 'S2'])
        [<Signal S1:
                samples=[1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3 3 3 3 3 3 3 3 3 4]
                timestamps=[1.  1.1 1.2 1.3 1.4 1.5 1.6 1.7 1.8 1.9 2.  2.1 2.2 2.3 2.4 2.5 2.6 2.7
         2.8 2.9 3.  3.1 3.2 3.3 3.4 3.5 3.6 3.7 3.8 3.9 4. ]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        , <Signal S2:
                samples=[1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 2 2 3 3 4 4]
                timestamps=[1.  1.1 1.2 1.3 1.4 1.5 1.6 1.7 1.8 1.9 2.  2.1 2.2 2.3 2.4 2.5 2.6 2.7
         2.8 2.9 3.  3.1 3.2 3.3 3.4 3.5 3.6 3.7 3.8 3.9 4. ]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        ]
        >>> resampled = mdf.resample(raster='S2')
        >>> resampled.select(['S1', 'S2'])
        [<Signal S1:
                samples=[1 3 3 3]
                timestamps=[1.1 3.5 3.7 3.9]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        , <Signal S2:
                samples=[1 2 3 4]
                timestamps=[1.1 3.5 3.7 3.9]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        ]
        >>> resampled = mdf.resample(raster=[1.9, 2.0, 2.1])
        >>> resampled.select(['S1', 'S2'])
        [<Signal S1:
                samples=[1 2 2]
                timestamps=[1.9 2.  2.1]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        , <Signal S2:
                samples=[1 1 1]
                timestamps=[1.9 2.  2.1]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        ]
        >>> resampled = mdf.resample(raster='S2', time_from_zero=True)
        >>> resampled.select(['S1', 'S2'])
        [<Signal S1:
                samples=[1 3 3 3]
                timestamps=[0.  2.4 2.6 2.8]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        , <Signal S2:
                samples=[1 2 3 4]
                timestamps=[0.  2.4 2.6 2.8]
                invalidation_bits=None
                unit=""
                conversion=None
                source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0)
                comment=""
                mastermeta="('time', 1)"
                raw=True
                display_names={}
                attachment=()>
        ]
        """

        if version is None:
            version = self.version
        else:
            version = validate_version_argument(version)

        mdf = MDF(
            version=version,
            **self._kwargs,
        )

        integer_interpolation_mode = self._integer_interpolation
        float_interpolation_mode = self._float_interpolation
        mdf.configure(from_other=self)

        mdf.header.start_time = self.header.start_time

        groups_nr = len(self.virtual_groups)

        if self._callback:
            self._callback(0, groups_nr)

        try:
            raster = float(raster)
            assert raster > 0
        except (TypeError, ValueError):
            if isinstance(raster, str):
                raster = self.get(raster).timestamps
            else:
                raster = np.array(raster)
        else:
            raster = master_using_raster(self, raster)

        if time_from_zero and len(raster):
            delta = raster[0]
            new_raster = raster - delta
            t_epoch = self.header.start_time.timestamp() + delta
            mdf.header.start_time = datetime.fromtimestamp(t_epoch)
        else:
            delta = 0
            new_raster = None
            mdf.header.start_time = self.header.start_time

        for i, (group_index, virtual_group) in enumerate(self.virtual_groups.items()):
            channels = [
                (None, gp_index, ch_index)
                for gp_index, channel_indexes in self.included_channels(group_index)[
                    group_index
                ].items()
                for ch_index in channel_indexes
            ]
            sigs = self.select(channels, raw=True)

            sigs = [
                sig.interp(
                    raster,
                    integer_interpolation_mode=integer_interpolation_mode,
                    float_interpolation_mode=float_interpolation_mode,
                )
                for sig in sigs
            ]

            if new_raster is not None:
                for sig in sigs:
                    if len(sig):
                        sig.timestamps = new_raster

            cg = self.groups[group_index].channel_group
            dg_cntr = mdf.append(
                sigs,
                common_timebase=True,
                comment=cg.comment,
            )
            MDF._transfer_channel_group_data(mdf.groups[dg_cntr].channel_group, cg)

            if self._callback:
                self._callback(i + 1, groups_nr)

            if self._terminate:
                return

        if self._callback:
            self._callback(groups_nr, groups_nr)

        mdf._transfer_metadata(self, message=f"Resampled from {self.name}")
        if self._callback:
            mdf._callback = mdf._mdf._callback = self._callback
        return mdf

    def select(
        self,
        channels: ChannelsType,
        record_offset: int = 0,
        raw: bool = False,
        copy_master: bool = True,
        ignore_value2text_conversions: bool = False,
        record_count: int | None = None,
        validate: bool = False,
    ) -> list[Signal]:
        """retrieve the channels listed in *channels* argument as *Signal*
        objects

        .. note:: the *dataframe* argument was removed in version 5.8.0
                  use the ``to_dataframe`` method instead

        Parameters
        ----------
        channels : list
            list of items to be filtered; each item can be :

                * a channel name string
                * (channel name, group index, channel index) list or tuple
                * (channel name, group index) list or tuple
                * (None, group index, channel index) list or tuple

        record_offset : int
            record number offset; optimization to get the last part of signal samples
        raw : bool
            get raw channel samples; default *False*
        copy_master : bool
            option to get a new timestamps array for each selected Signal or to
            use a shared array for channels of the same channel group; default *True*
        ignore_value2text_conversions (False) : bool
            valid only for the channels that have value to text conversions and
            if *raw=False*. If this is True then the raw numeric values will be
            used, and the conversion will not be applied.

            .. versionadded:: 5.8.0

        validate (False) : bool
            consider the invalidation bits

            .. versionadded:: 5.16.0

        Returns
        -------
        signals : list
            list of *Signal* objects based on the input channel list

        Examples
        --------
        >>> from asammdf import MDF, Signal
        >>> import numpy as np
        >>> t = np.arange(5)
        >>> s = np.ones(5)
        >>> mdf = MDF()
        >>> for i in range(4):
        ...     sigs = [Signal(s*(i*10+j), t, name='SIG') for j in range(1,4)]
        ...     mdf.append(sigs)
        ...
        >>> # select SIG group 0 default index 1 default, SIG group 3 index 1, SIG group 2 index 1 default and channel index 2 from group 1
        ...
        >>> mdf.select(['SIG', ('SIG', 3, 1), ['SIG', 2],  (None, 1, 2)])
        [<Signal SIG:
                samples=[ 1.  1.  1.  1.  1.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        , <Signal SIG:
                samples=[ 31.  31.  31.  31.  31.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        , <Signal SIG:
                samples=[ 21.  21.  21.  21.  21.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        , <Signal SIG:
                samples=[ 12.  12.  12.  12.  12.]
                timestamps=[0 1 2 3 4]
                unit=""
                info=None
                comment="">
        ]

        """

        virtual_groups = self.included_channels(
            channels=channels, minimal=False, skip_master=False
        )

        output_signals = {}

        for virtual_group, groups in virtual_groups.items():
            cycles_nr = self._mdf.virtual_groups[virtual_group].cycles_nr
            pairs = [
                (gp_index, ch_index)
                for gp_index, channel_indexes in groups.items()
                for ch_index in channel_indexes
            ]

            if record_count is None:
                cycles = cycles_nr - record_offset
            else:
                if cycles_nr < record_count + record_offset:
                    cycles = cycles_nr - record_offset
                else:
                    cycles = record_count

            signals = []

            current_pos = 0

            for idx, sigs in enumerate(
                self._yield_selected_signals(
                    virtual_group,
                    groups=groups,
                    record_offset=record_offset,
                    record_count=record_count,
                )
            ):
                if not sigs:
                    break
                if idx == 0:
                    next_pos = current_pos + len(sigs[0])

                    master = np.empty(cycles, dtype=sigs[0].timestamps.dtype)
                    master[current_pos:next_pos] = sigs[0].timestamps

                    for sig in sigs:
                        shape = (cycles,) + sig.samples.shape[1:]
                        signal = np.empty(shape, dtype=sig.samples.dtype)
                        signal[current_pos:next_pos] = sig.samples
                        sig.samples = signal
                        signals.append(sig)

                        if sig.invalidation_bits is not None:
                            inval = np.empty(cycles, dtype=sig.invalidation_bits.dtype)
                            inval[current_pos:next_pos] = sig.invalidation_bits
                            sig.invalidation_bits = inval

                else:
                    sig, _ = sigs[0]
                    next_pos = current_pos + len(sig)
                    master[current_pos:next_pos] = sig

                    for signal, (sig, inval) in zip(signals, sigs[1:]):
                        signal.samples[current_pos:next_pos] = sig
                        if signal.invalidation_bits is not None:
                            signal.invalidation_bits[current_pos:next_pos] = inval

                current_pos = next_pos

            for signal, pair in zip(signals, pairs):
                signal.timestamps = master
                output_signals[pair] = signal

        indexes = []

        for item in channels:
            if not isinstance(item, (list, tuple)):
                item = [item]
            indexes.append(self._validate_channel_selection(*item))

        signals = [output_signals[pair] for pair in indexes]

        if copy_master:
            for signal in signals:
                signal.timestamps = signal.timestamps.copy()

        if not raw:
            if ignore_value2text_conversions:
                for signal in signals:
                    conversion = signal.conversion
                    if conversion:
                        samples = conversion.convert(signal.samples)
                        if samples.dtype.kind not in "US":
                            signal.samples = samples
                    signal.raw = True
                    signal.conversion = None
            else:
                for signal in signals:
                    conversion = signal.conversion
                    if conversion:
                        signal.samples = conversion.convert(signal.samples)
                    signal.raw = False
                    signal.conversion = None
                    if signal.samples.dtype.kind == "S":
                        signal.encoding = (
                            "utf-8" if self.version >= "4.00" else "latin-1"
                        )

        if validate:
            signals = [sig.validate(copy=False) for sig in signals]

        for signal, channel in zip(signals, channels):
            if isinstance(channel, str):
                signal.name = channel
            else:
                name = channel[0]
                if name is not None:
                    signal.name = name

        return signals

    @staticmethod
    def scramble(name: StrPathType, skip_attachments: bool = False, **kwargs) -> Path:
        """scramble text blocks and keep original file structure

        Parameters
        ----------
        name : str | pathlib.Path
            file name
        skip_attachments : bool
            skip scrambling of attachments data if True

            .. versionadded:: 5.9.0

        Returns
        -------
        name : pathlib.Path
            scrambled file name

        """

        name = Path(name)

        mdf = MDF(name)
        texts = {}

        callback = kwargs.get("callback", None)
        if callback:
            callback(0, 100)

        count = len(mdf.groups)

        if mdf.version >= "4.00":
            try:
                ChannelConversion = ChannelConversionV4

                stream = mdf._file

                if mdf.header.comment_addr:
                    stream.seek(mdf.header.comment_addr + 8)
                    size = UINT64_u(stream.read(8))[0] - 24
                    texts[mdf.header.comment_addr] = randomized_string(size)

                for fh in mdf.file_history:
                    addr = fh.comment_addr
                    if addr and addr not in texts:
                        stream.seek(addr + 8)
                        size = UINT64_u(stream.read(8))[0] - 24
                        texts[addr] = randomized_string(size)

                for ev in mdf.events:
                    for addr in (ev.comment_addr, ev.name_addr):
                        if addr and addr not in texts:
                            stream.seek(addr + 8)
                            size = UINT64_u(stream.read(8))[0] - 24
                            texts[addr] = randomized_string(size)

                for at in mdf.attachments:
                    for addr in (at.comment_addr, at.file_name_addr):
                        if addr and addr not in texts:
                            stream.seek(addr + 8)
                            size = UINT64_u(stream.read(8))[0] - 24
                            texts[addr] = randomized_string(size)
                    if not skip_attachments and at.embedded_data:
                        texts[at.address + v4c.AT_COMMON_SIZE] = randomized_string(
                            at.embedded_size
                        )

                for idx, gp in enumerate(mdf.groups, 1):

                    addr = gp.data_group.comment_addr
                    if addr and addr not in texts:
                        stream.seek(addr + 8)
                        size = UINT64_u(stream.read(8))[0] - 24
                        texts[addr] = randomized_string(size)

                    cg = gp.channel_group
                    for addr in (cg.acq_name_addr, cg.comment_addr):
                        if cg.flags & v4c.FLAG_CG_BUS_EVENT:
                            continue

                        if addr and addr not in texts:
                            stream.seek(addr + 8)
                            size = UINT64_u(stream.read(8))[0] - 24
                            texts[addr] = randomized_string(size)

                        source = cg.acq_source_addr
                        if source:
                            source = SourceInformation(
                                address=source, stream=stream, mapped=False, tx_map={}
                            )
                            for addr in (
                                source.name_addr,
                                source.path_addr,
                                source.comment_addr,
                            ):
                                if addr and addr not in texts:
                                    stream.seek(addr + 8)
                                    size = UINT64_u(stream.read(8))[0] - 24
                                    texts[addr] = randomized_string(size)

                    for ch in gp.channels:

                        for addr in (ch.name_addr, ch.unit_addr, ch.comment_addr):
                            if addr and addr not in texts:
                                stream.seek(addr + 8)
                                size = UINT64_u(stream.read(8))[0] - 24
                                texts[addr] = randomized_string(size)

                        source = ch.source_addr
                        if source:
                            source = SourceInformation(
                                address=source, stream=stream, mapped=False, tx_map={}
                            )
                            for addr in (
                                source.name_addr,
                                source.path_addr,
                                source.comment_addr,
                            ):
                                if addr and addr not in texts:
                                    stream.seek(addr + 8)
                                    size = UINT64_u(stream.read(8))[0] - 24
                                    texts[addr] = randomized_string(size)

                        conv = ch.conversion_addr
                        if conv:
                            conv = ChannelConversion(
                                address=conv,
                                stream=stream,
                                mapped=False,
                                tx_map={},
                                si_map={},
                            )
                            for addr in (
                                conv.name_addr,
                                conv.unit_addr,
                                conv.comment_addr,
                            ):
                                if addr and addr not in texts:
                                    stream.seek(addr + 8)
                                    size = UINT64_u(stream.read(8))[0] - 24
                                    texts[addr] = randomized_string(size)
                            if conv.conversion_type == v4c.CONVERSION_TYPE_ALG:
                                addr = conv.formula_addr
                                if addr and addr not in texts:
                                    stream.seek(addr + 8)
                                    size = UINT64_u(stream.read(8))[0] - 24
                                    texts[addr] = randomized_string(size)

                            if conv.referenced_blocks:
                                for key, block in conv.referenced_blocks.items():
                                    if block:
                                        if isinstance(block, bytes):
                                            addr = conv[key]
                                            if addr not in texts:
                                                stream.seek(addr + 8)
                                                size = len(block)
                                                texts[addr] = randomized_string(size)

                    if callback:
                        callback(int(idx / count * 66), 100)

            except:
                print(
                    f"Error while scrambling the file: {format_exc()}.\nWill now use fallback method"
                )
                texts = MDF._fallback_scramble_mf4(name)

            mdf.close()

            dst = name.with_suffix(".scrambled.mf4")

            copy(name, dst)

            with open(dst, "rb+") as mdf:
                count = len(texts)
                chunk = max(count // 34, 1)
                idx = 0
                for index, (addr, bts) in enumerate(texts.items()):
                    mdf.seek(addr + 24)
                    mdf.write(bts)
                    if index % chunk == 0:
                        if callback:
                            callback(66 + idx, 100)

            if callback:
                callback(100, 100)

        else:
            ChannelConversion = ChannelConversionV3

            stream = mdf._file

            if mdf.header.comment_addr:
                stream.seek(mdf.header.comment_addr + 2)
                size = UINT16_u(stream.read(2))[0] - 4
                texts[mdf.header.comment_addr + 4] = randomized_string(size)
            texts[36 + 0x40] = randomized_string(32)
            texts[68 + 0x40] = randomized_string(32)
            texts[100 + 0x40] = randomized_string(32)
            texts[132 + 0x40] = randomized_string(32)

            for idx, gp in enumerate(mdf.groups, 1):

                cg = gp.channel_group
                addr = cg.comment_addr

                if addr and addr not in texts:
                    stream.seek(addr + 2)
                    size = UINT16_u(stream.read(2))[0] - 4
                    texts[addr + 4] = randomized_string(size)

                if gp.trigger:
                    addr = gp.trigger.text_addr
                    if addr:
                        stream.seek(addr + 2)
                        size = UINT16_u(stream.read(2))[0] - 4
                        texts[addr + 4] = randomized_string(size)

                for ch in gp.channels:

                    for key in ("long_name_addr", "display_name_addr", "comment_addr"):
                        if hasattr(ch, key):
                            addr = getattr(ch, key)
                        else:
                            addr = 0
                        if addr and addr not in texts:
                            stream.seek(addr + 2)
                            size = UINT16_u(stream.read(2))[0] - 4
                            texts[addr + 4] = randomized_string(size)

                    texts[ch.address + 26] = randomized_string(32)
                    texts[ch.address + 58] = randomized_string(128)

                    source = ch.source_addr
                    if source:
                        source = ChannelExtension(address=source, stream=stream)
                        if source.type == v23c.SOURCE_ECU:
                            texts[source.address + 12] = randomized_string(80)
                            texts[source.address + 92] = randomized_string(32)
                        else:
                            texts[source.address + 14] = randomized_string(36)
                            texts[source.address + 50] = randomized_string(36)

                    conv = ch.conversion_addr
                    if conv:
                        texts[conv + 22] = randomized_string(20)

                        conv = ChannelConversion(address=conv, stream=stream)

                        if conv.conversion_type == v23c.CONVERSION_TYPE_FORMULA:
                            texts[conv + 36] = randomized_string(conv.block_len - 36)

                        if conv.referenced_blocks:
                            for key, block in conv.referenced_blocks.items():
                                if block:
                                    if isinstance(block, bytes):
                                        addr = conv[key]
                                        if addr and addr not in texts:
                                            stream.seek(addr + 2)
                                            size = UINT16_u(stream.read(2))[0] - 4
                                            texts[addr + 4] = randomized_string(size)
                if callback:
                    callback(int(idx / count * 66), 100)

            mdf.close()

            dst = name.with_suffix(".scrambled.mdf")

            copy(name, dst)

            with open(dst, "rb+") as mdf:
                chunk = count // 34
                idx = 0
                for index, (addr, bts) in enumerate(texts.items()):
                    mdf.seek(addr)
                    mdf.write(bts)
                    if chunk and index % chunk == 0:
                        if callback:
                            callback(66 + idx, 100)

            if callback:
                callback(100, 100)

        return dst

    @staticmethod
    def _fallback_scramble_mf4(name: StrOrBytesPathType) -> dict[int, bytes]:
        """scramble text blocks and keep original file structure

        Parameters
        ----------
        name : pathlib.Path
            file name

        Returns
        -------
        name : pathlib.Path
            scrambled file name

        """

        pattern = re.compile(
            rb"(?P<block>##(TX|MD))",
            re.DOTALL | re.MULTILINE,
        )

        texts = {}

        with open(name, "rb") as stream:

            stream.seek(0, 2)
            file_limit = stream.tell()
            stream.seek(0)

            for match in re.finditer(pattern, stream.read()):
                start = match.start()

                if file_limit - start >= 24:
                    stream.seek(start + 8)
                    (size,) = UINT64_u(stream.read(8))

                    if start + size <= file_limit:
                        texts[start + 24] = randomized_string(size - 24)

        return texts

    def get_group(
        self,
        index: int,
        raster: RasterType | None = None,
        time_from_zero: bool = True,
        empty_channels: EmptyChannelsType = "skip",
        keep_arrays: bool = False,
        use_display_names: bool = False,
        time_as_date: bool = False,
        reduce_memory_usage: bool = False,
        raw: bool = False,
        ignore_value2text_conversions: bool = False,
        only_basenames: bool = False,
    ) -> pd.DataFrame:
        """get channel group as pandas DataFrames. If there are multiple
        occurrences for the same channel name, then a counter will be used to
        make the names unique (<original_name>_<counter>)

        Parameters
        ----------
        index : int
            channel group index
        use_display_names : bool
            use display name instead of standard channel name, if available.
        reduce_memory_usage : bool
            reduce memory usage by converting all float columns to float32 and
            searching for minimum dtype that can reprezent the values found
            in integer columns; default *False*
        raw (False) : bool
            the dataframe will contain the raw channel values

            .. versionadded:: 5.7.0

        ignore_value2text_conversions (False) : bool
            valid only for the channels that have value to text conversions and
            if *raw=False*. If this is True then the raw numeric values will be
            used, and the conversion will not be applied.

            .. versionadded:: 5.8.0

        keep_arrays (False) : bool
            keep arrays and structure channels as well as the
            component channels. If *True* this can be very slow. If *False*
            only the component channels are saved, and their names will be
            prefixed with the parent channel.

            .. versionadded:: 5.8.0

        empty_channels ("skip") : str
            behaviour for channels without samples; the options are *skip* or
            *zeros*; default is *skip*

            .. versionadded:: 5.8.0

        only_basenames (False) : bool
            use just the field names, without prefix, for structures and channel
            arrays

            .. versionadded:: 5.13.0

        raster : float | np.array | str
            new raster that can be

            * a float step value
            * a channel name who's timestamps will be used as raster (starting with asammdf 5.5.0)
            * an array (starting with asammdf 5.5.0)

            see `resample` for examples of using this argument

        Returns
        -------
        df : pandas.DataFrame

        """

        channels = [
            (None, gp_index, ch_index)
            for gp_index, channel_indexes in self.included_channels(index)[
                index
            ].items()
            for ch_index in channel_indexes
        ]

        return self.to_dataframe(
            channels=channels,
            raster=raster,
            time_from_zero=time_from_zero,
            empty_channels=empty_channels,
            keep_arrays=keep_arrays,
            use_display_names=use_display_names,
            time_as_date=time_as_date,
            reduce_memory_usage=reduce_memory_usage,
            raw=raw,
            ignore_value2text_conversions=ignore_value2text_conversions,
            only_basenames=only_basenames,
        )

    def iter_to_dataframe(
        self,
        channels: ChannelsType | None = None,
        raster: RasterType | None = None,
        time_from_zero: bool = True,
        empty_channels: EmptyChannelsType = "skip",
        keep_arrays: bool = False,
        use_display_names: bool = False,
        time_as_date: bool = False,
        reduce_memory_usage: bool = False,
        raw: bool = False,
        ignore_value2text_conversions: bool = False,
        use_interpolation: bool = True,
        only_basenames: bool = False,
        chunk_ram_size: int = 200 * 1024 * 1024,
        interpolate_outwards_with_nan: bool = False,
        numeric_1D_only: bool = False,
    ) -> Iterator[pd.DataFrame]:
        """generator that yields pandas DataFrame's that should not exceed
        200MB of RAM

        .. versionadded:: 5.15.0

        Parameters
        ----------
        channels : list
            list of items to be filtered (default None); each item can be :

                * a channel name string
                * (channel name, group index, channel index) list or tuple
                * (channel name, group index) list or tuple
                * (None, group index, channel index) list or tuple

        raster : float | np.array | str
            new raster that can be

            * a float step value
            * a channel name who's timestamps will be used as raster (starting with asammdf 5.5.0)
            * an array (starting with asammdf 5.5.0)

            see `resample` for examples of using this argument

        time_from_zero : bool
            adjust time channel to start from 0; default *True*
        empty_channels : str
            behaviour for channels without samples; the options are *skip* or
            *zeros*; default is *skip*
        use_display_names : bool
            use display name instead of standard channel name, if available.
        keep_arrays : bool
            keep arrays and structure channels as well as the
            component channels. If *True* this can be very slow. If *False*
            only the component channels are saved, and their names will be
            prefixed with the parent channel.
        time_as_date : bool
            the dataframe index will contain the datetime timestamps
            according to the measurement start time; default *False*. If
            *True* then the argument ``time_from_zero`` will be ignored.
        reduce_memory_usage : bool
            reduce memory usage by converting all float columns to float32 and
            searching for minimum dtype that can reprezent the values found
            in integer columns; default *False*
        raw (False) : bool
            the columns will contain the raw values
        ignore_value2text_conversions (False) : bool
            valid only for the channels that have value to text conversions and
            if *raw=False*. If this is True then the raw numeric values will be
            used, and the conversion will not be applied.
        use_interpolation (True) : bool
            option to perform interpoaltions when multiple timestamp raster are
            present. If *False* then dataframe columns will be automatically
            filled with NaN's were the dataframe index values are not found in
            the current column's timestamps
        only_basenames (False) : bool
            use jsut the field names, without prefix, for structures and channel
            arrays
        interpolate_outwards_with_nan : bool
            use NaN values for the samples that lie outside of the original
            signal's timestamps
        chunk_ram_size : int
            desired data frame RAM usage in bytes; default 200 MB
        numeric_1D_only (False) : bool
            only keep the 1D-columns that have numeric values

            .. versionadded:: 7.0.0


        Returns
        -------
        dataframe : pandas.DataFrame
            yields pandas DataFrame's that should not exceed 200MB of RAM

        """

        if channels:
            mdf = self.filter(channels)

            result = mdf.iter_to_dataframe(
                raster=raster,
                time_from_zero=time_from_zero,
                empty_channels=empty_channels,
                keep_arrays=keep_arrays,
                use_display_names=use_display_names,
                time_as_date=time_as_date,
                reduce_memory_usage=reduce_memory_usage,
                raw=raw,
                ignore_value2text_conversions=ignore_value2text_conversions,
                use_interpolation=use_interpolation,
                only_basenames=only_basenames,
                chunk_ram_size=chunk_ram_size,
                interpolate_outwards_with_nan=interpolate_outwards_with_nan,
                numeric_1D_only=numeric_1D_only,
            )

            for df in result:
                yield df

            mdf.close()

        df = {}
        self._set_temporary_master(None)

        masters = {index: self.get_master(index) for index in self.virtual_groups}

        if raster is not None:
            try:
                raster = float(raster)
                assert raster > 0
            except (TypeError, ValueError):
                if isinstance(raster, str):
                    raster = self.get(
                        raster, raw=True, ignore_invalidation_bits=True
                    ).timestamps
                else:
                    raster = np.array(raster)
            else:
                raster = master_using_raster(self, raster)
            master = raster
        else:

            if masters:
                master = reduce(np.union1d, masters.values())
            else:
                master = np.array([], dtype="<f4")

        master_ = master
        channel_count = sum(len(gp.channels) - 1 for gp in self.groups) + 1
        # approximation with all float64 dtype
        itemsize = channel_count * 8
        # use 200MB DataFrame chunks
        chunk_count = chunk_ram_size // itemsize or 1

        chunks, r = divmod(len(master), chunk_count)
        if r:
            chunks += 1

        for i in range(chunks):

            master = master_[chunk_count * i : chunk_count * (i + 1)]
            start = master[0]
            end = master[-1]

            df = {}
            self._set_temporary_master(None)

            used_names = UniqueDB()
            used_names.get_unique_name("timestamps")

            groups_nr = len(self.virtual_groups)

            for group_index, virtual_group in self.virtual_groups.items():
                group_cycles = virtual_group.cycles_nr
                if group_cycles == 0 and empty_channels == "skip":
                    continue

                record_offset = max(
                    np.searchsorted(masters[group_index], start).flatten()[0] - 1, 0
                )
                stop = np.searchsorted(masters[group_index], end).flatten()[0]
                record_count = min(stop - record_offset + 1, group_cycles)

                channels = [
                    (None, gp_index, ch_index)
                    for gp_index, channel_indexes in self.included_channels(
                        group_index
                    )[group_index].items()
                    for ch_index in channel_indexes
                ]
                signals = [
                    signal
                    for signal in self.select(
                        channels,
                        raw=True,
                        copy_master=False,
                        record_offset=record_offset,
                        record_count=record_count,
                        validate=False,
                    )
                ]

                if not signals:
                    continue

                group_master = signals[0].timestamps

                for sig in signals:
                    if len(sig) == 0:
                        if empty_channels == "zeros":
                            sig.samples = np.zeros(
                                len(master)
                                if virtual_group.cycles_nr == 0
                                else virtual_group.cycles_nr,
                                dtype=sig.samples.dtype,
                            )
                            sig.timestamps = (
                                master if virtual_group.cycles_nr == 0 else group_master
                            )

                if not raw:
                    if ignore_value2text_conversions:
                        if self.version < "4.00":
                            text_conversion = 11
                        else:
                            text_conversion = 7

                        for signal in signals:
                            conversion = signal.conversion
                            if (
                                conversion
                                and conversion.conversion_type < text_conversion
                            ):
                                signal.samples = conversion.convert(signal.samples)

                    else:
                        for signal in signals:
                            if signal.conversion:
                                signal.samples = signal.conversion.convert(
                                    signal.samples
                                )

                for s_index, sig in enumerate(signals):
                    sig = sig.validate(copy=False)

                    if len(sig) == 0:
                        if empty_channels == "zeros":
                            sig.samples = np.zeros(
                                len(master)
                                if virtual_group.cycles_nr == 0
                                else virtual_group.cycles_nr,
                                dtype=sig.samples.dtype,
                            )
                            sig.timestamps = (
                                master if virtual_group.cycles_nr == 0 else group_master
                            )

                    signals[s_index] = sig

                if use_interpolation:
                    same_master = np.array_equal(master, group_master)

                    if not same_master and interpolate_outwards_with_nan:
                        idx = np.argwhere(
                            (master >= group_master[0]) & (master <= group_master[-1])
                        ).flatten()

                    cycles = len(group_master)

                    signals = [
                        signal.interp(master, self._integer_interpolation)
                        if not same_master or len(signal) != cycles
                        else signal
                        for signal in signals
                    ]

                    if not same_master and interpolate_outwards_with_nan:
                        for sig in signals:
                            sig.timestamps = sig.timestamps[idx]
                            sig.samples = sig.samples[idx]

                    group_master = master

                signals = [sig for sig in signals if len(sig)]

                if signals:
                    diffs = np.diff(group_master, prepend=-np.inf) > 0

                    if group_master.dtype.byteorder not in target_byte_order:
                        group_master = group_master.byteswap().newbyteorder()

                    if np.all(diffs):
                        index = pd.Index(group_master, tupleize_cols=False)

                    else:
                        idx = np.argwhere(diffs).flatten()
                        group_master = group_master[idx]

                        index = pd.Index(group_master, tupleize_cols=False)

                        for sig in signals:
                            sig.samples = sig.samples[idx]
                            sig.timestamps = sig.timestamps[idx]

                size = len(index)
                for k, sig in enumerate(signals):

                    if sig.timestamps.dtype.byteorder not in target_byte_order:
                        sig.timestamps = sig.timestamps.byteswap().newbyteorder()

                    sig_index = (
                        index
                        if len(sig) == size
                        else pd.Index(sig.timestamps, tupleize_cols=False)
                    )

                    # byte arrays
                    if len(sig.samples.shape) > 1:

                        if use_display_names:
                            channel_name = (
                                list(sig.display_names)[0]
                                if sig.display_names
                                else sig.name
                            )
                        else:
                            channel_name = sig.name

                        channel_name = used_names.get_unique_name(channel_name)

                        if sig.samples.dtype.byteorder not in target_byte_order:
                            sig.samples = sig.samples.byteswap().newbyteorder()

                        df[channel_name] = pd.Series(
                            list(sig.samples),
                            index=sig_index,
                        )

                    # arrays and structures
                    elif sig.samples.dtype.names:
                        for name, series in components(
                            sig.samples,
                            sig.name,
                            used_names,
                            master=sig_index,
                            only_basenames=only_basenames,
                        ):
                            df[name] = series

                    # scalars
                    else:
                        if use_display_names:
                            channel_name = (
                                list(sig.display_names)[0]
                                if sig.display_names
                                else sig.name
                            )
                        else:
                            channel_name = sig.name

                        channel_name = used_names.get_unique_name(channel_name)

                        if reduce_memory_usage and sig.samples.dtype.kind in "SU":
                            unique = np.unique(sig.samples)

                            if sig.samples.dtype.byteorder not in target_byte_order:
                                sig.samples = sig.samples.byteswap().newbyteorder()

                            if len(sig.samples) / len(unique) >= 2:
                                df[channel_name] = pd.Series(
                                    sig.samples,
                                    index=sig_index,
                                    dtype="category",
                                )
                            else:
                                df[channel_name] = pd.Series(
                                    sig.samples,
                                    index=sig_index,
                                    fastpath=True,
                                )
                        else:
                            if reduce_memory_usage:
                                sig.samples = downcast(sig.samples)

                            if sig.samples.dtype.byteorder not in target_byte_order:
                                sig.samples = sig.samples.byteswap().newbyteorder()

                            df[channel_name] = pd.Series(
                                sig.samples,
                                index=sig_index,
                                fastpath=True,
                            )

                if self._callback:
                    self._callback(group_index + 1, groups_nr)

            strings, nonstrings = {}, {}

            for col, series in df.items():
                if series.dtype.kind == "S":
                    strings[col] = series
                else:
                    nonstrings[col] = series

            df = pd.DataFrame(nonstrings, index=master)

            for col, series in strings.items():
                df[col] = series

            df.index.name = "timestamps"

            if time_as_date:
                new_index = np.array(df.index) + self.header.start_time.timestamp()
                new_index = pd.to_datetime(new_index, unit="s")

                df.set_index(new_index, inplace=True)
            elif time_from_zero and len(master):
                df.set_index(df.index - df.index[0], inplace=True)

            yield df

    def to_dataframe(
        self,
        channels: ChannelsType | None = None,
        raster: RasterType | None = None,
        time_from_zero: bool = True,
        empty_channels: EmptyChannelsType = "skip",
        keep_arrays: bool = False,
        use_display_names: bool = False,
        time_as_date: bool = False,
        reduce_memory_usage: bool = False,
        raw: bool = False,
        ignore_value2text_conversions: bool = False,
        use_interpolation: bool = True,
        only_basenames: bool = False,
        interpolate_outwards_with_nan: bool = False,
        numeric_1D_only: bool = False,
    ) -> pd.DataFrame:
        """generate pandas DataFrame

        Parameters
        ----------
        channels : list
            list of items to be filtered (default None); each item can be :

                * a channel name string
                * (channel name, group index, channel index) list or tuple
                * (channel name, group index) list or tuple
                * (None, group index, channel index) list or tuple

        raster : float | np.array | str
            new raster that can be

            * a float step value
            * a channel name who's timestamps will be used as raster (starting with asammdf 5.5.0)
            * an array (starting with asammdf 5.5.0)

            see `resample` for examples of using this argument

        time_from_zero : bool
            adjust time channel to start from 0; default *True*
        empty_channels : str
            behaviour for channels without samples; the options are *skip* or
            *zeros*; default is *skip*
        use_display_names : bool
            use display name instead of standard channel name, if available.
        keep_arrays : bool
            keep arrays and structure channels as well as the
            component channels. If *True* this can be very slow. If *False*
            only the component channels are saved, and their names will be
            prefixed with the parent channel.
        time_as_date : bool
            the dataframe index will contain the datetime timestamps
            according to the measurement start time; default *False*. If
            *True* then the argument ``time_from_zero`` will be ignored.
        reduce_memory_usage : bool
            reduce memory usage by converting all float columns to float32 and
            searching for minimum dtype that can reprezent the values found
            in integer columns; default *False*
        raw (False) : bool
            the columns will contain the raw values

            .. versionadded:: 5.7.0

        ignore_value2text_conversions (False) : bool
            valid only for the channels that have value to text conversions and
            if *raw=False*. If this is True then the raw numeric values will be
            used, and the conversion will not be applied.

            .. versionadded:: 5.8.0

        use_interpolation (True) : bool
            option to perform interpoaltions when multiple timestamp raster are
            present. If *False* then dataframe columns will be automatically
            filled with NaN's were the dataframe index values are not found in
            the current column's timestamps

            .. versionadded:: 5.11.0

        only_basenames (False) : bool
            use just the field names, without prefix, for structures and channel
            arrays

            .. versionadded:: 5.13.0

        interpolate_outwards_with_nan : bool
            use NaN values for the samples that lie outside of the original
            signal's timestamps

            .. versionadded:: 5.15.0

        Returns
        -------
        dataframe : pandas.DataFrame

        """
        if channels is not None:
            mdf = self.filter(channels)

            result = mdf.to_dataframe(
                raster=raster,
                time_from_zero=time_from_zero,
                empty_channels=empty_channels,
                keep_arrays=keep_arrays,
                use_display_names=use_display_names,
                time_as_date=time_as_date,
                reduce_memory_usage=reduce_memory_usage,
                raw=raw,
                ignore_value2text_conversions=ignore_value2text_conversions,
                use_interpolation=use_interpolation,
                only_basenames=only_basenames,
                interpolate_outwards_with_nan=interpolate_outwards_with_nan,
                numeric_1D_only=numeric_1D_only,
            )

            mdf.close()
            return result

        target_byte_order = "<=" if sys.byteorder == "little" else ">="

        df = {}

        self._set_temporary_master(None)

        if raster is not None:
            try:
                raster = float(raster)
                assert raster > 0
            except (TypeError, ValueError):
                if isinstance(raster, str):
                    raster = self.get(raster).timestamps
                else:
                    raster = np.array(raster)
            else:
                raster = master_using_raster(self, raster)
            master = raster
        else:
            masters = {index: self.get_master(index) for index in self.virtual_groups}

            if masters:
                master = reduce(np.union1d, masters.values())
            else:
                master = np.array([], dtype="<f4")

            del masters

        idx = np.argwhere(np.diff(master, prepend=-np.inf) > 0).flatten()
        master = master[idx]

        used_names = UniqueDB()
        used_names.get_unique_name("timestamps")

        groups_nr = len(self.virtual_groups)

        for group_index, (virtual_group_index, virtual_group) in enumerate(
            self.virtual_groups.items()
        ):
            if virtual_group.cycles_nr == 0 and empty_channels == "skip":
                continue

            channels = [
                (None, gp_index, ch_index)
                for gp_index, channel_indexes in self.included_channels(
                    virtual_group_index
                )[virtual_group_index].items()
                for ch_index in channel_indexes
                if ch_index != self.masters_db.get(gp_index, None)
            ]

            signals = [
                signal
                for signal in self.select(
                    channels, raw=True, copy_master=False, validate=False
                )
            ]

            if not signals:
                continue

            group_master = signals[0].timestamps

            for sig in signals:
                if len(sig) == 0:
                    if empty_channels == "zeros":
                        sig.samples = np.zeros(
                            len(master)
                            if virtual_group.cycles_nr == 0
                            else virtual_group.cycles_nr,
                            dtype=sig.samples.dtype,
                        )
                        sig.timestamps = (
                            master if virtual_group.cycles_nr == 0 else group_master
                        )

            if not raw:
                if ignore_value2text_conversions:

                    for signal in signals:
                        conversion = signal.conversion
                        if conversion:
                            samples = conversion.convert(signal.samples)
                            if samples.dtype.kind not in "US":
                                signal.samples = samples
                else:
                    for signal in signals:
                        if signal.conversion:
                            signal.samples = signal.conversion.convert(signal.samples)

            for s_index, sig in enumerate(signals):
                sig = sig.validate(copy=False)

                if len(sig) == 0:
                    if empty_channels == "zeros":
                        sig.samples = np.zeros(
                            len(master)
                            if virtual_group.cycles_nr == 0
                            else virtual_group.cycles_nr,
                            dtype=sig.samples.dtype,
                        )
                        sig.timestamps = (
                            master if virtual_group.cycles_nr == 0 else group_master
                        )

                signals[s_index] = sig

            if use_interpolation:
                same_master = np.array_equal(master, group_master)

                if not same_master and interpolate_outwards_with_nan:
                    idx = np.argwhere(
                        (master >= group_master[0]) & (master <= group_master[-1])
                    ).flatten()

                cycles = len(group_master)

                signals = [
                    signal.interp(
                        master,
                        integer_interpolation_mode=self._integer_interpolation,
                        float_interpolation_mode=self._float_interpolation,
                    )
                    if not same_master or len(signal) != cycles
                    else signal
                    for signal in signals
                ]

                if not same_master and interpolate_outwards_with_nan:
                    for sig in signals:
                        sig.timestamps = sig.timestamps[idx]
                        sig.samples = sig.samples[idx]

                group_master = master

            if any(len(sig) for sig in signals):
                signals = [sig for sig in signals if len(sig)]

            if group_master.dtype.byteorder not in target_byte_order:
                group_master = group_master.byteswap().newbyteorder()

            if signals:

                diffs = np.diff(group_master, prepend=-np.inf) > 0
                if np.all(diffs):
                    index = pd.Index(group_master, tupleize_cols=False)

                else:
                    idx = np.argwhere(diffs).flatten()
                    group_master = group_master[idx]

                    index = pd.Index(group_master, tupleize_cols=False)

                    for sig in signals:
                        sig.samples = sig.samples[idx]
                        sig.timestamps = sig.timestamps[idx]
            else:
                index = pd.Index(group_master, tupleize_cols=False)

            size = len(index)
            for k, sig in enumerate(signals):
                if sig.timestamps.dtype.byteorder not in target_byte_order:
                    sig.timestamps = sig.timestamps.byteswap().newbyteorder()

                sig_index = (
                    index
                    if len(sig) == size
                    else pd.Index(sig.timestamps, tupleize_cols=False)
                )

                # byte arrays
                if len(sig.samples.shape) > 1:

                    if use_display_names:
                        channel_name = (
                            list(sig.display_names)[0]
                            if sig.display_names
                            else sig.name
                        )
                    else:
                        channel_name = sig.name

                    channel_name = used_names.get_unique_name(channel_name)

                    if sig.samples.dtype.byteorder not in target_byte_order:
                        sig.samples = sig.samples.byteswap().newbyteorder()

                    df[channel_name] = pd.Series(
                        list(sig.samples),
                        index=sig_index,
                    )

                # arrays and structures
                elif sig.samples.dtype.names:
                    for name, series in components(
                        sig.samples,
                        sig.name,
                        used_names,
                        master=sig_index,
                        only_basenames=only_basenames,
                    ):
                        df[name] = series

                # scalars
                else:
                    if use_display_names:
                        channel_name = (
                            list(sig.display_names)[0]
                            if sig.display_names
                            else sig.name
                        )
                    else:
                        channel_name = sig.name

                    channel_name = used_names.get_unique_name(channel_name)

                    if reduce_memory_usage and sig.samples.dtype.kind not in "SU":
                        sig.samples = downcast(sig.samples)

                    if sig.samples.dtype.byteorder not in target_byte_order:
                        sig.samples = sig.samples.byteswap().newbyteorder()

                    df[channel_name] = pd.Series(
                        sig.samples, index=sig_index, fastpath=True
                    )

            if self._callback:
                self._callback(group_index + 1, groups_nr)

        strings, nonstrings = {}, {}

        for col, series in df.items():
            if series.dtype.kind == "S":
                strings[col] = series
            else:
                nonstrings[col] = series

        if numeric_1D_only:
            nonstrings = {
                col: series
                for col, series in nonstrings.items()
                if series.dtype.kind in "uif"
            }
            strings = {}

        df = pd.DataFrame(nonstrings, index=master)

        for col, series in strings.items():
            df[col] = series

        df.index.name = "timestamps"

        if time_as_date:
            new_index = np.array(df.index) + self.header.start_time.timestamp()
            new_index = (
                pd.to_datetime(new_index, unit="s")
                .tz_localize("UTC")
                .tz_convert(LOCAL_TIMEZONE)
            )

            df.set_index(new_index, inplace=True)
        elif time_from_zero and len(master):
            df.set_index(df.index - df.index[0], inplace=True)

        return df

    def extract_bus_logging(
        self,
        database_files: dict[BusType, Iterable[DbcFileType]],
        version: str | None = None,
        ignore_invalid_signals: bool | None = None,
        consolidated_j1939: bool = True,
        ignore_value2text_conversion: bool = True,
        prefix: str = "",
    ) -> MDF:
        """extract all possible CAN signal using the provided databases.

        Changed in version 6.0.0 from `extract_can_logging`

        Parameters
        ----------
        database_files : dict
            each key will contain an iterable of database files for that bus type. The
            supported bus types are "CAN", "LIN". The iterables will contain the
            (databases, valid bus) pairs. The database can be a  str, pathlib.Path or canamtrix.CanMatrix object.
            The valid bus is an integer specifying for which bus channel the database
            can be applied; 0 means any bus channel.

            .. versionchanged:: 6.0.0 added canmatrix.CanMatrix type

            .. versionchanged:: 6.3.0 added bus channel fileter

        version (None) : str
            output file version
        ignore_invalid_signals (False) : bool
            ignore signals that have all samples equal to their maximum value

            .. versionadded:: 5.7.0

            .. deprecated:: 7.0.2
                this argument is no longer used and will be removed in the future

        consolidated_j1939 (True) : bool
            handle PGNs from all the messages as a single instance

            .. versionadded:: 5.7.0

        ignore_value2text_conversion (True): bool
            ignore value to text conversions

            .. versionadded:: 5.23.0

        prefix ("") : str
            prefix that will added to the channel group names and signal names in
            the output file

            .. versionadded:: 6.3.0


        Returns
        -------
        mdf : MDF
            new MDF file that contains the succesfully extracted signals

        Examples
        --------
        >>> "extrac CAN and LIN bus logging"
        >>> mdf = asammdf.MDF(r'bus_logging.mf4')
        >>> databases = {
        ...     "CAN": [("file1.dbc", 0), ("file2.arxml", 2)],
        ...     "LIN": [("file3.dbc", 0)],
        ... }
        >>> extracted = mdf.extract_bus_logging(database_files=database_files)
        >>> ...
        >>> "extrac just LIN bus logging"
        >>> mdf = asammdf.MDF(r'bus_logging.mf4')
        >>> databases = {
        ...     "LIN": [("file3.dbc", 0)],
        ... }
        >>> extracted = mdf.extract_bus_logging(database_files=database_files)

        """
        if ignore_invalid_signals is not None:
            warn(
                "The argument `ignore_invalid_signals` from the method `extract_bus_logging` is no longer used and will be removed in the future"
            )

        if version is None:
            version = self.version
        else:
            version = validate_version_argument(version)

        out = MDF(
            version=version,
            password=self._password,
            callback=self._callback,
            use_display_names=True,
        )
        out.header.start_time = self.header.start_time

        if self._callback:
            out._callback = out._mdf._callback = self._callback

        self.last_call_info = {}

        if database_files.get("CAN", None):
            out = self._extract_can_logging(
                out,
                database_files["CAN"],
                consolidated_j1939,
                ignore_value2text_conversion,
                prefix,
            )

        if database_files.get("LIN", None):
            out = self._extract_lin_logging(
                out,
                database_files["LIN"],
                ignore_value2text_conversion,
                prefix,
            )

        return out

    def _extract_can_logging(
        self,
        output_file: MDF,
        dbc_files: Iterable[DbcFileType],
        consolidated_j1939: bool = True,
        ignore_value2text_conversion: bool = True,
        prefix: str = "",
    ) -> MDF:

        out = output_file

        max_flags = []

        valid_dbc_files = []
        unique_name = UniqueDB()
        for (dbc_name, bus_channel) in dbc_files:
            if isinstance(dbc_name, CanMatrix):
                valid_dbc_files.append(
                    (
                        dbc_name,
                        unique_name.get_unique_name("UserProvidedCanMatrix"),
                        bus_channel,
                    )
                )
            else:
                dbc = load_can_database(Path(dbc_name))
                if dbc is None:
                    continue
                else:
                    valid_dbc_files.append((dbc, dbc_name, bus_channel))

        count = sum(
            1
            for group in self.groups
            if group.channel_group.flags & v4c.FLAG_CG_BUS_EVENT
            and group.channel_group.acq_source.bus_type == v4c.BUS_TYPE_CAN
        )
        count *= len(valid_dbc_files)

        cntr = 0

        total_unique_ids = set()
        found_ids = defaultdict(set)
        not_found_ids = defaultdict(list)
        unknown_ids = defaultdict(list)

        for dbc, dbc_name, bus_channel in valid_dbc_files:
            is_j1939 = dbc.contains_j1939
            if is_j1939:
                messages = {message.arbitration_id.pgn: message for message in dbc}
            else:
                messages = {message.arbitration_id.id: message for message in dbc}

            current_not_found_ids = {
                (msg_id, message.name) for msg_id, message in messages.items()
            }

            msg_map = {}

            for i, group in enumerate(self.groups):
                if (
                    not group.channel_group.flags & v4c.FLAG_CG_BUS_EVENT
                    or not group.channel_group.acq_source.bus_type == v4c.BUS_TYPE_CAN
                    or not "CAN_DataFrame" in [ch.name for ch in group.channels]
                ):
                    continue

                self._prepare_record(group)
                data = self._load_data(group, optimize_read=False)

                for fragment_index, fragment in enumerate(data):

                    self._set_temporary_master(None)
                    self._set_temporary_master(self.get_master(i, data=fragment))

                    bus_ids = self.get(
                        "CAN_DataFrame.BusChannel",
                        group=i,
                        data=fragment,
                        samples_only=True,
                    )[0].astype("<u1")

                    msg_ids = (
                        self.get("CAN_DataFrame.ID", group=i, data=fragment).astype(
                            "<u4"
                        )
                        & 0x1FFFFFFF
                    )

                    original_ids = msg_ids.samples.copy()

                    if is_j1939:
                        tmp_pgn = msg_ids.samples >> 8
                        ps = tmp_pgn & 0xFF
                        pf = (msg_ids.samples >> 16) & 0xFF
                        _pgn = tmp_pgn & 0x3FF00
                        msg_ids.samples = np.where(pf >= 240, _pgn + ps, _pgn)

                    data_bytes = self.get(
                        "CAN_DataFrame.DataBytes",
                        group=i,
                        data=fragment,
                        samples_only=True,
                    )[0]

                    buses = np.unique(bus_ids)

                    for bus in buses:
                        if bus_channel and bus != bus_channel:
                            continue
                        idx = np.argwhere(bus_ids == bus).ravel()
                        bus_t = msg_ids.timestamps[idx]
                        bus_msg_ids = msg_ids.samples[idx]
                        bus_data_bytes = data_bytes[idx]
                        original_msg_ids = original_ids[idx]

                        if is_j1939:

                            if consolidated_j1939:
                                unique_ids = np.unique(
                                    np.core.records.fromarrays(
                                        [bus_msg_ids, bus_msg_ids]
                                    )
                                )
                            else:
                                unique_ids = np.unique(
                                    np.core.records.fromarrays(
                                        [bus_msg_ids, original_msg_ids]
                                    )
                                )
                        else:
                            unique_ids = np.unique(
                                np.core.records.fromarrays([bus_msg_ids, bus_msg_ids])
                            )

                        total_unique_ids = total_unique_ids | set(
                            tuple(int(e) for e in f) for f in unique_ids
                        )

                        for msg_id_record in unique_ids:
                            msg_id = int(msg_id_record[0])
                            original_msg_id = int(msg_id_record[1])
                            message = messages.get(msg_id, None)
                            if message is None:
                                unknown_ids[msg_id].append(True)
                                continue

                            found_ids[dbc_name].add((msg_id, message.name))
                            try:
                                current_not_found_ids.remove((msg_id, message.name))
                            except KeyError:
                                pass

                            unknown_ids[msg_id].append(False)

                            if is_j1939 and not consolidated_j1939:
                                idx = np.argwhere(
                                    (bus_msg_ids == msg_id)
                                    & (original_msg_ids == original_msg_id)
                                ).ravel()
                            else:
                                idx = np.argwhere(bus_msg_ids == msg_id).ravel()
                            payload = bus_data_bytes[idx]
                            t = bus_t[idx]

                            extracted_signals = extract_mux(
                                payload,
                                message,
                                msg_id,
                                bus,
                                t,
                                original_message_id=original_msg_id
                                if is_j1939 and not consolidated_j1939
                                else None,
                                ignore_value2text_conversion=ignore_value2text_conversion,
                                is_j1939=is_j1939,
                            )

                            for entry, signals in extracted_signals.items():
                                if len(next(iter(signals.values()))["samples"]) == 0:
                                    continue
                                if entry not in msg_map:
                                    sigs = []

                                    index = len(out.groups)
                                    msg_map[entry] = index

                                    for name_, signal in signals.items():
                                        signal_name = f"{prefix}{signal['name']}"
                                        sig = Signal(
                                            samples=signal["samples"],
                                            timestamps=signal["t"],
                                            name=signal_name,
                                            comment=signal["comment"],
                                            unit=signal["unit"],
                                            invalidation_bits=signal[
                                                "invalidation_bits"
                                            ],
                                        )

                                        sig.comment = f"""\
<CNcomment>
<TX>{sig.comment}</TX>
<names>
    <display>CAN{bus}.{message.name}.{signal_name}</display>
</names>
</CNcomment>"""
                                        sigs.append(sig)

                                    if is_j1939:
                                        if consolidated_j1939:
                                            if prefix:
                                                acq_name = (
                                                    comment
                                                ) = f"{prefix}: CAN{bus} consolidated PGN=0x{msg_id:X} {message}"
                                            else:
                                                acq_name = (
                                                    comment
                                                ) = f"CAN{bus} consolidated PGN=0x{msg_id:X} {message}"

                                        else:
                                            source_adddress = original_msg_id & 0xFF
                                            if prefix:
                                                comment = f"{prefix}: CAN{bus} PGN=0x{msg_id:X} {message} from ID=0x{original_msg_id:X} SA=0x{source_adddress:X}"
                                            else:
                                                comment = f"CAN{bus} PGN=0x{msg_id:X} {message} from ID=0x{original_msg_id:X} SA=0x{source_adddress:X}"
                                            acq_name = (
                                                f"SourceAddress = 0x{source_adddress}"
                                            )
                                    else:

                                        if prefix:
                                            acq_name = f"{prefix}: CAN{bus} message ID=0x{msg_id:X}"
                                            comment = f'{prefix}: CAN{bus} - message "{message}" 0x{msg_id:X}'
                                        else:
                                            acq_name = (
                                                f"CAN{bus} message ID=0x{msg_id:X}"
                                            )
                                            comment = f"CAN{bus} - message {message} 0x{msg_id:X}"

                                    acq_source = Source(
                                        name=acq_name,
                                        path=f"CAN{int(bus)}.CAN_DataFrame.ID=0x{message.arbitration_id.id:X}",
                                        comment=f"""\
<SIcomment>
    <TX>CAN{bus} data frame 0x{message.arbitration_id.id:X} - {message.name}</TX>
    <bus name="CAN{int(bus)}"/>
    <common_properties>
        <e name="ChannelNo" type="integer">{int(bus)}</e>
    </common_properties>
</SIcomment>""",
                                        source_type=v4c.SOURCE_BUS,
                                        bus_type=v4c.BUS_TYPE_CAN,
                                    )

                                    for sig in sigs:
                                        sig.source = acq_source

                                    cg_nr = out.append(
                                        sigs,
                                        acq_name=acq_name,
                                        acq_source=acq_source,
                                        comment=comment,
                                        common_timebase=True,
                                    )

                                    out.groups[
                                        cg_nr
                                    ].channel_group.flags = v4c.FLAG_CG_BUS_EVENT

                                    if is_j1939:
                                        max_flags.append([False])
                                        for ch_index, sig in enumerate(sigs, 1):
                                            max_flags[cg_nr].append(
                                                np.all(sig.invalidation_bits)
                                            )
                                    else:
                                        max_flags.append([False] * (len(sigs) + 1))

                                else:

                                    index = msg_map[entry]

                                    sigs = []

                                    for name_, signal in signals.items():

                                        sigs.append(
                                            (
                                                signal["samples"],
                                                signal["invalidation_bits"],
                                            )
                                        )

                                        t = signal["t"]

                                    if is_j1939:
                                        for ch_index, sig in enumerate(sigs, 1):
                                            max_flags[index][ch_index] = max_flags[
                                                index
                                            ][ch_index] or np.all(sig[1])

                                    sigs.insert(0, (t, None))

                                    out.extend(index, sigs)
                    self._set_temporary_master(None)

                cntr += 1
                if self._callback:
                    self._callback(cntr, count)

            if current_not_found_ids:
                not_found_ids[dbc_name] = list(current_not_found_ids)

        unknown_ids = {
            msg_id for msg_id, not_found in unknown_ids.items() if all(not_found)
        }

        self.last_call_info["CAN"] = {
            "dbc_files": dbc_files,
            "total_unique_ids": total_unique_ids,
            "unknown_id_count": len(unknown_ids),
            "not_found_ids": not_found_ids,
            "found_ids": found_ids,
            "unknown_ids": unknown_ids,
        }

        if self._callback:
            self._callback(100, 100)
        if not out.groups:
            logger.warning(
                f'No CAN signals could be extracted from "{self.name}". The'
                "output file will be empty."
            )

        return out

    def _extract_lin_logging(
        self,
        output_file: MDF,
        dbc_files: Iterable[DbcFileType],
        ignore_value2text_conversion: bool = True,
        prefix: str = "",
    ) -> MDF:

        out = output_file

        max_flags = []

        valid_dbc_files = []
        unique_name = UniqueDB()
        for (dbc_name, bus_channel) in dbc_files:
            if isinstance(dbc_name, CanMatrix):
                valid_dbc_files.append(
                    (
                        dbc_name,
                        unique_name.get_unique_name("UserProvidedCanMatrix"),
                        bus_channel,
                    )
                )
            else:
                dbc = load_can_database(Path(dbc_name))
                if dbc is None:
                    continue
                else:
                    valid_dbc_files.append((dbc, dbc_name, bus_channel))

        count = sum(
            1
            for group in self.groups
            if group.channel_group.flags & v4c.FLAG_CG_BUS_EVENT
            and group.channel_group.acq_source.bus_type == v4c.BUS_TYPE_LIN
        )
        count *= len(valid_dbc_files)

        cntr = 0

        total_unique_ids = set()
        found_ids = defaultdict(set)
        not_found_ids = defaultdict(list)
        unknown_ids = defaultdict(list)

        for dbc, dbc_name, bus_channel in valid_dbc_files:
            messages = {message.arbitration_id.id: message for message in dbc}

            current_not_found_ids = {
                (msg_id, message.name) for msg_id, message in messages.items()
            }

            msg_map = {}

            for i, group in enumerate(self.groups):
                if (
                    not group.channel_group.flags & v4c.FLAG_CG_BUS_EVENT
                    or not group.channel_group.acq_source.bus_type == v4c.BUS_TYPE_LIN
                    or not "LIN_Frame" in [ch.name for ch in group.channels]
                ):
                    continue

                self._prepare_record(group)
                data = self._load_data(group, optimize_read=False)

                for fragment_index, fragment in enumerate(data):

                    self._set_temporary_master(None)
                    self._set_temporary_master(self.get_master(i, data=fragment))

                    msg_ids = (
                        self.get("LIN_Frame.ID", group=i, data=fragment).astype("<u4")
                        & 0x1FFFFFFF
                    )

                    original_ids = msg_ids.samples.copy()

                    data_bytes = self.get(
                        "LIN_Frame.DataBytes",
                        group=i,
                        data=fragment,
                        samples_only=True,
                    )[0]

                    try:
                        bus_ids = self.get(
                            "LIN_Frame.BusChannel",
                            group=i,
                            data=fragment,
                            samples_only=True,
                        )[0].astype("<u1")
                    except:
                        bus_ids = np.ones(len(original_ids), dtype="u1")

                    bus_t = msg_ids.timestamps
                    bus_msg_ids = msg_ids.samples
                    bus_data_bytes = data_bytes
                    original_msg_ids = original_ids

                    unique_ids = np.unique(
                        np.core.records.fromarrays([bus_msg_ids, bus_msg_ids])
                    )

                    total_unique_ids = total_unique_ids | set(
                        tuple(int(e) for e in f) for f in unique_ids
                    )

                    buses = np.unique(bus_ids)

                    for bus in buses:
                        if bus_channel and bus != bus_channel:
                            continue

                        for msg_id_record in unique_ids:
                            msg_id = int(msg_id_record[0])
                            original_msg_id = int(msg_id_record[1])
                            message = messages.get(msg_id, None)
                            if message is None:
                                unknown_ids[msg_id].append(True)
                                continue

                            found_ids[dbc_name].add((msg_id, message.name))
                            try:
                                current_not_found_ids.remove((msg_id, message.name))
                            except KeyError:
                                pass

                            unknown_ids[msg_id].append(False)

                            idx = np.argwhere(bus_msg_ids == msg_id).ravel()
                            payload = bus_data_bytes[idx]
                            t = bus_t[idx]

                            extracted_signals = extract_mux(
                                payload,
                                message,
                                msg_id,
                                bus,
                                t,
                                original_message_id=None,
                                ignore_value2text_conversion=ignore_value2text_conversion,
                            )

                            for entry, signals in extracted_signals.items():
                                if len(next(iter(signals.values()))["samples"]) == 0:
                                    continue
                                if entry not in msg_map:
                                    sigs = []

                                    index = len(out.groups)
                                    msg_map[entry] = index

                                    for name_, signal in signals.items():
                                        signal_name = f"{prefix}{signal['name']}"
                                        sig = Signal(
                                            samples=signal["samples"],
                                            timestamps=signal["t"],
                                            name=signal_name,
                                            comment=signal["comment"],
                                            unit=signal["unit"],
                                            invalidation_bits=signal[
                                                "invalidation_bits"
                                            ],
                                        )

                                        sig.comment = f"""\
<CNcomment>
    <TX>{sig.comment}</TX>
    <names>
        <display>LIN.{message.name}.{signal_name}</display>
    </names>
</CNcomment>"""
                                        sigs.append(sig)

                                    if prefix:
                                        acq_name = f"{prefix}: from LIN{bus} message ID=0x{msg_id:X}"
                                    else:
                                        acq_name = (
                                            f"from LIN{bus} message ID=0x{msg_id:X}"
                                        )

                                    acq_source = Source(
                                        name=acq_name,
                                        path=f"LIN{int(bus)}.LIN_Frame.ID=0x{message.arbitration_id.id:X}",
                                        comment=f"""\
<SIcomment>
    <TX>LIN{bus} data frame 0x{message.arbitration_id.id:X} - {message.name}</TX>
    <bus name="LIN{int(bus)}"/>
    <common_properties>
        <e name="ChannelNo" type="integer">{int(bus)}</e>
    </common_properties>
</SIcomment>""",
                                        source_type=v4c.SOURCE_BUS,
                                        bus_type=v4c.BUS_TYPE_LIN,
                                    )

                                    for sig in sigs:
                                        sig.source = acq_source

                                    cg_nr = out.append(
                                        sigs,
                                        acq_name=acq_name,
                                        acq_source=acq_source,
                                        comment=f"from LIN{bus} - message {message} 0x{msg_id:X}",
                                        common_timebase=True,
                                    )

                                    out.groups[
                                        cg_nr
                                    ].channel_group.flags = v4c.FLAG_CG_BUS_EVENT

                                else:

                                    index = msg_map[entry]

                                    sigs = []

                                    for name_, signal in signals.items():
                                        sigs.append(
                                            (
                                                signal["samples"],
                                                signal["invalidation_bits"],
                                            )
                                        )

                                        t = signal["t"]

                                    sigs.insert(0, (t, None))

                                    out.extend(index, sigs)
                    self._set_temporary_master(None)

                cntr += 1
                if self._callback:
                    self._callback(cntr, count)

            if current_not_found_ids:
                not_found_ids[dbc_name] = list(current_not_found_ids)

        unknown_ids = {
            msg_id for msg_id, not_found in unknown_ids.items() if all(not_found)
        }

        self.last_call_info["LIN"] = {
            "dbc_files": dbc_files,
            "total_unique_ids": total_unique_ids,
            "unknown_id_count": len(unknown_ids),
            "not_found_ids": not_found_ids,
            "found_ids": found_ids,
            "unknown_ids": unknown_ids,
        }

        if self._callback:
            self._callback(100, 100)
        if not out.groups:
            logger.warning(
                f'No LIN signals could be extracted from "{self.name}". The'
                "output file will be empty."
            )

        return out

    @property
    def start_time(self) -> datetime:
        """getter and setter the measurement start timestamp

        Returns
        -------
        timestamp : datetime.datetime
            start timestamp

        """

        return self.header.start_time

    @start_time.setter
    def start_time(self, timestamp: datetime) -> None:
        self.header.start_time = timestamp

    def cleanup_timestamps(
        self,
        minimum: float,
        maximum: float,
        exp_min: int = -15,
        exp_max: int = 15,
        version: str | None = None,
    ) -> MDF:
        """convert *MDF* to other version

        .. versionadded:: 5.22.0

        Parameters
        ----------
        minimum : float
            minimum plausible time stamp
        maximum : float
            maximum plausible time stamp
        exp_min (-15) : int
            minimum plausible exponent used for the time stamps float values
        exp_max (15) : int
            maximum plausible exponent used for the time stamps float values
        version : str
            new mdf file version from ('2.00', '2.10', '2.14', '3.00', '3.10',
            '3.20', '3.30', '4.00', '4.10', '4.11', '4.20'); default the same as
            the input file

        Returns
        -------
        out : MDF
            new *MDF* object

        """

        if version is None:
            version = self.version
        else:
            version = validate_version_argument(version)

        out = MDF(version=version)

        out.header.start_time = self.header.start_time

        groups_nr = len(self.virtual_groups)

        if self._callback:
            self._callback(0, groups_nr)

        cg_nr = None

        self.configure(copy_on_get=False)

        # walk through all groups and get all channels
        for i, virtual_group in enumerate(self.virtual_groups):

            for idx, sigs in enumerate(
                self._yield_selected_signals(virtual_group, version=version)
            ):
                if idx == 0:
                    if sigs:

                        t = sigs[0].timestamps
                        if len(t):
                            all_ok, idx = plausible_timestamps(
                                t, minimum, maximum, exp_min, exp_max
                            )
                            if not all_ok:
                                t = t[idx]
                                if len(t):
                                    for sig in sigs:
                                        sig.samples = sig.samples[idx]
                                        sig.timestamps = t
                                        if sig.invalidation_bits is not None:
                                            sig.invalidation_bits = (
                                                sig.invalidation_bits[idx]
                                            )
                        cg = self.groups[virtual_group].channel_group
                        cg_nr = out.append(
                            sigs,
                            acq_name=getattr(cg, "acq_name", None),
                            acq_source=getattr(cg, "acq_source", None),
                            comment=f"Timestamps cleaned up and converted from {self.version} to {version}",
                            common_timebase=True,
                        )
                    else:
                        break
                else:
                    t, _ = sigs[0]
                    if len(t):
                        all_ok, idx = plausible_timestamps(
                            t, minimum, maximum, exp_min, exp_max
                        )
                        if not all_ok:
                            t = t[idx]
                            if len(t):
                                for i, (samples, invalidation_bits) in enumerate(sigs):
                                    if invalidation_bits is not None:
                                        invalidation_bits = invalidation_bits[idx]
                                    samples = samples[idx]

                                    sigs[i] = (samples, invalidation_bits)

                    out.extend(cg_nr, sigs)

            if self._callback:
                self._callback(i + 1, groups_nr)

            if self._terminate:
                return

        out._transfer_metadata(self)
        self.configure(copy_on_get=True)
        if self._callback:
            out._callback = out._mdf._callback = self._callback
        return out

    def whereis(
        self,
        channel: str,
        source_name: str | None = None,
        source_path: str | None = None,
        acq_name: str | None = None,
    ) -> tuple[tuple[int, int], ...]:
        """get occurrences of channel name in the file

        Parameters
        ----------
        channel : str
            channel name string
        source_name : str, optional
            filter occurrences on source name, by default None
        source_path : str, optional
            filter occurrences on source path, by default None
        acq_name : str, optional
            filter occurrences on channel group acquisition name, by default None

            .. versionadded:: 6.0.0

        Returns
        -------
        tuple[tuple[int, int], ...]
            (gp_idx, cn_idx) pairs

        Examples
        --------
        >>> mdf = MDF(file_name)
        >>> mdf.whereis('VehicleSpeed') # "VehicleSpeed" exists in the file
        ((1, 2), (2, 4))
        >>> mdf.whereis('VehicleSPD') # "VehicleSPD" doesn't exist in the file
        ()
        """
        occurrences = tuple(
            self._filter_occurrences(
                self.channels_db.get(channel, []),
                source_name=source_name,
                source_path=source_path,
                acq_name=acq_name,
            )
        )
        return occurrences

    def search(
        self,
        pattern: str,
        mode: Literal["plain", "regex", "wildcard"] | SearchMode = SearchMode.plain,
        case_insensitive: bool = False,
    ) -> list[str]:
        """search channels

        .. versionadded:: 7.0.0

        Parameters
        ----------
        pattern : str
            search pattern
        mode : Literal["plain", "regex", "wildcard"] or SearchMode, optional
            search mode, by default SearchMode.plain

                * `plain` : normal name search
                * `regex` : regular expression based search
                * `wildcard` : wildcard based search
        case_insensitive : bool, optional
            case sensitivity for the channel name search, by default False

        Returns
        -------
        list[str]
            name of the channels

        Raises
        ------
        ValueError
            unsupported search mode

        Examples
        --------
        >>> mdf = MDF(file_name)
        >>> mdf.search('*veh*speed*', case_insensitive=True, mode='wildcard') # case insensitive wildcard based search
        ['vehicleAverageSpeed', 'vehicleInstantSpeed', 'targetVehicleAverageSpeed', 'targetVehicleInstantSpeed']
        >>> mdf.search('^vehicle.*Speed$', case_insensitive=False, mode='regex') # case sensitive regex based search
        ['vehicleAverageSpeed', 'vehicleInstantSpeed']
        """
        search_mode = SearchMode(mode)

        if search_mode is SearchMode.plain:
            if case_insensitive:
                pattern = pattern.casefold()
                channels = [
                    name for name in self.channels_db if pattern in name.casefold()
                ]
            else:
                channels = [name for name in self.channels_db if pattern in name]
        elif search_mode is SearchMode.regex:
            flags = re.IGNORECASE if case_insensitive else 0
            compiled_pattern = re.compile(pattern, flags=flags)
            channels = [
                name for name in self.channels_db if compiled_pattern.search(name)
            ]
        elif search_mode is SearchMode.wildcard:
            if case_insensitive:
                channels = fnmatch.filter(self.channels_db, pattern)
            else:
                channels = [
                    name
                    for name in self.channels_db
                    if fnmatch.fnmatchcase(name, pattern)
                ]
        else:
            raise ValueError(f"unsupported mode {search_mode}")

        return channels class Signal(object):
    """
    The *Signal* represents a channel described by it's samples and timestamps.
    It can perform arithmetic operations against other *Signal* or numeric types.
    The operations are computed in respect to the timestamps (time correct).
    The non-float signals are not interpolated, instead the last value relative
    to the current timestamp is used.
    *samples*, *timestamps* and *name* are mandatory arguments.

    Parameters
    ----------
    samples : numpy.array | list | tuple
        signal samples
    timestamps : numpy.array | list | tuple
        signal timestamps
    unit : str
        signal unit
    name : str
        signal name
    conversion : dict | channel conversion block
        dict that contains extra conversion information about the signal ,
        default *None*
    comment : str
        signal comment, default ''
    raw : bool
        signal samples are raw values, with no physical conversion applied
    master_metadata : list
        master name and sync type
    display_names : dict
        display names used by mdf version 3
    attachment : bytes, name
        channel attachment and name from MDF version 4
    source : Source
        source information named tuple
    bit_count : int
        bit count; useful for integer channels
    stream_sync : bool
        the channel is a synchronisation for the attachment stream (mdf v4 only)
    invalidation_bits : numpy.array | None
        channel invalidation bits, default *None*
    encoding : str | None
        encoding for string signals; default *None*

    """

    def __init__(
        self,
        samples: ArrayLike | None = None,
        timestamps: ArrayLike | None = None,
        unit: str = "",
        name: str = "",
        conversion: dict[str, Any] | ChannelConversionType | None = None,
        comment: str = "",
        raw: bool = True,
        master_metadata: tuple[str, SyncType] | None = None,
        display_names: dict[str, str] | None = None,
        attachment: tuple[bytes, str | None, str | None] | None = None,
        source: SourceType | None = None,
        bit_count: int | None = None,
        stream_sync: bool = False,
        invalidation_bits: ArrayLike | None = None,
        encoding: str | None = None,
        group_index: int = -1,
        channel_index: int = -1,
    ) -> None:

        if samples is None or timestamps is None or not name:
            message = (
                '"samples", "timestamps" and "name" are mandatory '
                f"for Signal class __init__: samples={samples}\n"
                f"timestamps={timestamps}\nname={name}"
            )
            raise MdfException(message)
        else:
            if not isinstance(samples, np.ndarray):
                samples = np.array(samples)
                if samples.dtype.kind == "U":
                    if encoding is None:
                        encodings = ["utf-8", "latin-1"]
                    else:
                        encodings = [encoding, "utf-8", "latin-1"]
                    for encoding in encodings:
                        try:
                            samples = encode(samples, encoding)
                            break
                        except:
                            continue
                    else:
                        samples = encode(samples, encodings[0], errors="ignore")

            if not isinstance(timestamps, np.ndarray):
                timestamps = np.array(timestamps, dtype=np.float64)
            if samples.shape[0] != timestamps.shape[0]:
                message = "{} samples and timestamps length mismatch ({} vs {})"
                message = message.format(name, samples.shape[0], timestamps.shape[0])
                logger.exception(message)
                raise MdfException(message)

            self.samples = samples
            self.timestamps = timestamps
            self.unit = unit
            self.name = name
            self.comment = comment
            self._plot_axis = None
            self.raw = raw
            self.master_metadata = master_metadata
            self.display_names = display_names or {}
            self.attachment = attachment
            self.encoding = encoding
            self.group_index = group_index
            self.channel_index = channel_index

            if source:
                if not isinstance(source, Source):
                    source = Source.from_source(source)
            self.source = source

            if bit_count is None:
                self.bit_count = samples.dtype.itemsize * 8
            else:
                self.bit_count = bit_count

            self.stream_sync = stream_sync

            if invalidation_bits is not None:
                if not isinstance(invalidation_bits, np.ndarray):
                    invalidation_bits = np.array(invalidation_bits)
                if invalidation_bits.shape[0] != samples.shape[0]:
                    message = (
                        "{} samples and invalidation bits length mismatch ({} vs {})"
                    )
                    message = message.format(
                        name, samples.shape[0], invalidation_bits.shape[0]
                    )
                    logger.exception(message)
                    raise MdfException(message)
            self.invalidation_bits = invalidation_bits

            if conversion:
                if not isinstance(
                    conversion, (v4b.ChannelConversion, v3b.ChannelConversion)
                ):
                    conversion = from_dict(conversion)

            self.conversion = conversion

    def __repr__(self):
        return f"""<Signal {self.name}:
\tsamples={self.samples}
\ttimestamps={self.timestamps}
\tinvalidation_bits={self.invalidation_bits}
\tunit="{self.unit}"
\tconversion={self.conversion}
\tsource={self.source}
\tcomment="{self.comment}"
\tmastermeta="{self.master_metadata}"
\traw={self.raw}
\tdisplay_names={self.display_names}
\tattachment={self.attachment}>
"""

    def plot(self, validate: bool = True, index_only: bool = False) -> None:
        """plot Signal samples. Pyqtgraph is used if it is available; in this
        case see the GUI plot documentation to see the available commands

        Parameters
        ----------
        validate (True): bool
            apply the invalidation bits

        index_only (False) : bool
            use index based X axis. This can be useful if the master (usually
            time based) is corrupted with NaN, inf or if it is not strictly
            increasing

        """
        try:

            from .gui.plot import plot

            plot(self, validate=True, index_only=False)
            return

        except:
            try:
                import matplotlib.pyplot as plt
                from matplotlib.widgets import Slider
            except ImportError:
                logging.warning("Signal plotting requires pyqtgraph or matplotlib")
                return

        if len(self.samples.shape) <= 1 and self.samples.dtype.names is None:
            fig = plt.figure()
            fig.canvas.set_window_title(self.name)
            fig.text(
                0.95,
                0.05,
                f"asammdf {__version__}",
                fontsize=8,
                color="red",
                ha="right",
                va="top",
                alpha=0.5,
            )

            name = self.name

            if self.comment:
                comment = self.comment.replace("$", "")
                comment = extract_cncomment_xml(comment)
                comment = fill(comment, 120).replace("\\n", " ")

                title = f"{name}\n({comment})"
                plt.title(title)
            else:
                plt.title(name)
            try:
                if not self.master_metadata:
                    plt.xlabel("Time [s]")
                    plt.ylabel(f"[{self.unit}]")
                    plt.plot(self.timestamps, self.samples, "b")
                    plt.plot(self.timestamps, self.samples, "b.")
                    plt.grid(True)
                    plt.show()
                else:
                    master_name, sync_type = self.master_metadata
                    if sync_type in (0, 1):
                        plt.xlabel(f"{master_name} [s]")
                    elif sync_type == 2:
                        plt.xlabel(f"{master_name} [deg]")
                    elif sync_type == 3:
                        plt.xlabel(f"{master_name} [m]")
                    elif sync_type == 4:
                        plt.xlabel(f"{master_name} [index]")
                    plt.ylabel(f"[{self.unit}]")
                    plt.plot(self.timestamps, self.samples, "b")
                    plt.plot(self.timestamps, self.samples, "b.")
                    plt.grid(True)
                    plt.show()
            except ValueError:
                plt.close(fig)
        else:
            try:
                names = self.samples.dtype.names
                if self.samples.dtype.names is None or len(names) == 1:

                    if names:
                        samples = self.samples[names[0]]
                    else:
                        samples = self.samples

                    shape = samples.shape[1:]

                    fig = plt.figure()
                    fig.canvas.set_window_title(self.name)
                    fig.text(
                        0.95,
                        0.05,
                        f"asammdf {__version__}",
                        fontsize=8,
                        color="red",
                        ha="right",
                        va="top",
                        alpha=0.5,
                    )

                    if self.comment:
                        comment = self.comment.replace("$", "")
                        plt.title(f"{self.name}\n({comment})")
                    else:
                        plt.title(self.name)

                    ax = fig.add_subplot(111, projection="3d")

                    # Grab some test data.
                    X = np.array(range(shape[1]))
                    Y = np.array(range(shape[0]))
                    X, Y = np.meshgrid(X, Y)

                    Z = samples[0]

                    # Plot a basic wireframe.
                    self._plot_axis = ax.plot_wireframe(X, Y, Z, rstride=1, cstride=1)

                    # Place Sliders on Graph
                    ax_a = plt.axes([0.25, 0.1, 0.65, 0.03])

                    # Create Sliders & Determine Range
                    sa = Slider(
                        ax_a,
                        "Time [s]",
                        self.timestamps[0],
                        self.timestamps[-1],
                        valinit=self.timestamps[0],
                    )

                    def update(val):
                        self._plot_axis.remove()
                        idx = np.searchsorted(self.timestamps, sa.val, side="right")
                        Z = samples[idx - 1]
                        self._plot_axis = ax.plot_wireframe(
                            X, Y, Z, rstride=1, cstride=1
                        )
                        fig.canvas.draw_idle()

                    sa.on_changed(update)

                    plt.show()

                else:
                    fig = plt.figure()
                    fig.canvas.set_window_title(self.name)
                    fig.text(
                        0.95,
                        0.05,
                        f"asammdf {__version__}",
                        fontsize=8,
                        color="red",
                        ha="right",
                        va="top",
                        alpha=0.5,
                    )

                    if self.comment:
                        comment = self.comment.replace("$", "")
                        plt.title(f"{self.name}\n({comment})")
                    else:
                        plt.title(self.name)

                    ax = fig.add_subplot(111, projection="3d")

                    samples = self.samples[names[0]]
                    axis1 = self.samples[names[1]]
                    axis2 = self.samples[names[2]]

                    # Grab some test data.
                    X, Y = np.meshgrid(axis2[0], axis1[0])

                    Z = samples[0]

                    # Plot a basic wireframe.
                    self._plot_axis = ax.plot_wireframe(X, Y, Z, rstride=1, cstride=1)

                    # Place Sliders on Graph
                    ax_a = plt.axes([0.25, 0.1, 0.65, 0.03])

                    # Create Sliders & Determine Range
                    sa = Slider(
                        ax_a,
                        "Time [s]",
                        self.timestamps[0],
                        self.timestamps[-1],
                        valinit=self.timestamps[0],
                    )

                    def update(val):
                        self._plot_axis.remove()
                        idx = np.searchsorted(self.timestamps, sa.val, side="right")
                        Z = samples[idx - 1]
                        X, Y = np.meshgrid(axis2[idx - 1], axis1[idx - 1])
                        self._plot_axis = ax.plot_wireframe(
                            X, Y, Z, rstride=1, cstride=1
                        )
                        fig.canvas.draw_idle()

                    sa.on_changed(update)

                    plt.show()

            except Exception as err:
                print(err)

    def cut(
        self,
        start: float | None = None,
        stop: float | None = None,
        include_ends: bool = True,
        integer_interpolation_mode: IntInterpolationModeType
        | IntegerInterpolation = IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE,
        float_interpolation_mode: FloatInterpolationModeType
        | FloatInterpolation = FloatInterpolation.LINEAR_INTERPOLATION,
    ) -> Signal:
        """
        Cuts the signal according to the *start* and *stop* values, by using
        the insertion indexes in the signal's *time* axis.

        Parameters
        ----------
        start : float
            start timestamp for cutting
        stop : float
            stop timestamp for cutting
        include_ends : bool
            include the *start* and *stop* timestamps after cutting the signal.
            If *start* and *stop* are found in the original timestamps, then
            the new samples will be computed using interpolation. Default *True*

        integer_interpolation_mode : int
            interpolation mode for integer signals; default 0

                * 0 - repeat previous samples
                * 1 - linear interpolation
                * 2 - hybrid interpolation: channels with integer data type (raw values) that have a
                  conversion that outputs float values will use linear interpolation, otherwise
                  the previous sample is used

                .. versionadded:: 6.2.0

        float_interpolation_mode : int
            interpolation mode for float channels; default 1

                * 0 - repeat previous sample
                * 1 - use linear interpolation

                .. versionadded:: 6.2.0

        Returns
        -------
        result : Signal
            new *Signal* cut from the original

        Examples
        --------
        >>> new_sig = old_sig.cut(1.0, 10.5)
        >>> new_sig.timestamps[0], new_sig.timestamps[-1]
        0.98, 10.48

        """

        integer_interpolation_mode = IntegerInterpolation(integer_interpolation_mode)
        float_interpolation_mode = FloatInterpolation(float_interpolation_mode)

        original_start, original_stop = (start, stop)

        if len(self) == 0:
            result = Signal(
                np.array([], dtype=self.samples.dtype),
                np.array([], dtype=self.timestamps.dtype),
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

        elif start is None and stop is None:
            # return the channel uncut
            result = Signal(
                self.samples.copy(),
                self.timestamps.copy(),
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                invalidation_bits=self.invalidation_bits.copy()
                if self.invalidation_bits is not None
                else None,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

        else:
            if start is None:
                # cut from begining to stop
                if stop < self.timestamps[0]:
                    result = Signal(
                        np.array([], dtype=self.samples.dtype),
                        np.array([], dtype=self.timestamps.dtype),
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

                else:
                    stop = np.searchsorted(self.timestamps, stop, side="right")
                    if (
                        include_ends
                        and original_stop not in self.timestamps
                        and original_stop < self.timestamps[-1]
                    ):
                        interpolated = self.interp(
                            [original_stop],
                            integer_interpolation_mode=integer_interpolation_mode,
                            float_interpolation_mode=float_interpolation_mode,
                        )

                        if len(interpolated):
                            samples = np.append(
                                self.samples[:stop], interpolated.samples, axis=0
                            )
                            timestamps = np.append(
                                self.timestamps[:stop], interpolated.timestamps
                            )
                            if self.invalidation_bits is not None:
                                invalidation_bits = np.append(
                                    self.invalidation_bits[:stop],
                                    interpolated.invalidation_bits,
                                )
                            else:
                                invalidation_bits = None
                    else:
                        samples = self.samples[:stop].copy()
                        timestamps = self.timestamps[:stop].copy()
                        if self.invalidation_bits is not None:
                            invalidation_bits = self.invalidation_bits[:stop].copy()
                        else:
                            invalidation_bits = None
                    result = Signal(
                        samples,
                        timestamps,
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        invalidation_bits=invalidation_bits,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

            elif stop is None:
                # cut from start to end
                if start > self.timestamps[-1]:
                    result = Signal(
                        np.array([], dtype=self.samples.dtype),
                        np.array([], dtype=self.timestamps.dtype),
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

                else:
                    start = np.searchsorted(self.timestamps, start, side="left")
                    if (
                        include_ends
                        and original_start not in self.timestamps
                        and original_start > self.timestamps[0]
                    ):
                        interpolated = self.interp(
                            [original_start],
                            integer_interpolation_mode=integer_interpolation_mode,
                            float_interpolation_mode=float_interpolation_mode,
                        )
                        if len(interpolated):
                            samples = np.append(
                                interpolated.samples, self.samples[start:], axis=0
                            )
                            timestamps = np.append(
                                interpolated.timestamps, self.timestamps[start:]
                            )
                            if self.invalidation_bits is not None:
                                invalidation_bits = np.append(
                                    interpolated.invalidation_bits,
                                    self.invalidation_bits[start:],
                                )
                            else:
                                invalidation_bits = None
                    else:
                        samples = self.samples[start:].copy()
                        timestamps = self.timestamps[start:].copy()
                        if self.invalidation_bits is not None:
                            invalidation_bits = self.invalidation_bits[start:].copy()
                        else:
                            invalidation_bits = None
                    result = Signal(
                        samples,
                        timestamps,
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        invalidation_bits=invalidation_bits,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

            else:
                # cut between start and stop
                if start > self.timestamps[-1] or stop < self.timestamps[0]:
                    result = Signal(
                        np.array([], dtype=self.samples.dtype),
                        np.array([], dtype=self.timestamps.dtype),
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )
                else:
                    start = np.searchsorted(self.timestamps, start, side="left")
                    stop = np.searchsorted(self.timestamps, stop, side="right")

                    if start == stop:
                        if include_ends:
                            if original_start == original_stop:
                                ends = np.array(
                                    [original_start], dtype=self.timestamps.dtype
                                )
                            else:
                                ends = np.array(
                                    [original_start, original_stop],
                                    dtype=self.timestamps.dtype,
                                )

                            interpolated = self.interp(
                                ends,
                                integer_interpolation_mode=integer_interpolation_mode,
                                float_interpolation_mode=float_interpolation_mode,
                            )
                            samples = interpolated.samples
                            timestamps = ends
                            invalidation_bits = interpolated.invalidation_bits
                        else:
                            samples = np.array([], dtype=self.samples.dtype)
                            timestamps = np.array([], dtype=self.timestamps.dtype)
                            if self.invalidation_bits is not None:
                                invalidation_bits = np.array([], dtype=bool)
                            else:
                                invalidation_bits = None
                    else:
                        samples = self.samples[start:stop].copy()
                        timestamps = self.timestamps[start:stop].copy()
                        if self.invalidation_bits is not None:
                            invalidation_bits = self.invalidation_bits[
                                start:stop
                            ].copy()
                        else:
                            invalidation_bits = None

                        if (
                            include_ends
                            and original_stop not in self.timestamps
                            and original_stop < self.timestamps[-1]
                        ):
                            interpolated = self.interp(
                                [original_stop],
                                integer_interpolation_mode=integer_interpolation_mode,
                                float_interpolation_mode=float_interpolation_mode,
                            )

                            if len(interpolated):
                                samples = np.append(
                                    samples, interpolated.samples, axis=0
                                )
                                timestamps = np.append(
                                    timestamps, interpolated.timestamps
                                )
                                if invalidation_bits is not None:
                                    invalidation_bits = np.append(
                                        invalidation_bits,
                                        interpolated.invalidation_bits,
                                    )

                        if (
                            include_ends
                            and original_start not in self.timestamps
                            and original_start > self.timestamps[0]
                        ):
                            interpolated = self.interp(
                                [original_start],
                                integer_interpolation_mode=integer_interpolation_mode,
                                float_interpolation_mode=float_interpolation_mode,
                            )

                            if len(interpolated):
                                samples = np.append(
                                    interpolated.samples, samples, axis=0
                                )
                                timestamps = np.append(
                                    interpolated.timestamps, timestamps
                                )

                                if invalidation_bits is not None:
                                    invalidation_bits = np.append(
                                        interpolated.invalidation_bits,
                                        invalidation_bits,
                                    )

                    result = Signal(
                        samples,
                        timestamps,
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        invalidation_bits=invalidation_bits,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

        return result

    def extend(self, other: Signal) -> Signal:
        """extend signal with samples from another signal

        Parameters
        ----------
        other : Signal

        Returns
        -------
        signal : Signal
            new extended *Signal*

        """
        if len(self.timestamps):
            last_stamp = self.timestamps[-1]
        else:
            last_stamp = 0
        if len(other):
            other_first_sample = other.timestamps[0]
            if last_stamp >= other_first_sample:
                timestamps = other.timestamps + last_stamp
            else:
                timestamps = other.timestamps

            if self.invalidation_bits is None and other.invalidation_bits is None:
                invalidation_bits = None
            elif self.invalidation_bits is None and other.invalidation_bits is not None:
                invalidation_bits = np.concatenate(
                    (np.zeros(len(self), dtype=bool), other.invalidation_bits)
                )
            elif self.invalidation_bits is not None and other.invalidation_bits is None:
                invalidation_bits = np.concatenate(
                    (self.invalidation_bits, np.zeros(len(other), dtype=bool))
                )
            else:
                invalidation_bits = np.append(
                    self.invalidation_bits, other.invalidation_bits
                )

            result = Signal(
                np.append(self.samples, other.samples, axis=0),
                np.append(self.timestamps, timestamps),
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                invalidation_bits=invalidation_bits,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )
        else:
            result = self

        return result

    def interp(
        self,
        new_timestamps: NDArray[Any],
        integer_interpolation_mode: IntInterpolationModeType
        | IntegerInterpolation = IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE,
        float_interpolation_mode: FloatInterpolationModeType
        | FloatInterpolation = FloatInterpolation.LINEAR_INTERPOLATION,
    ) -> Signal:
        """returns a new *Signal* interpolated using the *new_timestamps*

        Parameters
        ----------
        new_timestamps : np.array
            timestamps used for interpolation

        integer_interpolation_mode : int
            interpolation mode for integer signals; default 0

                * 0 - repeat previous samples
                * 1 - linear interpolation
                * 2 - hybrid interpolation: channels with integer data type (raw values) that have a
                  conversion that outputs float values will use linear interpolation, otherwise
                  the previous sample is used

                .. versionadded:: 6.2.0

        float_interpolation_mode : int
            interpolation mode for float channels; default 1

                * 0 - repeat previous sample
                * 1 - use linear interpolation

                .. versionadded:: 6.2.0

        Returns
        -------
        signal : Signal
            new interpolated *Signal*

        """

        integer_interpolation_mode = IntegerInterpolation(integer_interpolation_mode)
        float_interpolation_mode = FloatInterpolation(float_interpolation_mode)

        if not len(self.samples) or not len(new_timestamps):

            return Signal(
                self.samples[:0].copy(),
                self.timestamps[:0].copy(),
                self.unit,
                self.name,
                comment=self.comment,
                conversion=self.conversion,
                raw=self.raw,
                master_metadata=self.master_metadata,
                display_names=self.display_names,
                attachment=self.attachment,
                stream_sync=self.stream_sync,
                invalidation_bits=None,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )
        else:

            # # we need to validate first otherwise we can get false invalid data
            # # if the new timebase and the invaldiation bits are alligned in an
            # # infavorable way
            #
            # if self.invalidation_bits is not None:
            #     signal = self.validate()
            #     has_invalidation = True
            # else:
            #     signal = self
            #     has_invalidation = False

            signal = self
            invalidation_bits = signal.invalidation_bits

            if not len(signal.samples):
                return Signal(
                    self.samples[:0].copy(),
                    self.timestamps[:0].copy(),
                    self.unit,
                    self.name,
                    comment=self.comment,
                    conversion=self.conversion,
                    raw=self.raw,
                    master_metadata=self.master_metadata,
                    display_names=self.display_names,
                    attachment=self.attachment,
                    stream_sync=self.stream_sync,
                    invalidation_bits=None
                    if invalidation_bits is None
                    else np.array([], dtype=bool),
                    encoding=self.encoding,
                    group_index=self.group_index,
                    channel_index=self.channel_index,
                )

            if len(signal.samples.shape) > 1:
                idx = np.searchsorted(signal.timestamps, new_timestamps, side="right")
                idx -= 1
                idx[idx < 0] = 0
                s = signal.samples[idx]
                if invalidation_bits is not None:
                    invalidation_bits = invalidation_bits[idx]
            else:

                kind = signal.samples.dtype.kind

                if kind == "f":

                    if (
                        float_interpolation_mode
                        == FloatInterpolation.REPEAT_PREVIOUS_SAMPLE
                    ):
                        idx = np.searchsorted(
                            signal.timestamps, new_timestamps, side="right"
                        )
                        idx -= 1
                        idx[idx < 0] = 0
                        s = signal.samples[idx]

                        if invalidation_bits is not None:
                            invalidation_bits = invalidation_bits[idx]

                    else:
                        s = np.interp(new_timestamps, signal.timestamps, signal.samples)

                        if invalidation_bits is not None:
                            idx = np.searchsorted(
                                signal.timestamps, new_timestamps, side="right"
                            )
                            idx -= 1
                            idx[idx < 0] = 0
                            invalidation_bits = invalidation_bits[idx]

                elif kind in "ui":
                    if (
                        integer_interpolation_mode
                        == IntegerInterpolation.HYBRID_INTERPOLATION
                    ):
                        if signal.raw and signal.conversion:
                            kind = signal.conversion.convert(
                                signal.samples[:1]
                            ).dtype.kind
                            if kind == "f":
                                integer_interpolation_mode = (
                                    IntegerInterpolation.LINEAR_INTERPOLATION
                                )

                    if (
                        integer_interpolation_mode
                        == IntegerInterpolation.HYBRID_INTERPOLATION
                    ):
                        integer_interpolation_mode = (
                            IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE
                        )

                    if (
                        integer_interpolation_mode
                        == IntegerInterpolation.LINEAR_INTERPOLATION
                    ):
                        s = np.interp(
                            new_timestamps, signal.timestamps, signal.samples
                        ).astype(signal.samples.dtype)

                        if invalidation_bits is not None:
                            idx = np.searchsorted(
                                signal.timestamps, new_timestamps, side="right"
                            )
                            idx -= 1
                            idx[idx < 0] = 0
                            invalidation_bits = invalidation_bits[idx]

                    elif (
                        integer_interpolation_mode
                        == IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE
                    ):
                        idx = np.searchsorted(
                            signal.timestamps, new_timestamps, side="right"
                        )
                        idx -= 1
                        idx[idx < 0] = 0

                        s = signal.samples[idx]

                        if invalidation_bits is not None:
                            invalidation_bits = invalidation_bits[idx]

                else:
                    idx = np.searchsorted(
                        signal.timestamps, new_timestamps, side="right"
                    )
                    idx -= 1
                    idx[idx < 0] = 0
                    s = signal.samples[idx]

                    if invalidation_bits is not None:
                        invalidation_bits = invalidation_bits[idx]

            return Signal(
                s,
                new_timestamps,
                self.unit,
                self.name,
                comment=self.comment,
                conversion=self.conversion,
                source=self.source,
                raw=self.raw,
                master_metadata=self.master_metadata,
                display_names=self.display_names,
                attachment=self.attachment,
                stream_sync=self.stream_sync,
                invalidation_bits=invalidation_bits,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

    def __apply_func(
        self, other: Signal | NDArray[Any] | None, func_name: str
    ) -> Signal:
        """delegate operations to the *samples* attribute, but in a time
        correct manner by considering the *timestamps*

        """

        if isinstance(other, Signal):
            if len(self) and len(other):
                start = max(self.timestamps[0], other.timestamps[0])
                stop = min(self.timestamps[-1], other.timestamps[-1])
                s1 = self.physical().cut(start, stop)
                s2 = other.physical().cut(start, stop)
            else:
                s1 = self
                s2 = other
            time = np.union1d(s1.timestamps, s2.timestamps)
            s = s1.interp(time).samples
            o = s2.interp(time).samples
            func = getattr(s, func_name)
            conversion = None
            s = func(o)
        elif other is None:
            s = self.samples
            conversion = self.conversion
            time = self.timestamps
        else:
            func = getattr(self.samples, func_name)
            s = func(other)
            conversion = self.conversion
            time = self.timestamps
        return Signal(
            samples=s,
            timestamps=time,
            unit=self.unit,
            name=self.name,
            conversion=conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
            group_index=self.group_index,
            channel_index=self.channel_index,
        )

    def __pos__(self) -> Signal:
        return self

    def __neg__(self) -> Signal:
        return Signal(
            np.negative(self.samples),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def __round__(self, n: int) -> Signal:
        return Signal(
            np.around(self.samples, n),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def __sub__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__sub__")

    def __isub__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__sub__(other)

    def __rsub__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return -self.__sub__(other)

    def __add__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__add__")

    def __iadd__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__add__(other)

    def __radd__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__add__(other)

    def __truediv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__truediv__")

    def __itruediv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__truediv__(other)

    def __rtruediv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__rtruediv__")

    def __mul__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__mul__")

    def __imul__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__mul__(other)

    def __rmul__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__mul__(other)

    def __floordiv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__floordiv__")

    def __ifloordiv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__truediv__(other)

    def __rfloordiv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return 1 / self.__apply_func(other, "__rfloordiv__")

    def __mod__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__mod__")

    def __pow__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__pow__")

    def __and__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__and__")

    def __or__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__or__")

    def __xor__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__xor__")

    def __invert__(self) -> Signal:
        s = ~self.samples
        time = self.timestamps
        return Signal(
            s,
            time,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def __lshift__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__lshift__")

    def __rshift__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__rshift__")

    def __lt__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__lt__")

    def __le__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__le__")

    def __gt__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__gt__")

    def __ge__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__ge__")

    def __eq__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__eq__")

    def __ne__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__ne__")

    def __iter__(self) -> Iterator[Any]:
        for item in (self.samples, self.timestamps, self.unit, self.name):
            yield item

    def __reversed__(self) -> Iterator[tuple[int, tuple[Any, Any]]]:
        return enumerate(zip(reversed(self.samples), reversed(self.timestamps)))

    def __len__(self) -> int:
        return len(self.samples)

    def __abs__(self) -> Signal:
        return Signal(
            np.fabs(self.samples),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
        )

    def __getitem__(self, val: int) -> Any:
        return self.samples[val]

    def __setitem__(self, idx: int, val: Any) -> None:
        self.samples[idx] = val

    def astype(self, np_type: DTypeLike) -> Signal:
        """returns new *Signal* with samples of dtype *np_type*

        Parameters
        ----------
        np_type : np.dtype
            new numpy dtye

        Returns
        -------
        signal : Signal
            new *Signal* with the samples of *np_type* dtype

        """
        return Signal(
            self.samples.astype(np_type),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def physical(self) -> Signal:
        """
        get the physical samples values

        Returns
        -------
        phys : Signal
            new *Signal* with physical values

        """

        if not self.raw or self.conversion is None:
            samples = self.samples.copy()
            encoding = None
        else:
            samples = self.conversion.convert(self.samples)
            if samples.dtype.kind == "S":
                encoding = "utf-8" if self.conversion.id == b"##CC" else "latin-1"
            else:
                encoding = None

        return Signal(
            samples,
            self.timestamps.copy(),
            unit=self.unit,
            name=self.name,
            conversion=None,
            raw=False,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=encoding,
            group_index=self.group_index,
            channel_index=self.channel_index,
        )

    def validate(self, copy: bool = True) -> Signal:
        """appply invalidation bits if they are available for this signal

        Parameters
        ----------
        copy (True) : bool
            return a copy of the result

            .. versionadded:: 5.12.0

        """
        if self.invalidation_bits is None:
            signal = self

        else:
            idx = np.nonzero(~self.invalidation_bits)[0]
            signal = Signal(
                self.samples[idx],
                self.timestamps[idx],
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                invalidation_bits=None,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

        if copy:
            signal = signal.copy()

        return signal

    def copy(self) -> Signal:
        """copy all attributes to a new Signal"""
        return Signal(
            self.samples.copy(),
            self.timestamps.copy(),
            self.unit,
            self.name,
            self.conversion,
            self.comment,
            self.raw,
            self.master_metadata,
            self.display_names,
            self.attachment,
            self.source,
            self.bit_count,
            self.stream_sync,
            invalidation_bits=self.invalidation_bits.copy()
            if self.invalidation_bits is not None
            else None,
            encoding=self.encoding,
            group_index=self.group_index,
            channel_index=self.channel_index,
        ) class ChannelInfoDialog(QtWidgets.QDialog):
    def __init__(self, channel, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.setWindowFlags(QtCore.Qt.Window)

        layout = QtWidgets.QVBoxLayout()
        self.setLayout(layout)

        self.setWindowTitle(channel.name)

        layout.addWidget(ChannelInfoWidget(channel, self))

        self.setStyleSheet('font: 8pt "Consolas";}')

        icon = QtGui.QIcon()
        icon.addPixmap(QtGui.QPixmap(":/info.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off)

        self.setWindowIcon(icon)
        self.setGeometry(240, 60, 1200, 600)

        screen = QtWidgets.QApplication.desktop().screenGeometry()
        self.move((screen.width() - 1200) // 2, (screen.height() - 600) // 2) class WindowSelectionDialog(Ui_WindowSelectionDialog, QtWidgets.QDialog):
    def __init__(self, options=("Plot", "Numeric", "Tabular"), *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setupUi(self)

        for i, name in enumerate(options):
            radio = QtWidgets.QRadioButton(name)
            self.selection_layout.addWidget(radio)
            if i == 0:
                radio.setChecked(True)

    def selected_type(self):
        for i in range(self.selection_layout.count()):
            radio = self.selection_layout.itemAt(i).widget()
            if radio.isChecked():
                return radio.text()
        return "" def compute_signal(description, measured_signals, all_timebase):
    type_ = description["type"]

    if type_ == "arithmetic":
        op = description["op"]

        operand1 = description["operand1"]
        if isinstance(operand1, dict):
            operand1 = compute_signal(operand1, measured_signals, all_timebase)
        elif isinstance(operand1, str):
            operand1 = measured_signals[operand1]

        operand2 = description["operand2"]
        if isinstance(operand2, dict):
            operand2 = compute_signal(operand2, measured_signals, all_timebase)
        elif isinstance(operand2, str):
            operand2 = measured_signals[operand2]

        result = eval(f"operand1 {op} operand2")
        if not hasattr(result, "name"):
            result = Signal(
                name="_",
                samples=np.ones(len(all_timebase)) * result,
                timestamps=all_timebase,
            )

    elif type_ == "function":
        function = description["name"]
        args = description["args"]

        channel = description["channel"]

        if isinstance(channel, dict):
            channel = compute_signal(channel, measured_signals, all_timebase)
        else:
            channel = measured_signals[channel]

        func = getattr(np, function)

        if function in [
            "arccos",
            "arcsin",
            "arctan",
            "cos",
            "deg2rad",
            "degrees",
            "rad2deg",
            "radians",
            "sin",
            "tan",
            "floor",
            "rint",
            "fix",
            "trunc",
            "cumprod",
            "cumsum",
            "diff",
            "exp",
            "log10",
            "log",
            "log2",
            "absolute",
            "cbrt",
            "sqrt",
            "square",
            "gradient",
        ]:

            samples = func(channel.samples)
            if function == "diff":
                timestamps = channel.timestamps[1:]
            else:
                timestamps = channel.timestamps

        elif function == "around":
            samples = func(channel.samples, *args)
            timestamps = channel.timestamps
        elif function == "clip":
            samples = func(channel.samples, *args)
            timestamps = channel.timestamps

        result = Signal(samples=samples, timestamps=timestamps, name="_")

    elif type_ == "expression":
        expression_string = description["expression"]
        expression_string = "".join(expression_string.splitlines())
        names = [match.group("name") for match in SIG_RE.finditer(expression_string)]
        positions = [
            (i, match.start(), match.end())
            for i, match in enumerate(SIG_RE.finditer(expression_string))
        ]
        positions.reverse()

        expression = expression_string
        for idx, start, end in positions:
            expression = expression[:start] + f"X_{idx}" + expression[end:]

        signals = [measured_signals[name] for name in names]
        common_timebase = reduce(np.union1d, [sig.timestamps for sig in signals])
        signals = {
            f"X_{i}": sig.interp(common_timebase).samples
            for i, sig in enumerate(signals)
        }

        samples = evaluate(expression, local_dict=signals)

        result = Signal(
            name="_",
            samples=samples,
            timestamps=common_timebase,
        )

    return result
 def copy_ranges(ranges):
    if ranges:
        new_ranges = []
        for range_info in ranges:
            range_info = dict(range_info)
            for color_name in ("background_color", "font_color"):
                color = range_info[color_name]
                if isinstance(color, QtGui.QBrush):
                    range_info[color_name] = QtGui.QBrush(color)
                elif isinstance(color, QtGui.QColor):
                    range_info[color_name] = QtGui.QColor(color)
            new_ranges.append(range_info)

        return new_ranges
    else:
        return ranges
 def extract_mime_names(data):
    def fix_comparison_name(data):
        for i, (
            name,
            group_index,
            channel_index,
            mdf_uuid,
            item_type,
            ranges,
        ) in enumerate(data):
            if item_type == "channel":
                if (group_index, channel_index) != (-1, -1):
                    name = COMPARISON_NAME.match(name).group("name").strip()
                    data[i][0] = name
            else:
                fix_comparison_name(channel_index)

    names = []
    if data.hasFormat("application/octet-stream-asammdf"):
        data = bytes(data.data("application/octet-stream-asammdf")).decode("utf-8")
        data = json.loads(data)
        fix_comparison_name(data)
        names = data

    return names
 class Bar(Ui_BarDisplay, QtWidgets.QWidget):
    add_channels_request = QtCore.pyqtSignal(list)
    timestamp_changed_signal = QtCore.pyqtSignal(object, float)

    def __init__(self, signals, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setupUi(self)
        for sig in signals:
            sig.timestamps = np.around(sig.timestamps, 9)
        self.signals = {}
        self._min = float("inf")
        self._max = -float("inf")
        self.add_new_channels(signals)

        self.timestamp.valueChanged.connect(self._timestamp_changed)
        self.timestamp_slider.valueChanged.connect(self._timestamp_slider_changed)

        self._timestamp = self._min
        self.set_timestamp()

        self._inhibit = False

    def items_deleted(self, names):
        for name in names:
            self.signals.pop(name)
        self.build()

    def _timestamp_changed(self, stamp):
        if not self._inhibit:
            self.set_timestamp(stamp)

    def _timestamp_slider_changed(self, stamp):
        if not self._inhibit:
            factor = stamp / 99999
            stamp = (self._max - self._min) * factor + self._min
            self.set_timestamp(stamp)

    def set_timestamp(self, stamp=None):
        if stamp is None:
            stamp = self._timestamp

        if not (self._min <= stamp <= self._max):
            return

        idx_cache = {}

        for row in range(self.channels.count()):
            item = self.channels.item(row)
            widget = self.channels.itemWidget(item)
            uuid = widget.uuid
            sig = self.signals[uuid]

            if len(sig):
                if (sig.group_index, sig.mdf_uuid) in idx_cache:
                    idx = idx_cache[(sig.group_index, sig.mdf_uuid)]
                else:
                    idx = min(sig.size - 1, searchsorted(sig.timestamps, stamp))
                    idx_cache[(sig.group_index, sig.mdf_uuid)] = idx
                value = sig.samples[idx]

                widget.set_value(value)
                widget.bar.update()

        self._inhibit = True
        if self._min != self._max:
            val = int((stamp - self._min) / (self._max - self._min) * 99999)
            self.timestamp_slider.setValue(val)
        self.timestamp.setValue(stamp)
        self._inhibit = False
        self.timestamp_changed_signal.emit(self, stamp)

    def add_new_channels(self, channels):
        for index, sig in enumerate(channels):
            sig.uuid = os.urandom(6).hex()
            sig.computed = False
            sig.computation = None
            sig.color = COLORS[index % 10]
            sig.size = len(sig)
            sig.kind = sig.samples.dtype.kind

        invalid = []

        for channel in channels:
            diff = np.diff(channel.timestamps)
            invalid_indexes = np.argwhere(diff <= 0).ravel()
            if len(invalid_indexes):
                invalid_indexes = invalid_indexes[:10] + 1
                idx = invalid_indexes[0]
                ts = channel.timestamps[idx - 1 : idx + 2]
                invalid.append(
                    f"{channel.name} @ index {invalid_indexes[:10] - 1} with first time stamp error: {ts}"
                )

        if invalid:
            errors = "\n".join(invalid)
            QtWidgets.QMessageBox.warning(
                self,
                "The following channels do not have monotonous increasing time stamps:",
                f"The following channels do not have monotonous increasing time stamps:\n{errors}",
            )
            self.plot._can_trim = False

        valid = []
        invalid = []
        for channel in channels:
            if len(channel):
                samples = channel.samples
                if samples.dtype.kind not in "SUV" and np.all(np.isnan(samples)):
                    invalid.append(channel.name)
                elif channel.conversion:
                    samples = channel.physical().samples
                    if samples.dtype.kind not in "SUV" and np.all(np.isnan(samples)):
                        invalid.append(channel.name)
                    else:
                        valid.append(channel)
                else:
                    valid.append(channel)
            else:
                valid.append(channel)

        if invalid:
            QtWidgets.QMessageBox.warning(
                self,
                "All NaN channels will not be plotted:",
                f"The following channels have all NaN samples and will not be plotted:\n{', '.join(invalid)}",
            )

        channels = valid

        for sig in channels:
            sig = PlotSignal(sig)
            if len(sig):
                self._min = min(self._min, sig.timestamps[0])
                self._max = max(self._max, sig.timestamps[-1])

            item = ListItem(
                (sig.group_index, sig.channel_index),
                sig.name,
                sig.computation,
                self.channels,
                sig.mdf_uuid,
            )
            item.setData(QtCore.Qt.UserRole, sig.name)
            tooltip = getattr(sig, "tooltip", "") or sig.comment

            it = ChannelBarDisplay(
                sig.uuid,
                0,
                (sig.min, sig.max),
                sig.max + 1,
                sig.color,
                sig.unit,
                3,
                tooltip,
                self,
            )
            it.setAttribute(QtCore.Qt.WA_StyledBackground)

            if sig.computed:
                font = QtGui.QFont()
                font.setItalic(True)
                it.name.setFont(font)

            it.set_name(sig.name)
            it.set_color(sig.color)
            item.setSizeHint(it.sizeHint())
            self.channels.addItem(item)
            self.channels.setItemWidget(item, it)

            self.signals[sig.uuid] = sig

    def keyPressEvent(self, event):
        key = event.key()
        modifier = event.modifiers()

        if (
            key in (QtCore.Qt.Key_H, QtCore.Qt.Key_B, QtCore.Qt.Key_P)
            and modifier == QtCore.Qt.ControlModifier
        ):
            if key == QtCore.Qt.Key_H:
                self.format_selection.setCurrentText("hex")
            elif key == QtCore.Qt.Key_B:
                self.format_selection.setCurrentText("bin")
            else:
                self.format_selection.setCurrentText("phys")
            event.accept()
        elif key == QtCore.Qt.Key_Right and modifier == QtCore.Qt.NoModifier:
            self.timestamp_slider.setValue(self.timestamp_slider.value() + 1)
        elif key == QtCore.Qt.Key_Left and modifier == QtCore.Qt.NoModifier:
            self.timestamp_slider.setValue(self.timestamp_slider.value() - 1)
        else:
            super().keyPressEvent(event)

    def to_config(self):

        channels = []
        iterator = QtWidgets.QTreeWidgetItemIterator(self.channels)
        while 1:
            item = iterator.value()
            if not item:
                break
            channels.append(item.text(0))
            iterator += 1

        config = {
            "format": self.format,
            "channels": channels if not self.pattern else [],
            "pattern": self.pattern,
        }

        return config

    def search_forward(self):
        if (
            self.op.currentIndex() < 0
            or not self.target.text().strip()
            or not self.pattern_match.text().strip()
        ):
            self.match.setText("invalid input values")
            return

        operator = self.op.currentText()

        pattern = self.pattern_match.text().strip().replace("*", "_WILDCARD_")
        pattern = re.escape(pattern)
        pattern = pattern.replace("_WILDCARD_", ".*")

        pattern = re.compile(f"(?i){pattern}")
        matches = [name for name in self.signals if pattern.search(name)]

        if not matches:
            self.match.setText("the pattern does not match any channel name")
            return

        try:
            target = float(self.target.text().strip())
        except:
            self.match.setText("the target must a numeric value")
        else:

            if target.is_integer():
                target = int(target)

            start = self.timestamp.value()

            timestamp = None
            signal_name = ""
            for name in matches:
                sig = self.signals[name].cut(start=start)
                samples = sig.samples

                op = getattr(samples, OPS[operator])
                try:
                    idx = np.argwhere(op(target)).flatten()
                    if len(idx):
                        if len(idx) == 1 or sig.timestamps[idx[0]] != start:
                            timestamp_ = sig.timestamps[idx[0]]
                        else:
                            timestamp_ = sig.timestamps[idx[1]]

                        if timestamp is None or timestamp_ < timestamp:
                            timestamp = timestamp_
                            signal_name = name
                except:
                    continue

            if timestamp is not None:
                self.timestamp.setValue(timestamp)
                self.match.setText(f"condition found for {signal_name}")
            else:
                self.match.setText("condition not found")

    def search_backward(self):
        if (
            self.op.currentIndex() < 0
            or not self.target.text().strip()
            or not self.pattern_match.text().strip()
        ):
            self.match.setText("invalid input values")
            return

        operator = self.op.currentText()

        pattern = self.pattern_match.text().strip().replace("*", "_WILDCARD_")
        pattern = re.escape(pattern)
        pattern = pattern.replace("_WILDCARD_", ".*")

        pattern = re.compile(f"(?i){pattern}")
        matches = [name for name in self.signals if pattern.search(name)]

        if not matches:
            self.match.setText("the pattern does not match any channel name")
            return

        try:
            target = float(self.target.text().strip())
        except:
            self.match.setText(f"the target must a numeric value")
        else:

            if target.is_integer():
                target = int(target)

            stop = self.timestamp.value()

            timestamp = None
            signal_name = ""
            for name in matches:
                sig = self.signals[name].cut(stop=stop)
                samples = sig.samples[:-1]

                op = getattr(samples, OPS[operator])
                try:
                    idx = np.argwhere(op(target)).flatten()
                    if len(idx):

                        if len(idx) == 1 or sig.timestamps[idx[-1]] != stop:
                            timestamp_ = sig.timestamps[idx[-1]]
                        else:
                            timestamp_ = sig.timestamps[idx[-2]]

                        if timestamp is None or timestamp_ > timestamp:
                            timestamp = timestamp_
                            signal_name = name
                except:
                    continue

            if timestamp is not None:
                self.timestamp.setValue(timestamp)
                self.match.setText(f"condition found for {signal_name}")
            else:
                self.match.setText(f"condition not found")

    def set_format(self, fmt):
        self.format = fmt
        self._settings.setValue("numeric_format", fmt)
        self._update_values() class CANBusTrace(TabularBase):
    add_channels_request = QtCore.pyqtSignal(list)

    def __init__(
        self, signals=None, start=0, format="phys", ranges=None, *args, **kwargs
    ):
        ranges = ranges or {name: [] for name in signals.columns}
        if not ranges["Event Type"]:
            ranges["Event Type"] = [
                {
                    "background_color": QtGui.QColor("#ff0000"),
                    "font_color": QtGui.QColor("#000000"),
                    "op1": "==",
                    "op2": "==",
                    "value1": "Error Frame",
                    "value2": None,
                },
                {
                    "background_color": QtGui.QColor("#00ff00"),
                    "font_color": QtGui.QColor("#000000"),
                    "op1": "==",
                    "op2": "==",
                    "value1": "Remote Frame",
                    "value2": None,
                },
            ]

        super().__init__(signals, ranges)

        self.signals_descr = {name: 0 for name in signals.columns}
        self.start = start
        self.pattern = {}
        self.format = format
        self.format_selection.setCurrentText(format)

        self._original_timestamps = signals["timestamps"]
        self._original_ts_series = pd.Series(
            self._original_timestamps,
            index=self.tree.pgdf.df_unfiltered.index,
        )

        prefixes = set()
        for name in self.tree.pgdf.df_unfiltered.columns:
            while "." in name:
                name = name.rsplit(".", 1)[0]
                prefixes.add(f"{name}.")

        self.filters.minimal_menu = True

        self.prefix.insertItems(0, sorted(prefixes, key=lambda x: (-len(x), x)))
        self.prefix.setEnabled(False)

        self.prefix.currentIndexChanged.connect(self.prefix_changed)

        if prefixes:
            self.remove_prefix.setCheckState(QtCore.Qt.Checked)

        self._settings = QtCore.QSettings()
        integer_mode = self._settings.value("tabular_format", "phys")

        self.format_selection.setCurrentText(integer_mode) class ChannelDisplay(Ui_ChannelDiplay, QtWidgets.QWidget):

    color_changed = QtCore.pyqtSignal(object, str)
    enable_changed = QtCore.pyqtSignal(object, int)
    ylink_changed = QtCore.pyqtSignal(object, int)
    individual_axis_changed = QtCore.pyqtSignal(object, int)
    unit_changed = QtCore.pyqtSignal(object, str)
    name_changed = QtCore.pyqtSignal(object, str)

    def __init__(
        self,
        uuid,
        unit="",
        kind="f",
        precision=3,
        tooltip="",
        details="",
        ranges=None,
        item=None,
        *args,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        self.setupUi(self)

        self.item = item

        self.color = "#ff0000"
        self._value_prefix = ""
        self._value = "n.a."
        self._name = ""
        self.double_clicked_enabled = True

        self.details.setText(details or "\tSource not available")

        self.details.setVisible(False)

        self.uuid = uuid
        self.set_ranges(ranges or [])
        self.resolved_ranges = None
        self._unit = unit.strip()
        self.kind = kind
        self.precision = precision

        self._transparent = True
        self._tooltip = tooltip

        self.color_btn.clicked.connect(self.select_color)
        # self.display.stateChanged.connect(self.display_changed)
        self.ylink.stateChanged.connect(self._ylink_changed)
        self.individual_axis.stateChanged.connect(self._individual_axis)

        self.fm = QtGui.QFontMetrics(self.name.font())

        self.setToolTip(self._tooltip or self._name)

        if kind in "SUVui" or self.precision == -1:
            self.fmt = "{}"
        else:
            self.fmt = f"{{:.{self.precision}f}}"

        self.setAutoFillBackground(True)
        self._back_ground_color = self.palette().color(QtGui.QPalette.Base)
        self._selected_color = self.palette().color(QtGui.QPalette.Highlight)
        self._selected_font_color = self.palette().color(QtGui.QPalette.HighlightedText)
        self._font_color = QtGui.QColor(self.color)

        self._current_background_color = self._back_ground_color
        self._current_font_color = self._font_color = QtGui.QColor(self.color)

        self.exists = True

    def set_double_clicked_enabled(self, state):
        self.double_clicked_enabled = state

    def set_unit(self, unit):
        unit = str(unit)
        if unit != self._unit:
            self._unit = unit
            self.unit_changed.emit(self.uuid, unit)

    def copy(self):
        new = ChannelDisplay(
            self.uuid,
            self._unit,
            self.kind,
            self.precision,
            self._tooltip,
            self.details.text(),
            ranges=copy_ranges(self.ranges),
        )

        new._value_prefix = self._value_prefix
        new.fmt = self.fmt
        new.set_name(self._name)
        new.individual_axis.setCheckState(self.individual_axis.checkState())
        new.ylink.setCheckState(self.ylink.checkState())
        new.set_color(self.color)
        new.set_value(self._value)

        return new

    def set_precision(self, precision):
        self.precision = precision
        if self.kind == "f" and precision >= 0:
            self.fmt = f"{{:.{self.precision}f}}"
        else:
            self.fmt = "{}"

    def _individual_axis(self, state):
        state = self.individual_axis.checkState()
        self.individual_axis_changed.emit(self.uuid, state)

    def _ylink_changed(self, state):
        state = self.ylink.checkState()
        self.ylink_changed.emit(self.uuid, state)

    def mouseDoubleClickEvent(self, event):
        if self.double_clicked_enabled:
            dlg = RangeEditor(self._name, self._unit, self.ranges, parent=self)
            dlg.exec_()
            if dlg.pressed_button == "apply":
                self.set_ranges(dlg.result)
                self.set_value(self._value, update=True)

    def select_color(self):
        color = QtWidgets.QColorDialog.getColor(QtGui.QColor(self.color))
        if color.isValid():
            self.set_color(color.name())
            self.color_changed.emit(self.uuid, color.name())

    def set_fmt(self, fmt):
        if self.kind in "SUV":
            self.fmt = "{}"
        elif self.kind == "f":
            self.fmt = f"{{:.{self.precision}f}}"
        else:
            if fmt == "hex":
                self.fmt = "0x{:X}"
            elif fmt == "bin":
                self.fmt = "0b{:b}"
            elif fmt == "phys":
                self.fmt = "{}"

    def set_color(self, color):
        self.color = color
        self.color_btn.setStyleSheet(f"background-color: {color};")

        self._font_color = QtGui.QColor(self.color)

        if self._current_font_color.name() != self._selected_font_color.name():
            self._current_font_color = self._font_color

        palette = self.palette()
        palette.setColor(QtGui.QPalette.Text, self._current_font_color)

        self.setPalette(palette)

        self.set_name(self._name)
        if self.item is not None:
            self.set_value(update=True)

    def set_selected(self, on):
        if on:
            self._current_background_color = self._selected_color
            self._current_font_color = self._selected_font_color
            self.set_value(update=True, force=True)
        else:
            self._current_background_color = self._back_ground_color
            self._current_font_color = self._font_color
            self.set_value(update=True, force=True)

    def set_name(self, text=""):
        self.setToolTip(self._tooltip or text)
        self._name = text
        self.name_changed.emit(self.uuid, text)

    def set_prefix(self, text=""):
        self._value_prefix = text

    def update_information(self):
        width = self.name.size().width()
        if self._unit:
            self.name.setText(
                self.fm.elidedText(
                    f"{self._name} ({self._unit})", QtCore.Qt.ElideMiddle, width
                )
            )
        else:
            self.name.setText(
                self.fm.elidedText(self._name, QtCore.Qt.ElideMiddle, width)
            )
        self.set_value(update=True)

    def set_value(self, value=None, update=False, force=False):
        if value is not None:
            if self._value == value and update is False:
                return
            else:
                self._value = value
        else:
            value = self._value

        default_background_color = self._current_background_color
        default_font_color = self._current_font_color

        new_background_color, new_font_color = get_colors_using_ranges(
            value,
            ranges=self.get_ranges(),
            default_background_color=default_background_color,
            default_font_color=default_font_color,
        )

        if (
            force
            or new_background_color is not default_background_color
            or new_font_color is not default_font_color
        ):
            p = self.palette()
            p.setColor(QtGui.QPalette.Base, new_background_color)
            p.setColor(QtGui.QPalette.Text, new_font_color)
            self.setPalette(p)

        template = "{{}}{}"
        if value not in ("", "n.a."):
            template = template.format(self.fmt)
        else:
            template = template.format("{}")
        try:
            self.value.setText(template.format(self._value_prefix, value))
        except (ValueError, TypeError):
            template = "{}{}"
            self.value.setText(template.format(self._value_prefix, value))

    def keyPressEvent(self, event):
        key = event.key()
        modifier = event.modifiers()
        if modifier == QtCore.Qt.ControlModifier and key == QtCore.Qt.Key_C:
            QtWidgets.QApplication.instance().clipboard().setText(self._name)

        elif (
            modifier == (QtCore.Qt.ControlModifier | QtCore.Qt.ShiftModifier)
            and key == QtCore.Qt.Key_C
        ):
            QtWidgets.QApplication.instance().clipboard().setText(
                self.get_display_properties()
            )

        elif (
            modifier == (QtCore.Qt.ControlModifier | QtCore.Qt.ShiftModifier)
            and key == QtCore.Qt.Key_P
        ):
            info = QtWidgets.QApplication.instance().clipboard().text()
            try:
                info = json.loads(info)
                self.set_color(info["color"])
                self.color_changed.emit(self.uuid, info["color"])
                self.set_fmt(info["format"])
                self.individual_axis.setCheckState(
                    QtCore.Qt.Checked
                    if info["individual_axis"]
                    else QtCore.Qt.Unchecked
                )
                self.ylink.setCheckState(
                    QtCore.Qt.Checked if info["ylink"] else QtCore.Qt.Unchecked
                )
                self.set_precision(info["precision"])

                parent = self.parent().parent().parent().parent().parent()
                sig, index = parent.plot.signal_by_uuid(self.uuid)
                viewbox = parent.plot.view_boxes[index]
                viewbox.setYRange(info["min"], info["max"], padding=0)

                # self.display.setCheckState(
                #     QtCore.Qt.Checked if info["display"] else QtCore.Qt.Unchecked
                # )

                self.set_ranges(info["ranges"])

            except:
                pass

        else:
            super().keyPressEvent(event)

    def resizeEvent(self, event):
        width = self.name.size().width()
        if self._unit:
            self.name.setText(
                self.fm.elidedText(
                    f"{self._name} ({self._unit})", QtCore.Qt.ElideMiddle, width
                )
            )
        else:
            self.name.setText(
                self.fm.elidedText(self._name, QtCore.Qt.ElideMiddle, width)
            )

    def text(self):
        return self._name

    def get_display_properties(self):
        info = {
            "color": self.color,
            "precision": self.precision,
            "ylink": self.ylink.checkState() == QtCore.Qt.Checked,
            "individual_axis": self.individual_axis.checkState() == QtCore.Qt.Checked,
            "format": "hex"
            if self.fmt.startswith("0x")
            else "bin"
            if self.fmt.startswith("0b")
            else "phys",
            # "display": self.display.checkState() == QtCore.Qt.Checked,
            "ranges": self.ranges,
        }

        parent = self.parent().parent().parent().parent().parent()

        sig, index = parent.plot.signal_by_uuid(self.uuid)

        min_, max_ = parent.plot.view_boxes[index].viewRange()[1]

        info["min"] = float(min_)
        info["max"] = float(max_)

        return json.dumps(info)

    def does_not_exist(self, exists=False):
        if not exists:
            icon = utils.ERROR_ICON
            if icon is None:
                utils.ERROR_ICON = QtGui.QIcon()
                utils.ERROR_ICON.addPixmap(
                    QtGui.QPixmap(":/error.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off
                )

                utils.NO_ERROR_ICON = QtGui.QIcon()

                icon = utils.ERROR_ICON

            self.color_btn.setIcon(icon)
            self.color_btn.setFlat(True)
            try:
                self.color_btn.clicked.disconnect()
            except:
                pass
        else:
            icon = utils.NO_ERROR_ICON
            if icon is None:
                utils.ERROR_ICON = QtGui.QIcon()
                utils.ERROR_ICON.addPixmap(
                    QtGui.QPixmap(":/error.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off
                )

                utils.NO_ERROR_ICON = QtGui.QIcon()

                icon = utils.NO_ERROR_ICON

            self.color_btn.setIcon(icon)
            self.color_btn.setFlat(False)
            self.color_btn.clicked.connect(self.select_color)

        self.exists = exists

        if self.item:
            tree = self.item.treeWidget()
            if tree:
                tree.update_hidden_states()

    def disconnect_slots(self):
        self.color_changed.disconnect()
        self.enable_changed.disconnect()
        self.ylink_changed.disconnect()
        self.individual_axis_changed.disconnect()

    def get_ranges(self):
        if self.resolved_ranges is None:
            if self.item is None:
                return self.ranges
            else:
                return self.item.get_ranges()
        else:
            return self.resolved_ranges

    def set_ranges(self, ranges):
        if ranges:
            self.range_indicator.setHidden(False)
        else:
            self.range_indicator.setHidden(True)
        self.ranges = ranges class FlexRayBusTrace(TabularBase):
    add_channels_request = QtCore.pyqtSignal(list)

    def __init__(
        self, signals=None, start=0, format="phys", ranges=None, *args, **kwargs
    ):
        ranges = ranges or {name: [] for name in signals.columns}
        if not ranges["Event Type"]:
            ranges["Event Type"] = [
                {
                    "background_color": QtGui.QColor("#ff0000"),
                    "font_color": QtGui.QColor("#000000"),
                    "op1": "==",
                    "op2": "==",
                    "value1": "Error Frame",
                    "value2": None,
                },
                {
                    "background_color": QtGui.QColor("#00ff00"),
                    "font_color": QtGui.QColor("#000000"),
                    "op1": "==",
                    "op2": "==",
                    "value1": "Remote Frame",
                    "value2": None,
                },
            ]

        super().__init__(signals, ranges)

        self.signals_descr = {name: 0 for name in signals.columns}
        self.start = start
        self.pattern = {}
        self.format = format
        self.format_selection.setCurrentText(format)

        self._original_timestamps = signals["timestamps"]
        self._original_ts_series = pd.Series(
            self._original_timestamps,
            index=signals.index,
        )

        prefixes = set()
        for name in signals.columns:
            while "." in name:
                name = name.rsplit(".", 1)[0]
                prefixes.add(f"{name}.")

        self.filters.minimal_menu = True

        self.prefix.insertItems(0, sorted(prefixes, key=lambda x: (-len(x), x)))
        self.prefix.setEnabled(False)

        self.prefix.currentIndexChanged.connect(self.prefix_changed)

        if prefixes:
            self.remove_prefix.setCheckState(QtCore.Qt.Checked)

        self._settings = QtCore.QSettings()
        integer_mode = self._settings.value("tabular_format", "phys")

        self.format_selection.setCurrentText(integer_mode) class GPS(Ui_GPSDisplay, QtWidgets.QWidget):
    timestamp_changed_signal = QtCore.pyqtSignal(object, float)

    def __init__(self, latitude_channel, longitude_channel, zoom=15, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setupUi(self)

        timebase = np.around(
            np.union1d(latitude_channel.timestamps, longitude_channel.timestamps), 9
        )
        self.latitude_signal = latitude_channel.interp(timebase)
        self.longitude_signal = longitude_channel.interp(timebase)
        if len(timebase):
            self.latitude = self.latitude_signal.samples[0]
            self.longitude = self.longitude_signal.samples[0]
        else:
            self.latitude = self.longitude = None

        self._min = self._max = 0

        self._inhibit = False

        if len(timebase):
            self._min = timebase[0]
            self._max = timebase[-1]
        else:
            self._min = float("inf")
            self._max = -float("inf")

        if self._min == float("inf"):
            self._min = self._max = 0

        self._timestamp = self._min

        self.timestamp.setRange(self._min, self._max)
        self.timestamp.setValue(self._min)
        self.min_t.setText(f"{self._min:.6f}s")
        self.max_t.setText(f"{self._max:.6f}s")

        self.mapWidget = MapWidget()
        self.map_layout.insertWidget(0, self.mapWidget)
        self.map_layout.setStretch(0, 1)
        self.map = L.map(self.mapWidget)
        self.map.setView([50.1364092, 8.5991296], zoom)

        L.tileLayer("https://{s}.tile.openstreetmap.fr/osmfr/{z}/{x}/{y}.png").addTo(
            self.map
        )

        if len(timebase):
            line = L.polyline(
                np.column_stack(
                    [self.latitude_signal.samples, self.longitude_signal.samples]
                ).tolist()
            )
            line.addTo(self.map)

            self.map.setView([self.latitude, self.longitude], zoom)
            self.marker = L.marker([self.latitude, self.longitude])
            self.map.addLayer(self.marker)
        else:
            self.marker = None

        self.timestamp.valueChanged.connect(self._timestamp_changed)
        self.timestamp_slider.valueChanged.connect(self._timestamp_slider_changed)
        self.set_timestamp()

        self.show()

    def _timestamp_changed(self, stamp):
        if not self._inhibit:
            self.set_timestamp(stamp)

    def _timestamp_slider_changed(self, stamp):
        if not self._inhibit:
            factor = stamp / 99999
            stamp = (self._max - self._min) * factor + self._min
            self.set_timestamp(stamp)

    def set_timestamp(self, stamp=None):
        if stamp is None:
            stamp = self._timestamp

        if not (self._min <= stamp <= self._max):
            return

        try:
            self.latitude = self.latitude_signal.cut(stamp, stamp).samples[0]
            self.longitude = self.longitude_signal.cut(stamp, stamp).samples[0]
        except:
            return
        if self.marker is not None:
            self.marker.setLatLng([self.latitude, self.longitude])
            app = QtWidgets.QApplication.instance()
            app.processEvents()

        self._inhibit = True
        if self._min != self._max:
            val = int((stamp - self._min) / (self._max - self._min) * 99999)
            self.timestamp_slider.setValue(val)
        self.timestamp.setValue(stamp)
        self._inhibit = False
        self.timestamp_changed_signal.emit(self, stamp)

    def to_config(self):

        config = {
            "latitude_channel": self.latitude_signal.name,
            "longitude_channel": self.longitude_signal.name,
            "zoom": 5,
        }

        return config class LINBusTrace(TabularBase):
    add_channels_request = QtCore.pyqtSignal(list)

    def __init__(
        self, signals=None, start=0, format="phys", ranges=None, *args, **kwargs
    ):
        ranges = ranges or {name: [] for name in signals.columns}
        if not ranges["Event Type"]:
            ranges["Event Type"] = [
                {
                    "background_color": QtGui.QColor("#ff0000"),
                    "font_color": QtGui.QColor("#000000"),
                    "op1": "!=",
                    "op2": "==",
                    "value1": "LIN Frame",
                    "value2": None,
                },
            ]

        super().__init__(signals, ranges)

        self.signals_descr = {name: 0 for name in signals.columns}
        self.start = start
        self.pattern = {}
        self.format = format
        self.format_selection.setCurrentText(format)

        self._original_timestamps = signals["timestamps"]
        self._original_ts_series = pd.Series(
            self._original_timestamps,
            index=signals.index,
        )

        prefixes = set()
        for name in signals.columns:
            while "." in name:
                name = name.rsplit(".", 1)[0]
                prefixes.add(f"{name}.")

        self.filters.minimal_menu = True

        self.prefix.insertItems(0, sorted(prefixes, key=lambda x: (-len(x), x)))
        self.prefix.setEnabled(False)

        self.prefix.currentIndexChanged.connect(self.prefix_changed)

        if prefixes:
            self.remove_prefix.setCheckState(QtCore.Qt.Checked)

        self._settings = QtCore.QSettings()
        integer_mode = self._settings.value("tabular_format", "phys")

        self.format_selection.setCurrentText(integer_mode) class Numeric(QtWidgets.QWidget):
    add_channels_request = QtCore.pyqtSignal(list)
    timestamp_changed_signal = QtCore.pyqtSignal(object, float)

    def __init__(
        self,
        channels=None,
        format=None,
        mode="offline",
        float_precision=None,
        *args,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)

        if mode == "offline":
            uic.loadUi(HERE.joinpath("..", "ui", "numeric_offline.ui"), self)
        else:
            uic.loadUi(HERE.joinpath("..", "ui", "numeric_online.ui"), self)

        self.mode = mode

        self._settings = QtCore.QSettings()

        if mode == "offline":
            backend = OfflineBackEnd(None, self)
        else:
            backend = OnlineBackEnd(None, self)
        self.channels = NumericViewer(backend)

        self.channels.dataView.ranges = {}

        self.verticalLayout.insertWidget(1, self.channels)
        self.verticalLayout.setStretch(1, 1)

        self.float_precision.addItems(
            ["Full float precision"] + [f"{i} float decimals" for i in range(16)]
        )
        self.float_precision.setCurrentIndex(-1)
        self.format_selection.setCurrentIndex(-1)

        self.format_selection.currentTextChanged.connect(self.set_format)
        self.float_precision.currentIndexChanged.connect(self.set_float_precision)

        format = format or self._settings.value("numeric_format", "Physical")
        if format not in ("Physical", "Hex", "Binary"):
            format = "Physical"
            self._settings.setValue("numeric_format", format)

        self.format_selection.setCurrentText(format)

        if float_precision is None:
            float_precision = self._settings.value("numeric_float_precision", -1)
        self.float_precision.setCurrentIndex(float_precision + 1)

        if channels:
            self.add_new_channels(channels)

        self.channels.dataView.add_channels_request.connect(self.add_channels_request)

        self.channels.auto_size_header()
        self.double_clicked_enabled = True

        if self.mode == "offline":
            self.pattern = {}

            self.timestamp.valueChanged.connect(self._timestamp_changed)
            self.timestamp_slider.valueChanged.connect(self._timestamp_slider_changed)

            self._inhibit = False

            self.forward.clicked.connect(self.search_forward)
            self.backward.clicked.connect(self.search_backward)
            self.op.addItems([">", ">=", "<", "<=", "==", "!="])

    def add_new_channels(self, channels, mime_data=None):

        if self.mode == "online":
            others = []
            for sig in channels:
                if sig is not None:
                    entry = (sig.mdf_uuid, sig.name)

                    others.append(
                        SignalOnline(
                            name=sig.name,
                            conversion=sig.conversion,
                            entry=entry,
                            unit=sig.unit,
                        )
                    )

                    self.channels.dataView.ranges[entry] = []

        else:
            others = []
            for sig in channels:
                if sig is not None:
                    sig.computed = False
                    sig.computation = None
                    sig = PlotSignal(sig)
                    sig.entry = sig.group_index, sig.channel_index

                    others.append(
                        SignalOffline(
                            signal=sig,
                        )
                    )

                    self.channels.dataView.ranges[sig.entry] = []

        self.channels.backend.update(others)

        if self.mode == "offline":

            numeric = self
            numeric._min = float("inf")
            numeric._max = -float("inf")

            for sig in self.channels.backend.signals:
                timestamps = sig.signal.timestamps
                if timestamps.size:
                    numeric._min = min(numeric._min, timestamps[0])
                    numeric._max = max(numeric._max, timestamps[-1])

            if numeric._min == float("inf"):
                numeric._min = numeric._max = 0

            numeric._timestamp = numeric._min

            numeric.timestamp.setRange(numeric._min, numeric._max)
            numeric.min_t.setText(f"{numeric._min:.9f}s")
            numeric.max_t.setText(f"{numeric._max:.9f}s")
            numeric.set_timestamp(numeric._min)

        self.channels.auto_size_header()

    def reset(self):
        self.channels.backend.reset()
        self.channels.dataView.double_clicked_enabled = True

    def set_values(self, values=None):
        self.channels.backend.set_values(values)

    def to_config(self):

        if self.mode == "online":

            channels = []
            for signal in self.channels.backend.signals:
                ranges = self.channels.dataView.ranges[signal.entry]
                ranges = copy_ranges(ranges)

                for range_info in ranges:
                    range_info["font_color"] = range_info["font_color"].color().name()
                    range_info["background_color"] = (
                        range_info["background_color"].color().name()
                    )

                channels.append(
                    {
                        "mdf_uuid": str(signal.entry[0]),
                        "name": signal.name,
                        "ranges": ranges,
                    }
                )

            config = {
                "format": self.format_selection.currentText(),
                "mode": "",
                "channels": channels,
                "float_precision": self.float_precision.currentIndex() - 1,
                "header_sections_width": [
                    self.channels.columnHeader.horizontalHeader().sectionSize(i)
                    for i in range(
                        self.channels.columnHeader.horizontalHeader().count()
                    )
                ],
            }

        else:
            channels = {}
            for signal in self.channels.backend.signals:
                ranges = self.channels.dataView.ranges[signal.entry]
                ranges = copy_ranges(ranges)

                for range_info in ranges:
                    range_info["font_color"] = range_info["font_color"].color().name()
                    range_info["background_color"] = (
                        range_info["background_color"].color().name()
                    )

                channels[signal.name] = ranges

            pattern = self.pattern
            if pattern:
                ranges = copy_ranges(pattern["ranges"])

                for range_info in ranges:
                    range_info["font_color"] = range_info["font_color"].color().name()
                    range_info["background_color"] = (
                        range_info["background_color"].color().name()
                    )

                pattern["ranges"] = ranges

            config = {
                "format": self.format_selection.currentText(),
                "mode": self.mode,
                "channels": list(channels) if not self.pattern else [],
                "ranges": list(channels.values()) if not self.pattern else [],
                "pattern": pattern,
                "float_precision": self.float_precision.currentIndex() - 1,
                "header_sections_width": [
                    self.channels.columnHeader.horizontalHeader().sectionSize(i)
                    for i in range(
                        self.channels.columnHeader.horizontalHeader().count()
                    )
                ],
            }

        return config

    def does_not_exist(self, entry, exists=False):
        self.channels.backend.does_not_exist(entry, exists)

    def set_format(self, fmt):
        if fmt == "Phys":
            fmt = "Physical"
        elif fmt == "Binary":
            fmt = "Bin"
        if fmt not in ("Physical", "Hex", "Bin"):
            fmt = "Physical"

        self.channels.dataView.model().format = fmt
        self._settings.setValue("numeric_format", fmt)
        self.channels.backend.data_changed()

    def set_float_precision(self, index):
        self._settings.setValue("numeric_float_precision", index - 1)
        self.channels.dataView.model().float_precision = index - 1
        self.channels.backend.data_changed()

    def visible_entries(self):
        visible = set()

        if self.channels.backend.sorted_column_index in (1, 2):
            visible = set(self.channels.backend.map)

        else:
            rect = self.channels.dataView.viewport().rect()

            top = self.channels.dataView.indexAt(rect.topLeft()).row()
            bottom = self.channels.dataView.indexAt(rect.bottomLeft()).row()
            if top == -1:
                pass
            elif bottom == -1:
                visible = set(self.channels.backend.map)
            else:
                for row in range(top, bottom + 1):
                    visible.add(self.channels.backend.signals[row].entry)

        return visible

    def _timestamp_changed(self, stamp):
        if not self._inhibit:
            self.set_timestamp(stamp)

    def _timestamp_slider_changed(self, stamp):
        if not self._inhibit:
            factor = stamp / 99999
            stamp = (self._max - self._min) * factor + self._min
            actual_stamp = self.channels.backend.get_timestamp(stamp)
            self.set_timestamp(actual_stamp)

    def set_timestamp(self, stamp=None):
        if stamp is None:
            stamp = self._timestamp

        if not (self._min <= stamp <= self._max):
            return

        self.channels.backend.set_timestamp(stamp)

        self._inhibit = True
        if self._min != self._max:
            val = int((stamp - self._min) / (self._max - self._min) * 99999)
            self.timestamp_slider.setValue(val)
        self.timestamp.setValue(stamp)
        self._inhibit = False
        self.timestamp_changed_signal.emit(self, stamp)

    def search_forward(self):
        if (
            self.op.currentIndex() < 0
            or not self.target.text().strip()
            or not self.pattern_match.text().strip()
        ):
            self.match.setText("invalid input values")
            return

        operator = self.op.currentText()

        pattern = self.pattern_match.text().strip().replace("*", "_WILDCARD_")
        pattern = re.escape(pattern)
        pattern = pattern.replace("_WILDCARD_", ".*")

        pattern = re.compile(f"(?i){pattern}")
        matches = [
            sig for sig in self.channels.backend.signals if pattern.fullmatch(sig.name)
        ]

        mode = self.match_mode.currentText()

        if not matches:
            self.match.setText("the pattern does not match any channel name")
            return

        try:
            target = float(self.target.text().strip())
        except:
            self.match.setText("the target must a numeric value")
        else:

            if target.is_integer():
                target = int(target)

            start = self.timestamp.value()

            timestamp = None
            signal_name = ""
            for sig in matches:
                sig = sig.signal.cut(start=start)
                if mode == "Raw" or sig.conversion is None:
                    samples = sig.raw_samples
                else:
                    samples = sig.phys_samples

                op = getattr(samples, OPS[operator])
                try:
                    idx = np.argwhere(op(target)).flatten()
                    if len(idx):
                        if len(idx) == 1 or sig.timestamps[idx[0]] != start:
                            timestamp_ = sig.timestamps[idx[0]]
                        else:
                            timestamp_ = sig.timestamps[idx[1]]

                        if timestamp is None or timestamp_ < timestamp:
                            timestamp = timestamp_
                            signal_name = sig.name
                except:
                    continue

            if timestamp is not None:
                self.timestamp.setValue(timestamp)
                self.match.setText(f"condition found for {signal_name}")
            else:
                self.match.setText("condition not found")

    def search_backward(self):
        if (
            self.op.currentIndex() < 0
            or not self.target.text().strip()
            or not self.pattern_match.text().strip()
        ):
            self.match.setText("invalid input values")
            return

        operator = self.op.currentText()

        pattern = self.pattern_match.text().strip().replace("*", "_WILDCARD_")
        pattern = re.escape(pattern)
        pattern = pattern.replace("_WILDCARD_", ".*")

        pattern = re.compile(f"(?i){pattern}")
        matches = [
            sig for sig in self.channels.backend.signals if pattern.fullmatch(sig.name)
        ]

        mode = self.match_mode.currentText()

        if not matches:
            self.match.setText("the pattern does not match any channel name")
            return

        try:
            target = float(self.target.text().strip())
        except:
            self.match.setText(f"the target must a numeric value")
        else:

            if target.is_integer():
                target = int(target)

            stop = self.timestamp.value()

            timestamp = None
            signal_name = ""
            for sig in matches:
                sig = sig.signal.cut(stop=stop)
                if mode == "raw values" or sig.conversion is None:
                    samples = sig.raw_samples[:-1]
                else:
                    samples = sig.phys_samples[:-1]

                op = getattr(samples, OPS[operator])
                try:
                    idx = np.argwhere(op(target)).flatten()
                    if len(idx):

                        if len(idx) == 1 or sig.timestamps[idx[-1]] != stop:
                            timestamp_ = sig.timestamps[idx[-1]]
                        else:
                            timestamp_ = sig.timestamps[idx[-2]]

                        if timestamp is None or timestamp_ > timestamp:
                            timestamp = timestamp_
                            signal_name = sig.name
                except:
                    continue

            if timestamp is not None:
                self.timestamp.setValue(timestamp)
                self.match.setText(f"condition found for {signal_name}")
            else:
                self.match.setText(f"condition not found")

    def keyPressEvent(self, event):
        key = event.key()
        modifier = event.modifiers()

        if (
            key in (QtCore.Qt.Key_H, QtCore.Qt.Key_B, QtCore.Qt.Key_P)
            and modifier == QtCore.Qt.ControlModifier
        ):
            if key == QtCore.Qt.Key_H:
                self.format_selection.setCurrentText("Hex")
            elif key == QtCore.Qt.Key_B:
                self.format_selection.setCurrentText("Binary")
            else:
                self.format_selection.setCurrentText("Physical")
            event.accept()
        elif (
            key == QtCore.Qt.Key_Right
            and modifier == QtCore.Qt.NoModifier
            and self.mode == "offline"
        ):
            self.timestamp_slider.setValue(self.timestamp_slider.value() + 1)

        elif (
            key == QtCore.Qt.Key_Left
            and modifier == QtCore.Qt.NoModifier
            and self.mode == "offline"
        ):
            self.timestamp_slider.setValue(self.timestamp_slider.value() - 1)

        elif (
            key == QtCore.Qt.Key_S
            and modifier == QtCore.Qt.ControlModifier
            and self.mode == "offline"
        ):
            file_name, _ = QtWidgets.QFileDialog.getSaveFileName(
                self,
                "Select output measurement file",
                "",
                "MDF version 4 files (*.mf4)",
            )

            if file_name:
                signals = [signal for signal in self.signals if signal.enable]
                if signals:
                    with MDF() as mdf:
                        groups = {}
                        for sig in signals:
                            id_ = id(sig.timestamps)
                            group_ = groups.setdefault(id_, [])
                            group_.append(sig)

                        for signals in groups.values():
                            sigs = []
                            for signal in signals:
                                if ":" in signal.name:
                                    sig = signal.copy()
                                    sig.name = sig.name.split(":")[-1].strip()
                                    sigs.append(sig)
                                else:
                                    sigs.append(signal)
                            mdf.append(sigs, common_timebase=True)
                        mdf.save(file_name, overwrite=True)
        else:
            self.channels.dataView.keyPressEvent(event)
 class Plot(QtWidgets.QWidget):

    add_channels_request = QtCore.pyqtSignal(list)
    close_request = QtCore.pyqtSignal()
    clicked = QtCore.pyqtSignal()
    cursor_moved_signal = QtCore.pyqtSignal(object, float)
    cursor_removed_signal = QtCore.pyqtSignal(object)
    region_moved_signal = QtCore.pyqtSignal(object, list)
    region_removed_signal = QtCore.pyqtSignal(object)
    show_properties = QtCore.pyqtSignal(list)
    splitter_moved = QtCore.pyqtSignal(object, int)
    pattern_group_added = QtCore.pyqtSignal(object, object)

    def __init__(
        self,
        signals,
        with_dots=False,
        origin=None,
        mdf=None,
        line_interconnect="line",
        hide_missing_channels=False,
        hide_disabled_channels=False,
        x_axis="time",
        *args,
        **kwargs,
    ):
        events = kwargs.pop("events", None)
        super().__init__(*args, **kwargs)
        self.line_interconnect = line_interconnect
        self.setContentsMargins(0, 0, 0, 0)
        self.pattern = {}
        self.mdf = mdf

        self.x_name = "t" if x_axis == "time" else "f"
        self.x_unit = "s" if x_axis == "time" else "Hz"

        self.info_uuid = None

        self._range_start = None
        self._range_stop = None

        self._can_switch_mode = True

        self._accept_cursor_update = False
        self._last_cursor_update = perf_counter()

        main_layout = QtWidgets.QVBoxLayout(self)
        main_layout.setSpacing(1)
        main_layout.setContentsMargins(1, 1, 1, 1)

        vbox = QtWidgets.QVBoxLayout()
        vbox.setSpacing(1)
        vbox.setContentsMargins(1, 1, 1, 1)
        widget = QtWidgets.QWidget()
        self.channel_selection = ChannelsTreeWidget(
            hide_missing_channels=hide_missing_channels,
            hide_disabled_channels=hide_disabled_channels,
            parent=self,
        )

        widget.setLayout(vbox)

        self.splitter = QtWidgets.QSplitter()
        self.splitter.addWidget(widget)
        self.splitter.setOpaqueResize(False)

        self.plot = _Plot(
            with_dots=with_dots,
            line_interconnect=self.line_interconnect,
            parent=self,
            events=events,
            origin=origin,
            mdf=self.mdf,
            x_axis=x_axis,
        )

        self.cursor_info = CursorInfo(
            precision=QtCore.QSettings().value("plot_cursor_precision", 6),
            unit=self.x_unit,
            name=self.x_name,
            plot=self.plot,
        )

        vbox.addWidget(self.channel_selection)
        vbox.addWidget(self.cursor_info)

        self.plot.range_modified.connect(self.range_modified)
        self.plot.range_removed.connect(self.range_removed)
        self.plot.range_modified_finished.connect(self.range_modified_finished)
        self.plot.cursor_removed.connect(self.cursor_removed)
        self.plot.cursor_moved.connect(self.cursor_moved)
        self.plot.cursor_move_finished.connect(self.cursor_move_finished)
        self.plot.xrange_changed.connect(self.xrange_changed)
        self.plot.computation_channel_inserted.connect(
            self.computation_channel_inserted
        )
        self.plot.curve_clicked.connect(self.curve_clicked)

        self.splitter.addWidget(self.plot)

        self.info = ChannelStats(self.x_unit)
        self.info.hide()
        self.splitter.addWidget(self.info)

        self.splitter.setStretchFactor(0, 0)
        self.splitter.setStretchFactor(1, 1)
        self.splitter.setStretchFactor(2, 0)

        self.plot.add_channels_request.connect(self.add_channels_request)
        self.setAcceptDrops(True)

        main_layout.addWidget(self.splitter)

        self.show()
        self.hide()

        if signals:
            self.add_new_channels(signals)

        self.channel_selection.itemsDeleted.connect(self.channel_selection_reduced)
        self.channel_selection.itemPressed.connect(self.channel_selection_modified)
        self.channel_selection.currentItemChanged.connect(
            self.channel_selection_row_changed
        )
        self.channel_selection.add_channels_request.connect(self.add_channels_request)
        self.channel_selection.set_time_offset.connect(self.plot.set_time_offset)
        self.channel_selection.show_properties.connect(self._show_properties)
        self.channel_selection.insert_computation.connect(self.plot.insert_computation)

        self.channel_selection.itemChanged.connect(self.channel_selection_item_changed)
        self.channel_selection.items_rearranged.connect(
            self.channel_selection_rearranged
        )
        self.channel_selection.pattern_group_added.connect(self.pattern_group_added_req)
        self.channel_selection.itemDoubleClicked.connect(
            self.channel_selection_item_double_clicked
        )
        self.channel_selection.compute_fft_request.connect(self.compute_fft)
        self.channel_selection.itemExpanded.connect(self.update_current_values)
        self.channel_selection.verticalScrollBar().valueChanged.connect(
            self.update_current_values
        )

        self.keyboard_events = (
            set(
                [
                    (QtCore.Qt.Key_M, QtCore.Qt.NoModifier),
                    (QtCore.Qt.Key_B, QtCore.Qt.ControlModifier),
                    (QtCore.Qt.Key_H, QtCore.Qt.ControlModifier),
                    (QtCore.Qt.Key_P, QtCore.Qt.ControlModifier),
                ]
            )
            | self.plot.keyboard_events
        )

        self.splitter.splitterMoved.connect(self.set_splitter)

        self.show()

    def curve_clicked(self, uuid):
        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            widget = self.channel_selection.itemWidget(item, 1)
            if isinstance(widget, ChannelDisplay) and widget.uuid == uuid:
                self.channel_selection.setCurrentItem(item)
                break

            iterator += 1

    def channel_selection_rearranged(self, uuids):
        uuids = set(uuids)

        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            widget = self.channel_selection.itemWidget(item, 1)
            if isinstance(widget, ChannelDisplay) and widget.uuid in uuids:
                widget.color_changed.connect(self.plot.set_color)
                widget.enable_changed.connect(self.plot.set_signal_enable)
                widget.ylink_changed.connect(self.plot.set_common_axis)
                widget.individual_axis_changed.connect(self.plot.set_individual_axis)

            iterator += 1

    def channel_selection_item_changed(self, item, column):
        if item is not None and column == 0:
            state = item.checkState(0)
            widget = self.channel_selection.itemWidget(item, 1)
            if isinstance(widget, ChannelDisplay):
                widget.enable_changed.emit(widget.uuid, state)

    def channel_selection_item_double_clicked(self, item, column):
        if isinstance(item, ChannelsGroupTreeItem):
            item.show_info()

    def mousePressEvent(self, event):
        self.clicked.emit()
        super().mousePressEvent(event)

    def channel_selection_modified(self, item):
        if isinstance(item, ChannelsTreeItem):
            uuid = self.channel_selection.itemWidget(item, 1).uuid
            self.info_uuid = uuid

            sig, index = self.plot.signal_by_uuid(uuid)
            if sig.enable:

                self.plot.curves[index].hide()
                self.plot.update_signal_curve(sig, index)
                for i in range(3):
                    QtWidgets.QApplication.processEvents()
                    sleep(0.01)
                self.plot.curves[index].show()
                self.plot.update_signal_curve(sig, index)

                self.plot.set_current_uuid(self.info_uuid)
                if self.info.isVisible():
                    stats = self.plot.get_stats(self.info_uuid)
                    self.info.set_stats(stats)

    def channel_selection_row_changed(self, current, previous):
        if isinstance(current, ChannelsTreeItem):
            item = current
            uuid = self.channel_selection.itemWidget(item, 1).uuid
            self.info_uuid = uuid

            sig, index = self.plot.signal_by_uuid(uuid)
            if sig.enable:

                self.plot.curves[index].hide()
                self.plot.update_signal_curve(sig, index)
                for i in range(3):
                    QtWidgets.QApplication.processEvents()
                    sleep(0.01)
                self.plot.curves[index].show()
                self.plot.update_signal_curve(sig, index)

                self.plot.set_current_uuid(self.info_uuid)
                if self.info.isVisible():
                    stats = self.plot.get_stats(self.info_uuid)
                    self.info.set_stats(stats)

    def channel_selection_reduced(self, deleted):

        self.plot.delete_channels(deleted)

        if self.info_uuid in deleted:
            self.info_uuid = None
            self.info.hide()

        count = 0
        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            count += 1
            iterator += 1

        if not count:
            self.close_request.emit()

    def cursor_move_finished(self):
        x = self.plot.get_current_timebase()
        if x.size:
            dim = len(x)
            position = self.plot.cursor1.value()

            right = np.searchsorted(x, position, side="right")
            if right == 0:
                next_pos = x[0]
            elif right == dim:
                next_pos = x[-1]
            else:
                if position - x[right - 1] < x[right] - position:
                    next_pos = x[right - 1]
                else:
                    next_pos = x[right]
            self._accept_cursor_update = True
            self.plot.cursor1.setPos(next_pos)

        # self.plot.cursor_hint.setData(x=[], y=[])
        self.plot.cursor_hint.hide()

    def cursor_moved(self):
        if self.plot.cursor1 is None or (
            perf_counter() - self._last_cursor_update < 0.030
            and not self._accept_cursor_update
        ):
            return

        position = self.plot.cursor1.value()

        if not self.plot.region:

            self.cursor_info.update_value()

            iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
            while iterator.value():
                item = iterator.value()
                iterator += 1
                if not self.channel_selection.is_item_visible(item):
                    continue

                widget = self.channel_selection.itemWidget(item, 1)
                if isinstance(widget, ChannelDisplay):

                    signal, idx = self.plot.signal_by_uuid(widget.uuid)
                    value, kind, fmt = signal.value_at_timestamp(position)

                    widget.set_prefix("= ")
                    widget.kind = kind
                    widget.set_fmt(fmt)

                    widget.set_value(value, update=True)

        if self.info.isVisible():
            stats = self.plot.get_stats(self.info_uuid)
            self.info.set_stats(stats)

        self._last_cursor_update = perf_counter()
        self._accept_cursor_update = False
        self.cursor_moved_signal.emit(self, position)

    def cursor_removed(self):

        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            widget = self.channel_selection.itemWidget(item, 1)

            if isinstance(widget, ChannelDisplay) and not self.plot.region:
                self.cursor_info.update_value()
                widget.set_prefix("")
                widget.set_value("")

            iterator += 1

        if self.info.isVisible():
            stats = self.plot.get_stats(self.info_uuid)
            self.info.set_stats(stats)

        self.cursor_removed_signal.emit(self)

    def range_modified(self):
        if self.plot.region is None or (
            perf_counter() - self._last_cursor_update < 0.030
            and not self._accept_cursor_update
        ):
            return

        start, stop = self.plot.region.getRegion()

        self.cursor_info.update_value()

        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            iterator += 1

            if not self.channel_selection.is_item_visible(item):
                continue

            widget = self.channel_selection.itemWidget(item, 1)
            if isinstance(widget, ChannelDisplay):

                signal, i = self.plot.signal_by_uuid(widget.uuid)

                start_v, kind, fmt = signal.value_at_timestamp(start)
                stop_v, kind, fmt = signal.value_at_timestamp(stop)

                widget.set_prefix("Δ = ")
                widget.set_fmt(signal.format)

                if "n.a." not in (start_v, stop_v):
                    if kind in "ui":
                        delta = np.int64(stop_v) - np.int64(start_v)
                        widget.kind = kind
                        widget.set_value(delta)
                        widget.set_fmt(fmt)
                    elif kind == "f":
                        delta = stop_v - start_v
                        widget.kind = kind
                        widget.set_value(delta)
                        widget.set_fmt(fmt)
                    else:
                        widget.set_value("n.a.")
                else:
                    widget.set_value("n.a.")

        if self.info.isVisible():
            stats = self.plot.get_stats(self.info_uuid)
            self.info.set_stats(stats)

        self._last_cursor_update = perf_counter()
        self._accept_cursor_update = False

        self.region_moved_signal.emit(self, [start, stop])

    def xrange_changed(self):
        if self.info.isVisible():
            stats = self.plot.get_stats(self.info_uuid)
            self.info.set_stats(stats)

    def range_modified_finished(self):
        if not self.plot.region:
            return
        start, stop = self.plot.region.getRegion()

        timebase = self.plot.get_current_timebase()

        if timebase.size:
            timebase = self.plot.timebase
            dim = len(timebase)

            right = np.searchsorted(timebase, start, side="right")
            if right == 0:
                next_pos = timebase[0]
            elif right == dim:
                next_pos = timebase[-1]
            else:
                if start - timebase[right - 1] < timebase[right] - start:
                    next_pos = timebase[right - 1]
                else:
                    next_pos = timebase[right]
            start = next_pos

            right = np.searchsorted(timebase, stop, side="right")
            if right == 0:
                next_pos = timebase[0]
            elif right == dim:
                next_pos = timebase[-1]
            else:
                if stop - timebase[right - 1] < timebase[right] - stop:
                    next_pos = timebase[right - 1]
                else:
                    next_pos = timebase[right]
            stop = next_pos

            self._accept_cursor_update = True
            self.plot.region.setRegion((start, stop))

    def keyPressEvent(self, event):
        key = event.key()
        modifiers = event.modifiers()

        if key == QtCore.Qt.Key_M and modifiers == QtCore.Qt.NoModifier:
            ch_size, plt_size, info_size = self.splitter.sizes()

            if self.info.isVisible():
                self.info.hide()
                self.splitter.setSizes((ch_size, plt_size + info_size, 0))

            else:

                self.info.show()
                self.splitter.setSizes(
                    (
                        ch_size,
                        int(0.8 * (plt_size + info_size)),
                        int(0.2 * (plt_size + info_size)),
                    )
                )

        elif (
            key in (QtCore.Qt.Key_B, QtCore.Qt.Key_H, QtCore.Qt.Key_P)
            and modifiers == QtCore.Qt.ControlModifier
        ):
            selected_items = self.channel_selection.selectedItems()
            if not selected_items:
                signals = [(sig, i) for i, sig in enumerate(self.plot.signals)]

            else:
                uuids = [
                    self.channel_selection.itemWidget(item, 1).uuid
                    for item in selected_items
                ]

                signals = [self.plot.signal_by_uuid(uuid) for uuid in uuids]

            if key == QtCore.Qt.Key_B:
                fmt = "bin"
            elif key == QtCore.Qt.Key_H:
                fmt = "hex"
            else:
                fmt = "phys"

            for signal, idx in signals:
                if signal.plot_samples.dtype.kind in "ui":
                    signal.format = fmt

                    value, kind, fmt = signal.value_at_timestamp(0)

                    widget = self.widget_by_uuid(signal.uuid)
                    widget.kind = kind
                    widget.set_fmt(fmt)
                    widget.set_value(update=True)

                    if self.plot.current_uuid == signal.uuid:
                        self.plot.y_axis.format = fmt
                        self.plot.y_axis.picture = None
                        self.plot.y_axis.update()

                    axis = self.plot.axes[idx]
                    axis.format = fmt
                    axis.picture = None
                    axis.update()

            if self.plot.cursor1:
                self.plot.cursor_moved.emit()

        elif (
            key in (QtCore.Qt.Key_R, QtCore.Qt.Key_S)
            and modifiers == QtCore.Qt.AltModifier
            and self._can_switch_mode
        ):
            selected_items = self.channel_selection.selectedItems()
            if not selected_items:
                signals = [(sig, i) for i, sig in enumerate(self.plot.signals)]

            else:
                uuids = [
                    self.channel_selection.itemWidget(item, 1).uuid
                    for item in selected_items
                ]

                signals = [self.plot.signal_by_uuid(uuid) for uuid in uuids]

            if key == QtCore.Qt.Key_R:
                mode = "raw"
                style = QtCore.Qt.DashLine

            else:
                mode = "phys"
                style = QtCore.Qt.SolidLine

            for signal, idx in signals:
                if signal.mode != mode:
                    signal.pen = pg.mkPen(color=signal.color, style=style)

                    view = self.plot.view_boxes[idx]
                    _, (buttom, top) = view.viewRange()

                    try:
                        min_, max_ = float(signal.min), float(signal.max)
                    except:
                        min_, max_ = 0, 1

                    if max_ != min_ and top != buttom:

                        factor = (top - buttom) / (max_ - min_)
                        offset = (buttom - min_) / (top - buttom)
                    else:
                        factor = 1
                        offset = 0

                    signal.mode = mode

                    try:
                        min_, max_ = float(signal.min), float(signal.max)
                    except:
                        min_, max_ = 0, 1

                    if max_ != min_:

                        delta = (max_ - min_) * factor
                        buttom = min_ + offset * delta
                        top = buttom + delta
                    else:
                        buttom, top = max_ - 1, max_ + 1

                    view.setYRange(buttom, top, padding=0, update=True)

                    if self.plot.current_uuid == signal.uuid:
                        self.plot.y_axis.mode = mode
                        self.plot.y_axis.hide()
                        self.plot.y_axis.show()

            self.plot.update_lines()

            if self.plot.cursor1:
                self.plot.cursor_moved.emit()

        elif key == QtCore.Qt.Key_I and modifiers == QtCore.Qt.ControlModifier:
            if self.plot.cursor1:
                position = self.plot.cursor1.value()
                comment, _ = QtWidgets.QInputDialog.getMultiLineText(
                    self,
                    "Insert comments",
                    f"Enter the comments for cursor position {position:.9f}s:",
                    "",
                )
                line = pg.InfiniteLine(
                    pos=position,
                    label=f"t = {position}s\n\n{comment}",
                    pen={"color": "#FF0000", "width": 2, "style": QtCore.Qt.DashLine},
                    labelOpts={
                        "border": {
                            "color": "#FF0000",
                            "width": 2,
                            "style": QtCore.Qt.DashLine,
                        },
                        "fill": "ff9b37",
                        "color": "#000000",
                        "movable": True,
                    },
                )
                self.plot.plotItem.addItem(line, ignoreBounds=True)

        elif key == QtCore.Qt.Key_I and modifiers == QtCore.Qt.AltModifier:
            visible = None
            for item in self.plot.plotItem.items:
                if not isinstance(item, pg.InfiniteLine) or item is self.plot.cursor1:
                    continue

                if visible is None:
                    visible = item.label.isVisible()

                try:
                    if visible:
                        item.label.setVisible(False)
                    else:
                        item.label.setVisible(True)
                except:
                    pass

        elif key == QtCore.Qt.Key_R and modifiers == QtCore.Qt.NoModifier:
            iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
            while iterator.value():
                item = iterator.value()
                widget = self.channel_selection.itemWidget(item, 1)
                if isinstance(widget, ChannelDisplay):
                    widget.set_prefix("")
                    widget.set_value("")

                iterator += 1

            self.plot.keyPressEvent(event)

        elif (key, modifiers) in self.plot.keyboard_events:
            self.plot.keyPressEvent(event)

        else:
            event.ignore()

    def range_removed(self):
        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            widget = self.channel_selection.itemWidget(item, 1)
            if isinstance(widget, ChannelDisplay):
                widget.set_prefix("")
                widget.set_value("")

            iterator += 1

        self._range_start = None
        self._range_stop = None

        self.cursor_info.update_value()

        if self.plot.cursor1:
            self.plot.cursor_moved.emit()
        if self.info.isVisible():
            stats = self.plot.get_stats(self.info_uuid)
            self.info.set_stats(stats)

        self.region_removed_signal.emit(self)

    def computation_channel_inserted(self):
        sig = self.plot.signals[-1]

        name, unit = sig.name, sig.unit
        item = ChannelsTreeItem((-1, -1), name, sig.computation)
        tooltip = getattr(sig, "tooltip", "")
        it = ChannelDisplay(
            sig.uuid,
            unit,
            sig.samples.dtype.kind,
            3,
            tooltip,
            "",
            item=item,
            parent=self,
        )
        it.setAttribute(QtCore.Qt.WA_StyledBackground)

        font = QtGui.QFont()
        font.setItalic(True)
        it.name.setFont(font)

        it.set_name(name)
        it.set_value("")
        it.set_color(sig.color)
        item.setSizeHint(1, it.sizeHint())
        self.channel_selection.addTopLevelItem(item)
        self.channel_selection.setItemWidget(item, 1, it)

        it.color_changed.connect(self.plot.set_color)
        it.unit_changed.connect(self.plot.set_unit)
        it.name_changed.connect(self.plot.set_name)
        it.enable_changed.connect(self.plot.set_signal_enable)
        it.ylink_changed.connect(self.plot.set_common_axis)
        it.individual_axis_changed.connect(self.plot.set_individual_axis)
        it.unit_changed.connect(self.plot.set_unit)

        it.enable_changed.emit(sig.uuid, 1)
        it.enable_changed.emit(sig.uuid, 0)
        it.enable_changed.emit(sig.uuid, 1)

        self.info_uuid = sig.uuid

        self.plot.set_current_uuid(self.info_uuid, True)

    def add_new_channels(self, channels, mime_data=None, destination=None):
        def add_new_items(tree, root, items, items_pool):
            for (name, group_index, channel_index, mdf_uuid, type_, ranges) in items:

                ranges = deepcopy(ranges)
                for range_info in ranges:
                    range_info["font_color"] = QtGui.QColor(range_info["font_color"])
                    range_info["background_color"] = QtGui.QColor(
                        range_info["background_color"]
                    )

                if type_ == "group":
                    pattern = group_index
                    item = ChannelsGroupTreeItem(name, pattern)
                    widget = ChannelGroupDisplay(
                        name, pattern, item=item, ranges=ranges
                    )
                    widget.item = item
                    root.addChild(item)
                    tree.setItemWidget(item, 1, widget)

                    add_new_items(tree, item, channel_index, items_pool)

                else:

                    key = (name, group_index, channel_index, mdf_uuid)

                    if isinstance(name, dict):
                        key = (name["name"],) + key[1:]
                        if key in items_pool:
                            item, widget = items_pool[key]
                            widget.item = item
                            widget.set_ranges(ranges)

                            root.addChild(item)
                            tree.setItemWidget(item, 1, widget)

                            del items_pool[key]
                    else:

                        if key in items_pool:
                            item, widget = items_pool[key]
                            widget.item = item
                            widget.set_ranges(ranges)

                            root.addChild(item)
                            tree.setItemWidget(item, 1, widget)

                            del items_pool[key]

        for sig in channels:
            sig.uuid = os.urandom(6).hex()

        invalid = []

        can_trim = True
        for channel in channels:
            diff = np.diff(channel.timestamps)
            invalid_indexes = np.argwhere(diff <= 0).ravel()
            if len(invalid_indexes):
                invalid_indexes = invalid_indexes[:10] + 1
                idx = invalid_indexes[0]
                ts = channel.timestamps[idx - 1 : idx + 2]
                invalid.append(
                    f"{channel.name} @ index {invalid_indexes[:10] - 1} with first time stamp error: {ts}"
                )
                if len(np.argwhere(diff < 0).ravel()):
                    can_trim = False

        if invalid:
            errors = "\n".join(invalid)
            try:
                mdi_title = self.parent().windowTitle()
                title = f"plot <{mdi_title}>"
            except:
                title = "plot window"

            QtWidgets.QMessageBox.warning(
                self,
                f"Channels with corrupted time stamps added to {title}",
                f"The following channels do not have monotonous increasing time stamps:\n{errors}",
            )
            self.plot._can_trim = can_trim

        valid = []
        invalid = []
        for channel in channels:
            if len(channel):
                samples = channel.samples
                if samples.dtype.kind not in "SUV" and np.all(np.isnan(samples)):
                    invalid.append(channel.name)
                elif channel.conversion:
                    samples = channel.physical().samples
                    if samples.dtype.kind not in "SUV" and np.all(np.isnan(samples)):
                        invalid.append(channel.name)
                    else:
                        valid.append(channel)
                else:
                    valid.append(channel)
            else:
                valid.append(channel)

        if invalid:
            QtWidgets.QMessageBox.warning(
                self,
                "All NaN channels will not be plotted:",
                f"The following channels have all NaN samples and will not be plotted:\n{', '.join(invalid)}",
            )

        channels = valid

        self.adjust_splitter(channels)

        channels = self.plot.add_new_channels(channels)

        enforce_y_axis = False
        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            widget = self.channel_selection.itemWidget(item, 1)
            if isinstance(widget, ChannelDisplay):
                if widget.ylink.checkState() == QtCore.Qt.Unchecked:
                    enforce_y_axis = False
                    break
                else:
                    enforce_y_axis = True

            iterator += 1

        new_items = {}
        for sig in channels:

            item = ChannelsTreeItem(
                (sig.group_index, sig.channel_index),
                sig.name,
                sig.computation,
                None,
                sig.mdf_uuid,
                check=QtCore.Qt.Checked if sig.enable else QtCore.Qt.Unchecked,
            )

            # item.setData(QtCore.Qt.UserRole, sig.name)
            tooltip = getattr(sig, "tooltip", "") or f"{sig.name}\n{sig.comment}"
            if sig.source:
                details = sig.source.get_details()
            else:
                details = ""

            if len(sig.samples) and sig.conversion:
                kind = sig.conversion.convert(sig.samples[:1]).dtype.kind
            else:
                kind = sig.samples.dtype.kind
            it = ChannelDisplay(
                sig.uuid, sig.unit, kind, 3, tooltip, details, parent=self, item=item
            )
            if self.channel_selection.details_enabled:
                it.details.setVisible(True)
            it.setAttribute(QtCore.Qt.WA_StyledBackground)

            if len(sig):
                it.set_value(sig.samples[0])

            if sig.computed:
                font = QtGui.QFont()
                font.setItalic(True)
                it.name.setFont(font)

            it.set_name(sig.name)
            it.set_color(sig.color)
            item.setSizeHint(1, it.sizeHint())

            if mime_data is None:

                if destination is None:
                    self.channel_selection.addTopLevelItem(item)
                else:
                    destination.addChild(item)

                self.channel_selection.setItemWidget(item, 1, it)
            else:
                new_items[(item.name, *item.entry, item.mdf_uuid)] = (item, it)

            it.color_changed.connect(self.plot.set_color)
            it.enable_changed.connect(self.plot.set_signal_enable)
            it.ylink_changed.connect(self.plot.set_common_axis)
            it.unit_changed.connect(self.plot.set_unit)
            it.name_changed.connect(self.plot.set_name)
            if enforce_y_axis:
                it.ylink.setCheckState(QtCore.Qt.Checked)
            it.individual_axis_changed.connect(self.plot.set_individual_axis)

            item.setFlags(
                item.flags() | QtCore.Qt.ItemIsUserCheckable | QtCore.Qt.ItemIsEnabled
            )
            self.info_uuid = sig.uuid

        if mime_data:
            add_new_items(
                self.channel_selection,
                destination or self.channel_selection.invisibleRootItem(),
                mime_data,
                new_items,
            )

            # still have simple signals to add
            if new_items:
                for item, widget in new_items.values():
                    self.channel_selection.addTopLevelItem(item)
                    widget.item = item
                    self.channel_selection.setItemWidget(item, 1, widget)

        self.channel_selection.update_channel_groups_count()
        self.channel_selection.refresh()

    def to_config(self):
        def item_to_config(tree, root):
            channels = []

            for i in range(root.childCount()):
                item = root.child(i)
                widget = tree.itemWidget(item, 1)
                if isinstance(widget, ChannelDisplay):

                    channel = {"type": "channel"}

                    sig, idx = self.plot.signal_by_uuid(widget.uuid)

                    channel["name"] = sig.name
                    channel["unit"] = sig.unit
                    channel["enabled"] = item.checkState(0) == QtCore.Qt.Checked

                    if widget.individual_axis.checkState() == QtCore.Qt.Checked:
                        channel["individual_axis"] = True
                        channel["individual_axis_width"] = (
                            self.plot.axes[idx].boundingRect().width()
                        )
                    else:
                        channel["individual_axis"] = False

                    channel["common_axis"] = (
                        widget.ylink.checkState() == QtCore.Qt.Checked
                    )
                    channel["color"] = sig.color
                    channel["computed"] = sig.computed
                    channel["ranges"] = copy_ranges(widget.ranges)

                    for range_info in channel["ranges"]:
                        range_info["background_color"] = range_info[
                            "background_color"
                        ].name()
                        range_info["font_color"] = range_info["font_color"].name()

                    channel["precision"] = widget.precision
                    channel["fmt"] = widget.fmt
                    channel["mode"] = sig.mode
                    if sig.computed:
                        channel["computation"] = sig.computation

                    view = self.plot.view_boxes[idx]
                    channel["y_range"] = [float(e) for e in view.viewRange()[1]]
                    channel["mdf_uuid"] = str(sig.mdf_uuid)

                    if sig.computed and sig.conversion:
                        channel["user_defined_name"] = sig.name
                        channel["name"] = sig.computation["expression"].strip("}{")

                        channel["conversion"] = {}
                        for i in range(sig.conversion.val_param_nr):
                            channel["conversion"][
                                f"text_{i}"
                            ] = sig.conversion.referenced_blocks[f"text_{i}"].decode(
                                "utf-8"
                            )
                            channel["conversion"][f"val_{i}"] = sig.conversion[
                                f"val_{i}"
                            ]

                elif isinstance(widget, ChannelGroupDisplay):
                    pattern = item.pattern
                    if pattern:
                        pattern = dict(pattern)
                        ranges = copy_ranges(pattern["ranges"])

                        for range_info in ranges:
                            range_info["font_color"] = range_info["font_color"].name()
                            range_info["background_color"] = range_info[
                                "background_color"
                            ].name()

                        pattern["ranges"] = ranges

                    ranges = copy_ranges(widget.ranges)

                    for range_info in ranges:
                        range_info["font_color"] = range_info["font_color"].name()
                        range_info["background_color"] = range_info[
                            "background_color"
                        ].name()

                    channel = {
                        "type": "group",
                        "name": widget.name.text().rsplit("\t[")[0],
                        "channels": item_to_config(tree, item)
                        if item.pattern is None
                        else [],
                        "enabled": item.checkState(0) == QtCore.Qt.Checked,
                        "pattern": pattern,
                        "ranges": ranges,
                    }

                channels.append(channel)

            return channels

        pattern = self.pattern
        if pattern:
            ranges = copy_ranges(pattern["ranges"])

            for range_info in ranges:
                range_info["font_color"] = range_info["font_color"].name()
                range_info["background_color"] = range_info["background_color"].name()

            pattern["ranges"] = ranges

        config = {
            "channels": item_to_config(
                self.channel_selection, self.channel_selection.invisibleRootItem()
            )
            if not self.pattern
            else [],
            "pattern": pattern,
            "splitter": [int(e) for e in self.splitter.sizes()[:2]]
            + [
                0,
            ],
            "x_range": [float(e) for e in self.plot.viewbox.viewRange()[0]],
            "y_axis_width": self.plot.y_axis.boundingRect().width(),
            "grid": [
                self.plot.plotItem.ctrl.xGridCheck.isChecked(),
                self.plot.plotItem.ctrl.yGridCheck.isChecked(),
            ],
            "cursor_precision": self.cursor_info.precision,
        }

        return config

    def dragEnterEvent(self, e):
        if e.mimeData().hasFormat("application/octet-stream-asammdf"):
            e.accept()
        super().dragEnterEvent(e)

    def dropEvent(self, e):
        if e.source() is self.channel_selection:
            super().dropEvent(e)
        else:
            data = e.mimeData()
            if data.hasFormat("application/octet-stream-asammdf"):
                names = extract_mime_names(data)
                self.add_channels_request.emit(names)
            else:
                super().dropEvent(e)

    def widget_by_uuid(self, uuid):
        iterator = QtWidgets.QTreeWidgetItemIterator(self.channel_selection)
        while iterator.value():
            item = iterator.value()
            widget = self.channel_selection.itemWidget(item, 1)

            if isinstance(widget, ChannelDisplay) and widget.uuid == uuid:
                break

            iterator += 1
        else:
            widget = None
        return widget

    def _show_properties(self, uuid):
        for sig in self.plot.signals:
            if sig.uuid == uuid:
                if sig.computed:
                    view = ComputedChannelInfoWindow(sig, self)
                    view.show()

                else:
                    self.show_properties.emit(
                        [sig.group_index, sig.channel_index, sig.mdf_uuid]
                    )

    def set_splitter(self, pos, index):
        self.splitter_moved.emit(self, pos)

    def pattern_group_added_req(self, group):
        self.pattern_group_added.emit(self, group)

    def set_timestamp(self, stamp):
        if self.plot.cursor1 is None:
            event = QtGui.QKeyEvent(
                QtCore.QEvent.KeyPress,
                QtCore.Qt.Key_C,
                QtCore.Qt.NoModifier,
            )
            self.plot.keyPressEvent(event)

        self.plot.cursor1.setPos(stamp)
        self.cursor_move_finished()

    def compute_fft(self, uuid):
        signal, index = self.plot.signal_by_uuid(uuid)
        window = FFTWindow(PlotSignal(signal), parent=self)
        window.show()

    def clear(self):
        event = QtGui.QKeyEvent(
            QtCore.QEvent.KeyPress, QtCore.Qt.Key_A, QtCore.Qt.ControlModifier
        )
        self.channel_selection.keyPressEvent(event)
        event = QtGui.QKeyEvent(
            QtCore.QEvent.KeyPress, QtCore.Qt.Key_Delete, QtCore.Qt.NoModifier
        )
        self.channel_selection.keyPressEvent(event)

    def update_current_values(self, *args):
        if self.plot.region:
            self.range_modified()
        else:
            self.cursor_moved()

    def adjust_splitter(self, channels=None):
        channels = channels or self.plot.signals

        size = sum(self.splitter.sizes())

        width = 0
        for ch in channels:
            width = max(
                width,
                self.channel_selection.fontMetrics()
                .boundingRect(f"{ch.name} ({ch.unit})")
                .width(),
            )
        width += 170

        if width > self.splitter.sizes()[0]:

            if size - width >= 300:
                self.splitter.setSizes([width, size - width, 0])
            else:
                if size >= 350:
                    self.splitter.setSizes([size - 300, 300, 0])
                elif size >= 100:
                    self.splitter.setSizes([50, size - 50, 0]) class Tabular(TabularBase):
    add_channels_request = QtCore.pyqtSignal(list)

    def __init__(
        self, signals=None, start=0, format="phys", ranges=None, *args, **kwargs
    ):
        # super().__init__(*args, **kwargs)

        self.signals_descr = {}
        self.start = start
        self.pattern = {}
        self.format = format

        if signals is None:
            signals = pd.DataFrame()
        else:
            index = pd.Series(np.arange(len(signals), dtype="u8"), index=signals.index)
            signals["Index"] = index

            signals["timestamps"] = signals.index
            signals.set_index(index, inplace=True)
            dropped = {}

            for name_ in signals.columns:
                col = signals[name_]
                if col.dtype.kind == "O":
                    if name_.endswith("DataBytes"):
                        try:
                            sizes = signals[name_.replace("DataBytes", "DataLength")]
                        except:
                            sizes = None
                        dropped[name_] = pd.Series(
                            csv_bytearray2hex(
                                col,
                                sizes,
                            ),
                            index=signals.index,
                        )

                    elif name_.endswith("Data Bytes"):
                        try:
                            sizes = signals[name_.replace("Data Bytes", "Data Length")]
                        except:
                            sizes = None
                        dropped[name_] = pd.Series(
                            csv_bytearray2hex(
                                col,
                                sizes,
                            ),
                            index=signals.index,
                        )

                    elif col.dtype.name != "category":
                        try:
                            dropped[name_] = pd.Series(
                                csv_bytearray2hex(col), index=signals.index
                            )
                        except:
                            pass

                    self.signals_descr[name_] = 0

                elif col.dtype.kind == "S":
                    try:
                        dropped[name_] = pd.Series(
                            npchar.decode(col, "utf-8"), index=signals.index
                        )
                    except:
                        dropped[name_] = pd.Series(
                            npchar.decode(col, "latin-1"), index=signals.index
                        )
                    self.signals_descr[name_] = 0
                else:
                    self.signals_descr[name_] = 0

            signals = signals.drop(columns=["Index", *list(dropped)])
            for name, s in dropped.items():
                signals[name] = s

            names = list(signals.columns)
            names = [
                "timestamps",
                *[name for name in names if name.endswith((".ID", ".DataBytes"))],
                *[
                    name
                    for name in names
                    if name != "timestamps" and not name.endswith((".ID", ".DataBytes"))
                ],
            ]
            signals = signals[names]

        super().__init__(signals, ranges)
        self.format_selection.setCurrentText(format)

        self._original_timestamps = signals["timestamps"]
        self._original_ts_series = pd.Series(
            self._original_timestamps,
            index=index,
        )

        # self.build(self.signals, True, ranges=ranges)

        prefixes = set()
        for name in signals.columns:
            while "." in name:
                name = name.rsplit(".", 1)[0]
                prefixes.add(f"{name}.")

        self.filters.minimal_menu = True

        self.prefix.insertItems(0, sorted(prefixes, key=lambda x: (-len(x), x)))
        self.prefix.setEnabled(False)

        self.prefix.currentIndexChanged.connect(self.prefix_changed)

        if prefixes:
            self.remove_prefix.setCheckState(QtCore.Qt.Checked)

        self._settings = QtCore.QSettings()
        integer_mode = self._settings.value("tabular_format", "phys")

        self.format_selection.setCurrentText(integer_mode)

        self.tree.dataView.setAcceptDrops(True)
        self.tree.dataView.setDragDropMode(QtWidgets.QAbstractItemView.InternalMove)
        self.tree.dataView.setDropIndicatorShown(True)
        self.tree.dataView.setDefaultDropAction(QtCore.Qt.MoveAction) class ChannelsGroupTreeItem(QtWidgets.QTreeWidgetItem):
    def __init__(self, name="", pattern=None):
        super().__init__(["", ""])
        self.name = name.split("\t[")[0]
        self.pattern = pattern
        self._is_visible = True

        self.setFlags(
            self.flags()
            | QtCore.Qt.ItemIsUserCheckable
            | QtCore.Qt.ItemIsEnabled
            | QtCore.Qt.ItemIsAutoTristate
        )

        self.setCheckState(0, QtCore.Qt.Checked)

    def copy(self):
        x = ChannelsGroupTreeItem(self.name, self.pattern)
        return x

    def show_info(self):
        widget = self.treeWidget().itemWidget(self, 1)
        if widget:
            ranges = widget.ranges
            name = widget._name or self.name
        else:
            ranges = []
            name = self.name
        ChannnelGroupDialog(name, self.pattern, ranges, self.treeWidget()).show()

    def get_ranges(self, tree=None):
        tree = tree or self.treeWidget()
        widget = tree.itemWidget(self, 1)
        parent = self.parent()
        if widget is None:
            if parent is None:
                return []
            else:
                return parent.get_ranges(tree)
        else:
            if parent is None:
                return widget.ranges
            else:
                return [*widget.ranges, *parent.get_ranges(tree)]

    def update_child_values(self, tree=None):
        tree = tree or self.treeWidget()
        count = self.childCount()
        for i in range(count):
            item = self.child(i)
            item.update_child_values(tree)
from copy import deepcopy
from functools import partial
from traceback import format_exc
from natsort import natsorted
from PyQt5 import QtCore, QtGui, QtWidgets
from ...blocks import v4_constants as v4c
from ...blocks.conversion_utils import from_dict
from ...blocks.utils import (
    csv_bytearray2hex,
    extract_cncomment_xml,
    load_can_database,
    MdfException,
)
from ...mdf import MDF
from ...signal import Signal
from ..dialogs.channel_info import ChannelInfoDialog
from ..dialogs.window_selection_dialog import WindowSelectionDialog
from ..utils import compute_signal, copy_ranges, extract_mime_names
from .bar import Bar
from .can_bus_trace import CANBusTrace
from .channel_display import ChannelDisplay
from .flexray_bus_trace import FlexRayBusTrace
from .gps import GPS
from .lin_bus_trace import LINBusTrace
from .numeric import Numeric
from .plot import Plot
from .tabular import Tabular
from .tree import ChannelsGroupTreeItem
import itertools
import json
import os
import re
import sys
import numpy as np
import pandas as pd
                    for sig, sig_info in zip(
                        self.mdf.select(
                            measured_signals_,
                            ignore_value2text_conversions=self.ignore_value2text_conversions,
                            copy_master=False,
                            validate=True,
                            raw=True,
                        ),
                        measured_signals_,
                    )
                }

                for signal, entry_info, channel in zip(
                    measured_signals.values(), measured_signals_, found_signals
                ):
                    signal.computed = False
                    signal.computation = {}
                    signal.color = channel["color"]
                    signal.group_index = entry_info[1]
                    signal.channel_index = entry_info[2]
                    signal.mdf_uuid = uuid
                    signal.name = entry_info[0]

                matrix_components = []
                for name in not_found:
                    name, indexes = parse_matrix_component(name)
                    if indexes and name in self.mdf:
                        matrix_components.append((name, indexes))

                matrix_signals = {
                    str(matrix_element): sig
                    for sig, matrix_element in zip(
                        self.mdf.select(
                            [el[0] for el in matrix_components],
                            ignore_value2text_conversions=self.ignore_value2text_conversions,
                            copy_master=False,
                        ),
                        matrix_components,
                    )
                }

                new_matrix_signals = {}
                for signal_mat, (_n, indexes) in zip(
                    matrix_signals.values(), matrix_components
                ):
                    signal = deepcopy(signal_mat)
                    signal.computed = False
                    signal.computation = {}
                    signal.group_index, signal.channel_index = self.mdf.whereis(
                        signal.name
                    )[0]

                    indexes_string = "".join(f"[{_index}]" for _index in indexes)

                    samples = signal.samples
                    if samples.dtype.names:
                        samples = samples[signal.name]

                    for idx in indexes:
                        samples = samples[:, idx]
                    sig_name = f"{signal.name}{indexes_string}"
                    signal.name = sig_name
                    signal.samples = samples

                    new_matrix_signals[signal.name] = signal

                measured_signals.update(
                    {name: sig for name, sig in new_matrix_signals.items()}
                )

                if measured_signals:
                    all_timebase = np.unique(
                        np.concatenate(
                            [sig.timestamps for sig in measured_signals.values()]
                        )
                    )
                else:
                    all_timebase = []

                required_channels = []
                for ch in computed_signals_descriptions:
                    required_channels.extend(get_required_from_computed(ch))

                required_channels = set(required_channels)
                required_channels = [
                    (None, *self.mdf.whereis(channel)[0])
                    for channel in required_channels
                    if channel not in list(measured_signals) and channel in self.mdf
                ]
                required_channels = {
                    sig.name: sig
                    for sig in self.mdf.select(
                        required_channels,
                        ignore_value2text_conversions=self.ignore_value2text_conversions,
                        copy_master=False,
                    )
                }

                required_channels.update(measured_signals)

                computed_signals = {}

                for channel in computed_signals_descriptions:
                    computation = channel["computation"]

                    try:

                        signal = compute_signal(
                            computation, required_channels, all_timebase
                        )
                        signal.color = channel["color"]
                        signal.computed = True
                        signal.computation = channel["computation"]
                        signal.name = channel["name"]
                        signal.unit = channel["unit"]
                        signal.group_index = -1
                        signal.channel_index = -1
                        signal.mdf_uuid = uuid

                        if "conversion" in channel:

                            signal.conversion = from_dict(channel["conversion"])