Parsing¶

Chromstream presently supports parsing of the folling types of files:

Chromatograms¶

• .txt files exported from chromeleon
• .ch files (Agilent)
• .d directories containing multiple .ch files
• .dx files (Agilent)
• .h5 files written by ChromStream

If your file formats varies, you can create your own chromatogram object a shown below.

Logfiles¶

• Logfiles generated by Labview (limited support, preferebly add otehr data as pandas DataFrame)

Parsing single chromatograms¶

There are in general two ways of parsing a chromatogram. 1) parsing it directly to a chromatogram object 2) Adding a Chromatogram to a experiment or channel object. You can also construct your own chromatogram object.

For agilent .d directories and .dx files, the simple parsing commands return lists, as these formats can contain multiple chromatograms.

Parsing to an object¶

In [1]:

# Chromeleon
from pathlib import Path
import chromstream as cs
import pandas as pd
chrompaths = Path("../..") /"tests" / "testdata" / "chroms" 
chromeleon_test = chrompaths / "format_6.txt"
chromatogram_1 = cs.parse_chromatogram(chromeleon_test)

# Agilent
agilent_test_path = chrompaths / "agilent.d" / "FID1A.ch"
chromatogram_2 = cs.parse_chromatogram(agilent_test_path)

# if you know the specific format, you can also use the more specific parser
chromatogram_1 = cs.parse_chromatogram_txt(chromeleon_test) #chromeleon .txt format
chromatogram_2 = cs.parse_agilent_ch(agilent_test_path) # Agilent .ch format

# custom chromatogram
mychromatogram = cs.Chromatogram(
        data=pd.DataFrame({"time": [0, 1, 2], "signal": [10, 20, 15]}), # must be a dataframe with "time" and "signal" columns
        injection_time=pd.Timestamp("2024-01-01 12:00:00"), # pandas Timestamp for injection time
        metadata={"time_unit": "s", "signal_unit": "mV"}, # any dict
        channel="FID",
        path=None
    )

# Agilent .d directory
agilent_d_path = chrompaths / "agilent.d" # returns a list of chromatograms from the .d directory
chromlist_1 = cs.parse_agilent_dot_d(agilent_d_path)

# Agilent .dx file
agilent_dx_path = chrompaths / "test_dx.dx"
chromlist_2 = cs.parse_agilent_dx(agilent_dx_path)

# Chromeleon from pathlib import Path import chromstream as cs import pandas as pd chrompaths = Path("../..") /"tests" / "testdata" / "chroms" chromeleon_test = chrompaths / "format_6.txt" chromatogram_1 = cs.parse_chromatogram(chromeleon_test) # Agilent agilent_test_path = chrompaths / "agilent.d" / "FID1A.ch" chromatogram_2 = cs.parse_chromatogram(agilent_test_path) # if you know the specific format, you can also use the more specific parser chromatogram_1 = cs.parse_chromatogram_txt(chromeleon_test) #chromeleon .txt format chromatogram_2 = cs.parse_agilent_ch(agilent_test_path) # Agilent .ch format # custom chromatogram mychromatogram = cs.Chromatogram( data=pd.DataFrame({"time": [0, 1, 2], "signal": [10, 20, 15]}), # must be a dataframe with "time" and "signal" columns injection_time=pd.Timestamp("2024-01-01 12:00:00"), # pandas Timestamp for injection time metadata={"time_unit": "s", "signal_unit": "mV"}, # any dict channel="FID", path=None ) # Agilent .d directory agilent_d_path = chrompaths / "agilent.d" # returns a list of chromatograms from the .d directory chromlist_1 = cs.parse_agilent_dot_d(agilent_d_path) # Agilent .dx file agilent_dx_path = chrompaths / "test_dx.dx" chromlist_2 = cs.parse_agilent_dx(agilent_dx_path)

Note on Chromeleon date parsing: the Inject Time field's day/month order is read from the exporting machine's OS locale. Where the date is unambiguous (one of the two components is >12), the order is detected automatically. When it's genuinely ambiguous (both components <=12), dayfirst=True is assumed by default.

Adding to an Experiment or Channel¶

You can also start by inilizing an experiment and using a method to add the chromatogram. Note that if you want to add multiple chromatograms at the same time, you need a seperate method. The channel is inferred automatically from the metadata. When parsing to a channel, the injeciton number needs to be specified.

In [2]:

myexp = cs.Experiment(name="My Experiment")

# From a path
myexp.add_chromatogram(chromeleon_test)

# From a Chromatogram object
myexp.add_chromatogram(chromatogram_1)

# From a dx file or a .d directory (adds all chromatograms in the file/directory)
myexp.add_mult_chromatograms(agilent_d_path)

# adding to channel works analogously
mychannel = cs.ChannelChromatograms(channel="My Channel")
mychannel.add_chromatogram(1, chromatogram_1)

myexp = cs.Experiment(name="My Experiment") # From a path myexp.add_chromatogram(chromeleon_test) # From a Chromatogram object myexp.add_chromatogram(chromatogram_1) # From a dx file or a .d directory (adds all chromatograms in the file/directory) myexp.add_mult_chromatograms(agilent_d_path) # adding to channel works analogously mychannel = cs.ChannelChromatograms(channel="My Channel") mychannel.add_chromatogram(1, chromatogram_1)

Parsing a ChromStream HDF5 experiment¶

ChromStream can also read back the HDF5 experiment format written with Experiment.to_hdf5() or write_experiment_hdf5(). The parser is strict and only supports known ChromStream schemas.

In [3]:

hdf5_path = chrompaths / "test_dx.h5"
hdf5_experiment = cs.parse_experiment_hdf5(hdf5_path)

hdf5_experiment.name, hdf5_experiment.channel_names

hdf5_path = chrompaths / "test_dx.h5" hdf5_experiment = cs.parse_experiment_hdf5(hdf5_path) hdf5_experiment.name, hdf5_experiment.channel_names

Out[3]:

('Agilent DX Example', ['RID1A'])

Parsing standalone HDF5 channels and chromatograms¶

ChromStream also reads back single ChannelChromatograms and Chromatogram files written with to_hdf5(). Chromatograms saved with an injection_index can be merged into an Experiment via add_chromatogram_hdf5(), which restores their original injection numbering.

In [ ]:

# write a standalone channel and chromatogram, then read them back
mychannel.to_hdf5("my_channel.h5", overwrite=True)
mychromatogram.to_hdf5("my_chromatogram.h5", injection_index=0, overwrite=True)

channel_from_hdf5 = cs.parse_channel_hdf5("my_channel.h5")
chrom_from_hdf5 = cs.parse_chromatogram_hdf5("my_chromatogram.h5")

# reassemble chromatogram files into an experiment (restores the stored injection index)
reassembled = cs.Experiment(name="Reassembled")
reassembled.add_chromatogram_hdf5("my_chromatogram.h5")

# write a standalone channel and chromatogram, then read them back mychannel.to_hdf5("my_channel.h5", overwrite=True) mychromatogram.to_hdf5("my_chromatogram.h5", injection_index=0, overwrite=True) channel_from_hdf5 = cs.parse_channel_hdf5("my_channel.h5") chrom_from_hdf5 = cs.parse_chromatogram_hdf5("my_chromatogram.h5") # reassemble chromatogram files into an experiment (restores the stored injection index) reassembled = cs.Experiment(name="Reassembled") reassembled.add_chromatogram_hdf5("my_chromatogram.h5")