Refactor 36

docs/examples/pyzx_random.py

@@ -67,7 +67,7 @@ def run_random(pyzx_graph: BaseGraph | GraphS, fig_data: matplotlib.figure.Figur
     kwargs = {
         "weights": VALUE_FUNCTION_HYPERPARAMS,
         "first_id_strategy": "first_spider",
-        "seed": None,
+        "seed": 42,
         "vis_options": ("final", None),
         "max_attempts": 1,
         "stop_on_first_success": True,
@@ -77,7 +77,7 @@ def run_random(pyzx_graph: BaseGraph | GraphS, fig_data: matplotlib.figure.Figur
 
     # General description of circuit
     qubit_n = 5
-    depth = 15
+    depth = 100
     circuit_name = f"random_{kwargs['seed'] if kwargs.get('seed') else 'noseed'}_{qubit_n}_{depth}"
 
     # Get a valid random PyZX circuit graph

src/topologiq/core/beams.py

@@ -0,0 +1,146 @@
+"""Beams and related classes used across graph_manager and pathfinder.
+
+Usage:
+    Call any required class from a separate script.
+
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+
+from topologiq.core.pathfinder.utils import get_manhattan
+from topologiq.kwargs import BEAMS_SHORT_LEN
+from topologiq.utils.classes import StandardCoord
+
+
+@dataclass
+class BeamAxisComponent:
+    """Class representing the beam coordinates for any given axis.
+
+    Attributes:
+        start: The starting point for the segment.
+        end: The end point for the segment (== start if segment is a point).
+        direction: Whether segment grows towards the positive or negative end of its axis.
+
+    """
+
+    start: int | float = -np.inf
+    end: int | float = np.inf
+    direction: int = 0 if start == end else 1 if end > start else -1
+
+    def __hash__(self) -> int:
+        """Return start and end for hashing."""
+        return hash((self.start, self.end, self.direction))
+
+    def __eq__(self, other: object) -> bool:
+        """Check equality against any other segments."""
+        return (
+            isinstance(other, BeamAxisComponent)
+            and self.start == other.start
+            and self.end == other.end
+        )
+
+    def __str__(self) -> str:
+        """Return a readable representation."""
+        return f"[{self.start} => {self.end})"
+
+    def contains(self, point: int) -> bool:
+        """Check if a given point is contained in the segment."""
+        if self.direction == 0 and (self.start == point == self.end):
+            return True
+        if self.direction == 1 and (self.start < point <= self.end):
+            return True
+        if self.direction == -1 and (self.start > point >= self.end):
+            return True
+
+        return False
+
+    def to_array(self, len_of_materialised_beam: int) -> list[int] | None:
+        """Convert segment into an array of arbitrary length."""
+        if self.direction != 0:
+            return [self.start + i * self.direction for i in range(len_of_materialised_beam)]
+
+    def get_length(self) -> int:
+        """Get the length of the beam."""
+        return abs(self.start - self.end)
+
+
+@dataclass
+class SingleBeam:
+    """Class representing a single beam.
+
+    Attributes:
+        x: The beam for the x-axis (a point if x-axis has no beam).
+        y: The beam for the y-axis (a point if y-axis has no beam).
+        z: The beam for the y-axis (a point if z-axis has no beam).
+
+    """
+
+    x: BeamAxisComponent
+    y: BeamAxisComponent
+    z: BeamAxisComponent
+
+    def __post_init__(self) -> None:
+        """Ensure beam runs only along a single dimension."""
+        if (abs(self.x.direction) + abs(self.y.direction) + abs(self.z.direction)) != 1:
+            raise ValueError("Malformed beam. Beam must run only along a single dimension.")
+
+    def __hash__(self) -> int:
+        """Return start and end for hashing."""
+        return hash(self.coords)
+
+    def __eq__(self, other: object) -> bool:
+        """Check equality against any other segments."""
+        return isinstance(other, SingleBeam) and self.coords == other.coords
+
+    def __str__(self) -> str:
+        """Return a readable representation."""
+        return f"({self.x!s}, {self.y!s}, {self.z!s})"
+
+    def coords(self) -> StandardCoord:
+        """Return the beam coordinates across all axes."""
+        return self.x.start, self.y.start, self.z.start
+
+    def direction(self) -> StandardCoord:
+        """Return te beam direction as a coordinate tuple."""
+        return self.x.direction, self.y.direction, self.z.direction
+
+    def contains(self, coords_to_check: StandardCoord) -> bool:
+        """Check if beam contains a given coordinate."""
+        x, y, z = coords_to_check
+        return self.x.contains(x) and self.y.contains(y) and self.z.contains(z)
+
+    def intersects(self, other: SingleBeam, short_beams: bool = True) -> bool:
+        """Check if two beams intersect one another."""
+
+        # Get source coords for both beams
+        p1 = self.coords()
+        p2 = other.coords()
+
+        # If checking on short mode,
+        # exit if beams' sources are further than LEN_SHORT_BEAMS
+        if short_beams and get_manhattan(p1, p2) > BEAMS_SHORT_LEN:
+            return False
+
+        # Check if beams are parallel or orthogonal
+        # No clashes possible if beams are parallel
+        d1 = self.direction()
+        d2 = other.direction()
+        orientation = np.dot(d1, d2)
+
+        if orientation != 0:
+            return False
+
+        # Evaluate clash if beams are orthogonal
+        # Source of the other beam must be in the positive quadrant of the span of {d1, -d2}
+        # Sigma is the position of the source of the other beam relative to the source of this beam
+        sigma = np.subtract(p2, p1)
+        basis = np.subtract(d1, d2)
+
+        return np.all((sigma == 0) | (np.sign(sigma) == np.sign(basis)))
+
+
+CubeBeams = list[SingleBeam]

src/topologiq/core/graph_manager/beams.py

@@ -8,9 +8,11 @@
 import networkx as nx
 import numpy as np
 
+from topologiq.core.beams import CubeBeams
 from topologiq.core.graph_manager.utils import get_node_degree
-from topologiq.core.pathfinder.utils import get_manhattan
-from topologiq.utils.classes import CubeBeams, StandardCoord
+
+# from topologiq.core.pathfinder.utils import get_manhattan
+from topologiq.utils.classes import StandardCoord
 
 
 ##################
@@ -164,8 +166,7 @@ def check_tgt_beam_clashes(
                 for cube_beam in nx_g.nodes[cube_id]["beams_short"]:
                     if nx_g.nodes[cube_id]["beams_short"]:
                         intersections = [
-                            tgt_beam.intersects(cube_beam, kwargs["beams_len_short"])
-                            for tgt_beam in tgt_beams_short
+                            tgt_beam.intersects(cube_beam) for tgt_beam in tgt_beams_short
                         ]
                         tgt_clash_tracker = tgt_clash_tracker + np.array(intersections)
                         cube_clash_count += 1 if any(intersections) else 0
@@ -328,7 +329,6 @@ def check_need_twins_beams(
         if (
             nx_g.nodes[out_id]["beams"] if strict else nx_g.nodes[out_id]["beams_short"]
         ) and nx_g.nodes[out_id]["coords"]:
-            out_coords = nx_g.nodes[out_id]["coords"]
             out_beams = nx_g.nodes[out_id]["beams"] if strict else nx_g.nodes[out_id]["beams_short"]
             out_beams_num = len(out_beams)
             out_degree = get_node_degree(nx_g, out_id)
@@ -341,7 +341,6 @@ def check_need_twins_beams(
                     if (
                         nx_g.nodes[out_id]["beams"] if strict else nx_g.nodes[in_id]["beams_short"]
                     ) and nx_g.nodes[in_id]["coords"]:
-                        in_coords = nx_g.nodes[in_id]["coords"]
                         in_beams = (
                             nx_g.nodes[in_id]["beams"]
                             if strict
@@ -350,11 +349,10 @@ def check_need_twins_beams(
                         in_beams_num = len(in_beams)
                         in_degree = get_node_degree(nx_g, in_id)
                         in_pending = in_degree - nx_g.nodes[in_id]["completed"]
-                        manhattan_between = get_manhattan(out_coords, in_coords)
 
                         for beam in in_beams:
                             broken_beams = [
-                                beam.intersects(out_beam, manhattan_between)
+                                beam.intersects(out_beam, short_beams=False)
                                 for out_beam in out_beams
                             ]
                             out_tracker = out_tracker + np.array(broken_beams)

src/topologiq/core/graph_manager/callers.py

@@ -10,9 +10,11 @@
 import matplotlib
 import networkx as nx
 
+from topologiq.core.beams import CubeBeams
 from topologiq.core.graph_manager.utils import reindex_path_dict
 from topologiq.core.pathfinder.pathfinder import pathfinder
-from topologiq.utils.classes import CubeBeams, PathBetweenNodes, StandardBlock, StandardCoord
+from topologiq.core.paths import PathBetweenNodes
+from topologiq.utils.classes import StandardBlock, StandardCoord
 from topologiq.utils.read_write import prep_stats_n_log
 from topologiq.vis.animation import create_animation
 from topologiq.vis.blockgraph import vis_3d
@@ -29,7 +31,7 @@ def call_pathfinder(
     taken: list[StandardCoord],
     tgt_block_info: StandardCoord | None = None,
     hdm: bool = False,
-    critical_beams: dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]] = {},
+    critical_beams: dict[int, tuple[StandardCoord, int, CubeBeams, CubeBeams]] = {},
     src_tgt_ids: tuple[int, int] | None = None,
     **kwargs,
 ) -> tuple[

src/topologiq/core/graph_manager/edge_handlers.py

@@ -11,15 +11,15 @@
 import matplotlib
 import networkx as nx
 
+from topologiq.core.beams import CubeBeams
 from topologiq.core.graph_manager.beams import check_path_to_beam_clashes, check_tgt_beam_clashes
 from topologiq.core.graph_manager.callers import call_debug_vis, call_pathfinder
 from topologiq.core.graph_manager.utils import get_node_degree, prune_beams, update_edge_paths
 from topologiq.core.pathfinder.spatial import get_taken_coords
 from topologiq.core.pathfinder.symbolic import check_exits
+from topologiq.core.paths import PathBetweenNodes
 from topologiq.utils.classes import (
     Colors,
-    CubeBeams,
-    PathBetweenNodes,
     StandardBlock,
     StandardCoord,
 )

src/topologiq/core/graph_manager/utils.py

@@ -13,10 +13,10 @@
 import networkx as nx
 
 from topologiq.core.pathfinder.spatial import get_taken_coords
+from topologiq.core.paths import PathBetweenNodes
 from topologiq.input.simple_graphs import check_zx_types, get_zx_type_fam
 from topologiq.utils.classes import (
     Colors,
-    PathBetweenNodes,
     SimpleDictGraph,
     StandardBlock,
     StandardCoord,

src/topologiq/core/pathfinder/beams.py

@@ -12,8 +12,8 @@
 
 import numpy as np
 
-from topologiq.core.pathfinder.utils import get_manhattan
-from topologiq.utils.classes import CubeBeams, StandardCoord
+from topologiq.core.beams import CubeBeams
+from topologiq.utils.classes import StandardCoord
 
 ##################
 # STANDARD EDGES #
@@ -31,7 +31,7 @@
 # CROSS EDGES #
 ###############
 def check_critical_beams(
-    critical_beams: dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]],
+    critical_beams: dict[int, tuple[StandardCoord, int, CubeBeams, CubeBeams]],
     full_path_coords: list[StandardCoord],
     nxt_coords: StandardCoord,
     tgt_coords: StandardCoord,
@@ -69,9 +69,7 @@ def check_critical_beams(
             for in_id, (_, in_min_exit_num, _, in_beams_short) in critical_beams.items():
                 in_clash_tracker = 0
                 for in_beam in in_beams_short:
-                    intersections = [
-                        out_beam.intersects(in_beam, 9) for out_beam in out_beams_short
-                    ]
+                    intersections = [out_beam.intersects(in_beam) for out_beam in out_beams_short]
                     # out_clash_tracker = out_clash_tracker + np.array(intersections)
                     in_clash_tracker += any(intersections)
 
@@ -93,11 +91,11 @@ def check_critical_beams(
 # CROSS EDGES NOT CURRENTLY IN USED BUT NOT DISCARDED YET #
 ###########################################################
 def split_critical_beams(
-    critical_beams: dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]],
-    src_tgt_ids: tuple[int, int] | None,
+    critical_beams: dict[int, tuple[StandardCoord, int, CubeBeams, CubeBeams]],
+    src_tgt_ids: tuple[int, int],
 ) -> tuple[
-    dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]],
-    dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]],
+    dict[int, tuple[StandardCoord, int, CubeBeams, CubeBeams]],
+    dict[int, tuple[StandardCoord, int, CubeBeams, CubeBeams]],
 ]:
     """Split critical beams into simple and verbose object containing different kinds of beams.
 
@@ -143,10 +141,10 @@ def split_critical_beams(
 
 
 def check_unbreakable_beams(
-    unbreakable_beams: dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]],
+    unbreakable_beams: dict[int, tuple[StandardCoord, int, CubeBeams, CubeBeams]],
     full_path_coords: list[StandardCoord],
     src_tgt_ids: tuple[int, int],
-) -> bool:
+) -> tuple[bool, list[StandardCoord]]:
     """Check that move does not break any beams of cubes that need all their exits.
 
     Args:
@@ -177,63 +175,3 @@ def check_unbreakable_beams(
                 broken_beams += 1
 
     return True, clash_coords
-
-
-def check_negotiable_beams(
-    negotiable_beams: dict[StandardCoord, int, tuple[int, CubeBeams], tuple[int, CubeBeams]],
-    full_path_coords: list[StandardCoord],
-    src_tgt_ids: tuple[int, int],
-) -> bool:
-    """Check that move does not break any beams of cubes that need all their exits.
-
-    Args:
-        negotiable_beams: A minified `critical_beams` object containing beams for nodes that can lose some beams.
-        full_path_coords: All coordinates occupied by current path.
-        src_tgt_ids: The exact IDs of the source and target cubes.
-
-    Return:
-        (bool): True if move clears all checks, False otherwise.
-
-    """
-
-    for node_id, (
-        node_coords,
-        min_exit_num,
-        cube_beams,
-        cube_beams_short,
-    ) in negotiable_beams.items():
-        # For each beam of current cube, check if path breaks the beam
-        out_broken_beams = 0
-        for single_beam in cube_beams_short:
-            if any([single_beam.contains(coord) for coord in full_path_coords]):
-                out_broken_beams += 1
-
-            # If beam is broken, add pre-existing beam-to-beam clashes to consider previously-used allowances
-            for other_node_id, (
-                other_node_coords,
-                other_min_exit_num,
-                other_cube_beams,
-                other_cube_beams_short,
-            ) in negotiable_beams.items():
-                adjust = 1 if other_node_id in src_tgt_ids else 0
-                manhattan_between = get_manhattan(node_coords, other_node_coords)
-                intersections = [
-                    single_beam.intersects(negotiable_beam, manhattan_between)
-                    for negotiable_beam in other_cube_beams_short
-                ]
-                if intersections:
-                    in_broken_beams = sum(intersections) - adjust
-                    out_broken_beams += in_broken_beams
-
-                # Flip check to false if number of broken beams exceeds tolerance
-                if len(other_cube_beams) - in_broken_beams < (other_min_exit_num - adjust):
-                    return False
-
-        # Adjust to consider the broken beam of outgoing/incoming edge in src and tgt cubes
-        adjust = 1 if node_id in src_tgt_ids else 0
-
-        # Flip check to false if number of broken beams exceeds tolerance
-        if len(cube_beams) - out_broken_beams < (min_exit_num - adjust):
-            return False
-
-    return True