如何用Python从反应堆通量文件提取指定位置数据并导出CSV

1. 解析目标位置标识 #

先把输入的位置（如b27）拆成列索引和行索引，方便后续定位数据：

def parse_location(location_str):
    col_char = location_str[0].lower()
    row_num = int(location_str[1:])
    # 字母转列索引：a对应0，y对应24
    col_idx = ord(col_char) - ord('a')
    # 行号转文件内的行索引：行号从-2开始，所以索引=行号+2
    row_idx = row_num + 2
    return col_idx, row_idx

2. 遍历目标文件夹的所有文件 #

用glob批量获取inputs目录下的文本文件：

import os
import glob

input_dir = "./inputs"
file_paths = glob.glob(os.path.join(input_dir, "*.txt"))

3. 从文件名提取core标识 #

假设core标识是文件名中第一个下划线前的部分（如171203fl_someflux.txt里的171203fl），直接拆分提取：

def get_core_id(file_path):
    filename = os.path.basename(file_path)
    core_id = filename.split("_")[0]
    return core_id

4. 解析单个文件的通量数据 #

按文件结构（通量类型名称+网格数据块）提取目标位置的数值：

def extract_flux_data(file_path, col_idx, row_idx):
    results = []
    current_flux_type = None
    data_rows = []

    with open(file_path, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                # 空行分隔不同通量类型，处理当前类型数据
                if current_flux_type and data_rows:
                    if 0 <= row_idx < len(data_rows):
                        row_data = data_rows[row_idx].split()
                        if 0 <= col_idx < len(row_data):
                            results.append((current_flux_type, float(row_data[col_idx])))
                current_flux_type = None
                data_rows = []
                continue
            # 判断是否是通量类型名称（假设无数字的行是类型描述）
            if not any(char.isdigit() for char in line):
                current_flux_type = line
            else:
                data_rows.append(line)
        # 处理文件末尾的最后一个通量类型
        if current_flux_type and data_rows:
            if 0 <= row_idx < len(data_rows):
                row_data = data_rows[row_idx].split()
                if 0 <= col_idx < len(row_data):
                    results.append((current_flux_type, float(row_data[col_idx])))
    return results

5. 导出结果到CSV #

用Python内置的csv模块整理并写入结果：

import csv

def export_to_csv(all_results, output_file="flux_results.csv"):
    with open(output_file, 'w', newline='') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=['core_id', 'flux_type', 'location', 'flux_value'])
        writer.writeheader()
        for core_id, flux_type, location, value in all_results:
            writer.writerow({
                'core_id': core_id,
                'flux_type': flux_type,
                'location': location,
                'flux_value': value
            })

6. 整合所有逻辑的主函数 #

把上述步骤串起来，接收用户输入并执行：

def main(target_location):
    col_idx, row_idx = parse_location(target_location)
    all_results = []

    for file_path in file_paths:
        core_id = get_core_id(file_path)
        flux_data = extract_flux_data(file_path, col_idx, row_idx)
        for flux_type, value in flux_data:
            all_results.append((core_id, flux_type, target_location, value))
    
    export_to_csv(all_results)
    print(f"结果已导出到flux_results.csv")

if __name__ == "__main__":
    target_loc = input("请输入目标位置（如b27）：")
    main(target_loc)

关键调整点 #

• 如果通量类型的识别规则和示例不同（比如有特定前缀），修改extract_flux_data里的类型判断逻辑
• 核对文件内的行号顺序，确保row_idx的转换和实际文件匹配
• 若文件编码非默认，在open时添加encoding参数（如encoding='latin-1'）

内容的提问来源于stack exchange，提问作者Joe Blough