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.
94 lines
5.0 KiB
94 lines
5.0 KiB
"""Independent MDF 4.10 check of artifacts exported by Mf4RecordingRegressionTest.
|
|
|
|
Set DCT_RECORDING_FIXTURES_DIR when running the JUnit test, then run this script
|
|
with that directory. Requires asammdf; it is not an Android/build dependency.
|
|
"""
|
|
import argparse
|
|
import csv
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from asammdf import MDF
|
|
|
|
|
|
def verify(directory):
|
|
checks = []
|
|
with MDF(directory / 'relative.mf4') as recording:
|
|
assert recording.version == '4.10'
|
|
assert len(recording.groups) == 1
|
|
assert recording.groups[0].channel_group.cycles_nr == 2
|
|
temperature = recording.get('온도')
|
|
assert temperature.unit == '°C'
|
|
np.testing.assert_array_equal(temperature.samples, [-12.5, 3.75])
|
|
np.testing.assert_allclose(temperature.timestamps, [0.125, 0.5], atol=1e-6)
|
|
np.testing.assert_array_equal(recording.get('Speed').samples, [87.25, 88.0])
|
|
assert np.isnan(recording.get('Missing').samples).all()
|
|
checks.append('relative time, names, units, negative/fractional values, NaN')
|
|
for name, channels in [('empty.mf4', 2), ('time_only.mf4', 1)]:
|
|
with MDF(directory / name) as recording:
|
|
group = recording.groups[0]
|
|
assert group.channel_group.cycles_nr == 0
|
|
assert len(group.channels) == channels
|
|
assert len(recording.get_master(0)) == 0
|
|
checks.append(name)
|
|
with MDF(directory / 'wide.mf4') as recording:
|
|
assert len(recording.groups[0].channels) == 4940
|
|
assert recording.groups[0].channel_group.cycles_nr == 1
|
|
assert recording.get('DBC_0').samples.tolist() == [123.5]
|
|
assert np.isnan(recording.get('DBC_4938').samples).all()
|
|
np.testing.assert_allclose(recording.get_master(0), [0.25], atol=1e-6)
|
|
checks.append('4939-signal DBC width and missing tail')
|
|
with MDF(directory / 'restart.mf4') as recording:
|
|
assert recording.get('Signal').samples.tolist() == [2.0]
|
|
np.testing.assert_allclose(recording.get_master(0), [0.25], atol=1e-6)
|
|
checks.append('restart resets origin, rejects nonfinite timestamps')
|
|
with (directory / 'paired.csv').open(encoding='utf-8', newline='') as stream:
|
|
rows = list(csv.reader(stream))[7:]
|
|
with MDF(directory / 'paired.mf4') as recording:
|
|
signal = recording.get('Signal')
|
|
assert len(rows) == len(signal.samples) == 2
|
|
assert [int(row[1]) for row in rows] == [0, 1]
|
|
np.testing.assert_array_equal([float(row[2]) for row in rows], signal.samples)
|
|
assert [row[0][-4:] for row in rows] == ['.125', '.500']
|
|
np.testing.assert_allclose(signal.timestamps, [0.125, 0.5], atol=1e-6)
|
|
checks.append('CSV/MF4 rows, values and ingress timestamps match')
|
|
shared_origins = []
|
|
for source, first_offset in [('mdb', -0.010), ('dbc', -0.009), ('formula', -0.010)]:
|
|
with (directory / f'queued_{source}.csv').open(encoding='utf-8', newline='') as stream:
|
|
rows = list(csv.reader(stream))[7:]
|
|
with MDF(directory / f'queued_{source}.mf4') as recording:
|
|
signal = recording.get('Signal')
|
|
origin = recording.start_time.timestamp()
|
|
shared_origins.append(origin)
|
|
assert len(rows) == len(signal.samples) == 3
|
|
assert [int(row[1]) for row in rows] == [0, 1, 2]
|
|
np.testing.assert_array_equal(signal.samples, [0., 1., 2.])
|
|
np.testing.assert_array_equal([float(row[2]) for row in rows], signal.samples)
|
|
np.testing.assert_allclose(signal.timestamps, [first_offset, 0., 0.125], rtol=0, atol=1e-6)
|
|
np.testing.assert_allclose(origin, 1790923490.915, rtol=0, atol=1e-6)
|
|
for row, offset in zip(rows, signal.timestamps):
|
|
# UTC/local HH:mm:ss differences do not change the offset within a minute.
|
|
csv_second = float(row[0].rsplit(':', 1)[1])
|
|
mf4_second = (origin + offset) % 60
|
|
np.testing.assert_allclose(csv_second, mf4_second, rtol=0, atol=0.001)
|
|
checks.append(f'{source}: queued ingress preserved, CSV/MF4 absolute time and values match')
|
|
assert len(set(shared_origins)) == 1
|
|
with MDF(directory / 'queued_midnight.mf4') as recording:
|
|
signal = recording.get('Signal')
|
|
origin = recording.start_time.timestamp()
|
|
np.testing.assert_allclose(signal.timestamps, [-0.010], rtol=0, atol=1e-6)
|
|
np.testing.assert_array_equal(signal.samples, [7.])
|
|
absolute = origin + signal.timestamps[0]
|
|
np.testing.assert_allclose(absolute, 1790985599.995, rtol=0, atol=1e-6)
|
|
assert datetime.fromtimestamp(absolute, recording.start_time.tzinfo).date() < recording.start_time.date()
|
|
checks.append('queued ingress crossing UTC midnight preserves the previous date')
|
|
for check in checks:
|
|
print('PASS:', check)
|
|
print('PASS: ten MF4 files opened by independent reader')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('directory', type=Path)
|
|
verify(parser.parse_args().directory)
|
|
|