数值模式 | 如何将CSV转BUFR文件

用户11172986

发布于 2026-04-24 19:32:11

520

文章被收录于专栏：气python风雨气python风雨

数值模式 | 如何将CSV转BUFR文件

1. 前言：为什么需要CSV转BUFR？

前些日子一个读者问我怎么把csv转bufr格式，我做了一番探索找到了相关仓库。

在日常气象数据处理中，自动气象站、气候观测设备通常以CSV格式存储数据。然而，当需要将这些数据上报至国际气象组织（WMO）或与其他国家/地区进行数据交换时，就必须转换为BUFR格式。

BUFR（Binary Universal Form for the Representation of meteorological data）是WMO推荐的气象数据国际标准格式，具有以下特点：

高度压缩：二进制格式，数据体积小
自描述性：包含完整的元数据信息
全球通用：被各国气象部门广泛采用
标准化：严格遵循WMO代码手册规范

但BUFR格式的复杂性使得手动转换几乎不可能。本文介绍的csv2bufr正是WMO官方开发的Python工具，专用于解决CSV到BUFR的转换问题。

项目信息

GitHub仓库：https://github.com/World-Meteorological-Organization/csv2bufr
官方文档：https://csv2bufr.readthedocs.io
最新版本：v0.8.7（2026年1月发布）

2. csv2bufr核心功能

csv2bufr是世界气象组织（WMO）开源项目，主要特性包括：

特性	说明
双接口支持	同时提供CLI命令行和Python API
模板驱动	通过JSON配置文件定义映射关系
WIGOS支持	完整支持WIGOS站标识符体系
元数据输出	自动包含GeoJSON格式的位置信息
数据验证	内置范围校验和缺失值处理

典型应用场景：

自动气象站（AWS）数据实时上报
历史气候数据标准化归档
跨国气象数据交换格式转换
WMO全球观测系统（WIGOS）数据接入

3. 环境搭建与安装

3.1 前置依赖：ecCodes安装

csv2bufr依赖ECMWF的ecCodes库进行BUFR编解码，这是最容易踩坑的环节。

Ubuntu/Debian系统：

# 安装ecCodes
sudo apt-get install libeccodes-dev

# 验证安装
codes_info

macOS系统：

# 使用Homebrew安装
brew install eccodes

# 验证安装
codes_info

conda环境（推荐）：

# 创建独立环境
conda create -y -n csv2bufr-env python=3.9
conda activate csv2bufr-env
conda config --add channels conda-forge
conda install -c conda-forge eccodes

# 验证安装
codes_info

重要提示：

如遇到"ModuleNotFoundError: No module named 'eccodes'"错误，请确保ecCodes的Python绑定已正确安装
某些Linux发行版需要手动设置ECCODES_DEFINITION_PATH环境变量

3.2 安装csv2bufr

方式一：pip安装（推荐）

pip install csv2bufr

方式二：Docker安装（免配置）

# 拉取官方镜像
docker pull wmoim/csv2bufr:latest

# 运行转换（挂载本地目录）
docker run -v $(pwd):/data wmoim/csv2bufr \
    csv2bufr data transform /data/input.csv \
    --bufr-template /data/mapping.json \
    --output-dir /data/output

3.3 验证安装

# 查看版本和帮助信息
csv2bufr --version
csv2bufr --help

4. 快速入门：5分钟完成第一次转换

按照以下步骤快速上手：

步骤1：准备CSV数据

# 创建示例CSV数据文件
# 注意：WIGOS标识符需要拆分为4个独立列
csv_content = """wigosIdentifierSeries,wigosIssuerOfIdentifier,wigosIssueNumber,wigosLocalIdentifierCharacter,year,month,day,hour,minute,latitude,longitude,airTemperature
0,20000,0,12345,2024,1,15,12,0,39.9042,116.4074,288.65"""

with open("example-data.csv", "w") as f:
    f.write(csv_content)

print("CSV文件创建成功！")
print(csv_content)

数据要求：

使用英文逗号分隔（,）
第一行为列名（header）
时间字段需拆分为独立列（年/月/日/时/分）
温度单位为开尔文（K），如15.5°C = 288.65K
缺失值统一用None表示

步骤2：生成映射模板

# 使用BUFR描述符生成基础模板
# 301150=WIGOS标识符, 301011=日期, 301012=时间, 301021=经纬度高程, 012001=气温
csv2bufr mappings create 301150 301011 301012 301021 12001 --output my-mapping.json

步骤3：查看生成的映射模板

生成的 my-mapping.json 文件结构如下：

{
    "conformsTo": ["csv2bufr-template-v3.json"],
    "metadata": {
        "label": "",
        "description": "",
        "version": "0",
        "author": "",
        "editor": "",
        "dateCreated": "2026-03-27",
        "dateModified": "2026-03-27",
        "id": "xxx"
    },
    "inputDelayedDescriptorReplicationFactor": [],
    "inputShortDelayedDescriptorReplicationFactor": [],
    "inputExtendedDelayedDescriptorReplicationFactor": [],
    "number_header_rows": 1,
    "column_names_row": 1,
    "delimiter": ",",
    "quoting": "QUOTE_NONE",
    "quotechar": "",
    "header": [
        {"eccodes_key": "edition", "value": "const:4"},
        {"eccodes_key": "masterTableNumber", "value": "const:0"},
        {"eccodes_key": "dataCategory", "value": "const:0"},
        {"eccodes_key": "typicalYear", "value": "data:year"},
        ...
    ],
    "data": [
        {"eccodes_key": "#1#wigosIdentifierSeries", "value": "data:wigosIdentifierSeries", ...},
        {"eccodes_key": "#1#airTemperature", "value": "data:airTemperature", ...}
    ]
}

关键字段说明：

eccodes_key：ecCodes库中对应的BUFR元素名称
value：数据来源，data:列名或const:常量值
valid_min/valid_max：数据有效范围（数值型）
scale：缩放因子，用于单位转换
offset：偏移量，用于温度转换

步骤4：执行转换

# 创建输出目录
mkdir -p output

# 执行CSV到BUFR的转换
csv2bufr data transform example-data.csv --bufr-template my-mapping.json --output-dir ./output

查看生成的BUFR文件：

ls -la ./output/

验证BUFR文件内容：

bufr_dump -p ./output/*.bufr

5. Python API实战

5.1 基础用法

import json
from csv2bufr import transform

# 读取CSV数据
with open("example-data.csv") as fh:
    data = fh.read()

# 加载映射模板
with open("my-mapping.json") as fh:
    mapping = json.load(fh)

# 执行转换（注意：transform返回的是生成器，需要转换为列表）
result = list(transform(data, mapping))

# 处理结果
for item in result:
    # 注意：wigos_station_identifier在properties中
    wsid = item["_meta"]["properties"]["wigos_station_identifier"]
    timestamp = item["_meta"]["properties"]["datetime"]
    output_file = f"{wsid}_{timestamp.strftime('%Y%m%dT%H%MZ')}.bufr"
    
    with open(output_file, "wb") as fh:
        fh.write(item["bufr4"])
    
    print(f"已生成: {output_file}")

5.2 批量处理多站数据

import os
import glob
import json
from csv2bufr import transform

# 加载映射模板
with open("my-mapping.json") as fh:
    mapping = json.load(fh)

# 批量处理所有CSV文件
output_dir = "./bufr_output"
os.makedirs(output_dir, exist_ok=True)

# 模拟创建多个测试文件
for i in range(3):
    csv_data = f"""wigosIdentifierSeries,wigosIssuerOfIdentifier,wigosIssueNumber,wigosLocalIdentifierCharacter,year,month,day,hour,minute,latitude,longitude,airTemperature
0,20000,0,1234{i},2024,1,1{i},12,{i*10},39.{i+9042},116.4074,{288+i}.0"""
    with open(f"station_{i}.csv", "w") as f:
        f.write(csv_data)

# 处理文件
for csv_file in glob.glob("./station_*.csv"):
    with open(csv_file) as fh:
        data = fh.read()
    
    result = list(transform(data, mapping))
    
    for item in result:
        wsid = item["_meta"]["properties"]["wigos_station_identifier"]
        timestamp = item["_meta"]["properties"]["datetime"]
        filename = f"{wsid}_{timestamp.strftime('%Y%m%dT%H%MZ')}.bufr"
        filepath = os.path.join(output_dir, filename)
        
        with open(filepath, "wb") as fh:
            fh.write(item["bufr4"])
        
        print(f"[{os.path.basename(csv_file)}] -> {filename}")

5.3 集成到数据处理流水线

import pandas as pd
import json
from csv2bufr import transform

class CSV2BUFRProcessor:
    """CSV转BUFR处理器"""
    
    def __init__(self, mapping_file):
        with open(mapping_file) as fh:
            self.mapping = json.load(fh)
    
    def process_dataframe(self, df):
        """处理pandas DataFrame"""
        # DataFrame转CSV字符串
        csv_data = df.to_csv(index=False)
        
        # 转换
        result = list(transform(csv_data, self.mapping))
        
        bufr_files = []
        for item in result:
            wsid = item["_meta"]["properties"]["wigos_station_identifier"]
            timestamp = item["_meta"]["properties"]["datetime"]
            filename = f"{wsid}_{timestamp.strftime('%Y%m%dT%H%MZ')}.bufr"
            
            with open(filename, "wb") as fh:
                fh.write(item["bufr4"])
            
            bufr_files.append(filename)
        
        return bufr_files

# 使用示例
processor = CSV2BUFRProcessor("my-mapping.json")

# 创建示例DataFrame（注意列名要与映射模板匹配）
df = pd.DataFrame({
    "wigosIdentifierSeries": [0],
    "wigosIssuerOfIdentifier": [20000],
    "wigosIssueNumber": [0],
    "wigosLocalIdentifierCharacter": ["99999"],
    "year": [2024],
    "month": [1],
    "day": [15],
    "hour": [12],
    "minute": [0],
    "latitude": [39.9042],
    "longitude": [116.4074],
    "airTemperature": [288.65]  # 开尔文温度
})

bufr_files = processor.process_dataframe(df)
print(f"成功生成 {len(bufr_files)} 个BUFR文件")

6. 映射模板详解

6.1 BUFR描述符速查

BUFR描述符定义了数据结构，常用描述符：

描述符	含义	说明
301150	WIGOS标识符	站点唯一标识
301011	日期（年月日）	观测日期
301012	时间（时分秒）	观测时间
301021	经纬度高程	地理位置
302001	气压数据	海平面气压等
302031	温度数据	气温、露点等
302035	湿度数据	相对湿度
302047	风数据	风速风向
302053	降水数据	降水量
302056	能见度	能见度观测

6.2 映射文件结构

{
    "conformsTo": ["csv2bufr-template-v3.json"],
    "metadata": {
        "label": "模板名称",
        "description": "模板描述",
        "version": "1.0.0",
        "author": "作者",
        "editor": "编辑者",
        "dateCreated": "2024-01-15",
        "dateModified": "2024-01-15",
        "id": "uuid"
    },
    "inputShortDelayedDescriptorReplicationFactor": [],
    "inputDelayedDescriptorReplicationFactor": [],
    "inputExtendedDelayedDescriptorReplicationFactor": [],
    "number_header_rows": 1,
    "column_names_row": 1,
    "delimiter": ",",
    "quoting": "QUOTE_NONE",
    "quotechar": "",
    "header": [
        {"eccodes_key": "edition", "value": "const:4"},
        {"eccodes_key": "masterTableNumber", "value": "const:0"},
        {"eccodes_key": "dataCategory", "value": "const:0"}
    ],
    "data": [
        {
            "eccodes_key": "#1#year",
            "value": "data:year",
            "valid_min": "const:2020",
            "valid_max": "const:2030"
        }
    ]
}

关键字段说明：

conformsTo: 指定模板格式版本，必须为 ["csv2bufr-template-v3.json"]
metadata: 必须包含 label, description, version, author, editor, dateCreated, dateModified, id
inputShortDelayedDescriptorReplicationFactor/inputDelayedDescriptorReplicationFactor/inputExtendedDelayedDescriptorReplicationFactor: 复制因子，不能为空列表
eccodes_key：ecCodes库中对应的BUFR元素名称
value：数据来源，data:列名或const:常量值
valid_min/valid_max：数据有效范围（数值型）
scale：缩放因子，用于单位转换
offset：偏移量，用于温度转换（如摄氏度转开尔文）

6.3 温度转换示例

CSV中的摄氏温度需转换为BUFR的开尔文温度：

# 温度转换示例
temp_c = 15.5  # 摄氏度
temp_k = temp_c + 273.15  # 开尔文

print(f"CSV温度: {temp_c} 摄氏度")
print(f"BUFR温度: {temp_k} 开尔文")

注意：csv2bufr v0.8.7 使用 301021 描述符时，温度直接存储为开尔文，不需要额外的 scale 和 offset。

7. 内置模板与扩展

7.1 官方模板仓库

csv2bufr-templates提供多种现成模板，可通过以下方式获取：

# 克隆模板仓库
git clone https://github.com/wmo-im/csv2bufr-templates.git

可用模板：

模板	适用场景	关键描述符
aws-template.json	自动气象站	301150, 302001, 302031
daycli-template.json	日气候数据	301150, 302061
climat-template.json	月气候数据	301150, 302062
synop-template.json	SYNOP地面观测	301090系列

8. 常见问题与解决方案

问题1：ModuleNotFoundError: No module named 'eccodes'

原因：ecCodes Python绑定未正确安装

解决：

检查ecCodes安装：运行codes_info命令
如已安装但Python找不到，设置环境变量：export PYTHONPATH="/usr/local/lib/python3.9/site-packages:$PYTHONPATH"
使用conda安装：conda install -c conda-forge eccodes

问题2：ValueError: Invalid descriptor

原因：BUFR描述符拼写错误或不存在

解决：

确认描述符为6位数字
参考WMO代码手册验证描述符有效性
常用描述符：301150 (WIGOS), 301011 (日期), 301012 (时间), 301021 (经纬度), 12001 (气温)

问题3：数据验证失败

原因：CSV数据超出valid_min/valid_max范围

解决：

检查数据质量，识别异常值
调整valid_min/valid_max设置
添加数据清洗步骤

问题4：WIGOS标识符格式错误

原因：WIGOS ID格式不符合规范

解决：

标准格式：X-YYYYY-Z-ZZZZZ（4部分，用-分隔）
X：系列号（0-14）
Y：发布机构
Z：发布序号
本地标识符
重要：CSV中需要将WIGOS标识符拆分为4个独立列：wigosIdentifierSeries, wigosIssuerOfIdentifier, wigosIssueNumber, wigosLocalIdentifierCharacter

问题5：transform()返回生成器导致无法获取长度

原因：transform()函数返回的是生成器对象

解决：

# 错误写法
result = transform(data, mapping)
print(len(result))  # TypeError!

# 正确写法
result = list(transform(data, mapping))
print(len(result))  # 正常工作

问题6：KeyError: 'wigos_station_identifier'

原因：元数据结构中 wigos_station_identifier 的位置不正确

解决：

# 错误写法
wsid = item["_meta"]["wigos_station_identifier"]

# 正确写法
wsid = item["_meta"]["properties"]["wigos_station_identifier"]

9. 小结与选型建议

csv2bufr作为WMO官方工具，在气象数据标准化方面具有显著优势：

核心优势：

官方背书，严格遵循WMO标准
双接口设计，CLI快速上手，API深度集成
模板灵活，支持各种BUFR数据结构
持续维护，活跃的开源社区

适用场景：

需要将本地观测数据上报WMO
参与国际气象数据交换项目
建设符合WIGOS标准的数据系统
学习BUFR格式和编码规范

学习资源：

WMO代码手册：https://library.wmo.int/doc_num.php?explnum_id=10722
WIGOS指南：https://library.wmo.int/doc_num.php?explnum_id=10962
BUFR验证工具：https://apps.ecmwf.int/codes/bufr/validator/

10. 参考资料

csv2bufr GitHub仓库：https://github.com/World-Meteorological-Organization/csv2bufr
csv2bufr官方文档：https://csv2bufr.readthedocs.io
csv2bufr-templates仓库：https://github.com/wmo-im/csv2bufr-templates
WMO Manual on Codes, Volume I.2：https://library.wmo.int/doc_num.php?explnum_id=10722
WIGOS Station Identifier Guide：https://library.wmo.int/doc_num.php?explnum_id=10962
ecCodes文档：https://confluence.ecmwf.int/display/ECC/ecCodes+home
ECMWF BUFR Validator：https://apps.ecmwf.int/codes/bufr/validator/

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2026-03-27，如有侵权请联系 cloudcommunity@tencent.com 删除

data