# -*- coding: utf-8 -*-

""" Measured Data Format blocks parser for version 4.x

Created on Sun Dec 15 12:57:28 2013

:Author: `Aymeric Rateau <https://github.com/ratal/mdfreader>`__

mdfinfo4

--------------------------

"""

from struct import calcsize, unpack, pack, Struct

from os import remove

from warnings import warn

from zlib import compress, decompress

from numpy import zeros, array, append

from math import isnan

from time import time

from sys import getsizeof

from collections import OrderedDict

from xml.etree.ElementTree import Element, SubElement, \

tostring, register_namespace

from lxml import objectify

from .mdf import _open_mdf, dataField, descriptionField, unitField, \

masterField, masterTypeField, idField, _convert_name

# datatypes

_LINK = '<Q'

_REAL = '<d'

_BOOL = '<h'

_UINT8 = '<B'

_BYTE = '<c'

_INT16 = '<h'

_UINT16 = '<H'

_UINT32 = '<I'

_INT32 = '<i'

_UINT64 = '<Q'

_INT64 = '<q'

_CHAR = '<c'

_HeaderStruct = Struct('<4sI2Q')

_DZStruct = Struct('<2s2BI2Q')

_HLStruct = Struct('<QHB5s')

_SIStruct = Struct('<4sI5Q3B5s')

_DGStruct = Struct('<4sI6QB7s')

_CGStruct1 = Struct('<4sI8Q')

_CGStruct2 = Struct('<2Q2H3I')

_CNStruct1 = Struct('<4sI10Q')

_CNStruct2 = Struct('<4B4IBcH6d')

_CCStruct1 = Struct('<4sI6Q')

_CCStruct2 = Struct('<2B3H2d')

_SRStruct = Struct('<4sI5Qd2B6s')

# SI Types

si_type = {0: 'OTHER', 1: 'ECU', 2: 'BUS',

3: 'I/O', 4: 'TOOL', 5: 'USER'}

si_bus_type = {0: 'NONE', 1: 'OTHER', 2: 'CAN', 3: 'LIN',

4: 'MOST', 5: 'FLEXRAY', 6: 'K_LINE', 7: 'ETHERNET', 8: 'USB'}

# EV cause

ev_cause = {0: 'OTHER', 1: 'ERROR', 2: 'TOOL', 3: 'SCRIPT', 4: 'USER'}

# precompiled path for xmls to speed up parsing

CN_TX = objectify.ObjectPath('CNcomment.TX')

CN_names = objectify.ObjectPath('CNcomment.names')

CN_axis_monotony = objectify.ObjectPath('CNcomment.axis_monotony')

CN_raster = objectify.ObjectPath('CNcomment.raster')

CN_formula = objectify.ObjectPath('CNcomment.formula')

CN_linker_name = objectify.ObjectPath('CNcomment.linker_name')

CN_linker_address = objectify.ObjectPath('CNcomment.linker_address')

CN_address = objectify.ObjectPath('CNcomment.address')

CN_unit_TX = objectify.ObjectPath('CNunit.TX')

CC_TX = objectify.ObjectPath('CCcomment.TX')

CC_names = objectify.ObjectPath('CCcomment.names')

CC_COMPU_METHOD = objectify.ObjectPath('CCcomment.COMPU_METHOD')

CC_formula = objectify.ObjectPath('CCcomment.formula')

CC_unit_TX = objectify.ObjectPath('CCunit.TX')

SI_TX = objectify.ObjectPath('SIcomment.TX')

SI_names = objectify.ObjectPath('SIcomment.names')

SI_path = objectify.ObjectPath('SIcomment.path')

SI_bus = objectify.ObjectPath('SIcomment.bus')

SI_protocol = objectify.ObjectPath('SIcomment.protocol')

EV_TX = objectify.ObjectPath('EVcomment.TX')

EV_pre_trigger_interval = objectify.ObjectPath('EVcomment.pre_trigger_interval')

EV_post_trigger_interval = objectify.ObjectPath('EVcomment.post_trigger_interval')

EV_formula = objectify.ObjectPath('EVcomment.formula')

EV_timeout = objectify.ObjectPath('EVcomment.timeout')

chunk_size_writing = 4194304 # write by chunk of 4Mb, can be tuned for best performance

def _load_header(fid, pointer):

""" reads block's header and put in class dict

Parameters

----------------

fid : float

file identifier

pointer : int

position of block in file

"""

# All blocks have the same header

if pointer != 0 and pointer is not None:

fid.seek(pointer)

temp = dict()

(temp['id'],

reserved,

temp['length'],

temp['link_count']) = _HeaderStruct.unpack(fid.read(24))

temp['pointer'] = pointer

return temp

else:

return None

def _mdf_block_read(fid, data_type, count):

""" converts a byte array of length count to a given data Type

Parameters

----------------

data_type : str

C format data type

count : int

number of elements to sequentially read

Returns

-----------

array of values of 'Type' parameter

"""

value = fid.read(calcsize(data_type) * count)

if value:

if count == 1:

return unpack(data_type, value)[0]

else:

if '<' in data_type or '>' in data_type:

endian = data_type[0]

data_type = data_type.strip('<>')

return unpack(endian + count * data_type, value)

else:

return unpack(count * data_type, value)

else:

return None

def _calculate_block_start(current_position):

""" converts pointer position into one being multiple of 8

Parameters

----------------

current_position : int

position in file

Returns

-----------

position multiple of 8

"""

remain = current_position % 8

if not remain == 0:

return current_position - remain + 8

else:

return current_position

class IDBlock(dict):

""" reads or writes ID Block

"""

def __init__(self, fid=None):

if fid is not None:

self.read(fid)

def read(self, fid):

""" reads IDBlock

"""

fid.seek(0)

(self['id_file'],

self['id_vers'],

self['id_prog'],

id_reserved1,

self['id_ver'],

id_reserved2,

self['id_unfi_flags'],

self['id_custom_unfi_flags']) = unpack('<8s8s8sIH30s2H',

fid.read(64))

# treatment of unfinalised file

if self['id_ver'] > 410 and b'UnFin' in self['id_file']:

warn(' ! unfinalised file')

if self['id_unfi_flags'] & 1:

warn('Update of cycle counters for CG/CA blocks required')

if self['id_unfi_flags'] & (1 << 1):

warn('Update of cycle counters for SR blocks required')

if self['id_unfi_flags'] & (1 << 2):

warn('Update of length for last DT block required')

if self['id_unfi_flags'] & (1 << 3):

warn('Update of length for last RD block required')

if self['id_unfi_flags'] & (1 << 4):

warn('Update of last DL block in each chained list of '

'DL blocks required')

if self['id_unfi_flags'] & (1 << 5):

warn('Update of cg_data_bytes and cg_inval_bytes in VLSD '

'CG block required')

if self['id_unfi_flags'] & (1 << 6):

warn('Update of offset values for VLSD channel required '

'in case a VLSD CG block is used')

def write(self, fid):

""" Writes IDBlock

"""

# MDF versionTxt tool reserved version_int

head = (b'MDF ', self['id_vers'], b'MDFreadr', b'\0' * 4, self['id_ver'],

b'\0' * 30, 0, 0)

fid.write(pack('<8s8s8s4sH30s2H', *head))

class HDBlock(dict):

""" reads Header block and save in class dict

"""

def __init__(self, fid=None):

self['time'] = time()

if fid is not None:

self.read(fid)

def read(self, fid=None):

fid.seek(64)

(self['id'],

reserved,

self['length'],

self['link_count'],

self['hd_dg_first'],

self['hd_fh_first'],

self['hd_ch_first'],

self['hd_at_first'],

self['hd_ev_first'],

self['hd_md_comment'],

self['hd_start_time_ns'],

self['hd_tz_offset_min'],

self['hd_dst_offset_min'],

self['hd_time_flags'],

self['hd_time_class'],

self['hd_flags'],

hd_reserved,

self['hd_start_angle_rad'],

self['hd_start_distance']) = unpack('<4sI9Q2h4B2d', fid.read(104))

if self['hd_md_comment']: # if comments exist

self['Comment'] = {}

comment = CommentBlock()

comment.read_cm_hd(fid=fid, pointer=self['hd_md_comment'])

self['Comment'].update(comment)

def write(self, fid):

# write block header

fid.seek(self['block_start'])

# link section

# (first Data group pointer, first file history block pointer,

# pointer to hierarchy Block file, pointer to attachment Block,

# pointer to event Block, pointer to comment Block,

# time in ns, timezone offset in min, time daylight offset in min,

# time flags, time class, hd flags, reserved, start angle in radians

# start distance in meters)

data_bytes = (b'##HD', 0, 104, 6,

self['DG'], self['FH'], 0, 0, 0, 0,

int(self['time'] * 1E9),

0, 0, 1, 0, 0, b'\0', 0, 0)

try:

fid.write(pack('<4sI2Q7Q2h3Bs2d', *data_bytes))

except:

warn('time is Nan or malformed')

data_bytes = (b'##HD', 0, 104, 6,

self['DG'], self['FH'], 0, 0, 0, 0,

int(time() * 1E9),

0, 0, 1, 0, 0, b'\0', 0, 0)

fid.write(pack('<4sI2Q7Q2h3Bs2d', *data_bytes))

class FHBlock(dict):

""" reads File History block and save in class dict

"""

def __init__(self, fid=None, pointer=None):

if fid is not None:

self.read(fid, pointer)

def read(self, fid, pointer):

fid.seek(pointer)

(self['id'],

reserved,

self['length'],

self['link_count'],

self['fh_fh_next'],

self['fh_md_comment'],

self['fh_time_ns'],

self['fh_tz_offset_min'],

self['fh_dst_offset_min'],

self['fh_time_flags'],

fh_reserved) = unpack('<4sI5Q2hB3s', fid.read(56))

if self['fh_md_comment']: # comments exist

self['Comment'] = {}

comment = CommentBlock()

comment.read_cm_fh(fid=fid, pointer=self['fh_md_comment'])

self['Comment'].update(comment)

def write(self, fid):

# write block header

fid.seek(self['block_start'])

# link section

# (No next FH, comment block pointer

# time in ns, timezone offset in min, time daylight offest in min,

# time flags, reserved)

data_bytes = (b'##FH', 0, 56, 2,

0, self['MD'],

int(time()*1E9),

0, 0, 1, b'\0' * 3)

fid.write(pack('<4sI2Q3Q2hB3s', *data_bytes))

class CHBlock(dict):

""" reads Channel Hierarchy block and saves in class dict

"""

def __init__(self, fid, pointer):

# block header

self.update(_load_header(fid, pointer))

# Channel hierarchy block

(self['ch_ch_next'],

self['ch_ch_first'],

self['ch_tx_name'],

self['ch_md_comment']) = unpack('<4Q', fid.read(32))

n_links = self['link_count'] - 4

self['ch_element'] = unpack('<{}Q'.format(n_links),

fid.read(n_links * 8))

(self['ch_element_count'],

self['ch_type'],

ch_reserved) = unpack('<IB3s', fid.read(8))

if self['ch_md_comment']: # comments exist

self['Comment'] = {}

comment = CommentBlock()

comment.read_cm_ch(fid=fid, pointer=self['ch_md_comment'])

self['Comment'].update(comment)

if self['ch_tx_name']: # text block containing name of hierarchy level

self['ch_name_level'] = {}

comment = CommentBlock()

comment.read_tx(fid=fid, pointer=self['ch_tx_name'])

self['ch_name_level'].update(comment)

class CommentBlock(dict):

""" reads or writes Comment block and saves in class dict

"""

def __init__(self):

self = {}

def read_tx(self, fid, pointer):

""" reads TX block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

fid.seek(pointer)

# block header

(self['id'],

reserved,

self['length'],

self['link_count']) = _HeaderStruct.unpack(fid.read(24))

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_header(self, fid, pointer):

""" reads Comment block header

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

fid.seek(pointer)

# block header

(self['id'],

reserved,

self['length'],

self['link_count']) = _HeaderStruct.unpack(fid.read(24))

def read_xml(self, fid):

""" reads Comment block xml and objectifies it

Parameters

----------

fid:

file identifier

Metadata block

removes normal 0 at end

"""

try:

xml_string = fid.read(self['length'] - 24).rstrip(b'\x00')

xml_tree = objectify.fromstring(xml_string)

except:

warn('xml metadata malformed')

xml_tree = None

return xml_tree

def read_cm_hd(self, fid, pointer):

""" reads Comment block from header block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

try:

self['TX'] = xml_tree.TX.text

except AttributeError:

warn('Could not parse HD block TX tag')

try:

self['time_source'] = xml_tree.time_source.text

except AttributeError:

pass # optional

try:

tmp = xml_tree.common_properties

for t in tmp.e:

self[t.attrib.values()[0]] = t.text

except AttributeError:

pass # optional

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_fh(self, fid, pointer):

""" reads Comment block from file history block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

try:

self['TX'] = xml_tree.TX.text

except AttributeError:

warn('Could not parse FH block TX tag')

try:

self['tool_id'] = xml_tree.tool_id.text

except AttributeError:

warn('Could not parse FH block tool_id tag')

try:

self['tool_vendor'] = xml_tree.tool_vendor.text

except AttributeError:

warn('Could not parse extract HD block tool_vendor tag')

try:

self['tool_version'] = xml_tree.tool_version.text

except AttributeError:

warn('Could not parse extract HD block tool_vendor tag')

try:

self['user_name'] = xml_tree.user_name.text

except AttributeError:

pass # optional

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_ch(self, fid, pointer):

""" reads Comment block from file channel hierarchy block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

if xml_tree is not None:

try:

self['TX'] = xml_tree.TX.text

except AttributeError:

warn('Could not parse CH block TX tag')

try:

self['names'] = xml_tree.names.text

except AttributeError:

warn('Could not parse CH block names tag')

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_at(self, fid, pointer):

""" reads Comment block from attachment block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in (b'##MD', '##MD'):

xml_tree = self.read_xml(fid)

try:

self['TX'] = xml_tree.TX.text

except AttributeError:

warn('Could not parse AT block TX tag')

elif self['id'] in (b'##TX', '##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_ev(self, fid, pointer):

""" reads Comment block from event block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

if xml_tree is not None:

try:

self['TX'] = EV_TX(xml_tree).text

except AttributeError:

warn('Could not parse EV block TX tag')

try:

self['pre_trigger_interval'] = EV_pre_trigger_interval(xml_tree).text

except AttributeError:

pass # optional

try:

self['post_trigger_interval'] = EV_post_trigger_interval(xml_tree).text

except AttributeError:

pass # optional

try:

self['formula'] = EV_formula(xml_tree).text

except AttributeError:

pass # optional

try:

self['timeout'] = EV_timeout(xml_tree).text

except AttributeError:

pass # optional

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_dg(self, fid, pointer):

""" reads Comment block from data group block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

try:

self['TX'] = xml_tree.TX.text

except AttributeError:

warn('Could not parse AT block TX tag')

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_cg(self, fid, pointer):

""" reads Comment block from channel group block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

try:

self['TX'] = xml_tree.TX.text

except AttributeError:

warn('Could not parse CG block TX tag')

try:

self['names'] = xml_tree.names.text

except AttributeError:

warn('Could not parse CG block names tag')

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_si(self, fid, pointer):

""" reads Comment block from source information block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

if xml_tree is not None:

try:

self['TX'] = SI_TX(xml_tree).text

except AttributeError:

warn('Could not parse SI block TX tag')

try:

self['names'] = SI_names(xml_tree).text

except AttributeError:

pass # optional

try:

self['path'] = SI_path(xml_tree).text

except AttributeError:

pass # optional

try:

self['bus'] = SI_bus(xml_tree).text

except AttributeError:

pass # optional

try:

self['protocol'] = SI_protocol(xml_tree).text

except AttributeError:

pass # optional

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_cn(self, fid, pointer, minimal=True):

""" reads Comment block from channel block

Parameters

----------

fid:

file identifier

pointer: int

position in file

minimal: boolean

flag to reduce metadata parsing

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

if xml_tree is not None:

try:

self['description'] = CN_TX(xml_tree).text

except AttributeError:

warn('Could not parse CN block TX tag')

try:

self['names'] = CN_names(xml_tree).text

except AttributeError:

pass # optional

if minimal is False:

# not really used for the moment

try:

self['axis_monotony'] = CN_axis_monotony(xml_tree).text

except AttributeError:

pass # optional

try:

self['raster'] = CN_raster(xml_tree).text

except AttributeError:

pass # optional

try:

self['formula'] = CN_formula(xml_tree).text

except AttributeError:

pass # optional

try:

self['linker_name'] = CN_linker_name(xml_tree).text

except AttributeError:

pass # optional

try:

self['linker_address'] = CN_linker_address(xml_tree).text

except AttributeError:

pass # optional

try:

self['address'] = CN_address(xml_tree).text

except AttributeError:

pass # optional

elif self['id'] in ('##TX', b'##TX'):

self['name'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_cn_unit(self, fid, pointer):

""" reads Comment block for channel unit

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

try:

self['unit'] = CN_unit_TX(xml_tree).text

except AttributeError:

warn('Could not parse unit TX tag')

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_cc(self, fid, pointer):

""" reads Comment block from channel conversion block

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

if xml_tree is not None:

try:

self['TX'] = CC_TX(xml_tree).text

except AttributeError:

warn('Could not parse CC block TX tag')

try:

self['names'] = CC_names(xml_tree).text

except AttributeError:

pass # optional

try:

self['COMPU_METHOD'] = CC_COMPU_METHOD(xml_tree).text

except AttributeError:

pass # optional

try:

self['formula'] = CC_formula(xml_tree).text

except AttributeError:

pass # optional

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def read_cm_cc_unit(self, fid, pointer):

""" reads Comment block for channel conversion unit

Parameters

----------

fid:

file identifier

pointer: int

position in file

"""

if pointer > 0:

self.read_cm_header(fid, pointer)

if self['id'] in ('##MD', b'##MD'):

xml_tree = self.read_xml(fid)

try:

self['unit'] = CC_unit_TX(xml_tree).text

except AttributeError:

warn('Could not parse unit TX tag')

elif self['id'] in ('##TX', b'##TX'):

self['Comment'] = fid.read(self['length'] - 24).rstrip(b'\x00').decode('UTF-8', 'ignore')

def load(self, data, md_type):

if md_type == 'TX':

data = b''.join([data.encode('utf-8', 'replace'), b'\0'])

block_id = b'##TX'

else:

register_namespace('', 'http://www.asam.net/mdf/v4')

if md_type == 'HD':

root = Element('HDcomment')

root.set('xmlns', 'http://www.asam.net/mdf/v4')

tx = SubElement(root, 'TX')

tx.text = data['comment']

common_properties = SubElement(root, 'common_properties')

e = SubElement(common_properties, 'e',

attrib={'name': 'subject'})

e.text = data['subject']

e = SubElement(common_properties, 'e',

attrib={'name': 'project'})

e.text = data['project']

e = SubElement(common_properties, 'e',

attrib={'name': 'department'})

e.text = data['organisation']

e = SubElement(common_properties, 'e',

attrib={'name': 'author'})

e.text = data['author']

elif md_type == 'CN':

pass

elif md_type == 'FH':

root = Element('FHcomment')

root.set('xmlns', 'http://www.asam.net/mdf/v4')

tx = SubElement(root, 'TX')

tx.text = data['comment']

tool_id = SubElement(root, 'tool_id')

tool_id.text = 'mdfreader'

tool_vendor = SubElement(root, 'tool_vendor')

tool_vendor.text = 'mdfreader is under GPL V3'

tool_version = SubElement(root, 'tool_version')

tool_version.text = '2.8'

data = b''.join([tostring(root), b'\0'])

block_id = b'##MD'

# make sure block is multiple of 8

remain = len(data) % 8

if not remain == 0:

data = b''.join([data, b'\0' * (8 - (remain % 8))])

self['block_length'] = 24 + len(data)

self['Header'] = (block_id, 0, self['block_length'], 0)

self['data'] = data

def write(self, fid):

fid.write(_HeaderStruct.pack(*self['Header']))

fid.write(self['data'])

class DGBlock(dict):

""" reads Data Group block and saves in class dict

"""

def __init__(self, fid=None, pointer=None):

if fid is not None:

self.read_dg(fid, pointer)

def read_dg(self, fid, pointer):

fid.seek(pointer)

self['pointer'] = pointer

(self['id'],

reserved,

self['length'],

self['link_count'],

self['dg_dg_next'],

self['dg_cg_first'],

self['dg_data'],

self['dg_md_comment'],

self['dg_rec_id_size'],

dg_reserved) = _DGStruct.unpack(fid.read(64))

self['data_block_header'] = _load_header(fid, self['dg_data'])

if self['dg_md_comment']: # comments exist

self['Comment'] = {}

comment = CommentBlock()

comment.read_cm_dg(fid=fid, pointer=self['dg_md_comment'])

self['Comment'].update(comment)

def write(self, fid):

# write block header

fid.seek(self['block_start'])

# link section

# (Next Data group pointer, first channel group block pointer,

# data block pointer, comment block pointer,

# no recordID, reserved)

data_bytes = (b'##DG', 0, 64, 4,

self['DG'], self['CG'],

self['data'], 0, 0, b'\0' * 7)

fid.write(pack('<4sI2Q4QB7s', *data_bytes))

class CGBlock(dict):

""" reads Channel Group block and saves in class dict

"""

def __init__(self, fid=None, pointer=None):

if fid is not None:

self.read_cg(fid, pointer)

def read_cg(self, fid, pointer):

fid.seek(pointer)

self['pointer'] = pointer

(self['id'],

self['reserved'],

self['length'],

self['link_count'],

self['cg_cg_next'],

self['cg_cn_first'],

self['cg_tx_acq_name'],

self['cg_si_acq_source'],

self['cg_sr_first'],

self['cg_md_comment']) = _CGStruct1.unpack(fid.read(72))

if self['link_count'] > 6: # linked to another cg containing master

self['cg_cg_master'] = unpack(_LINK, fid.read(8))[0]

# data section

(self['cg_record_id'],

self['cg_cycle_count'],

self['cg_flags'],

self['cg_path_separator'],

self['cg_reserved'],

self['cg_data_bytes'],

self['cg_invalid_bytes']) = _CGStruct2.unpack(fid.read(32))

if self['cg_md_comment']: # comments exist

self['Comment'] = CommentBlock()

self['Comment'].read_cm_cg(fid=fid, pointer=self['cg_md_comment'])

if self['cg_tx_acq_name']: # comments exist

self['acq_name'] = {}

comment = CommentBlock()

comment.read_tx(fid=fid, pointer=self['cg_tx_acq_name'])

self['acq_name'].update(comment)

if self['cg_si_acq_source']: # comments exist

self['acq_source'] = {}

si = SIBlock()

si.read_si(fid=fid, pointer=self['cg_si_acq_source'])

self['acq_source'].update(si)

def write(self, fid):

# write block header

# fid.seek(self['block_start'])

# link section

# (Next channel group pointer, first channel block pointer,

# acquisition name pointer, acquisition source pointer,

# sample reduction pointer, comment pointer,

# no recordID, cycle count, no flags,

# no character specified for path separator,

# reserved, number of bytes taken by data in record,

# no invalid bytes)

if 'cg_cg_master' in self and self['cg_cg_master'] is not None:

data_bytes = (b'##CG', 0, self['length'], 7,

0, self['CN'], 0, 0, 0, 0, self['cg_cg_master'], 0,

self['cg_cycle_count'], 8, 0,

b'\0' * 4,

self['cg_data_bytes'], self['cg_inval_bytes'])

fid.write(pack('<4sI2Q9Q2H4s2I', *data_bytes))

else:

data_bytes = (b'##CG', 0, self['length'], 6,

0, self['CN'], 0, 0, 0, 0, 0,

self['cg_cycle_count'], 0, 0,

b'\0' * 4,

self['cg_data_bytes'], self['cg_inval_bytes'])

fid.write(pack('<4sI2Q8Q2H4s2I', *data_bytes))

class CNBlock(dict):

""" reads Channel block and saves in class dict

"""

def read_cn(self, **kargs):

if kargs['pointer'] != 0 and kargs['pointer'] is not None:

self['pointer'] = kargs['pointer']

kargs['fid'].seek(kargs['pointer'])

links = []

(self['id'],

reserved,

self['length'],

self['link_count'],

self['cn_cn_next'],

self['cn_composition'],

self['cn_tx_name'],

self['cn_si_source'],

self['cn_cc_conversion'],

self['cn_data'],

self['cn_md_unit'],

self['cn_md_comment']) = _CNStruct1.unpack(kargs['fid'].read(88))

if self['link_count'] > 8:

links = _mdf_block_read(kargs['fid'], _LINK, self['link_count'] - 8)

# data section

(self['cn_type'],

self['cn_sync_type'],

self['cn_data_type'],

self['cn_bit_offset'],

self['cn_byte_offset'],

self['cn_bit_count'],

self['cn_flags'],

self['cn_invalid_bit_pos'],

self['cn_precision'],

self['cn_reserved'],

self['cn_attachment_count'],

self['cn_val_range_min'],

self['cn_val_range_max'],

cn_limit_min,

cn_limit_max,

cn_limit_ext_min,

cn_limit_ext_max) = _CNStruct2.unpack(kargs['fid'].read(72))

if self['cn_attachment_count'] > 0:

self['attachment'] = {}

if self['cn_attachment_count'] > 1:

self['cn_at_reference'] = links[:self['cn_attachment_count']]

for at in range(self['cn_attachment_count']):

self['attachment'][at] = \

ATBlock(kargs['fid'], self['cn_at_reference'][at])

else:

self['cn_at_reference'] = links

self['attachment'][0] = \

ATBlock(kargs['fid'], self['cn_at_reference'])

if self['link_count'] > (8 + self['cn_attachment_count']):

self['cn_default_x'] = links[self['cn_attachment_count']:]

else: