Gerar dados falsos para testar

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)

utilsv2/mongo.py

@@ -3,14 +3,13 @@ from typing import Any
 
 from pymongo import MongoClient
 from pymongo.collection import Collection
+from pymongo.cursor import Cursor
 from pymongo.errors import ConnectionFailure
 
 try:
     from utilsv2.log import logger
-    from utilsv2.parser import massage_magnitudes
 except ModuleNotFoundError:
     from log import logger
-    from parser import massage_magnitudes
 
 logger = logging.getLogger(__name__)
 
@@ -32,8 +31,6 @@ def add_events(
     db = client["main"]
     coll = db[collection]
 
-    data = massage_magnitudes(data)
-
     _res = coll.insert_many(data)
 
     if _res.acknowledged:
@@ -65,7 +62,24 @@ def close(client: MongoClient) -> None:
     logger.info("Closed the DB.")
 
 
+def query_all(cli: MongoClient, collection: str) -> Any:
+    coll: Collection = cli.main[collection]
+
+    result = coll.find({})
+
+    for doc in result:
+        print(doc)
+
+
+def filter_query(cli: MongoClient, collection: str, filter_by):
+    coll: Collection = cli.main[collection]
+
+    res = coll.find({""})
+
+
 if __name__ == "__main__":
     v = connect("mongodb://localhost:27017")
 
+    query_all(v, "quakes")
+
     close(v)

utilsv2/nordic.py

@@ -25,7 +25,7 @@ class Mag:
         return f"mag: {self.mag}, type: {self.type}, agency: {self.agency}"
 
     def toJSON(self):
-        return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)
+        json.dumps({"Magnitude": self.mag, "Agency": self.agency})
 
 
 def parse_event(event: list[str]) -> evtype:
@@ -57,7 +57,6 @@ def parse_event(event: list[str]) -> evtype:
                 _ret.update(parse_type_i(v))
             case _:
                 pass
-    print(_ret)
     return _ret
 
 
@@ -94,10 +93,10 @@ def parse_type_1(lines: list[str]) -> dict[str, Any] | None:
             dt = parse_dt(line[:21])
             dist_ind = line[21]
             event_id = line[22]
-            lat = float(line[24:31])
-            long = float(line[30:39])
+            lat = float(line[23:30])
+            long = float(line[30:38])
             depth = float(line[38:44])
-            mags = parse_mag(line[56:79])
+            mags = parse_mag(line[55:79])
             line1.update(
                 {
                     "DateTime": dt,
@@ -111,7 +110,7 @@ def parse_type_1(lines: list[str]) -> dict[str, Any] | None:
             )
         else:
             mags = parse_mag(line[56:79])
-            line1["Magnitudes"] += mags
+            line1["Magnitudes"].union(mags)
 
         return line1
 
@@ -186,9 +185,11 @@ def parse_dt(_text: str, isStation=False) -> datetime | time:
         return dt
 
 
-def parse_mag(_text: str) -> list[Mag]:
-    mags = []
-    for mag in textwrap.wrap(_text, 8):
+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.append(Mag(float(mag[:3]), mag[3], mag[4:]))
+            mags[mag[3]] = {"Magnitude": float(mag[:4]), "Agency": mag[4:]}
     return mags

utilsv2/parser.py

@@ -2,9 +2,14 @@ import logging
 from io import TextIOWrapper
 from typing import Any
 
+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__)
 
@@ -63,15 +68,6 @@ def extract_event(
     return events, ev_stations
 
 
-def massage_magnitudes(data: list[evtype]) -> list[evtype]:
-    data = data
-    for idx, ev in enumerate(data):
-        for jdx, mag in enumerate(ev["Magnitudes"]):
-            data[idx]["Magnitudes"][jdx] = mag.toJSON()
-
-    return data
-
-
 def parse(fname: str):
     _ret = read_file(fname)
     if not isinstance(_ret, TextIOWrapper):