Merge pull request #24 from OxfordRSE/pyright_typing

Pyright typing not complete

l2gv2/anomaly_detection.py

@@ -1,21 +1,22 @@
 """Anomaly detection module."""
 
+from typing import Any
 import numpy as np
 from l2gv2.patch.patch import Patch
 
 
-def raw_anomaly_score_node_patch(aligned_patch_emb, emb, node) -> float:
+def raw_anomaly_score_node_patch(aligned_patch_emb, emb, node) -> np.floating[Any]:
     """TODO: docstring for `raw_anomaly_score_node_patch`
 
     Args:
-        aligned_patch_emb ([type]): [description]
+        aligned_patch_emb: [description]
 
-        emb ([type]): [description]
+        emb: [description]
 
-        node ([type]): [description]
+        node: [description]
 
     Returns:
-        float: Raw anomaly score of the node in the patch.
+        Raw anomaly score of the node in the patch.
     """
 
     return np.linalg.norm(aligned_patch_emb.get_coordinate(node) - emb[node])
@@ -25,10 +26,10 @@ def nodes_in_patches(patch_data: list[Patch]) -> list:
     """TODO: docstring for `nodes_in_patches`
 
     Args:
-        patch_data (list[Patch]): [description]
+        patch_data: [description]
 
     Returns:
-        list: [description]
+        [description]
     """
 
     return [set(p.nodes.numpy()) for p in patch_data]
@@ -40,14 +41,14 @@ def normalized_anomaly(
     """TODO: docstring for `normalized_anomaly`
 
     Args:
-        patch_emb (list[Patch]): [description]
+        patch_emb: [description]
 
-        patch_data (list[Patch]): [description]
+        patch_data: [description]
 
-        emb (np.array): [description]
+        emb: [description]
 
     Returns:
-        np.array: [description]
+        [description]
     """
 
     nodes = nodes_in_patches(patch_data)
@@ -95,16 +96,16 @@ def get_outliers(
     """TODO: docstring for `get_outliers`
 
     Args:
-        patch_emb (list): [description]
+        patch_emb: [description]
 
-        patch_data (list): [description]
+        patch_data: [description]
 
-        emb (np.array): [description]
+        emb: [description]
 
-        k (float): Threshold for outliers as multiplier of the standard deviation.
+        k: Threshold for outliers as multiplier of the standard deviation.
 
     Returns:
-        list[int]: [description]
+        [description]
     """
 
     out = []

l2gv2/induced_subgraph.py

@@ -8,14 +8,14 @@ def induced_subgraph(data: tg.data.Data, nodes, extend_hops: int = 0) -> tg.data
     """TODO: docstring for `induced_subgraph`
 
     Args:
-        data (torch_geometric.data.Data): [description]
+        data: [description]
 
-        nodes (int): [description]
+        nodes: [description]
 
-        extend_hops (int, optional): [description], default is 0.
+        extend_hops: [description], default is 0.
 
     Returns:
-        torch_geometric.data.Data: [description]
+        [description]
     """
 
     nodes = torch.as_tensor(nodes, dtype=torch.long)

l2gv2/manopt_optimization.py

@@ -2,7 +2,7 @@
 embeddings of the patches using the manopt library."""
 
 import random
-from typing import Tuple, Optional
+from typing import Tuple, Optional, Any, Literal
 import autograd.numpy as anp
 import pymanopt
 import pymanopt.manifolds
@@ -43,7 +43,7 @@ def total_loss(
     dim,
     k,
     rand: Optional[bool] = False,
-) -> float:
+) -> tuple[np.floating[Any] | float, dict]:
     """TODO: docstring for `total_loss`.
 
     R: list of orthogonal matrices for embeddings of patches.
@@ -116,7 +116,11 @@ def loss(
     consecutive: Optional[bool] = False,
     random_choice_in_intersections: Optional[bool] = False,
     fij: Optional[bool] = False,
-) -> Tuple[float, Optional[list[float]]]:
+) -> (
+    Tuple[np.floating[Any] | float | Literal[0], dict | None]
+    | np.floating[Any]
+    | Literal[0]
+):
     """TODO: docstring for `loss`.
 
     R: list of orthogonal matrices for embeddings of patches.
@@ -165,7 +169,7 @@ def loss(
         if fij:
             return l, f
 
-        return l, None
+        return l
 
     l, f = total_loss(
         rotations,
@@ -313,6 +317,8 @@ def ANPloss_nodes_consecutive_patches(
                 l += anp.linalg.norm(theta1 - theta2) ** 2
 
     return l  # , fij
+
+
 # pylint: enable=invalid-name
 # pylint: enable=no-member
 
@@ -383,9 +389,12 @@ def ANPloss_nodes(
                     # fij[(i, j+1+i, n)]=[theta1, theta2]
 
     return 1 / len(patches) * l  # fij
+
+
 # pylint enable=invalid-name
 # pylint enable=no-member
 
+
 # pylint: disable=no-member
 # pylint does not infer autograd.numpy.random.seed so disable no-member
 def optimization(
@@ -459,7 +468,11 @@ def cost(*R):
     scales = result.point[2 * n_patches :]
     emb_problem = l2g.AlignmentProblem(patches)
 
+    if emb_problem.n_nodes is None or emb_problem.dim is None:
+        raise ValueError("Both n_nodes and dim must be set to integer values.")
+
     embedding = np.empty((emb_problem.n_nodes, emb_problem.dim))
+
     for node, patch_list in enumerate(emb_problem.patch_index):
         embedding[node] = np.mean(
             [
@@ -471,8 +484,11 @@ def cost(*R):
         )
 
     return result, embedding
+
+
 # pylint enable=no-member
 
+
 def loss_dictionary(rs, ss, ts, nodes, patches, dim, k):
     """TODO: docstring for `loss_dictionary`.
 
@@ -508,8 +524,8 @@ def loss_dictionary(rs, ss, ts, nodes, patches, dim, k):
                 patches,
                 dim,
                 k,
-                consecutive=i,
-                random_choice_in_intersections=j,
+                consecutive=(i>0),
+                random_choice_in_intersections=(j>0),
                 fij=False,
             )
     return l