Add type hints and checks

.pre-commit-config.yaml

@@ -1,14 +1,20 @@
 repos:
 - repo: https://github.com/gitleaks/gitleaks
-  rev: v8.25.1
+  rev: v8.30.0
   hooks:
   - id: gitleaks
 - repo: https://github.com/pre-commit/pre-commit-hooks
-  rev: v5.0.0
+  rev: v6.0.0
   hooks:
   - id: end-of-file-fixer
   - id: trailing-whitespace
 - repo: https://github.com/pylint-dev/pylint
-  rev: v3.3.6
+  rev: v4.0.5
   hooks:
   - id: pylint
+- repo: https://github.com/pre-commit/mirrors-mypy
+  rev: v1.20.1
+  hooks:
+  - id: mypy
+    files: ^blitzortung/
+    additional_dependencies: [pyproj, shapely, psycopg2-binary, fasteners, txpostgres]

blitzortung/__init__.py

@@ -55,6 +55,7 @@ class Error(Exception):
 from . import util
 from . import base
 
+
 INJECTOR = injector.Injector(
     [config.ConfigModule(), db.DbModule()])
 

blitzortung/base.py

@@ -18,23 +18,26 @@
 
 """
 
+from __future__ import annotations
+
 import math
+from typing import Any, Tuple, Union
 
 import pyproj
 
 
 class EqualityAndHash:
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         if isinstance(other, self.__class__):
             return self.__dict__ == other.__dict__
-        return NotImplemented
+        return False
 
-    def __ne__(self, other):
+    def __ne__(self, other: object) -> bool:
         if isinstance(other, self.__class__):
             return not self.__eq__(other)
-        return NotImplemented
+        return True
 
-    def __hash__(self):
+    def __hash__(self) -> int:
         return hash(tuple(sorted(self.__dict__.items())))
 
 
@@ -49,39 +52,44 @@ class Point:
 
     __slots__ = ('x', 'y')
 
-    def __init__(self, x_coord_or_point, y_coord=None):
+    x: float
+    y: float
+
+    def __init__(self, x_coord_or_point: Union[float, Point], y_coord: float | None = None) -> None:
         (self.x, self.y) = self.__get_point_coordinates(x_coord_or_point, y_coord)
 
-    def distance_to(self, other):
+    def distance_to(self, other: Point) -> float:
         return self.geodesic_relation_to(other)[1]
 
-    def azimuth_to(self, other):
+    def azimuth_to(self, other: Point) -> float:
         return self.geodesic_relation_to(other)[0]
 
-    def geodesic_shift(self, azimuth, distance):
+    def geodesic_shift(self, azimuth: float, distance: float) -> Point:
         result = self.__geod.fwd(self.x, self.y, azimuth / self.__radians_factor, distance, radians=False)
         return Point(result[0], result[1])
 
-    def geodesic_relation_to(self, other):
+    def geodesic_relation_to(self, other: Point) -> Tuple[float, float]:
         result = self.__geod.inv(self.x, self.y, other.x, other.y, radians=False)
         return result[0] * self.__radians_factor, result[2]
 
     @staticmethod
-    def __get_point_coordinates(x_coord_or_point, y_coord):
+    def __get_point_coordinates(x_coord_or_point: Union[float, Point], y_coord: float | None) -> Tuple[float, float]:
         if isinstance(x_coord_or_point, Point):
             return x_coord_or_point.x, x_coord_or_point.y
         else:
-            return x_coord_or_point, y_coord
+            return x_coord_or_point, y_coord if y_coord is not None else 0.0
 
-    def __eq__(self, other):
+    def __eq__(self, other: Any) -> bool:
+        if not isinstance(other, Point):
+            return False
         return self.equal(self.x, other.x) and self.equal(self.y, other.y)
 
     @staticmethod
-    def equal(a, b):
+    def equal(a: float, b: float) -> bool:
         return abs(a - b) < 1e-4
 
-    def __str__(self):
+    def __str__(self) -> str:
         return "(%.4f, %.4f)" % (self.x, self.y)
 
-    def __hash__(self):
+    def __hash__(self) -> int:
         return hash(self.x) ^ hash(self.y)

blitzortung/builder/base.py

@@ -62,4 +62,6 @@ def set_y(self, y_coord):
         return self
 
     def build(self):
+        if self.timestamp is None:
+            raise BuilderError("Timestamp not set")
         return data.Event(self.timestamp, self.x_coord, self.y_coord)

blitzortung/builder/strike.py

@@ -85,7 +85,7 @@ def from_line(self, line):
             self.set_station_count(int(stations[0]))
             self.set_stations([int(station) for station in stations[2].split(',') if station])
         except (KeyError, ValueError, IndexError) as e:
-            raise BuilderError(e)
+            raise BuilderError(e) from e
 
         return self
 
@@ -107,5 +107,7 @@ def from_json(self, json_data: dict):
         return self
 
     def build(self):
+        if self.timestamp is None:
+            raise BuilderError("Timestamp not set")
         return data.Strike(self.id_value, self.timestamp, self.x_coord, self.y_coord, self.altitude,
                            self.amplitude, self.lateral_error, self.station_count, self.stations)

blitzortung/cache.py

@@ -54,7 +54,7 @@ def __init__(self, ttl_seconds=30, size=None, cleanup_period=None):
 
         self.cache = {}
         self.keys = {}
-        self.last_cleanup = 0
+        self.last_cleanup = 0.0
         self.cleanup_period = cleanup_period
 
     def get(self, cached_object_creator, *args, **kwargs):

blitzortung/cli/db.py

@@ -109,7 +109,7 @@ def main():
 
     start_time = parse_time(options.startdate, options.starttime, tz, "starttime")
     end_time = parse_time(options.enddate, options.endtime, tz, "endtime",
-                          is_end_time=True) if non_default_end else None
+                          is_end_time=True) if non_default_end else None  # type: ignore[assignment]
 
     area = None
     if options.area:

blitzortung/cli/imprt.py

@@ -20,12 +20,12 @@
 import logging
 import os
 import time
+from optparse import OptionParser
 from contextlib import nullcontext
 
 import requests
 import statsd
 import stopit
-from optparse import OptionParser
 
 import blitzortung.dataimport
 import blitzortung.db

blitzortung/cli/imprt_websocket.py

@@ -8,11 +8,11 @@
 
 import statsd
 import websocket
+from websocket import WebSocketConnectionClosedException
 
 import blitzortung.builder
 import blitzortung.data
 import blitzortung.logger
-from websocket import WebSocketConnectionClosedException
 
 from blitzortung.lock import LockWithTimeout, FailedToAcquireException
 from blitzortung.websocket import decode
@@ -30,11 +30,6 @@
 strike_count = 0
 last_commit_time = time.time()
 
-try:
-    import thread
-except ImportError:
-    import _thread as thread
-
 
 def on_message(ws, message):
     global strike_db

blitzortung/cli/update.py

@@ -19,11 +19,11 @@
 import datetime
 import logging
 import os
+from optparse import OptionParser
 from contextlib import nullcontext
 
 import requests
 import statsd
-from optparse import OptionParser
 
 import blitzortung.config
 import blitzortung.db

blitzortung/cli/webservice.py

@@ -7,6 +7,8 @@
 import platform
 import time
 
+from typing import Any
+
 from twisted.application import internet, service
 from twisted.internet.defer import succeed
 from twisted.internet.error import ReactorAlreadyInstalledError
@@ -28,9 +30,10 @@
 JSON_CONTENT_TYPE = 'text/json'
 
 try:
-    from twisted.internet import epollreactor as reactor, defer
+    from twisted.internet import epollreactor  # type: ignore[attr-defined, no-redef]
+    reactor = epollreactor
 except ImportError:
-    from twisted.internet import kqreactor as reactor
+    from twisted.internet import kqreactor as reactor  # type: ignore[assignment, no-redef]
 
 try:
     reactor.install()
@@ -49,7 +52,7 @@
 
 is_pypy = platform.python_implementation() == 'PyPy'
 
-FORBIDDEN_IPS = {}
+FORBIDDEN_IPS: dict[str, Any] = {}
 
 USER_AGENT_PREFIX = 'bo-android-'
 
@@ -338,12 +341,14 @@
     def parse_user_agent(self, request):
         """Parse user agent string to extract version information."""
         user_agent = request.getHeader("User-Agent")
+        user_agent_version = 0
         if user_agent and user_agent.startswith(USER_AGENT_PREFIX):
             user_agent_parts = user_agent.split(' ')[0].rsplit('-', 1)
-            version_string = user_agent_parts[1] if len(user_agent_parts) > 1 else None
-            user_agent_version = int(version_string) if user_agent_parts[0] == 'bo-android' else 0
-        else:
-            user_agent_version = 0
+            if len(user_agent_parts) > 1 and user_agent_parts[0] == 'bo-android':
+                try:
+                    user_agent_version = int(user_agent_parts[1])
+                except ValueError:
+                    pass
         return user_agent, user_agent_version
 
     def fix_bad_accept_header(self, request, user_agent):
@@ -375,7 +380,11 @@
         now = time.time()
         if now > self.next_memory_info:
             log.msg("### MEMORY INFO ###")
-            log.msg(gc.get_stats(True) if is_pypy else gc.get_stats())
+            # pylint: disable=no-member
+            if is_pypy:
+                log.msg(gc.get_stats(True))  # type: ignore[call-arg]
+            else:
+                log.msg(gc.get_stats())  # type: ignore[call-arg]
             self.next_memory_info = now + self.MEMORY_INFO_INTERVAL
 
 
@@ -406,11 +415,11 @@
 if os.environ.get('BLITZORTUNG_TEST'):
     import tempfile
 
-    log_directory = tempfile.mkdtemp()
+    log_directory: str | None = tempfile.mkdtemp()
     print("LOG_DIR", log_directory)
 else:
     log_directory = "/var/log/blitzortung"
-if os.path.exists(log_directory):
+if log_directory and os.path.exists(log_directory):
     logfile = DailyLogFile("webservice.log", log_directory)
     application.setComponent(ILogObserver, LogObserver(logfile).emit)
 else:

blitzortung/cli/webservice_insertlog.py

@@ -12,6 +12,7 @@
 import sys
 from optparse import OptionParser
 
+import geoip2.database
 import statsd
 
 from blitzortung.convert import value_to_string
@@ -20,8 +21,6 @@
 logger.addHandler(logging.StreamHandler(sys.stdout))
 logger.setLevel(logging.INFO)
 
-import geoip2.database
-
 
 
 def main():
@@ -70,13 +69,15 @@ def main():
                     data_area = None
 
                     user_agent = entry[7]
-                    version = None
+                    version: int | None = None
                     if user_agent:
                         user_agent_parts = user_agent.split(' ')[0].rsplit('-', 1)
                         version_prefix = user_agent_parts[0]
-                        version_string = user_agent_parts[1] if len(user_agent_parts) > 1 else None
-                        if version_prefix == 'bo-android':
-                            version = int(version_string)
+                        if version_prefix == 'bo-android' and len(user_agent_parts) > 1:
+                            try:
+                                version = int(user_agent_parts[1])
+                            except ValueError:
+                                pass
 
                     try:
                         geo_info = reader.city(remote_address)

blitzortung/config.py

@@ -18,6 +18,8 @@
 
 """
 
+from __future__ import annotations
+
 import configparser
 import os
 from typing import Optional
@@ -27,17 +29,19 @@
 
 @singleton
 class Config:
+    config_parser: configparser.ConfigParser
+
     @inject
-    def __init__(self, config_parser: configparser.ConfigParser):
+    def __init__(self, config_parser: configparser.ConfigParser) -> None:
         self.config_parser = config_parser
 
-    def get_username(self):
+    def get_username(self) -> str:
         return self.config_parser.get('auth', 'username')
 
-    def get_password(self):
+    def get_password(self) -> str:
         return self.config_parser.get('auth', 'password')
 
-    def get_db_connection_string(self):
+    def get_db_connection_string(self) -> str:
         host = self.config_parser.get('db', 'host')
         port = self.config_parser.get('db', 'port', fallback='5432')
         dbname = self.config_parser.get('db', 'dbname')
@@ -46,17 +50,18 @@ def get_db_connection_string(self):
 
         return "host='%s' port=%s dbname='%s' user='%s' password='%s'" % (host, port, dbname, username, password)
 
-    def get_webservice_port(self):
+    def get_webservice_port(self) -> int:
         return int(self.config_parser.get('webservice', 'port'))
 
-    def __str__(self):
+    def __str__(self) -> str:
         return "Config(user: %s, pass: %s)" % (self.get_username(), len(self.get_password()) * '*')
 
 
-def config():
+def config() -> Config:
     from blitzortung import INJECTOR
 
-    return INJECTOR.get(Config)
+    result: Config = INJECTOR.get(Config)
+    return result
 
 
 class ConfigModule(Module):
@@ -78,3 +83,4 @@ def find_config_file_path(self) -> Optional[str]:
             config_file_path = os.path.join(config_dir_name, config_file_name)
             if os.path.exists(config_file_path):
                 return config_file_path
+        return None