Read Dat File Content

Description

Example python script which loads an .adtfdat file and accesses its data using python directly.

Usage

python read_dat_file_content.py <adtfdat>

Source

# Copyright 2024 CARIAD SE.
#
#     This Source Code Form is subject to the terms of the Mozilla
#     Public License, v. 2.0. If a copy of the MPL was not distributed
#     with this file, You can obtain one at https://mozilla.org/MPL/2.0/.
#
# If it is not possible or desirable to put the notice in a particular file, then
# You may include the notice in a location (such as a LICENSE file in a
# relevant directory) where a recipient would be likely to look for such a notice.
#
# You may add additional accurate notices of copyright ownership.
 
import adtf_file
import sys
 
# a little helper to dump the content of a stream type
def dumpStreamType(stream_type, indent = "      "):
    print(indent + f'stream_meta_type: {stream_type.meta_type}')
    substreams = stream_type.substreams
    if len(substreams) == 0:
        print(indent + f'properties:')
        for name, value in stream_type.properties.items():
            print(indent + f'  {name}: {value.value}')
    else:
        print(indent + f'substreams:')
        for name, substream in substreams.items():
            print(indent + f'  {name}:')
            print(indent + f'    id: {substream.id}')
            dumpStreamType(substream.stream_type, indent + "    ")
 
# if you need any additional adtffileplugins you can load them with adtf_file.load_plugin(<filename>)
 
reader = adtf_file.create_reader(sys.argv[1])
 
print('header:')
print(f'  version: {reader.file_version}')
print(f'  item_count: {reader.item_count}')
print(f'  description: {reader.description}')
 
print('extensions:')
for extension in reader.extensions:
    print(f'  {extension.name}:')
    print(f'    stream_id: {extension.stream_id}')
    print(f'    user_id: {extension.user_id}')
    print(f'    type_id: {extension.type_id}')
    print(f'    version_id: {extension.version_id}')
    print(f'    data: {bytearray(extension).hex(" ")}')
 
print('streams:')
for stream in reader.streams:
    print(f'  {stream.name}:')
    print(f'    stream_id: {stream.stream_id}')
    print(f'    item_count: {stream.item_count}')
    print(f'    timestamp_of_first_item: {stream.timestamp_of_first_item}')
    print(f'    timestamp_of_last_item: {stream.timestamp_of_last_item}')
    print(f'    streamtype:')
    dumpStreamType(stream.initial_type)
 
print('items:')
item_index = 0
while True:
    item = reader.get_next_item()
 
    if not item:
        break
 
    print(f'  {item_index}:')
    print(f'    time_stamp: {item.time_stamp}')
    print(f'    stream_id: {item.stream_id}')
    print(f'    type: {item.type}')
 
    if item.type == adtf_file.ItemType.Sample:
        sample = item.sample
        print(f'    sample:')
        print(f'      time_stamp: {sample.time_stamp}')
        print(f'      substream_id: {sample.substream_id}')
        print(f'      flags: {sample.flags}')
        print(f'      buffer: {bytearray(sample.buffer).hex(" ")}')
 
    if item.type == adtf_file.ItemType.StreamType:
        print(f'    stream type:')
        dumpStreamType(item.stream_type)
 
    if item.type == adtf_file.ItemType.Trigger:
        print(f'    trigger')
    
    item_index = item_index + 1