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