Misc v2

fix: remover código morto ou desnecessário

.gitignore

@@ -1,3 +1,10 @@
+*.zip
+*.json
+*.csv
+
+stats-*.txt
+
+
 # ---> Python
 # Byte-compiled / optimized / DLL files
 __pycache__/

README.md

@@ -1,5 +1,11 @@
 # Project I - Earthquakes
 
+## Como utilizar
+Correr o ficheiro `earthquakes.py` usando `python earthquakes.py`
+Garantir que o ficheiro de dados está no mesmo diretório que o ficheiro `earthquakes.py`
+
+## Objectivos
+
 First, let's represent the data using Python's Pandas module and implement CRUD operations, including JSON's conversion. Then, let's implement some statistical operations with graphical representations using Python's Matplotlib module over data representation in Pandas data model.
 
 ## Tarefas:
@@ -8,12 +14,33 @@ First, let's represent the data using Python's Pandas module and implement CRUD
 - T2 - Implement CRUD operations through a text menu;
 - T3 - Implement statistical operations such as: average, variance, standard desviation, max, min, mode; through a text menu;
 - T4 - Convert from Pandas to JSON and save it in a text file;
-- T5 - to be continued ...
+- T5 - Calcular as seguintes estatísticas:
+  - Número de eventos por dia e por mês.
+  - Média e desvio padrão da profundidade e da magnitude por mês.
+  - Mediana, 1º quartil e  3º quartil da profundidade e da magnitude por mês.
+  - Máximo e mínimo a profundidade e da magnitude por mês.
+- T6 - Para a representação gráfica:
+  - Um gráfico de barras com o numero de eventos por dia.
+  - Um gráfico de barras com o numero de eventos por mês.
+  - Um gráfico linear com a média +/- o desvio padrão das profundidades por mês.
+  - Um gráfico linear com a média +/- a desvio padrão da magnitude L por mês.
+  - Um gráfico tipo "boxplot" com as profundidades por mês.
+  - Um gráfico tipo "boxplot" com as magnitudes L por mês.
+- T7 - Implementar os filtros de seleção de eventos para o cálculo / representação gráfica:
+  - Período temporal (Data inicial, Data final).
+  - Eventos com GAP menor que um determinado valor.
+  - Qualidade (EPI ou Todos).
+  - Zonas SZ.
+  - Zonas VZ.
+  - Limitar por Magnitudes L (mínimo, máximo).
+  - Limitar Profundidades  (mínimo, máximo).
 
 ## Prazos
 - T1 a T4 -> 10 de novembro
-- (a definir)
+- T5 a T7 -> 14 de dezembro
 
+## Apontamentos
+Dados parecem estar no formato [Nordic](https://seisan.info/v13/node259.html)
 
 ## Bibliografia
 - [Pandas lib](https://pandas.pydata.org/docs)

earthquakes.py

@@ -1,7 +0,0 @@
-#! /usr/bin/env python
-
-def main():
-    pass
-
-if __name__ == '__main__':
-    main()

ev2.py

@@ -0,0 +1,172 @@
+import logging
+import os
+import time
+
+from utilsv2 import mongo, parser, stats, utils
+from utilsv2.log import logger
+
+OS = os.name
+
+MAIN_MENU = {
+    "1": "Adicionar novos dados",
+    "2": "Aplicar filtros",
+    "3": "Estatísticas",
+    "4": "Limpar filtros",
+    "q": "Sair",
+}
+
+FILTER_MENU = {
+    "1": "Data início",
+    "2": "Magnitudes",
+    "3": "Profundidade",
+    "4": "GAP",
+    "6": "Limpar filtros",
+    "7": "Mostrar filtros",
+    "q": "Voltar ao menu principal",
+}
+
+
+def clear_screen():
+    os.system("cls" if OS == "nt" else "clear")
+
+
+def print_menu(menu: dict[str, str]):
+    clear_screen()
+    for k, v in menu.items():
+        print(f"[{k}]: {v}")
+
+
+def filter_menu(old_fiters):
+    filters = old_fiters
+    while True:
+        print_menu(FILTER_MENU)
+        usrIn = input()
+
+        match usrIn:
+            # datas
+            case "1":
+                clear_screen()
+                print(
+                    "Formato da data: YYYY-MM-DD\nInserir datas de corte, separadas por uma vírgula(,)"
+                )
+                aux = input()
+
+                d1, d2 = aux.split(",", maxsplit=1)
+
+                d1 = utils.toDateTime(d1)
+                d2 = utils.toDateTime(d2)
+                filters["DateTime"] = {}
+
+                if d1 != -1:
+                    filters["DateTime"]["$gte"] = d1
+
+                if d2 != -1:
+                    filters["DateTime"]["$lte"] = d2
+
+            # magnitudes
+            case "2":
+                clear_screen()
+                print("Inserir magnitudes de corte, separadas por uma vírgula(,)")
+                aux = input()
+
+                d1, d2 = aux.split(",", maxsplit=1)
+
+                d1 = utils.toFloat(d1)
+                d2 = utils.toFloat(d2)
+                filters["Magnitudes.L.Magnitude"] = {}
+
+                if d1 != -1:
+                    filters["Magnitudes.L.Magnitude"]["$gte"] = d1
+
+                if d2 != -1:
+                    filters["Magnitudes.L.Magnitude"]["$lte"] = d2
+
+            # Profundidades
+            case "3":
+                clear_screen()
+                print("Inserir profundidades de corte, separadas por uma vírgula(,)")
+                aux = input()
+
+                d1, d2 = aux.split(",", maxsplit=1)
+
+                d1 = utils.toFloat(d1)
+                d2 = utils.toFloat(d2)
+                filters["Depth"] = {}
+
+                if d1 != -1:
+                    filters["Depth"]["$gte"] = d1
+
+                if d2 != -1:
+                    filters["Depth"]["$lte"] = d2
+
+            # GAP
+            case "4":
+                clear_screen()
+                print("Inserir GAP")
+                aux = input()
+
+                gap = utils.toInt(aux)
+                filters["GAP"] = {}
+                if aux:
+                    filters["GAP"]["$lte"] = gap
+
+            case "6":
+                fliters = {}
+
+            case "7":
+                print(filters)
+                time.sleep(2.0)
+
+            case "q":
+                return filters
+
+
+def graph_menu():
+    pass
+
+
+def main():
+    cli = mongo.connect("mongodb://localhost:27017")
+
+    filters = {}
+
+    while True:
+        print_menu(MAIN_MENU)
+        usrIn = input()
+
+        match usrIn:
+            case "1":
+                aux = input("Ficheiro a ler:")
+                if utils.fileExists(aux):
+                    logger.info(f"Parsing the file {aux}")
+                    ev, st = parser.parse(aux)
+
+                    mongo.add_events(cli, "quakes", ev, "main")
+                    mongo.add_stations(cli, "stations", st, "main")
+
+                else:
+                    print(f"Could not open the file {aux}")
+                    logger.error(f"Could not open the file {aux}")
+                    time.sleep(2.0)
+
+            case "2":
+                filters = filter_menu(filters)
+            case "3":
+                print(filters)
+                v = mongo.filter_query(cli, "quakes", filters, "test")
+                stats.stats(v)
+                time.sleep(2.0)
+
+            case "q":
+                break
+
+    mongo.close(cli)
+
+
+if __name__ == "__main__":
+    logger = logging.getLogger(__name__)
+
+    # initialization
+    logger.info("Started")
+    main()
+    logger.info("Ended")

generator/gen-data.py

@@ -0,0 +1,113 @@
+import argparse
+import random
+import re
+import time
+from datetime import datetime
+
+PAD = re.compile(r"^0?([1-9]\d?)$")
+NORDIC_7 = (
+    " STAT SP IPHASW D HRMM SECON CODA AMPLIT PERI AZIMU VELO AIN AR TRES W  DIS CAZ7"
+)
+
+
+def generate_event():
+    ts = random.randint(946684800000, 1767225600000)
+    dt = datetime.fromtimestamp(ts / 1e3)
+    line_3 = " OBS: Dados falsos".ljust(79) + "3"
+    stat = generate_station(dt)
+
+    return "\n".join(
+        [
+            gen_line_1(dt),
+            line_3,
+            generate_line6(dt),
+            generate_line_i(dt),
+            NORDIC_7,
+            stat,
+            "\n",
+        ]
+    )
+
+
+def generate_agency(size: int = 3) -> str:
+    return "".join([chr(random.randint(65, 90)) for _ in range(size)])
+
+
+def remove_pad(v: str) -> str:
+    aux = PAD.search(v)
+    if aux:
+        return aux.group(0)
+    return ""
+
+
+def fmt_date(dt: datetime) -> str:
+    return f"{dt.year} {dt.month:2d}{dt.day:2d} {dt.hour:2d}{dt.minute:2d} {dt.second:2d}.0"
+
+
+# 1D Lerp
+def generate_number(lb: float, ub: float, precision: int = 2) -> float:
+    x = random.random()
+    return round(lb * (1.0 - x) + (ub * x), precision)
+
+
+def gen_line_1(dt: datetime) -> str:
+    lat = generate_number(-90.0, 90.0, 3)
+    long = generate_number(-180.0, 180.0, 3)
+    ev = random.choices(("E", "V", "Q"))[0]
+    depth = generate_number(0.0, 30.0, 1)
+    agency = generate_agency()
+
+    mag = generate_number(1.0, 6.0, 1)
+
+    return (
+        f" {fmt_date(dt)} L{ev}{lat: >7.3f}{long: >8.3f}{depth:>5.1f}  {agency}  1    {mag: >4.1f}L{agency}{mag: >4.1f}C{agency}".ljust(
+            79
+        )
+        + "1"
+    )
+
+
+def generate_line6(dt: datetime) -> str:
+    return f" {dt.strftime('%Y-%m-%d-%H%M-%S')}-FAKE___001".ljust(79) + "6"
+
+
+def generate_line_i(dt: datetime) -> str:
+    return " " * 57 + f"ID:{dt.strftime('%Y%m%d%H%M%S')}     I"
+
+
+def generate_station(dt: datetime) -> str:
+    st = generate_agency(4)
+
+    return "\n".join(
+        [
+            f" {st} EZ EP       {dt.hour:2d}{dt.minute:2d} {dt.second: >5.2f}".ljust(
+                80
+            ),
+            f" {st} EZ ES       {dt.hour:2d}{dt.minute:2d} {dt.second: >5.2f}".ljust(
+                80
+            ),
+        ]
+    )
+
+
+def main(argv: int):
+    fp = open("falsos.txt", "w", newline="\n")
+    for _ in range(argv):
+        aux = generate_event()
+        fp.write(aux)
+
+    fp.close()
+
+
+if __name__ == "__main__":
+    random.seed(time.time())
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "n", action="store", type=int, help="Generates n amount of events"
+    )
+
+    args = parser.parse_args()
+
+    if args.n:
+        main(args.n)

requirements.txt

@@ -0,0 +1,3 @@
+matplotlib==3.10.8
+numpy==2.4.0
+pymongo==4.15.5

utilsv2/graphs.py

@@ -0,0 +1,74 @@
+from datetime import datetime
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+from utilsv2 import stats, utils
+
+
+class Plotter:
+    def __init__(self) -> None:
+        self.x = []
+        self.y = []
+        self.figure = None
+        self.ax = None
+
+    def plot_bars(self, title: str, isDepth: bool = False):
+        self.figure, self.ax = plt.subplots()
+
+        self.ax.bar(self.x, self.y)
+        self.ax.set_title(title)
+        plt.show()
+
+    def plot_lin(self):
+        pass
+
+    def plot_box(self):
+        pass
+
+    def adjust_x(self):
+        pass
+
+    def add_x_values(self, xvalues):
+        self.x = xvalues
+
+    def add_y_values(self, yvalues):
+        self.y = yvalues
+
+    @staticmethod
+    def concat_data_day(data):
+        pass
+
+    @staticmethod
+    def concat_data_month(data):
+        x = []
+        y_vals = {"e": [], "m": [], "d": []}
+
+        currMonth: datetime = data[0]["DateTime"]
+        currMonth_str = utils.print_ym(currMonth)
+
+        x.append(currMonth_str)
+        e = 0
+        m = []
+        d = []
+        idx = 0
+        while idx <= len(data):
+            if data[idx]["DateTime"].month == currMonth.month and idx < len(data):
+                e += 1
+                m.append(data[idx]["Magnitudes"]["L"]["Magnitude"])
+                d.append(data[idx]["Depth"])
+                idx += 1
+            else:
+                y_vals["e"].append(e)
+                y_vals["m"].append(np.average(m))
+                y_vals["d"].append(np.average(d))
+
+                currMonth = data[idx]["DateTime"]
+                currMonth_str = utils.print_ym(currMonth)
+
+                x.append(currMonth_str)
+                e = 0
+                m = []
+                d = []
+
+        return x, y_vals

utilsv2/log.py

@@ -0,0 +1,10 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+logging.basicConfig(
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+    level=logging.INFO,
+    filename="ev.log",
+)

utilsv2/mongo.py

@@ -0,0 +1,91 @@
+import logging
+from typing import Any
+
+from pymongo import MongoClient
+from pymongo.collection import Collection
+from pymongo.errors import ConnectionFailure
+
+try:
+    from utilsv2.log import logger
+except ModuleNotFoundError:
+    from log import logger
+
+logger = logging.getLogger(__name__)
+
+
+def connect(uri: str) -> MongoClient:
+    try:
+        client = MongoClient(uri)
+        logger.info("Connected to the DB")
+    except ConnectionFailure as e:
+        logger.critical("Could not connect to the MongoDB")
+        raise e
+
+    return client
+
+
+def add_events(
+    client: MongoClient, collection: str, data: list[dict[str, Any]], db: str = "main"
+) -> None:
+    coll: Collection = client[db][collection]
+
+    _res = coll.insert_many(data)
+
+    if _res.acknowledged:
+        logger.info(f"Added {len(_res.inserted_ids)} events to {db}.{collection}")
+    else:
+        logger.info("Could not add events to the database.")
+
+
+def add_stations(
+    client: MongoClient, collection: str, data: list[dict[str, Any]], db: str = "main"
+) -> None:
+    coll: Collection = client[db][collection]
+
+    _res = coll.insert_many(data)
+
+    if _res.acknowledged:
+        logger.info(f"Added {len(_res.inserted_ids)} stations to {db}.{collection}")
+    else:
+        logger.info("Could not add events to the database.")
+
+
+def get_ids(collection: Collection) -> set[Any]:
+    return set(collection.distinct("ID"))
+
+
+def close(client: MongoClient) -> None:
+    client.close()
+    logger.info("Closed the DB.")
+
+
+def query_all(client: MongoClient, collection: str, db: str = "main") -> Any:
+    coll: Collection = client[db][collection]
+
+    result = coll.find({})
+
+    return list(result)
+
+
+def filter_query(
+    client: MongoClient, collection: str, filter_by: dict[str, Any], db: str = "main"
+):
+    coll: Collection = client[db][collection]
+
+    res = coll.find(
+        filter_by, {"DateTime": 1, "Magnitudes": 1, "Depth": 1, "GAP": 1}
+    ).sort({"DateTime": 1})
+
+    if not res._empty:
+        res = list(res)
+        logger.info(f"Retrieved {len(res)} elements.")
+
+    return res
+
+
+if __name__ == "__main__":
+    v = connect("mongodb://localhost:27017")
+
+    query_all(v, "quakes")
+
+    close(v)

utilsv2/nordic.py

@@ -0,0 +1,194 @@
+import json
+import logging
+from collections import defaultdict
+from datetime import datetime, time
+from typing import Any
+
+from utilsv2 import utils
+from utilsv2.log import logger
+
+logger = logging.getLogger(__name__)
+
+type evtype = dict[str, Any]
+type sttype = dict[str, Any]
+
+
+# INFO: Don't think we really need this
+class Mag:
+    def __init__(self, mag: float, type: str, agency: str):
+        self.mag = mag
+        self.type = type
+        self.agency = agency
+
+    def __str__(self):
+        return f"mag: {self.mag}, type: {self.type}, agency: {self.agency}"
+
+    def toJSON(self):
+        json.dumps({"Magnitude": self.mag, "Agency": self.agency})
+
+
+def parse_event(event: list[str]) -> evtype:
+    # nordic must always have the first line a type 1 line
+    # but a type 1 line can have the id ommited if it's the first line
+    # if event[0][-1] != "1" or event[0][-1] != " ":
+    #     return {}
+
+    toParse: dict[str, list[str]] = defaultdict(list)
+
+    for line in event:
+        toParse[line[-1]].append(line)
+
+    _ret = {}
+
+    for k, v in toParse.items():
+        match k:
+            case "1":
+                aux = parse_type_1(v)
+                if aux:
+                    _ret.update(aux)
+            case "3":
+                _ret.update(parse_type_3(v))
+            case "6":
+                _ret.update(parse_type_6(v))
+            case "E":
+                _ret.update(parse_type_e(v))
+            case "I":
+                _ret.update(parse_type_i(v))
+            case _:
+                pass
+    return _ret
+
+
+def parse_stations_V1(lines: list[str], event_id: int) -> sttype:
+    _ret = {"ID": event_id, "stations": {}}
+    for st in lines:
+        try:
+            ampl = float(st[35:40])
+        except ValueError:
+            ampl = None
+
+        station = st[1:6].strip()
+        if station not in _ret["stations"].keys():
+            _ret["stations"][station] = []
+
+        _ret["stations"][station].append(
+            {
+                "Component": st[6:9].strip(),
+                "I": None if st[9] == " " else st[9],
+                "Time": parse_dt(st[18:30], True).strftime("%H:%M:%S.%f%z"),
+                "Phase": st[10:15].strip(),
+                "Weigth": None if st[15] == " " else st[15],
+                "Amplitude": ampl,
+            }
+        )
+
+    return _ret
+
+
+def parse_type_1(lines: list[str]) -> dict[str, Any] | None:
+    line1 = {}
+    for line in lines:
+        if "Date" not in line1.keys():
+            dt = parse_dt(line[:21])
+            dist_ind = line[21]
+            event_id = line[22]
+            lat = float(line[23:30])
+            long = float(line[30:38])
+            depth = float(line[38:44])
+            mags = parse_mag(line[55:79])
+            line1.update(
+                {
+                    "DateTime": dt,
+                    "Distance Indicator": dist_ind,
+                    "Event ID": event_id,
+                    "Latitude": lat,
+                    "Longitude": long,
+                    "Depth": depth,
+                    "Magnitudes": mags,
+                }
+            )
+        else:
+            mags = parse_mag(line[56:79])
+            line1["Magnitudes"].union(mags)
+
+        return line1
+
+
+def parse_type_3(lines: list[str]) -> dict[str, Any]:
+    comments = {"Sentido": "", "Regiao": "", "VZ": None, "SZ": None, "FE": None}
+
+    for line in lines:
+        if line.startswith(" SENTIDO"):
+            aux = line[:-2].split(":", maxsplit=1)
+            comments["Sentido"] = aux[1].strip()
+        elif line.startswith(" REGIAO"):
+            aux = line[:-2].split(":", maxsplit=1)
+            for item in aux[1].split(","):
+                if item.startswith("VZ"):
+                    comments["VZ"] = int(item[2:])
+                elif item.startswith("SZ"):
+                    comments["SZ"] = int(item[2:])
+                elif item.startswith("FE"):
+                    comments["FE"] = int(item[2:5])
+                else:
+                    comments["Regiao"] = item[1:]
+
+    return comments
+
+
+def parse_type_6(lines: list[str]) -> dict[str, list[str]]:
+    _ret = {"Wavename": []}
+    for line in lines:
+        _ret["Wavename"].append(line[:-2].strip())
+    return _ret
+
+
+def parse_type_e(lines: list[str]) -> dict[str, int]:
+    err = {}
+    for line in lines:
+        gap = int(line[5:8])
+        err["GAP"] = gap
+
+    return err
+
+
+def parse_type_i(lines: list[str]) -> dict[str, int]:
+    aux = {}
+    for line in lines:
+        aux["ID"] = int(line[60:75])
+
+    return aux
+
+
+def parse_dt(_text: str, isStation=False) -> datetime | time:
+    if not isStation:
+        y = int(_text[0:5])
+        mo = int(_text[6:8])
+        d = int(_text[8:10])
+        h = int(_text[11:13])
+        m = int(_text[13:15])
+        s_ms = int(float(_text[16:20]) * 1000)
+        s = s_ms // 1000
+        s_ms = s_ms % 1000
+        dt = datetime(
+            year=y, month=mo, day=d, hour=h, minute=m, second=s, microsecond=s_ms
+        )
+        return dt
+    else:
+        h = int(_text[:2])
+        m = int(_text[2:4])
+        s_ms = int(float(_text[5:]) * 1000)
+        s = s_ms // 1000
+        s_ms = s_ms % 1000
+        dt = time(hour=h, minute=m, second=s, microsecond=s_ms)
+        return dt
+
+
+def parse_mag(_text: str) -> dict[str, Mag]:
+    mags = {}
+
+    for i in range(0, 3):
+        mag = _text[8 * i : 8 * (i + 1) - 1]  # split every 8 chars
+        if not utils.is_empty(mag):
+            mags[mag[4]] = {"Magnitude": float(mag[:4]), "Agency": mag[5:]}
+    return mags

utilsv2/parser.py

@@ -0,0 +1,83 @@
+import logging
+from io import TextIOWrapper
+
+try:
+    from utilsv2 import utils
+    from utilsv2.log import logger
+    from utilsv2.nordic import evtype, parse_event, parse_stations_V1, sttype
+except ModuleNotFoundError:
+    import utils
+    from log import logger
+    from nordic import evtype, parse_event, parse_stations_V1, sttype
+
+logger = logging.getLogger(__name__)
+
+
+def read_file(fname: str) -> TextIOWrapper | OSError:
+    try:
+        fp = open(fname, "r", newline="\n")
+        return fp
+    except FileNotFoundError:
+        return FileNotFoundError("Nenhum ficheiro encontrado")
+    except PermissionError:
+        return PermissionError("Sem permissões para abrir")
+
+
+def find_events(fp: TextIOWrapper) -> list[tuple[int, int]]:
+    event_indices = []
+
+    event_start = -1
+    idx = 0
+    for line in fp.read().split("\n"):
+        if event_start == -1:
+            event_start = idx
+        if utils.is_empty(line):
+            event_indices.append((event_start, idx))
+            event_start = -1
+        idx += 1
+
+    logger.info("Found %d events", len(event_indices))
+    return event_indices
+
+
+def split_event(lines: list[str], start: int, end: int) -> int:
+    for idx in range(start, end):
+        if lines[idx].endswith("7"):
+            return idx
+    return -1
+
+
+def extract_event(
+    fp: TextIOWrapper, event_bounds: list[tuple[int, int]]
+) -> tuple[list[evtype], list[sttype]]:
+    lines = fp.read().split("\n")
+    events, ev_stations = [], []
+
+    for event_idx in event_bounds:
+        stations = split_event(lines, event_idx[0], event_idx[1])
+        if stations == -1:
+            logger.error(f"Could not parse event at pos {event_idx}")
+            continue
+        ev = parse_event(lines[event_idx[0] : stations])
+        events.append(ev)
+        ev_stations.append(
+            parse_stations_V1(lines[stations + 1 : event_idx[1]], ev["ID"])
+        )
+
+    return events, ev_stations
+
+
+def parse(fname: str):
+    _ret = read_file(fname)
+    if not isinstance(_ret, TextIOWrapper):
+        logger.critical(_ret.__str__())
+        raise _ret
+
+    events = find_events(_ret)
+    _ret.seek(0)
+    evs, stations = extract_event(_ret, events)
+
+    # cleanup
+    _ret.close()
+
+    return evs, stations

utilsv2/stats.py

@@ -0,0 +1,83 @@
+import time
+from datetime import datetime
+
+import numpy as np
+
+
+def print_filters(filters):
+    _res = ""
+    for k, v in filters.items():
+        _res += f"{k}: {v}\n"
+
+
+def pprint(v):
+    return f"\tMédia: {v[0]} \u00b1 {v[1]}; 1o Quartil: {v[3]}; Mediana: {v[2]}; 3o Quartil: {v[4]}; Máximo: {v[5]}; Mínimo: {v[6]}"
+
+
+def stats(data):
+    _stats = f"===Estatística==\nNúmero total de eventos: {len(data)}\n"
+    aux = []
+    currMonth: datetime = data[0]["DateTime"]
+    idx = 0
+    while idx < len(data):
+        if data[idx]["DateTime"].month == currMonth.month:
+            aux.append(data[idx])
+            idx += 1
+        else:
+            m = calc_mag(aux)
+            d = calc_depth(aux)
+            aux = []
+            _stats += f"{currMonth.strftime('%Y-%m')}:\n\tMagnitude: {pprint(m)}\n\tProfundidade: {pprint(d)}\n\n"
+            currMonth = data[idx]["DateTime"]
+
+    m = calc_mag(aux)
+    d = calc_depth(aux)
+    _stats += f"{currMonth.strftime('%Y-%m')}:\nMagnitude: {pprint(m)}\nProfundidade: {pprint(d)}\n"
+
+    fname = f"stats-{time.time_ns()}.txt"
+    with open(fname, "wb") as fp:
+        fp.write(_stats.encode("utf-8"))
+
+    # print(_stats)
+
+
+def calc_depth(data):
+    if len(data) == 0:
+        return 0
+    depths = np.array([v["Depth"] for v in data], dtype=float)
+    quantile = np.quantile(depths, [0.25, 0.75])
+    return list(
+        map(
+            float,
+            (
+                round(np.average(depths), 3),
+                round(np.std(depths), 3),
+                round(np.median(depths), 3),
+                round(quantile[0], 3),
+                round(quantile[1], 3),
+                np.min(depths),
+                np.max(depths),
+            ),
+        )
+    )
+
+
+def calc_mag(data):
+    if len(data) == 0:
+        return 0
+    mags = np.array([v["Magnitudes"]["L"]["Magnitude"] for v in data], dtype=float)
+    quantile = np.quantile(mags, [0.25, 0.75])
+    return list(
+        map(
+            float,
+            (
+                round(np.average(mags), 3),
+                round(np.std(mags), 3),
+                round(np.median(mags), 3),
+                round(quantile[0], 3),
+                round(quantile[1], 3),
+                np.min(mags),
+                np.max(mags),
+            ),
+        )
+    )

utilsv2/utils.py

@@ -0,0 +1,48 @@
+import os
+from datetime import datetime
+
+
+def is_empty(_str: str) -> bool:
+    return len(_str.strip(" ")) == 0
+
+
+def toDateTime(dt: str) -> datetime | int:
+    if len(dt) == 0:
+        return -1
+    try:
+        return datetime.strptime(dt, "%Y-%m-%d")
+    except ValueError:
+        return -1
+
+
+def toFloat(v: str) -> float:
+    if len(v) == 0:
+        return -1.0
+    try:
+        return float(v)
+    except ValueError:
+        return -1.0
+
+
+def toInt(v: str) -> int | None:
+    if len(v) == 0:
+        return None
+    try:
+        return int(v)
+    except ValueError:
+        return None
+
+
+def print_ym(dt: datetime) -> str:
+    return dt.strftime("%Y-%m")
+
+
+def fileExists(fname: str) -> bool:
+    files = set(os.listdir())
+
+    if fname not in files:
+        return False
+
+    if not os.path.isfile(fname):
+        return False
+    return True