refactor: use typing for workaround circular dependencies

mixnet/v2/sim/adversary.py

@@ -1,5 +1,6 @@
 import math
 from collections import defaultdict
+from typing import TYPE_CHECKING
 
 import simpy
 from simpy.core import SimTime
@@ -7,6 +8,9 @@ from simpy.core import SimTime
 from config import Config
 from sphinx import SphinxPacket
 
+if TYPE_CHECKING:
+    from node import Node
+
 
 class Adversary:
     def __init__(self, env: simpy.Environment, config: Config):
@@ -20,10 +24,10 @@ class Adversary:
     def inspect_message_size(self, msg: SphinxPacket | bytes):
         self.message_sizes.append(len(msg))
 
-    def observe_incoming_message(self, node):
+    def observe_incoming_message(self, node: "Node"):
         self.mixed_msgs_per_window[-1][node] += 1
 
-    def observe_outgoing_message(self, node):
+    def observe_outgoing_message(self, node: "Node"):
         self.mixed_msgs_per_window[-1][node] -= 1
         if self.is_around_message_interval(self.env.now):
             self.senders_around_interval[node] += 1
@@ -35,4 +39,4 @@ class Adversary:
     def update_observation_window(self):
         while True:
             self.mixed_msgs_per_window.append(defaultdict(int))
-            yield self.env.timeout(self.config.io_observation_window)
+            yield self.env.timeout(self.config.io_observation_window)

mixnet/v2/sim/node.py

@@ -100,7 +100,8 @@ class Node:
     # TODO: This is a dummy logic
     @classmethod
     def create_incentive_tx(cls, mix_public_key: X25519PublicKey) -> Attachment:
-        public_key = mix_public_key.public_bytes(encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw)
+        public_key = mix_public_key.public_bytes(encoding=serialization.Encoding.Raw,
+                                                 format=serialization.PublicFormat.Raw)
         public_key += bytes(cls.INCENTIVE_TX_SIZE - len(public_key))
         return Attachment(public_key)
 
@@ -108,4 +109,4 @@ class Node:
         return tx == Node.create_incentive_tx(self.public_key)
 
     def log(self, msg):
-        print("Node:%d at %g: %s" % (self.id, self.env.now, msg))
+        print("Node:%d at %g: %s" % (self.id, self.env.now, msg))

mixnet/v2/sim/p2p.py

@@ -1,4 +1,6 @@
+from __future__ import annotations
 import random
+from typing import TYPE_CHECKING
 
 import simpy
 
@@ -6,6 +8,9 @@ from adversary import Adversary
 from config import Config
 from sphinx import SphinxPacket
 
+if TYPE_CHECKING:
+    from node import Node
+
 
 class P2p:
     def __init__(self, env: simpy.Environment, config: Config):
@@ -14,10 +19,10 @@ class P2p:
         self.nodes = []
         self.adversary = Adversary(env, config)
 
-    def add_node(self, nodes):
+    def add_node(self, nodes: list["Node"]):
         self.nodes.extend(nodes)
 
-    def get_nodes(self, n: int):
+    def get_nodes(self, n: int) -> list["Node"]:
         return random.sample(self.nodes, n)
 
     # This should accept only bytes in practice,
@@ -45,4 +50,4 @@ class P2p:
         self.env.process(node.receive_message(msg))
 
     def log(self, msg):
-        print("P2P at %g: %s" % (self.env.now, msg))
+        print("P2P at %g: %s" % (self.env.now, msg))