gui.py

@@ -7,9 +7,9 @@ from PySide6.QtWidgets import (QDialog, QMainWindow, QMessageBox,
 
 from exceptions import NoSelectedQuakesError, ConnectDatabaseError, \
     FormatToStrError
-from quake_storages import save_quakes, storages
-from ui.main_window_ui import Ui_MainWindow
-from ui.db_conn_ui import Ui_Dialog
+from quake_storages import save_quakes, get_storage
+from ui.main_window_ui import Ui_MainWindow  # type: ignore
+from ui.db_conn_ui import Ui_Dialog  # type: ignore
 import config
 from quakes_from_db import get_quakes, QueryParams
 import logging.config
@@ -72,7 +72,8 @@ class Window(QMainWindow, Ui_MainWindow):
         ext = file.suffix
         try:
             quakes = self.get_selected_quakes()
-            save_quakes(quakes, storages[ext](file))
+            storage = get_storage(ext)
+            save_quakes(quakes, storage(file))
             self.statusBar().showMessage('Writing into the file '
                                          'completed successfully.')
             log.info(f'Writing into the file completed successfully.')
@@ -148,8 +149,8 @@ class Window(QMainWindow, Ui_MainWindow):
             self,
             'About getquakes ',
             '<p>Desktop application for obtaining a list of earthquakes from '
-            'a database and then saving it as a bulletin or a file of another '
-            'structure.</p>'
+            'a database and saving it as a bulletin or other file structure.'
+            '</p>'
             '<p>The app built with:</p>'
             '<p>- Qt Designer</p>'
             '<p>- PySide6</p>'

quake_storages.py

@@ -1,8 +1,7 @@
 # -*- coding: utf-8 -*-
-import string
 from datetime import datetime
 from pathlib import Path
-from typing import Protocol, Sequence, Iterable, Tuple, List
+from typing import Protocol, Sequence, Iterable, Tuple, List, Callable
 
 import openpyxl
 from openpyxl.styles import Alignment
@@ -41,6 +40,7 @@ class CatalogStorage(QuakesStorage):
                 continue
             self._init_sheet(quake)
             self._add_values_in_sheet(quake)
+        self._set_alignment()
         self.wb.save(self._file)
 
     def _init_storage(self) -> None:
@@ -58,25 +58,25 @@ class CatalogStorage(QuakesStorage):
         self.sheet.append(row)
 
     def _set_alignment(self) -> None:
-        for letter in (string.ascii_uppercase[:len(config.CATALOG_HEADER)]):
-            columns = self.sheet.column_dimensions[letter]
-            columns.alignment = Alignment(horizontal='center',
-                                          vertical='center')
+        for sheet in self.wb.sheetnames:
+            rows_of_cells = self.wb[sheet][self.wb[sheet].dimensions]
+            for row in rows_of_cells:
+                for cell in row:
+                    cell.alignment = Alignment(horizontal='center',
+                                               vertical='center')
 
     def _init_sheet(self, quake: Quake) -> None:
         month_num = quake.origin_dt.month - 1
         sheet_name = config.MONTHS[month_num]
         if sheet_name not in self.wb.sheetnames:
             self.sheet = self.wb.create_sheet(sheet_name, month_num)
-            self._set_alignment()
             self.sheet.append(config.CATALOG_HEADER)
         else:
             self.sheet = self.wb.get_sheet_by_name(sheet_name)
 
     def _del_init_empty_worksheet(self):
-        if ('Sheet' in self.wb.sheetnames) and (
-                (self.wb['Sheet'].max_row == 1) and
-                (self.wb['Sheet'].max_column == 1)):
+        if 'Sheet' in self.wb.sheetnames and \
+                self.wb['Sheet'].dimensions == 'A1:A1':
             del self.wb['Sheet']
 
 
@@ -91,6 +91,7 @@ class BulletinStorage(QuakesStorage):
         self.mag = ''
         self.avg_ml = ''
         self.avg_mpsp = ''
+        self.mag_type = '-'
         self.depth = ''
 
     def save(self, quakes: Iterable[Quake]) -> None:
@@ -98,7 +99,8 @@ class BulletinStorage(QuakesStorage):
             amnt_quakes = 0
             for quake in quakes:
                 (self.origin_dt, self.lat, self.lon, self.mag, self.avg_ml,
-                 self.avg_mpsp, self.depth) = _format_common_attrs(quake)
+                 self.avg_mpsp, self.depth,
+                 self.mag_type) = _format_common_attrs(quake)
                 rows = self._get_rows(quake)
                 f.write('\n'.join(rows))
                 amnt_quakes += 1
@@ -115,8 +117,7 @@ class BulletinStorage(QuakesStorage):
                 sta_hdr_describe, sta_strings)
 
     def _get_quake_hdr_describe(self) -> str:
-        mag_type = 'ML' if self.avg_ml != '-' else \
-            'MPSP' if self.avg_mpsp != '-' else 'Mag'
+        mag_type = 'Mag' if self.mag_type == '-' else self.mag_type
         columns_data = config.QUAKE_HEADER_DESCRIBE[:]
         columns_data.insert(5, mag_type)
         return _format_to_str(columns_data,
@@ -132,13 +133,14 @@ class BulletinStorage(QuakesStorage):
     def _get_stations_string(self, quake: Quake) -> str:
         res = ''
         for sta in quake.stations:
-            mag = sta.mag_ML if sta.mag_ML else \
-                sta.mag_MPSP if sta.mag_MPSP else '-'
-            mag_type = 'ML' if sta.mag_ML else 'MPSP' if sta.mag_MPSP else '-'
             phase_dt = datetime.strftime(sta.phase_dt,
                                          '%d.%m.%Y %H:%M:%S.%f')[:-3]
-            sta_data = (sta.name, sta.dist, sta.azimuth, sta.phase, sta.entry,
-                        phase_dt, sta.ampl, sta.period, mag, mag_type)
+            dist = f'{sta.dist:.2f}' if sta.dist else '-'
+            az = f'{sta.azimuth:.2f}' if sta.azimuth else '-'
+            ampl = f'{sta.ampl:.4f}' if sta.ampl else '-'
+            period = f'{sta.period:.2f}' if sta.period else '-'
+            sta_data = (sta.name, dist, az, sta.phase, sta.entry, phase_dt,
+                        ampl, period, self.mag, self.mag_type)
             res += _format_to_str(sta_data,
                                   config.AMNT_COLUMN_SYMBOLS['sta_hdr']) + '\n'
         return res + '\n'
@@ -165,9 +167,8 @@ class NASBulletinStorage(QuakesStorage):
         self.bltn_strings.clear()
         if (quake.lat is not None and quake.lon is not None) \
                 or len(quake.stations_name) > 4:
-            dt = datetime.strftime(quake.origin_dt, '%Y %m %d %H %M %S.%f')[:-3]
-            lat = f'{quake.lat:.2f}' if quake.lat else '-'
-            lon = f'{quake.lon:.2f}' if quake.lon else '-'
+            dt, lat, lon = _format_common_attrs(quake,
+                                                '%Y %m %d %H %M %S.%f')[:3]
             self.bltn_strings.append(f'Fi={lat}  LD={lon} T0={dt}')
             for sta in quake.stations:
                 phase_dt = datetime.strftime(sta.phase_dt,
@@ -204,8 +205,9 @@ class ArcGisStorage(QuakesStorage):
         return columns
 
 
-def _format_common_attrs(quake: Quake) -> Tuple[str, ...]:
-    origin_dt = datetime.strftime(quake.origin_dt, '%d.%m.%Y %H:%M:%S.%f')[:-3]
+def _format_common_attrs(quake: Quake,
+                         date_fmt='%d.%m.%Y %H:%M:%S.%f') -> Tuple[str, ...]:
+    origin_dt = datetime.strftime(quake.origin_dt, date_fmt)[:-3]
     lat = f'{quake.lat:.2f}' if quake.lat else '-'
     lon = f'{quake.lon:.2f}' if quake.lon else '-'
     mag = quake.magnitude
@@ -213,7 +215,8 @@ def _format_common_attrs(quake: Quake) -> Tuple[str, ...]:
     avg_mpsp = f'{mag.MPSP:.1f}' if mag.MPSP else '-'
     preferred_mag = avg_ml if avg_ml != '-' else avg_mpsp
     depth = f'{quake.depth:.2f}' if quake.depth else '-'
-    return origin_dt, lat, lon, preferred_mag, avg_ml, avg_mpsp, depth
+    mag_type = 'ML' if avg_ml != '-' else 'MPSP' if avg_mpsp != '-' else '-'
+    return origin_dt, lat, lon, preferred_mag, avg_ml, avg_mpsp, depth, mag_type
 
 
 def _format_to_str(columns_data: Sequence, hdr_type_config: Sequence) -> str:
@@ -228,7 +231,9 @@ def _format_to_str(columns_data: Sequence, hdr_type_config: Sequence) -> str:
     return res
 
 
-storages = {'.txt': BulletinStorage,
-            '.bltn': NASBulletinStorage,
-            '.GIS': ArcGisStorage,
-            '.xlsx': CatalogStorage}
+def get_storage(ext: str) -> Callable:
+    storage = {'.txt': BulletinStorage,
+               '.bltn': NASBulletinStorage,
+               '.xlsx': CatalogStorage,
+               '.GIS': ArcGisStorage}
+    return storage[ext]

quakes_from_db.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 from typing import Tuple, List, NamedTuple
-from mysql.connector import connect, Error
+from mysql.connector import connect, Error  # type: ignore
 import config
 from exceptions import ConnectDatabaseError
 from quake_structures import Quake, Sta
@@ -10,16 +10,24 @@ from datetime import datetime
 class QueryParams(NamedTuple):
     from_dt: str
     to_dt: str
-    comment: str
     sta: str
     from_mag: str
     to_mag: str
+    comment: str
 
 
 def _get_sql_query(params: QueryParams) -> str:
     sta = '' if params.sta.lower() == 'all' else params.sta
     from_dt = datetime.strptime(params.from_dt, '%Y-%m-%d %H:%M:%S').timestamp()
     to_dt = datetime.strptime(params.to_dt, '%Y-%m-%d %H:%M:%S').timestamp()
+    key_words = params.comment.split()
+    key_words_count = len(key_words)
+    if key_words_count > 1:
+        comment_query = "%' OR o.COMMENTS LIKE '%".join(key_words)
+    elif key_words_count == 1:
+        comment_query = key_words[0]
+    else:
+        comment_query = ""
     return f"SELECT" \
            f" o.EVENTID, o.ORIGINTIME, o.LAT, o.LON," \
            f" o.`DEPTH`," \
@@ -27,12 +35,12 @@ def _get_sql_query(params: QueryParams) -> str:
            f"        SUBSTR(o.COMMENTS, 20))," \
            f" a.ITIME, a.STA, ROUND(a.DIST, 3)," \
            f" ROUND(a.AZIMUTH, 3), a.IPHASE, CONCAT(a.IM_EM, a.FM)," \
-           f" ROUND(a.AMPL, 3), ROUND(a.PER, 2)," \
-           f" ROUND(a.ML, 1), ROUND(a.MPSP, 1)" \
+           f" ROUND(a.AMPL, 4), ROUND(a.PER, 3)," \
+           f" a.ML, a.MPSP" \
            f"FROM origin o " \
            f"INNER JOIN arrival a ON a.EVENTID = o.EVENTID " \
            f"WHERE" \
-           f" (o.COMMENTS LIKE '%{params.comment}%')" \
+           f" (o.COMMENTS LIKE '%{comment_query}%')" \
            f" AND" \
            f" (o.ORIGINTIME BETWEEN '{from_dt}' AND " \
            f"                                       '{to_dt}')" \
@@ -56,30 +64,35 @@ def get_quakes(params: QueryParams) -> Tuple[Quake, ...]:
     """Return tuple of Quake from db records"""
     quakes = []
     stations: List[Sta] = []
-    _id, origin_dt, lat, lon, depth, reg = \
-        '', datetime(year=1, month=1, day=1), 0.0, 0.0, 0.0, ''
+    origin_dt = datetime(year=1, month=1, day=1)
+    _id, origin_dtime, lat, lon, depth, reg = \
+        '', 0.0, 0.0, 0.0, 0.0, ''
     quake_records = get_data(params)
     prev_sta = None
     for quake_record in quake_records:
         if quake_record[0] != _id:
             if len(stations) != 0:
-                quake = Quake(_id, datetime.utcfromtimestamp(origin_dt), lat,
+                stations = sorted(stations, key=sort_stations)
+                quake = Quake(_id, origin_dt, lat,
                               lon, depth, reg, tuple(stations))
                 quakes.append(quake)
                 stations.clear()
                 prev_sta = None
-            _id, origin_dt, lat, lon, depth, reg = quake_record[:6]
+            _id, origin_dtime, lat, lon, depth, reg = quake_record[:6]
+            origin_dt = datetime.utcfromtimestamp(origin_dtime)
 
         sta_dt = datetime.utcfromtimestamp(quake_record[6])
         sta = Sta(sta_dt, *quake_record[7:])
         prev_sta = _add_sta(sta, stations, prev_sta)
-    quakes.append(Quake(_id, datetime.utcfromtimestamp(origin_dt), lat, lon,
+    stations = sorted(stations, key=sort_stations)
+    quakes.append(Quake(_id, origin_dt, lat, lon,
                         depth, reg, tuple(stations)))
     return tuple(_filter_magnitude(quakes, params))
 
 
 def _add_sta(sta: Sta, stations: List[Sta], prev_sta: Sta | None):
-    if prev_sta is None or (sta.phase_dt != prev_sta.phase_dt):
+    if prev_sta is None or (sta.phase_dt != prev_sta.phase_dt) \
+            or (sta.name != prev_sta.name):
         stations.append(sta)
         out_sta = sta
     else:
@@ -93,6 +106,10 @@ def _add_sta(sta: Sta, stations: List[Sta], prev_sta: Sta | None):
     return out_sta
 
 
+def sort_stations(sta: Sta) -> float:
+    return sta.dist if sta.dist else 0.0
+
+
 def _filter_magnitude(quakes: List[Quake], params: QueryParams) -> List[Quake]:
     from_mag, to_mag = float(params.from_mag), float(params.to_mag)
     return [quake for quake in quakes