You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
193 lines
12 KiB
193 lines
12 KiB
"""Export the supplied PC MF4 as common raw input, inspect/compare reader semantics.
|
|
|
|
Requires the independent asammdf reader, not an Android dependency. Original files
|
|
are read-only. The exported input also drives the actual bundled PC writer DLL.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import struct
|
|
import xml.etree.ElementTree as ET
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from asammdf import MDF
|
|
|
|
|
|
def blocks(path):
|
|
data = path.read_bytes()
|
|
visited, counts, compression = set(), Counter(), []
|
|
|
|
def visit(address):
|
|
if not address or address in visited:
|
|
return
|
|
visited.add(address)
|
|
kind, _, length, links_nr = struct.unpack_from('<4sIQQ', data, address)
|
|
assert address + length <= len(data), (address, kind, length)
|
|
counts[kind.decode()] += 1
|
|
links = struct.unpack_from('<' + 'Q' * links_nr, data, address + 24)
|
|
if kind == b'##DZ':
|
|
original, zip_type, _, parameter, raw_size, zip_size = struct.unpack_from('<2sBBIQQ', data, address + 24)
|
|
compression.append(dict(original=original.decode(), zip_type=zip_type,
|
|
parameter=parameter, raw_size=raw_size, zip_size=zip_size))
|
|
for link in links:
|
|
visit(link)
|
|
|
|
visit(64)
|
|
return dict(counts=dict(counts), compression=compression, file_size=len(data))
|
|
|
|
|
|
def channel(c):
|
|
conversion = c.conversion
|
|
cc = None if conversion is None else dict(
|
|
type=conversion.conversion_type,
|
|
parameters=[getattr(conversion, key) for key in ('b', 'a')]
|
|
if conversion.conversion_type == 1 else [])
|
|
source = None if c.source is None else {
|
|
key: getattr(c.source, key) for key in ('name', 'path', 'comment', 'source_type', 'bus_type')}
|
|
return dict(name=c.name, unit=c.unit, description=c.comment,
|
|
data_type=c.data_type, bit_count=c.bit_count, bit_offset=c.bit_offset,
|
|
byte_offset=c.byte_offset, flags=c.flags, conversion=cc, source=source,
|
|
min_raw=c.min_raw_value, max_raw=c.max_raw_value)
|
|
|
|
|
|
def export(source, destination):
|
|
with MDF(source) as m:
|
|
assert len(m.groups) == 1, 'The reference fixture requires one PC channel group'
|
|
g = m.groups[0]
|
|
signals = [m.get(group=0, index=i, raw=True) for i in range(1, len(g.channels))]
|
|
times = m.get_master(0)
|
|
result = dict(reference=str(source), version=m.version, start_ns=m.header.abs_time,
|
|
timezone_minutes=m.header.tz_offset, dst_minutes=m.header.daylight_save_time,
|
|
time_flags=m.header.time_flags, header_comment=m.header.comment,
|
|
history=[h.comment for h in m.file_history],
|
|
channels=[channel(c) for c in g.channels],
|
|
record_size=g.channel_group.samples_byte_nr,
|
|
records=[[float(t)] + [s.samples[i].item() for s in signals]
|
|
for i, t in enumerate(times)], blocks=blocks(source))
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
print('EXPORTED', len(result['channels']), 'channels,', len(result['records']), 'records,', result['record_size'], 'bytes')
|
|
print(json.dumps(result['blocks'], indent=2))
|
|
specs = [('U8', 0, 8, [0, 255, 17]), ('S8', 2, 8, [-128, 127, -3]),
|
|
('U16', 0, 16, [0, 65535, 12345]), ('S16', 2, 16, [-32768, 32767, -134]),
|
|
('U32', 0, 32, [0, 4294967295, 3735928559]), ('S32', 2, 32, [-2147483648, 2147483647, -10]),
|
|
('U64', 0, 64, [0, 9007199254740992, 123456789]), ('S64', 2, 64, [-9007199254740992, 9007199254740992, -9]),
|
|
('F32', 4, 32, [-12.5, 3.75, 1.25]), ('F64', 4, 64, [-1.25, 3.141592653589793, 0.125]),
|
|
('Temperature', 2, 16, [-300, 125, 1000])]
|
|
synthetic = {k: v for k, v in result.items() if k not in ('records', 'channels', 'blocks')}
|
|
offset = 8
|
|
synthetic['channels'] = [result['channels'][0]]
|
|
for name, dtype, bits, values in specs:
|
|
synthetic['channels'].append(dict(name=name, unit='degC' if name == 'Temperature' else 'V',
|
|
description='raw boundary '+name, data_type=dtype, bit_count=bits, bit_offset=0,
|
|
byte_offset=offset, flags=0, min_raw=0., max_raw=0., conversion=dict(type=1, parameters=[-40., .1])
|
|
if name == 'Temperature' else None,
|
|
source=dict(name=name, path='ECU.'+name, comment='raw boundary '+name, source_type=3, bus_type=0)))
|
|
offset += bits // 8
|
|
synthetic['record_size'] = offset
|
|
synthetic['records'] = [[time] + [s[3][i] for s in specs] for i, time in enumerate([-.01, 0., .125])]
|
|
synthetic['records'][2][1] = None # PC EnvSetting retains raw values and submits every field to the DLL.
|
|
synthetic['records'][2][-1] = None
|
|
destination.with_name('synthetic.json').write_text(json.dumps(synthetic, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
|
|
def compare(reference, candidate):
|
|
with MDF(reference) as left, MDF(candidate) as right:
|
|
assert left.version == right.version
|
|
assert len(left.groups) == len(right.groups)
|
|
assert left.header.abs_time == right.header.abs_time
|
|
assert (left.header.tz_offset, left.header.daylight_save_time, left.header.time_flags) == (
|
|
right.header.tz_offset, right.header.daylight_save_time, right.header.time_flags)
|
|
assert left.header.comment == right.header.comment
|
|
assert len(left.file_history) == len(right.file_history) > 0
|
|
for a, b in zip(left.file_history, right.file_history):
|
|
x, y = ET.fromstring(a.comment), ET.fromstring(b.comment)
|
|
for key in ('TX', 'tool_id', 'tool_vendor', 'tool_version'):
|
|
assert x.findtext(key) == y.findtext(key), ('File History', key)
|
|
original = x.find("./common_properties/e[@name='Free_MDF4_Lib']")
|
|
if original is not None:
|
|
same_writer = y.find("./common_properties/e[@name='Free_MDF4_Lib']")
|
|
if same_writer is not None:
|
|
assert original.text == same_writer.text
|
|
else:
|
|
# Android must identify its actual writer, not impersonate the Windows DLL.
|
|
assert y.findtext("./common_properties/e[@name='writer']") == 'DCT Android; MDF4Writer.dll-compatible'
|
|
assert y.findtext("./common_properties/e[@name='reference_writer']") == 'Free_MDF4_Lib ' + original.text
|
|
for group_index, (a, b) in enumerate(zip(left.groups, right.groups)):
|
|
for key in ('comment', 'record_id_len'):
|
|
assert getattr(a.data_group, key) == getattr(b.data_group, key), ('DG', key)
|
|
for key in ('acq_name', 'comment', 'flags', 'record_id', 'invalidation_bytes_nr'):
|
|
assert getattr(a.channel_group, key) == getattr(b.channel_group, key), ('CG', key)
|
|
assert a.channel_group.samples_byte_nr == b.channel_group.samples_byte_nr
|
|
assert a.channel_group.cycles_nr == b.channel_group.cycles_nr
|
|
assert len(a.channels) == len(b.channels)
|
|
for index, (x, y) in enumerate(zip(a.channels, b.channels)):
|
|
cx, cy = channel(x), channel(y)
|
|
for key in ('name', 'unit', 'description', 'data_type', 'bit_count', 'bit_offset',
|
|
'byte_offset', 'conversion', 'source', 'flags', 'min_raw', 'max_raw'):
|
|
assert cx[key] == cy[key], (index, x.name, key, cx[key], cy[key])
|
|
np.testing.assert_array_equal(left.get(group=group_index, index=index, raw=True).samples,
|
|
right.get(group=group_index, index=index, raw=True).samples)
|
|
np.testing.assert_allclose(left.get(group=group_index, index=index).samples,
|
|
right.get(group=group_index, index=index).samples, rtol=0, atol=1e-12)
|
|
np.testing.assert_array_equal(left.get_master(group_index), right.get_master(group_index))
|
|
print('PASS: version, HD/FH metadata (explicit actual writer exception), epoch/timezone, DG/CG metadata, channel order/types/bits/offsets, record size, count, raw/physical values, Time, units/descriptions/SI/conversion')
|
|
print('HEADER COMMENTS', left.header.comment == right.header.comment)
|
|
print('HISTORY LEFT', [h.comment for h in left.file_history])
|
|
print('HISTORY RIGHT', [h.comment for h in right.file_history])
|
|
reference_blocks, candidate_blocks = blocks(reference), blocks(candidate)
|
|
for kind in ('##HD', '##DG', '##CG', '##CN', '##SI', '##CC', '##FH', '##MD', '##HL'):
|
|
assert reference_blocks['counts'].get(kind, 0) == candidate_blocks['counts'].get(kind, 0), kind
|
|
sizes = {g.channel_group.samples_byte_nr for g in right.groups}
|
|
for info in (reference_blocks, candidate_blocks):
|
|
assert info['counts'].get('##DT', 0) == 0, 'PC profile requires compressed data blocks'
|
|
for dz in info['compression']:
|
|
assert dz['original'] == 'DT' and dz['zip_type'] == 1
|
|
assert dz['parameter'] in sizes and dz['raw_size'] % dz['parameter'] == 0
|
|
assert sum(d['raw_size'] for d in reference_blocks['compression']) == sum(
|
|
d['raw_size'] for d in candidate_blocks['compression'])
|
|
print('PASS: semantic block counts and transposed zlib compression; TX deduplication, DL chunking, compressed bytes and actual writer may differ')
|
|
print('REFERENCE BLOCKS', json.dumps(reference_blocks))
|
|
print('CANDIDATE BLOCKS', json.dumps(candidate_blocks))
|
|
|
|
|
|
def verify_fixtures(root):
|
|
# These files are generated by the Android implementation, but decoded only by asammdf.
|
|
for name, count, width in [('chunks.mf4', 10003, 10), ('wide-pc.mf4', 400, 19764),
|
|
('empty-pc.mf4', 0, 10), ('queued-pc.mf4', 1, 10)]:
|
|
with MDF(root / name) as m:
|
|
g = m.groups[0]
|
|
assert g.channel_group.cycles_nr == count
|
|
assert g.channel_group.samples_byte_nr == width
|
|
expected_time = [-.01] if name == 'queued-pc.mf4' else np.arange(count) / 1000
|
|
np.testing.assert_array_equal(m.get_master(0), expected_time)
|
|
if count:
|
|
expected_values = [-2] if name == 'queued-pc.mf4' else np.arange(count)
|
|
np.testing.assert_array_equal(m.get(group=0, index=1, raw=True).samples, expected_values)
|
|
if name == 'wide-pc.mf4':
|
|
assert len(g.channels) == 4940
|
|
np.testing.assert_array_equal(m.get(group=0, index=4939, raw=True).samples, np.zeros(count))
|
|
print('PASS independent reader:', name, 'samples', count, 'record bytes', width)
|
|
with MDF(root / 'conversions-pc.mf4') as m:
|
|
np.testing.assert_array_equal(m.get('Rational', raw=True).samples, [2, 5])
|
|
np.testing.assert_array_equal(m.get('Table', raw=True).samples, [2, 5])
|
|
np.testing.assert_allclose(m.get('Rational').samples, [11, 38], rtol=0, atol=1e-12)
|
|
np.testing.assert_allclose(m.get('Table').samples, [20, 50], rtol=0, atol=1e-12)
|
|
assert m.groups[0].channels[1].source.path == 'ECU.Rational'
|
|
assert m.groups[0].channels[1].unit == 'degC'
|
|
print('PASS independent reader: A2L factory, rational and interpolated CC, raw/physical values, original unit/SI')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('action', choices=['export', 'compare', 'fixtures'])
|
|
parser.add_argument('source', type=Path)
|
|
parser.add_argument('target', type=Path, nargs='?')
|
|
args = parser.parse_args()
|
|
if args.action == 'fixtures':
|
|
verify_fixtures(args.source)
|
|
else:
|
|
if args.target is None:
|
|
parser.error('export/compare requires target')
|
|
(export if args.action == 'export' else compare)(args.source, args.target)
|
|
|