是否有Python模块来处理带节的csv文件?

w1e3prcc  于 2023-04-09  发布在  Python
关注(0)|答案(3)|浏览(116)

我看到越来越多的csv文件包含多个部分,每个部分都包含自己的表。例如,来自10XGenomics的这个文件:

[gene-expression]
reference,/path/to/transcriptome

[libraries]
fastq_id,fastqs,feature_types
gex1,/path/to/fastqs,Gene Expression
mux1,/path/to/fastqs,Multiplexing Capture

[samples]
sample_id,cmo_ids
sample1,CMO301
sample2,CMO303

有时候,节标题甚至嵌入在它们自己的行中,例如。

[gene-expression],,
reference,/path/to/transcriptome,
[libraries],,
fastq_id,fastqs,feature_types
gex1,/path/to/fastqs,Gene Expression
mux1,/path/to/fastqs,Multiplexing Capture
[samples],,
sample_id,cmo_ids,
sample1,CMO301,
sample2,CMO303,

有没有一个Python模块可以直接处理这种划分?我找不到如何用Pandas或csv模块来做这件事。例如,从上面的两个例子中,我希望得到一个每节一个条目的字典,然后是每个节的列表列表。
有些部分有头,如果这也能被处理就好了,例如类似于csv.DictReader
虽然编写一个可以解析这个特定示例的解决方案并不是特别困难,但是产生一些在一般情况下工作的东西要困难得多,例如,解析一个简单的csv文件很容易用split完成,但是csv模块是400多行Python,还有更多的C行,所以我真正在这里寻找的是一个模块来处理这个问题。
PS:this question是相关的,但不幸的是,答案并没有解决csv解析器的问题

0md85ypi

0md85ypi1#

你可以使用configparser模块来读取你的文件:

from configparser import ConfigParser
import io
import pandas as pd

cfg = ConfigParser(allow_no_value=True)
cfg.optionxform = str
cfg.read('data.csv')

dfs = {}
for section in cfg.sections():
    buf = io.StringIO()
    buf.writelines('\n'.join(row.rstrip(',') for row in cfg[section]))
    buf.seek(0)
    dfs[section] = pd.read_csv(buf)

输出:

>>> dfs['gene-expression']
Empty DataFrame
Columns: [reference, /path/to/transcriptome]
Index: []

>>> dfs['libraries']
  fastq_id           fastqs         feature_types
0     gex1  /path/to/fastqs       Gene Expression
1     mux1  /path/to/fastqs  Multiplexing Capture

>>> dfs['samples']
  sample_id cmo_ids
0   sample1  CMO301
1   sample2  CMO303

现在你也可以只想提取一个部分:

cfg = ConfigParser(allow_no_value=True)
cfg.optionxform = str
cfg.read('data.csv')

def read_data(section):
    buf = io.StringIO()
    buf.writelines('\n'.join(row.rstrip(',') for row in cfg[section]))
    buf.seek(0)
    return pd.read_csv(buf)

df = read_data('samples')

输出:

>>> df
  sample_id cmo_ids
0   sample1  CMO301
1   sample2  CMO303
1u4esq0p

1u4esq0p2#

下面是使用pandas处理 * 两种格式 * 的建议:

import pandas as pd
    
df = (pd.read_fwf("input.txt", header=None, names=["data"])
        .assign(section=lambda x: x["data"].str.extract("\[(.*)\]").ffill())
)

d_dfs = { # type hint: Dict[str, pd.DataFrame]
    k: (g.iloc[1:,0].str.split(",", expand=True)
           .pipe(lambda df_: 
              df_.rename(columns=df_.iloc[0])
                 .drop(df_.index[0])))
    for k, g in df.groupby('section')
}

输出:

>>> print(d_dfs["libraries"])

  fastq_id           fastqs         feature_types
4     gex1  /path/to/fastqs       Gene Expression
5     mux1  /path/to/fastqs  Multiplexing Capture

>>> print(d_dfs["samples"])

  sample_id cmo_ids
8   sample1  CMO301
9   sample2  CMO303
wd2eg0qa

wd2eg0qa3#

对于csv标准库模块,使用itertools.groupby()来处理解析文件是相当容易的:

import csv
import io
import itertools

s = """\
[gene-expression]
reference,/path/to/transcriptome

[libraries]
fastq_id,fastqs,feature_types
gex1,/path/to/fastqs,Gene Expression
mux1,/path/to/fastqs,Multiplexing Capture

[samples]
sample_id,cmo_ids
sample1,CMO301
sample2,CMO303
"""

def is_header(l):
    return l.strip().startswith("[") and l.strip().endswith("]")

f = io.StringIO(s)
grouped = itertools.groupby(f, is_header)

try: 
    while True:
        _, header = next(grouped)
        header = list(csv.reader(header))[-1][0]
        _, section = next(grouped)
        section = list(csv.reader(section))

        print(header)
        print(section)
        
except StopIteration:
    pass

如果你有Python 3.10或更高版本,你可以将itertools.groupby()itertools.pairwise()结合使用,这会使这变得更简单:

s = """\
[gene-expression]
reference,/path/to/transcriptome

[libraries]
fastq_id,fastqs,feature_types
gex1,/path/to/fastqs,Gene Expression
mux1,/path/to/fastqs,Multiplexing Capture

[samples]
sample_id,cmo_ids
sample1,CMO301
sample2,CMO303
"""

import csv
import io
import itertools

f = io.StringIO(s)

def is_header(l):
    return l.strip().startswith("[") and l.strip().endswith("]")
    
grouped = itertools.groupby(f, is_header)
paired = itertools.pairwise(list(g) for k, g in grouped)

data = {header[-1].strip("[]\n"): list(csv.reader(section)) for header, section in paired}
print(data)

相关问题