prog-team-proj/utilsv2/nordic.py

import json
import logging
import textwrap
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