TDMS-Dateien speichern

General Usage
1

Does a package exist to write data in NI compatible TDMS format?
Here is my code, I am intersted to recieve your feedback:

using PythonCall
#: --- include Python-Package to write TDMS-File:
nptdms = pyimport("nptdms")

"""
    PythonTDMSWriter

A Wrapper around nptdms.TdmsWriter.
IMPORTANT: mutable struct, because fields (is_open, buffers) must be read/write.

"""
mutable struct PythonTDMSWriter
    filepath::String
    writer::Py
    buffer_size::Int
    buffers::Dict{UInt8, Dict{String, Vector}}
    schemas::Dict{UInt8, Dict{String, Dict{String, String}}}
    is_open::Bool
    
    function PythonTDMSWriter(filepath::String; buffer_size::Int=100)
        writer_obj = nptdms.TdmsWriter(filepath)
        buffers = Dict{UInt8, Dict{String, Vector}}()
        schemas = Dict{UInt8, Dict{String, Dict{String, String}}}()
        
        # new() funktioniert auch in mutable structs
        new(filepath, writer_obj, buffer_size, buffers, schemas, false)
    end
end

function open!(w::PythonTDMSWriter)
    if !w.is_open
        w.writer.__enter__()
        w.is_open = true  # ✅ Jetzt erlaubt!
        @info "TDMS Writer opened: $(w.filepath)"
    end
end

function add_data!(w::PythonTDMSWriter, struct_id::UInt8, 
                   field_names::Vector{String}, 
                   field_data::Vector, 
                   units::Vector{String} = String[],
                   descriptions::Vector{String} = String[])
    
    # Safety-Check: field_data must be a vector of vectors.
    if !(field_data isa Vector) || !all(x -> x isa Vector, field_data)
        error("field_data must be a vector of vectors.")
    end

    if !w.is_open
        open!(w)
    end
    
    if !haskey(w.buffers, struct_id)
        w.buffers[struct_id] = Dict{String, Vector}()
        w.schemas[struct_id] = Dict{String, Dict{String, String}}()
        
        for name in field_names
            w.buffers[struct_id][name] = Vector{Any}()
        end
        
        for (i, name) in enumerate(field_names)
            props = Dict{String, String}()
            if i <= length(units) && !isempty(units[i])
                props["unit"] = units[i]
            end
            if i <= length(descriptions) && !isempty(descriptions[i])
                props["description"] = descriptions[i]
            end
            w.schemas[struct_id][name] = props
        end
    end
    
    #: Write data into buffer.
    if length(field_names) != length(field_data)
        error("Number of field names ($length(field_names)) is unequal to number of vectors of data ($length(field_data)).")
    end
    
    for (i, name) in enumerate(field_names)
        data_vec = field_data[i]
        # Pay attention that all vectors have the same length
        if i > 1 && length(data_vec) != length(field_data[1])
            error("All data vectors must have the same length.")
        end
        append!(w.buffers[struct_id][name], data_vec)
    end
    
    #: If buffer full write data
    first_field_name = first(keys(w.buffers[struct_id]))
    if length(w.buffers[struct_id][first_field_name]) >= w.buffer_size
        flush_buffer!(w, struct_id)
    end
end


function flush_buffer!(w::PythonTDMSWriter, struct_id::UInt8)
    if !haskey(w.buffers, struct_id) return end
    
    first_field_name = first(keys(w.buffers[struct_id]))
    if isempty(w.buffers[struct_id][first_field_name]) return end
    
    group_name = "Struct_$(struct_id)"
    
    channel_class = nptdms.ChannelObject
    
    channels = []
    
    for name in keys(w.buffers[struct_id])
        data = w.buffers[struct_id][name]
        props = get(w.schemas[struct_id], name, Dict{String, String}())
        
        # Create Channel-Object via API
        # Channel(Group, Name, Data, Properties)
        channel_obj = channel_class(group_name, name, data, props)
        push!(channels, channel_obj)
    end
    
    try
        # write_segment expects channel-objects
        w.writer.write_segment(channels)
        
        for key in keys(w.buffers[struct_id])
            empty!(w.buffers[struct_id][key])
        end
    catch e
        @error "Could not write segment for StructID $struct_id: $e"
        rethrow(e)
    end
end

function close!(w::PythonTDMSWriter)
    for struct_id in keys(w.buffers)
        flush_buffer!(w, struct_id)
    end
    
    if w.is_open
        try
            w.writer.__exit__(nothing, nothing, nothing)
            @info "TDMS Writer closed: $(w.filepath)"
        catch e
            @warn "Issue to close TDMS-Writers: $e"
        end
    end
end

And here a snippet how to use it:

println("1. Creating PythonTDMSWriter instance...")
try
    test_file = joinpath(tempdir(), "test_tdms_writer.tdms")
    
    # Here the constructor is called!
    writer = PythonTDMSWriter(test_file; buffer_size=50)
    
    println("✅ Instance successfully created!")
    println("   Type: $(typeof(writer))")
    println("   Instance fields:")
    
    for field in fieldnames(typeof(writer))
        val = getfield(writer, field)
        println("   - $field: $(typeof(val)) = $(val)")
    end
    
    println("\n2. Opening the writer...")
    open!(writer)  # ✅ Should work now!
    println("   Status: is_open = $(writer.is_open)")
    
    println("\n3. Adding test data...")
    field_names = ["MsgID", "Temp1", "Status1"]
    field_data  = [UInt16[1, 2, 3], Float32[20.5, 21.0, 22.5], Int16[0, 1, 0]]
    units       = ["", "°C", "Count"]
    descs       = ["Message ID", "Temperature", "Status"]
    
    add_data!(writer, UInt8(1), field_names, field_data, units, descs)
    println("   Data buffered. Buffer size: $(length(writer.buffers[1]["MsgID"]))")
    
    println("\n4. Closing the writer (writes remaining data)...")
    close!(writer)  # ✅ Should work now!
    println("   Writer closed. File exists: $(isfile(test_file))")
    
catch e
    println("❌ Error during test: $e")
    showerror(stdout, e)
end
1 Like