422 lines
11 KiB
Python
422 lines
11 KiB
Python
import io
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
|
|
"""Parser de dados
|
|
|
|
A dataframe retornada tera multiplas linhas referentes ao mesmo evento
|
|
visto que se esta a guardar por linha cada estacao que registou o evento em questa
|
|
logo cada linha tem sempre a mesma informacao duplicada que se encontra no preambulo
|
|
para cada estacao
|
|
|
|
"""
|
|
|
|
|
|
# --- variáveis globais ---
|
|
DIST_IND = {"L": "Local", "R": "Regional", "D": "Distante"}
|
|
TYPE = {"Q": "Quake", "V": "Volcanic", "U": "Unknown", "E": "Explosion"}
|
|
|
|
|
|
# --- funções auxiliares ---
|
|
def is_blank(_str: str) -> bool:
|
|
"""Verifica se uma string tem ou nao conteudo
|
|
|
|
Args:
|
|
_str (str): str a verificar se esta vazia
|
|
|
|
Returns:
|
|
bool: True se str tem conteudo, False caso contrario
|
|
"""
|
|
return len(_str.strip(" ")) == 0
|
|
|
|
|
|
def parse_flt(value: str) -> float | None:
|
|
"""Formata str como float
|
|
|
|
Args:
|
|
value (str): nro em string para ser formatado
|
|
|
|
Returns:
|
|
float | None: Retorna um float se bem sucedido, None se excepcao
|
|
"""
|
|
try:
|
|
return float(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def parse_int(value: str) -> int | None:
|
|
"""Formata str como int
|
|
|
|
Args:
|
|
value (str): nro em string para ser formatado
|
|
|
|
Returns:
|
|
int | None: Retorna um int se bem sucedido, None se excepcao
|
|
"""
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def into_dataframe(data: dict[str, Any]) -> pd.DataFrame:
|
|
"""Transforma uma dict numa DataFrame
|
|
|
|
Args:
|
|
data (dict[str, Any]): [description]
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame
|
|
"""
|
|
if len(data) == 0:
|
|
return pd.DataFrame()
|
|
aux = {k: [] for k in data.keys()}
|
|
for k, v in data.items():
|
|
aux[k].append(v)
|
|
|
|
return pd.DataFrame(data=aux)
|
|
|
|
|
|
def _concat(preamble: dict[str, Any], df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Junta o preambulo, uma dict, na DataFrame
|
|
|
|
Args:
|
|
preamble (dict[str, Any]): Preambulo do evento a inserir
|
|
df (pd.DataFrame): DataFrame com eventos
|
|
|
|
Returns:
|
|
[type]: Nova DataFrame com o preambulo adicionado
|
|
"""
|
|
for k, v in preamble.items():
|
|
df.insert(len(df.columns) - 1, k, [v for _ in range(len(df))])
|
|
|
|
return df
|
|
|
|
|
|
# --- principal ---
|
|
def parse(fname: str) -> pd.DataFrame:
|
|
"""Faz o parse de todos os eventos no ficheiro.
|
|
|
|
A funcao separa em eventos singulares, e transforma cada evento numa DataFrame,
|
|
que sera concatenada com uma DataFrame que contem todos os eventos existentes
|
|
|
|
Args:
|
|
fname (str): nome do ficheiro que contem os dados
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame com os eventos formatados
|
|
"""
|
|
fp = open(fname)
|
|
data = [line for line in fp.read().split("\n")]
|
|
chunks = boundaries(data)
|
|
df = pd.DataFrame()
|
|
for c in chunks:
|
|
a = parse_chunk(data[c[0] : c[1]])
|
|
aux = pd.concat([df, a], axis=0, ignore_index=True)
|
|
df = aux
|
|
fp.close()
|
|
return df
|
|
|
|
|
|
def boundaries(data: list[str]) -> list[tuple[int, int]]:
|
|
"""Procura e guarda a posicao de cada evento.
|
|
|
|
O ficheiro tem os eventos separados por uma linha em branco
|
|
|
|
Args:
|
|
data (list[str]): lista dos dados
|
|
|
|
Returns:
|
|
list[tuple[int, int]]: lista com tuples dos indices de inicio
|
|
e fim de cada evento
|
|
"""
|
|
boundaries = []
|
|
eventStart = None
|
|
for idx, line in enumerate(data):
|
|
if eventStart is None:
|
|
if not is_blank(line):
|
|
eventStart = idx
|
|
else:
|
|
if is_blank(line):
|
|
boundaries.append((eventStart, idx))
|
|
eventStart = None
|
|
return boundaries
|
|
|
|
|
|
def parse_chunk(chunk_lines: list[str]) -> pd.DataFrame:
|
|
"""Parse de um evento no formato Nordic, separando num preambulo, e nas estacoes
|
|
Ambos sao enviados para as suas funcoes privadas para serem parsed
|
|
|
|
Args:
|
|
chunk_lines (list[str]): lista de str do evento, como slice da lista de todos os eventos
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame do evento
|
|
"""
|
|
separatorIdx = None
|
|
for idx, line in enumerate(chunk_lines):
|
|
if line[-1] == "7":
|
|
separatorIdx = idx
|
|
break
|
|
preambleRet = _parse_preamble(chunk_lines[:separatorIdx])
|
|
phaseRet = _parse_type_7(chunk_lines[separatorIdx:])
|
|
|
|
return _concat(preambleRet, phaseRet)
|
|
|
|
|
|
def _parse_preamble(hLines: list[str]) -> dict[str, Any]:
|
|
"""Transforma o preambulo numa dict com os valores que precisamos
|
|
|
|
Verifica cada linha e separa dentro de uma dict, com a chave sendo o tipo de linha
|
|
|
|
Args:
|
|
hLines (list[str]): slice da lista com apenas o preambulo
|
|
|
|
Returns:
|
|
dict[str, Any]: dict com os valores necessarios
|
|
"""
|
|
lineTypes = defaultdict(list)
|
|
|
|
for line in hLines:
|
|
match line[-1]:
|
|
case "1":
|
|
lineTypes[1].append(line)
|
|
case "3":
|
|
lineTypes[3].append(line)
|
|
case "6":
|
|
lineTypes[6].append(line)
|
|
case "E":
|
|
lineTypes["E"].append(line)
|
|
case "I":
|
|
lineTypes["I"].append(line)
|
|
case _:
|
|
pass
|
|
|
|
headerDict = dict()
|
|
for k, v in lineTypes.items():
|
|
if len(v) != 0:
|
|
# FUNCS[k] retorna o handle de cada funcao para cada tipo de linha
|
|
headerDict.update(FUNCS[k](v))
|
|
return headerDict
|
|
|
|
|
|
def _parse_type_1(data: list[str]) -> dict[str, Any]:
|
|
"""Transforma linhas tipo 1 (data, hora, latitude, longitude, profundidade
|
|
agencia, magnitudes e tipos e nro de estacoes que registaram o evento)
|
|
|
|
Args:
|
|
data (list[str]): lista de linhas tipo 1
|
|
|
|
Returns:
|
|
dict[str, Any]: dict com os valores necessarios
|
|
"""
|
|
y = int(data[0][1:5])
|
|
mo = int(data[0][6:8])
|
|
d = int(data[0][8:10])
|
|
h = int(data[0][11:13])
|
|
m = int(data[0][13:15])
|
|
s = int(data[0][16:18])
|
|
mil = int(data[0][19]) * 10**5
|
|
dt = datetime(y, mo, d, h, m, s, mil)
|
|
|
|
dist_ind = DIST_IND[data[0][21]]
|
|
ev_type = TYPE[data[0][22]]
|
|
lat = float(data[0][23:30])
|
|
long = float(data[0][30:38])
|
|
depth = float(data[0][38:43])
|
|
no_stat = int(data[0][48:51])
|
|
|
|
hypo = {
|
|
# NOTE: ANTES ERA UMA STRING, AGORA E O OBJECTO DATETIME
|
|
"Data": dt,
|
|
"Distancia": dist_ind,
|
|
"Tipo Evento": ev_type,
|
|
"Latitude": lat,
|
|
"Longitude": long,
|
|
"Profundidade": depth,
|
|
"Estacoes": no_stat,
|
|
"Magnitudes": [],
|
|
}
|
|
for line in data:
|
|
hypo["Magnitudes"] = hypo["Magnitudes"] + _parse_mag(line)
|
|
|
|
return hypo
|
|
|
|
|
|
def _parse_mag(line: str) -> list[dict[str, Any]]:
|
|
"""Transforma nos varios tipos de magnitudes
|
|
|
|
Args:
|
|
line (str): str das linhas tipo 1
|
|
|
|
Returns:
|
|
list[dict[str, Any]]: dict com os valores das magnitudes e o seu tipo
|
|
"""
|
|
magnitudes = []
|
|
base = 55
|
|
while base < 79:
|
|
m = line[base : base + 4]
|
|
mt = line[base + 4]
|
|
if not is_blank(m):
|
|
magnitudes.append({"Magnitude": m, "Tipo": mt})
|
|
base += 8
|
|
return magnitudes
|
|
|
|
|
|
def _parse_type_3(data: list[str]) -> dict[str, Any]:
|
|
"""Transforma linhas tipo 3 (observacoes)
|
|
|
|
Args:
|
|
data (list[str]): lista com linhas tipo 3
|
|
|
|
Returns:
|
|
dict[str, Any]: dict com valores necessarios
|
|
"""
|
|
comments = {}
|
|
for line in data:
|
|
if (
|
|
line.startswith(" SENTIDO")
|
|
or line.startswith(" REGIAO")
|
|
or line.startswith(" PUB")
|
|
):
|
|
chave, valor = line[:-2].strip().split(": ", maxsplit=1)
|
|
|
|
if chave == "REGIAO":
|
|
parts = valor.split(",")
|
|
comments["Regiao"] = parts[0].strip()
|
|
for p in parts[1:]:
|
|
p = p.strip()
|
|
if "SZ" in p:
|
|
comments["SZ"] = p
|
|
elif "VZ" in p:
|
|
comments["VZ"] = p
|
|
elif chave == "PUB":
|
|
comments["Pub"] = valor.strip()
|
|
else:
|
|
comments[chave.capitalize()] = valor.split(",")[0]
|
|
|
|
return comments
|
|
|
|
|
|
def _parse_type_6(data: list[str]) -> dict[str, list[str]]:
|
|
"""Transforma linhas tipo 6 (nome de onda)
|
|
|
|
[description]
|
|
|
|
Args:
|
|
data (list[str]): lista de linhas tipo 6
|
|
|
|
Returns:
|
|
dict[str, list[str]]: lista de nomes dos ficheiros das ondas
|
|
"""
|
|
waves = []
|
|
for line in data:
|
|
waves.append(line.strip().split(" ")[0])
|
|
return {"Onda": waves}
|
|
|
|
|
|
def _parse_type_7(data: list[str]) -> pd.DataFrame:
|
|
"""Transforma linhas tipo 7 (estacoes)
|
|
|
|
Args:
|
|
data (list[str]): linhas tipo 7
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame com as informacoes de cada estacao
|
|
"""
|
|
aux = io.StringIO("\n".join(data))
|
|
dados = pd.read_fwf(
|
|
aux,
|
|
colspecs=[
|
|
(1, 5),
|
|
(6, 8),
|
|
(10, 15),
|
|
(18, 20),
|
|
(20, 22),
|
|
(23, 28),
|
|
(34, 38),
|
|
(71, 75),
|
|
],
|
|
)
|
|
dados.rename(
|
|
columns={
|
|
"STAT": "Estacao",
|
|
"SP": "Componente",
|
|
"PHASW": "Tipo Onda",
|
|
"HR": "Hora",
|
|
"MM": "Min",
|
|
"SECON": "Seg",
|
|
"AMPL": "Amplitude",
|
|
" DIST": "Distancia Epicentro",
|
|
},
|
|
inplace=True,
|
|
)
|
|
return dados
|
|
|
|
|
|
def _parse_type_e(data: list[str]) -> dict[str, Any]:
|
|
"""Transformar linhas tipo E (erros)
|
|
|
|
Args:
|
|
data (list[str]): linhas tipo E
|
|
|
|
Returns:
|
|
dict[str, Any]: dict com os valores necessarios
|
|
"""
|
|
error = {
|
|
"Gap": int(data[0][5:8]),
|
|
"Origin": float(data[0][14:20]),
|
|
"Error_lat": float(data[0][24:30]),
|
|
"Error_long": float(data[0][32:38]),
|
|
"Error_depth": float(data[0][38:43]),
|
|
"Cov_xy": float(data[0][43:55]),
|
|
"Cov_xz": float(data[0][55:67]),
|
|
"Cov_yz": float(data[0][67:79]),
|
|
}
|
|
return error
|
|
|
|
|
|
def _parse_type_i(data: list[str]) -> dict[str, int]:
|
|
"""Transforma linhas tipo I(ID do evento)
|
|
|
|
Args:
|
|
data (list[str]): linhas tipo I
|
|
|
|
Returns:
|
|
dict[str, int]: dict com o valor do ID
|
|
"""
|
|
aux = data[0]
|
|
return {"ID": int(aux[60:74])}
|
|
|
|
|
|
FUNCS = {
|
|
1: _parse_type_1,
|
|
3: _parse_type_3,
|
|
6: _parse_type_6,
|
|
"E": _parse_type_e,
|
|
"I": _parse_type_i,
|
|
}
|
|
|
|
|
|
# -- Deprecated
|
|
|
|
|
|
def validate_station_numbers(expected: int, stationsDF: pd.DataFrame) -> bool:
|
|
"""[summary]
|
|
|
|
[description]
|
|
|
|
Args:
|
|
expected (int): [description]
|
|
stationsDF (pd.DataFrame): [description]
|
|
|
|
Returns:
|
|
bool: [description]
|
|
"""
|
|
uniqueStations = stationsDF["Estacao"].nunique()
|
|
return expected == uniqueStations
|