feat: Implementado formato JSON personalizado
This commit is contained in:
76
utils/utils.py
Normal file
76
utils/utils.py
Normal file
@@ -0,0 +1,76 @@
|
||||
#! /usr/bin/env python
|
||||
# pyright: basic
|
||||
|
||||
from datetime import time
|
||||
import json
|
||||
from math import modf
|
||||
from typing import Any
|
||||
|
||||
from numpy import nan
|
||||
import pandas as pd
|
||||
|
||||
|
||||
def save_as_json(info: dict[str, Any]) -> bool:
|
||||
with open("test.json", "w") as fp:
|
||||
json.dump(info, fp)
|
||||
|
||||
return True
|
||||
|
||||
# TODO: passar os nomes das colunas, para não haver problemas no futuro, caso se altere os nomes da dataframe
|
||||
def create_dict_struct(df: pd.DataFrame, event_cols, station_cols) -> dict[str, Any]:
|
||||
# get all events by their id
|
||||
uniqueIds = df["ID"].unique()
|
||||
|
||||
allEvents = {}
|
||||
|
||||
for id in uniqueIds:
|
||||
filteredDf = df.loc[df["ID"] == id]
|
||||
first_row = filteredDf.head(1)
|
||||
allEvents[int(id)] = create_event_info(first_row)
|
||||
allEvents[int(id)].update(create_stations_info_1(filteredDf))
|
||||
|
||||
return allEvents
|
||||
|
||||
|
||||
def create_event_info(info: pd.DataFrame) -> dict[str, Any]:
|
||||
return {"DataHora": info.iloc[0]["Data"], "Lat": float(info.iloc[0]["Lat"]), "Long": float(info.iloc[0]["Long"]),
|
||||
"Profundidade": float(info.iloc[0]["Prof"]), "Tipo Evento": info.iloc[0]["Tipo Ev"],
|
||||
"Magnitude": create_mag_info(info.iloc[0]["Magnitudes"]), "Regiao": info.iloc[0]["Regiao"],
|
||||
"Sentido": info.iloc[0]["Sentido"]}
|
||||
|
||||
|
||||
def create_stations_info_1(info: pd.DataFrame) -> dict[str, Any]:
|
||||
stationsDict = {}
|
||||
for idx in range(len(info)):
|
||||
aux = info.iloc[idx]
|
||||
|
||||
micro, sec = tuple(map(int, modf(aux["Seg"])))
|
||||
hms = time(hour=aux["Hora"],minute=aux["Min"], second=sec, microsecond=micro).strftime("%H:%M:%S.%f")
|
||||
station = {"Componente": aux["Componente"], "Hora": hms, "Distancia": float(aux["DIS"])}
|
||||
|
||||
if type(aux["Tipo Onda"]) != float:
|
||||
station.update({"Tipo Onda": aux["Tipo Onda"]})
|
||||
if aux["Tipo Onda"] == "IAML":
|
||||
station.update({"Amplitude": float(aux["Amplitude"])})
|
||||
|
||||
|
||||
if aux["Estacao"] not in stationsDict.keys():
|
||||
stationsDict[aux["Estacao"]] = [station]
|
||||
else:
|
||||
stationsDict[aux["Estacao"]].append(station)
|
||||
return {"Estacoes": stationsDict}
|
||||
|
||||
|
||||
def create_mag_info(magnitudes):
|
||||
mags = {}
|
||||
for value in magnitudes:
|
||||
mags[value["Tipo"]] = value["Magnitude"]
|
||||
return mags
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import parser
|
||||
|
||||
df = parser.parse("dados.txt")
|
||||
a = create_dict_struct(df, None, None)
|
||||
save_as_json(a)
|
||||
Reference in New Issue
Block a user