fix linting

Author: lfischerr

graphconstructor/operators/__init__.py

@@ -4,10 +4,10 @@
 from .knn_selector import KNNSelector
 from .locally_adaptive_sparsification import LocallyAdaptiveSparsification
 from .marginal_likelihood import MarginalLikelihoodFilter
+from .metric_distance import MetricDistanceFilter
 from .minimum_spanning_tree import MinimumSpanningTree
 from .noise_corrected import NoiseCorrected
 from .weight_threshold import WeightThreshold
-from .metric_distance import MetricDistanceFilter
 
 
 __all__ = [

graphconstructor/operators/metric_distance.py

@@ -1,13 +1,15 @@
 from dataclasses import dataclass
-from ..graph import Graph
-from .base import GraphOperator
+from typing import Literal
+import networkx as nx
 from distanceclosure.dijkstra import single_source_dijkstra_path_length
 from networkx.algorithms.shortest_paths.weighted import _weight_function
-import networkx as nx
-from typing import Literal
+from ..graph import Graph
+from .base import GraphOperator
+
 
 Mode = Literal["distance", "similarity"]
 
+
 @dataclass(slots=True)
 class MetricDistanceFilter(GraphOperator):
     """
@@ -23,65 +25,67 @@ class MetricDistanceFilter(GraphOperator):
     verbose : bool, optional
         Prints statements as it computes, by default False
     """
-    weight: str = 'weight'
+
+    weight: str = "weight"
     distortion: bool = False
     verbose: bool = False
     mode: Mode = "distance"
     supported_modes = ["distance", "similarity"]
 
     @staticmethod
-    def _compute_distortions(D: GraphOperator, B, weight='weight', disjunction=sum):
+    def _compute_distortions(D: GraphOperator, B, weight="weight", disjunction=sum):
         G = D.copy()
-        
+
         G.remove_edges_from(B.edges())
         weight_function = _weight_function(B, weight)
 
-        svals = dict()        
+        svals = dict()
         for u in G.nodes():
-            metric_dist = single_source_dijkstra_path_length(B, source=u, weight_function=weight_function, disjunction=disjunction)
+            metric_dist = single_source_dijkstra_path_length(
+                B, source=u, weight_function=weight_function, disjunction=disjunction
+            )
             for v in G.neighbors(u):
-                svals[(u, v)] = G[u][v][weight]/metric_dist[v]
-        
+                svals[(u, v)] = G[u][v][weight] / metric_dist[v]
+
         return svals
-        
+
     def _directed_filter(self, G: Graph) -> Graph:
-        raise NotImplementedError(
-            "MetricDistanceFilter is defined only for undirected graphs."
-        )
-        
-    def _undirected_filter(self, D):          
+        raise NotImplementedError("MetricDistanceFilter is defined only for undirected graphs.")
+
+    def _undirected_filter(self, D):
         disjunction = sum
-        
+
         D = D.to_networkx()
         G = D.copy()
         weight_function = _weight_function(G, self.weight)
-        
+
         if self.verbose:
             total = G.number_of_nodes()
             i = 0
-        
+
         for u, _ in sorted(G.degree(weight=self.weight), key=lambda x: x[1]):
             if self.verbose:
                 i += 1
-                per = i/total
+                per = i / total
                 print("Backbone: Dijkstra: {i:d} of {total:d} ({per:.2%})".format(i=i, total=total, per=per))
-            
-            metric_dist = single_source_dijkstra_path_length(G, source=u, weight_function=weight_function, disjunction=disjunction)
+
+            metric_dist = single_source_dijkstra_path_length(
+                G, source=u, weight_function=weight_function, disjunction=disjunction
+            )
             for v in list(G.neighbors(u)):
                 if metric_dist[v] < G[u][v][self.weight]:
                     G.remove_edge(u, v)
-        
 
         sparse_adj = nx.to_scipy_sparse_array(G)
         if self.distortion:
-            svals = self._compute_distortions(D, G, weight=self.weight, disjunction=disjunction)         
+            svals = self._compute_distortions(D, G, weight=self.weight, disjunction=disjunction)
             return Graph(sparse_adj, False, True, self.mode), svals
         else:
             return Graph(sparse_adj, False, True, self.mode)
-    
+
     def apply(self, G: Graph) -> Graph:
         self._check_mode_supported(G)
         if G.directed:
             return self._directed_filter(G)
         else:
-            return self._undirected_filter(G)
+            return self._undirected_filter(G)

tests/test_metric_distance.py

@@ -1,16 +1,18 @@
-import numpy as np
 import networkx as nx
-import scipy.sparse as sp
+import numpy as np
 import pytest
+import scipy.sparse as sp
 from graphconstructor import Graph
 from graphconstructor.operators import MetricDistanceFilter
 
+
 def _csr(data, rows, cols, n):
     return sp.csr_matrix(
         (np.asarray(data, float), (np.asarray(rows, int), np.asarray(cols, int))),
         shape=(n, n),
     )
 
+
 def simple_undirected_graph():
     A = _csr(
         data=[0.5, 0.5, 0.3, 0.3, 0.8, 0.8],
@@ -21,6 +23,7 @@ def simple_undirected_graph():
 
     return Graph.from_csr(A, directed=False, weighted=True, mode="similarity")
 
+
 def simple_directed_graph():
     A = _csr(
         data=[0.5, 0.5, 0.3],
@@ -31,19 +34,21 @@ def simple_directed_graph():
 
     return Graph.from_csr(A, directed=True, weighted=True, mode="similarity")
 
+
 def test_basic_undirected_filtering():
     G0 = simple_undirected_graph()
 
     out = MetricDistanceFilter(distortion=False, verbose=False).apply(G0)
 
     assert isinstance(out, Graph)
-    assert out.directed == False
-    assert out.weighted == True
+    assert not out.directed
+    assert out.weighted
 
     original_edges = G0.to_networkx().number_of_edges()
     result_edges = out.to_networkx().number_of_edges()
     assert result_edges <= original_edges
-    
+
+
 def test_undirected_filtering_distortion():
     G0 = simple_undirected_graph()
 
@@ -55,29 +60,32 @@ def test_undirected_filtering_distortion():
     filtered_graph, svals = out
     assert isinstance(filtered_graph, Graph)
     assert isinstance(svals, dict)
-    
+
     if svals:
         key = next(iter(svals.keys()))
         assert isinstance(key, tuple)
         assert len(key) == 2
 
+
 def test_directed_graph_not_implemented():
     G0 = simple_directed_graph()
     with pytest.raises(NotImplementedError):
         MetricDistanceFilter().apply(G0)
 
+
 def test_edge_removal_logic():
     G0 = simple_undirected_graph()
     out = MetricDistanceFilter().apply(G0)
 
     original_nx = G0.to_networkx()
-    out_nx = G0.to_networkx()
+    out_nx = out.to_networkx()
 
     assert out_nx.number_of_edges() <= original_nx.number_of_edges()
 
     if nx.is_connected(original_nx):
         assert nx.is_connected(out_nx)
 
+
 def test_isolated_nodes():
     A = _csr(
         data=[0.5, 0.5],
@@ -91,6 +99,7 @@ def test_isolated_nodes():
     assert out.to_networkx().number_of_nodes() == 3
     assert 2 in out.to_networkx().nodes()
 
+
 def test_empty_graph():
     A = _csr(data=[], rows=[], cols=[], n=3)
     G0 = Graph.from_csr(A, directed=False, weighted=True, mode="distance")
@@ -99,4 +108,3 @@ def test_empty_graph():
 
     assert out.to_networkx().number_of_edges() == 0
     assert out.to_networkx().number_of_nodes() == 3
-