modernize

Author: soodoku

.gitignore

@@ -13,3 +13,6 @@ dist
 build
 .DS_Store
 uv.lock
+coverage.xml
+.coverage
+sort-by-distance-output.csv

allocator/cluster_kahip.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Allocator by Karlsruhe High Quality Partitioning (KaHIP)
 """

allocator/cluster_kmeans.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Allocator by K-means
 """

allocator/compare_kahip_kmeans.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 import sys
 import shlex
@@ -60,16 +59,14 @@ def main(argv=sys.argv[1:]):
     buffoon = '--buffoon' if args.buffoon else ''
     balance_edges = '--balance-edges' if args.balance_edges else ''
 
-    kahip_cmd = "python -m allocator.cluster_kahip -n {k:d} --n-closest {n_closest:d} \
-{buffoon:s} {balance_edges:s} {input:s} -o tmpkahip{k:d}.csv -d {dfunc:s} \
-".format(k=n_clusters, input=args.input, n_closest=args.n_closest,
-         buffoon=buffoon, balance_edges=balance_edges, dfunc=args.distance_func)
+    kahip_cmd = (f"python -m allocator.cluster_kahip -n {n_clusters:d} --n-closest {args.n_closest:d} "
+                 f"{buffoon} {balance_edges} {args.input} -o tmpkahip{n_clusters:d}.csv -d {args.distance_func}")
 
-    print(("KaHIP command line '{:s}'".format(kahip_cmd)))
+    print(f"KaHIP command line '{kahip_cmd}'")
     out, err = execute(kahip_cmd)
-    print(("Output: {:s}".format(out)))
+    print(f"Output: {out}")
 
-    bdf = pd.read_csv('tmpkahip{k:d}.csv'.format(k=n_clusters))
+    bdf = pd.read_csv(f'tmpkahip{n_clusters:d}.csv')
 
     buffoon_w = []
     for cluster_id in sorted(bdf.assigned_points.unique()):
@@ -92,15 +89,14 @@ def main(argv=sys.argv[1:]):
     adf = pd.DataFrame(buffoon_w, columns=['label', 'n', 'graph_weight',
                                            'mst_weight'])
 
-    kmean_cmd = 'python -m allocator.cluster_kmeans -n {k:d} {input:s} \
--o tmpkmean{k:d}.csv -d {dfunc:s}'.format(k=n_clusters, input=args.input,
-                                          dfunc=args.distance_func)
+    kmean_cmd = (f'python -m allocator.cluster_kmeans -n {n_clusters:d} {args.input} '
+                 f'-o tmpkmean{n_clusters:d}.csv -d {args.distance_func}')
 
-    print(("K-mean command line '{:s}'".format(kmean_cmd)))
+    print(f"K-mean command line '{kmean_cmd}'")
     out, err = execute(kmean_cmd)
-    print(("Output: {:s}".format(out)))
+    print(f"Output: {out}")
 
-    kdf = pd.read_csv('tmpkmean{k:d}.csv'.format(k=n_clusters))
+    kdf = pd.read_csv(f'tmpkmean{n_clusters:d}.csv')
 
     kmean_w = []
     for cluster_id in sorted(kdf.assigned_points.unique()):

allocator/distance_matrix.py

@@ -1,6 +1,7 @@
 """
 Distance Matrix
 """
+from __future__ import annotations
 
 import os
 import math
@@ -20,15 +21,15 @@
 MAX_DISTANCE_MATRIX_SIZE = 100
 
 
-def pairwise_distances(X, Y=None):
+def pairwise_distances(X: np.ndarray, Y: np.ndarray | None = None) -> np.ndarray:
     """Pairwise euclidean distance calculation
     """
     if Y is None:
         Y = X
     return np.sqrt(((Y - X[:, np.newaxis])**2).sum(axis=2))
 
 
-def latlon2xy(lat, lon):
+def latlon2xy(lat: float, lon: float) -> list[float]:
     """Transform lat/lon to UTM coordinate
 
     Args:
@@ -44,7 +45,7 @@ def latlon2xy(lat, lon):
     return [utm_x, utm_y]
 
 
-def xy2latlog(x, y, zone_number, zone_letter=None):
+def xy2latlog(x: float, y: float, zone_number: int, zone_letter: str | None = None) -> tuple[float, float]:
     """Transform x, y coordinate to lat/lon coordinate
 
     Args:
@@ -57,10 +58,10 @@ def xy2latlog(x, y, zone_number, zone_letter=None):
 
     """
     lat, lon = utm.to_latlon(x, y, zone_number, zone_letter)
-    return [lat, lon]
+    return (lat, lon)
 
 
-def euclidean_distance_matrix(X, Y=None):
+def euclidean_distance_matrix(X: np.ndarray, Y: np.ndarray | None = None) -> np.ndarray:
     """Euclidean distance matrix calculation
     """
     if Y is None:
@@ -71,7 +72,7 @@ def euclidean_distance_matrix(X, Y=None):
     return pairwise_distances(X, Y)
 
 
-def haversine_distance_matrix(X, Y=None):
+def haversine_distance_matrix(X: np.ndarray, Y: np.ndarray | None = None) -> np.ndarray:
     """Harversine distance matrix calculation
     """
     if Y is None:

allocator/shortest_path_gm.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Shortest path using Google Direction API
 """

allocator/shortest_path_mst_tsp.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Shortest path using approximate TSP (Christofides's algorithm)
 """

allocator/shortest_path_ortools.py

@@ -1,16 +1,17 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Shortest path using Google's ortools TSP solver
 """
+from __future__ import annotations
 
 import os
 import sys
 import argparse
+from typing import Any
 from random import randint
 
 import pandas as pd
+import numpy as np
 
 from ortools.constraint_solver import pywrapcp
 # You need to import routing_enums_pb2 after pywrapcp!
@@ -24,10 +25,10 @@
 logger = get_logger(__name__)
 
 
-class DistanceMatrix(object):
-    """Random matrix."""
-
-    def __init__(self, A, args):
+class DistanceMatrix:
+    """Distance matrix for TSP calculations."""
+    
+    def __init__(self, A: np.ndarray, args) -> None:
         """Initialize distance matrix."""
         if args.distance_func == 'euclidean':
             distances = euclidean_distance_matrix(A)
@@ -38,7 +39,7 @@ def __init__(self, A, args):
                                              chunksize=args.osrm_max_table_size,
                                              osrm_base_url=args.osrm_base_url)
         (nx, ny) = distances.shape
-        self.matrix = {}
+        self.matrix: dict[int, dict[int, float]] = {}
         for from_node in range(nx):
             self.matrix[from_node] = {}
             for to_node in range(ny):
@@ -48,7 +49,7 @@ def __init__(self, A, args):
                     self.matrix[from_node][to_node] = distances[from_node,
                                                                 to_node]
 
-    def Distance(self, from_node, to_node):
+    def Distance(self, from_node: int, to_node: int) -> float:
         return self.matrix[from_node][to_node]
 
 

allocator/shortest_path_osrm.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Shortest path using OSRM trip API
 """

allocator/sort_by_distance.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 """
 Allocator by sorting distance from known worker location
 """