Collecting and Processing INMET-BDMEP Climate Data

气候数据在多个领域发挥着至关重要的作用，有助于影响农业、城市规划和自然资源管理等领域的研究和预测。

国家气象研究所 (INMET) 每月在其网站上提供气象数据库 (BDMEP)。该数据库包含分布在巴西各地的数百个测量站收集的一系列历史气候信息。在BDMEP中，您可以找到有关降雨量、温度、空气湿度和风速的详细数据。

每小时更新一次，数据量相当大，为详细分析和明智决策提供了丰富的基础。

在这篇文章中，我将展示如何从 INMET-BDMEP 收集和处理气候数据。我们将收集 INMET 网站上提供的原始数据文件，然后处理这些数据以方便分析。

1. 所需的Python包

要实现上述目标，您只需要安装三个软件包：

• httpx 发出 HTTP 请求
• Pandas 用于读取和处理数据
• tqdm 在程序下载或读取文件时在终端中显示友好的进度条

要安装必要的软件包，请在终端中运行以下命令：

pip install httpx pandas tqdm

例如，如果您使用诗歌虚拟环境 (venv)，请使用以下命令：

poetry add httpx pandas tqdm

2. 文件收集

2.1 文件 URL 模式

BDMEP 数据文件的地址遵循非常简单的模式。图案如下：

https://portal.inmet.gov.br/uploads/dadoshistoricos/{year}.zip

唯一改变的部分是文件名，它只是数据的参考年份。每月，最近（当前）年份的文件会替换为更新的数据。

这使得创建代码来自动收集所有可用年份的数据文件变得容易。

事实上，历史系列始于 2000 年。

2.2 采集策略

为了从 INMET-BDMEP 收集数据文件，我们将使用 httpx 库发出 HTTP 请求，并使用 tqdm 库在终端中显示友好的进度条。

首先，让我们导入必要的包：

import datetime as dt
from pathlib import Path

import httpx
from tqdm import tqdm

我们已经确定了 INMET-BDMEP 数据文件的 URL 模式。现在让我们创建一个接受年份作为参数并返回该年份文件的 URL 的函数。

def build_url(year):
    return f"https://portal.inmet.gov.br/uploads/dadoshistoricos/{year}.zip"

要检查 URL 文件是否已更新，我们可以使用 HTTP 请求返回的标头中存在的信息。在配置良好的服务器上，我们可以使用 HEAD 方法仅请求此标头。在这种情况下，服务器配置良好，我们可以使用此方法。

对 HEAD 请求的响应将采用以下格式：

Mon, 01 Sep 2024 00:01:00 GMT

为了解析这个日期/时间，我在 Python 中创建了以下函数，它接受 字符串 并返回一个日期时间对象：

def parse_last_modified(last_modified: str) -> dt.datetime:
    return dt.datetime.strptime(
        last_modified,
        "%a, %d %b %Y %H:%M:%S %Z"
    )

因此，我们可以使用字符串插值（f-strings），将上次修改的日期/时间包含在我们要下载的文件的名称中：

def build_local_filename(year: int, last_modified: dt.datetime) -> str:
    return f"inmet-bdmep_{year}_{last_modified:%Y%m%d}.zip"

这样，您可以轻松检查我们的本地文件系统中是否已存在包含最新数据的文件。如果文件已经存在，则可以终止程序；否则，我们必须继续收集文件，向服务器发出请求。

下面的 download_year 函数下载特定年份的文件。如果目标目录中已存在该文件，该函数将简单地返回而不执行任何操作。

请注意我们如何使用 tqdm 在下载文件时在终端中显示友好的进度条。

def download_year(
    year: int,
    destdirpath: Path,
    blocksize: int = 2048,
) -> None:

    if not destdirpath.exists():
        destdirpath.mkdir(parents=True)

    url = build_url(year)

    headers = httpx.head(url).headers
    last_modified = parse_last_modified(headers["Last-Modified"])
    file_size = int(headers.get("Content-Length", 0))

    destfilename = build_local_filename(year, last_modified)
    destfilepath = destdirpath / destfilename
    if destfilepath.exists():
        return

    with httpx.stream("GET", url) as r:
        pb = tqdm(
            desc=f"{year}",
            dynamic_ncols=True,
            leave=True,
            total=file_size,
            unit="iB",
            unit_scale=True,
        )
        with open(destfilepath, "wb") as f:
            for data in r.iter_bytes(blocksize):
                f.write(data)
                pb.update(len(data))
        pb.close()

2.3 文件收集

现在我们已经具备了所有必要的功能，我们可以收集 INMET-BDMEP 数据文件了。

使用 for 循环，我们可以下载所有可用年份的文件。下面的代码正是这样做的。从2000年开始至今年。

destdirpath = Path("data")
for year in range(2000, dt.datetime.now().year + 1):
    download_year(year, destdirpath)

3.数据的读取和处理

下载 INMET-BDMEP 原始数据文件后，我们现在可以读取和处理数据。

让我们导入必要的包：

import csv
import datetime as dt
import io
import re
import zipfile
from pathlib import Path

import numpy as np
import pandas as pd
from tqdm import tqdm

3.1 文件结构

在 INMET 提供的 ZIP 文件中，我们找到几个 CSV 文件，每个气象站一个。

Porém, nas primeiras linhas desses arquivos CSV encontramos informações sobre a estação, como a região, a unidade federativa, o nome da estação, o código WMO, as coordenadas geográficas (latitude e longitude), a altitude e a data de fundação. Vamos extrair essas informações para usar como metadados.

3.2 Leitura dos dados com pandas

A leitura dos arquivos será feita em duas partes: primeiro, será feita a leitura dos metadados das estações meteorológicas; depois, será feita a leitura dos dados históricos propriamente ditos.

3.2.1 Metadados

Para extrair os metadados nas primeiras 8 linhas do arquivo CSV vamos usar o pacote embutido csv do Python.

Para entender a função a seguir é necessário ter um conhecimento um pouco mais avançado de como funciona handlers de arquivos (open), iteradores (next) e expressões regulares (re.match).

def read_metadata(filepath: Path | zipfile.ZipExtFile) -> dict[str, str]:
    if isinstance(filepath, zipfile.ZipExtFile):
        f = io.TextIOWrapper(filepath, encoding="latin-1")
    else:
        f = open(filepath, "r", encoding="latin-1")
    reader = csv.reader(f, delimiter=";")
    _, regiao = next(reader)
    _, uf = next(reader)
    _, estacao = next(reader)
    _, codigo_wmo = next(reader)
    _, latitude = next(reader)
    try:
        latitude = float(latitude.replace(",", "."))
    except:
        latitude = np.nan
    _, longitude = next(reader)
    try:
        longitude = float(longitude.replace(",", "."))
    except:
        longitude = np.nan
    _, altitude = next(reader)
    try:
        altitude = float(altitude.replace(",", "."))
    except:
        altitude = np.nan
    _, data_fundacao = next(reader)
    if re.match("[0-9]{4}-[0-9]{2}-[0-9]{2}", data_fundacao):
        data_fundacao = dt.datetime.strptime(
            data_fundacao,
            "%Y-%m-%d",
        )
    elif re.match("[0-9]{2}/[0-9]{2}/[0-9]{2}", data_fundacao):
        data_fundacao = dt.datetime.strptime(
            data_fundacao,
            "%d/%m/%y",
        )
    f.close()
    return {
        "regiao": regiao,
        "uf": uf,
        "estacao": estacao,
        "codigo_wmo": codigo_wmo,
        "latitude": latitude,
        "longitude": longitude,
        "altitude": altitude,
        "data_fundacao": data_fundacao,
    }

Em resumo, a função read_metadata definida acima lê as primeiras oito linhas do arquivo, processa os dados e retorna um dicionário com as informações extraídas.

3.2.2 Dados históricos

Aqui, finalmente, veremos como fazer a leitura do arquivo CSV. Na verdade é bastante simples. Basta usar a função read_csv do Pandas com os argumentos certos.

A seguir está exposto a chamada da função com os argumentos que eu determinei para a correta leitura do arquivo.

pd.read_csv(
    "arquivo.csv",
    sep=";",
    decimal=",",
    na_values="-9999",
    encoding="latin-1",
    skiprows=8,
    usecols=range(19),
)

Primeiro é preciso dizer que o caractere separador das colunas é o ponto-e-vírgula (;), o separador de número decimal é a vírgula (,) e o encoding é latin-1, muito comum no Brasil.

Também é preciso dizer para pular as 8 primeiras linhas do arquivo (skiprows=8), que contém os metadados da estação), e usar apenas as 19 primeiras colunas (usecols=range(19)).

Por fim, vamos considerar o valor -9999 como sendo nulo (na_values="-9999").

3.3 Tratamento dos dados

Os nomes das colunas dos arquivos CSV do INMET-BDMEP são bem descritivos, mas um pouco longos. E os nomes não são consistentes entre os arquivos e ao longo do tempo. Vamos renomear as colunas para padronizar os nomes e facilitar a manipulação dos dados.

A seguinte função será usada para renomear as colunas usando expressões regulares (RegEx):

def columns_renamer(name: str) -> str:
    name = name.lower()
    if re.match(r"data", name):
        return "data"
    if re.match(r"hora", name):
        return "hora"
    if re.match(r"precipita[çc][ãa]o", name):
        return "precipitacao"
    if re.match(r"press[ãa]o atmosf[ée]rica ao n[íi]vel", name):
        return "pressao_atmosferica"
    if re.match(r"press[ãa]o atmosf[ée]rica m[áa]x", name):
        return "pressao_atmosferica_maxima"
    if re.match(r"press[ãa]o atmosf[ée]rica m[íi]n", name):
        return "pressao_atmosferica_minima"
    if re.match(r"radia[çc][ãa]o", name):
        return "radiacao"
    if re.match(r"temperatura do ar", name):
        return "temperatura_ar"
    if re.match(r"temperatura do ponto de orvalho", name):
        return "temperatura_orvalho"
    if re.match(r"temperatura m[áa]x", name):
        return "temperatura_maxima"
    if re.match(r"temperatura m[íi]n", name):
        return "temperatura_minima"
    if re.match(r"temperatura orvalho m[áa]x", name):
        return "temperatura_orvalho_maxima"
    if re.match(r"temperatura orvalho m[íi]n", name):
        return "temperatura_orvalho_minima"
    if re.match(r"umidade rel\. m[áa]x", name):
        return "umidade_relativa_maxima"
    if re.match(r"umidade rel\. m[íi]n", name):
        return "umidade_relativa_minima"
    if re.match(r"umidade relativa do ar", name):
        return "umidade_relativa"
    if re.match(r"vento, dire[çc][ãa]o", name):
        return "vento_direcao"
    if re.match(r"vento, rajada", name):
        return "vento_rajada"
    if re.match(r"vento, velocidade", name):
        return "vento_velocidade"

Agora que temos os nomes das colunas padronizados, vamos tratar a data/hora. Os arquivos CSV do INMET-BDMEP têm duas colunas separadas para data e hora. Isso é inconveniente, pois é mais prático ter uma única coluna de data/hora. Além disso existem inconsistências nos horários, que às vezes têm minutos e às vezes não.

As três funções a seguir serão usadas para criar uma única coluna de data/hora:

def convert_dates(dates: pd.Series) -> pd.DataFrame:
    dates = dates.str.replace("/", "-")
    return dates


def convert_hours(hours: pd.Series) -> pd.DataFrame:

    def fix_hour_string(hour: str) -> str:
        if re.match(r"^\d{2}\:\d{2}$", hour):
            return hour
        else:
            return hour[:2] + ":00"

    hours = hours.apply(fix_hour_string)
    return hours


def fix_data_hora(d: pd.DataFrame) -> pd.DataFrame:
    d = d.assign(
        data_hora=pd.to_datetime(
            convert_dates(d["data"]) + " " + convert_hours(d["hora"]),
            format="%Y-%m-%d %H:%M",
        ),
    )
    d = d.drop(columns=["data", "hora"])
    return d

Existe um problema com os dados do INMET-BDMEP que é a presença de linhas vazias. Vamos remover essas linhas vazias para evitar problemas futuros. O código a seguir faz isso:

# Remove empty rows
empty_columns = [
    "precipitacao",
    "pressao_atmosferica",
    "pressao_atmosferica_maxima",
    "pressao_atmosferica_minima",
    "radiacao",
    "temperatura_ar",
    "temperatura_orvalho",
    "temperatura_maxima",
    "temperatura_minima",
    "temperatura_orvalho_maxima",
    "temperatura_orvalho_minima",
    "umidade_relativa_maxima",
    "umidade_relativa_minima",
    "umidade_relativa",
    "vento_direcao",
    "vento_rajada",
    "vento_velocidade",
]
empty_rows = data[empty_columns].isnull().all(axis=1)
data = data.loc[~empty_rows]

Problema resolvido! (•̀ᴗ•́)و ̑̑

3.4 Encapsulando em funções

Para finalizar esta seção vamos encapsular o código de leitura e tratamento em funções.

Primeiro uma função para a leitura do arquivo CSV contino no arquivo comprimido.

def read_data(filepath: Path) -> pd.DataFrame:
    d = pd.read_csv(
        filepath,
        sep=";",
        decimal=",",
        na_values="-9999",
        encoding="latin-1",
        skiprows=8,
        usecols=range(19),
    )
    d = d.rename(columns=columns_renamer)

    # Remove empty rows
    empty_columns = [
        "precipitacao",
        "pressao_atmosferica",
        "pressao_atmosferica_maxima",
        "pressao_atmosferica_minima",
        "radiacao",
        "temperatura_ar",
        "temperatura_orvalho",
        "temperatura_maxima",
        "temperatura_minima",
        "temperatura_orvalho_maxima",
        "temperatura_orvalho_minima",
        "umidade_relativa_maxima",
        "umidade_relativa_minima",
        "umidade_relativa",
        "vento_direcao",
        "vento_rajada",
        "vento_velocidade",
    ]
    empty_rows = d[empty_columns].isnull().all(axis=1)
    d = d.loc[~empty_rows]

    d = fix_data_hora(d)

    return d

Tem um problema com a função acima. Ela não lida com arquivos ZIP.

Criamos, então, a função read_zipfile para a leitura de todos os arquivos contidos no arquivo ZIP. Essa função itera sobre todos os arquivos CSV no arquivo zipado, faz a leitura usando a função read_data e os metadados usando a função read_metadata, e depois junta os dados e os metadados em um único DataFrame.

def read_zipfile(filepath: Path) -> pd.DataFrame:
    data = pd.DataFrame()
    with zipfile.ZipFile(filepath) as z:
        files = [zf for zf in z.infolist() if not zf.is_dir()]
        for zf in tqdm(files):
            d = read_data(z.open(zf.filename))
            meta = read_metadata(z.open(zf.filename))
            d = d.assign(**meta)
            data = pd.concat((data, d), ignore_index=True)
    return data

No final, basta usar essa última função definida (read_zipfile) para fazer a leitura dos arquivos ZIP baixados do site do INMET. (. ❛ ᴗ ❛.)

df = reader.read_zipfile("inmet-bdmep_2023_20240102.zip")
# 100%|████████████████████████████████████████████████████████████████████████████████| 567/567 [01:46<00:00,  5.32it/s]
df
#         precipitacao  pressao_atmosferica  pressao_atmosferica_maxima  ...  longitude  altitude  data_fundacao
# 0                0.0                887.7                       887.7  ... -47.925833   1160.96     2000-05-07
# 1                0.0                888.1                       888.1  ... -47.925833   1160.96     2000-05-07
# 2                0.0                887.8                       888.1  ... -47.925833   1160.96     2000-05-07
# 3                0.0                887.8                       887.9  ... -47.925833   1160.96     2000-05-07
# 4                0.0                887.6                       887.9  ... -47.925833   1160.96     2000-05-07
# ...              ...                  ...                         ...  ...        ...       ...            ...
# 342078           0.0                902.6                       903.0  ... -51.215833    963.00     2019-02-15
# 342079           0.0                902.2                       902.7  ... -51.215833    963.00     2019-02-15
# 342080           0.2                902.3                       902.3  ... -51.215833    963.00     2019-02-15
# 342081           0.0                903.3                       903.3  ... -51.215833    963.00     2019-02-15
# 342082           0.0                903.8                       903.8  ... -51.215833    963.00     2019-02-15

# [342083 rows x 26 columns]
df.to_csv("inmet-bdmep_2023.csv", index=False)  # Salvando o DataFrame em um arquivo CSV

4. Gráfico de exemplo

Para finalizar, nada mais satisfatório do que fazer gráficos com os dados que coletamos e tratamos. ヾ(≧▽≦*)o

Nessa parte uso o R com o pacote tidyverse para fazer um gráfico combinando a temperatura horária e a média diária em São Paulo.

library(tidyverse)

dados <- read_csv("inmet-bdmep_2023.csv")

print(names(dados))
#  [1] "precipitacao"               "pressao_atmosferica"
#  [3] "pressao_atmosferica_maxima" "pressao_atmosferica_minima"
#  [5] "radiacao"                   "temperatura_ar"
#  [7] "temperatura_orvalho"        "temperatura_maxima"
#  [9] "temperatura_minima"         "temperatura_orvalho_maxima"
# [11] "temperatura_orvalho_minima" "umidade_relativa_maxima"
# [13] "umidade_relativa_minima"    "umidade_relativa"
# [15] "vento_direcao"              "vento_rajada"
# [17] "vento_velocidade"           "data_hora"
# [19] "regiao"                     "uf"
# [21] "estacao"                    "codigo_wmo"
# [23] "latitude"                   "longitude"
# [25] "altitude"                   "data_fundacao"

print(unique(dados$regiao))
# [1] "CO" "N"  "NE" "SE" "S"

print(unique(dados$uf))
#  [1] "DF" "GO" "MS" "MT" "AC" "AM" "AP" "AL" "BA" "CE" "MA" "PB" "PE" "PI" "RN"
# [16] "SE" "PA" "RO" "RR" "TO" "ES" "MG" "RJ" "SP" "PR" "RS" "SC"

dados_sp <- dados |> filter(uf == "SP")


# Temperatura horária em São Paulo
dados_sp_h <- dados_sp |>
  group_by(data_hora) |>
  summarise(
    temperatura_ar = mean(temperatura_ar, na.rm = TRUE),
  )


# Temperatura média diária em São Paulo
dados_sp_d <- dados_sp |>
  group_by(data = floor_date(data_hora, "day")) |>
  summarise(
    temperatura_ar = mean(temperatura_ar, na.rm = TRUE),
  )


# Gráfico combinando temperatura horária e média diária em São Paulo
dados_sp_h |>
  ggplot(aes(x = data_hora, y = temperatura_ar)) +
  geom_line(
    alpha = 0.5,
    aes(
      color = "Temperatura horária"
    )
  ) +
  geom_line(
    data = dados_sp_d,
    aes(
      x = data,
      y = temperatura_ar,
      color = "Temperatura média diária"
    ),
    linewidth = 1
  ) +
  labs(
    x = "Data",
    y = "Temperatura (°C)",
    title = "Temperatura horária e média diária em São Paulo",
    color = "Variável"
  ) +
  theme_minimal() +
  theme(legend.position = "top")
ggsave("temperatura_sp.png", width = 16, height = 8, dpi = 300)

Temperatura horária e média diária em São Paulo em 2023

5. Conclusão

Neste texto mostrei como coletar e tratar os dados climáticos do INMET-BDMEP. Os dados coletados são muito úteis para estudos e previsões nas mais variadas áreas. Com os dados tratados, é possível fazer análises e gráficos como o que mostrei no final.

Espero que tenha gostado do texto e que tenha sido útil para você.

I created a Python package with the functions I showed in this text. The package is available in my Git repository. If you want, you can download the package and use the functions in your own code.

Git repository: https://github.com/dankkom/inmet-bdmep-data

(～￣▽￣)～

The above is the detailed content of Collecting and Processing INMET-BDMEP Climate Data. For more information, please follow other related articles on the PHP Chinese website!