cbrkit.sim.graphs

15@dataclass(slots=True)
16class brute_force[K, N, E, G](
17    BaseGraphSimFunc[K, N, E, G], SimFunc[Graph[K, N, E, G], GraphSim[K]]
18):
19    """
20    Computes the similarity between two graphs by exhaustively computing all possible mappings
21    and selecting the one with the highest similarity score.
22
23    Args:
24        node_sim_func: A similarity function for graph nodes that receives two node objects.
25        edge_sim_func: A similarity function for graph edges that receives two edge objects.
26
27    Returns:
28        The similarity between the two graphs and the best mapping.
29    """
30
31    def expand(
32        self,
33        x: Graph[K, N, E, G],
34        y: Graph[K, N, E, G],
35        y_nodes: Sequence[K],
36        x_nodes: Sequence[K],
37    ) -> GraphSim[K] | None:
38        """Evaluates a single node mapping candidate and returns its similarity."""
39        node_mapping = frozendict(zip(y_nodes, x_nodes, strict=True))
40
41        # if one node can't be matched to another, skip this permutation
42        for y_key, x_key in node_mapping.items():
43            if not self.node_matcher(x.nodes[x_key].value, y.nodes[y_key].value):
44                return None
45
46        edge_mapping = self.induced_edge_mapping(x, y, node_mapping)
47
48        node_pair_sims = self.node_pair_similarities(x, y, list(node_mapping.items()))
49        edge_pair_sims = self.edge_pair_similarities(
50            x, y, node_pair_sims, list(edge_mapping.items())
51        )
52
53        return self.similarity(
54            x,
55            y,
56            frozendict(node_mapping),
57            frozendict(edge_mapping),
58            frozendict(node_pair_sims),
59            frozendict(edge_pair_sims),
60        )
61
62    def __call__(self, x: Graph[K, N, E, G], y: Graph[K, N, E, G]) -> GraphSim[K]:
63        y_node_keys = list(y.nodes.keys())
64        x_node_keys = list(x.nodes.keys())
65        best_sim: GraphSim[K] = GraphSim(
66            0.0, frozendict(), frozendict(), frozendict(), frozendict()
67        )
68
69        # iterate all possible subset sizes of query (up to target size)
70        for k in range(1, min(len(y_node_keys), len(x_node_keys)) + 1):
71            # all subsets of query nodes of size k
72            for y_candidates in itertools.combinations(y_node_keys, k):
73                # all injective mappings from this subset to target nodes
74                for x_candidates in itertools.permutations(x_node_keys, k):
75                    next_sim = self.expand(x, y, y_candidates, x_candidates)
76
77                    if next_sim and (
78                        next_sim.value > best_sim.value
79                        or (
80                            next_sim.value >= best_sim.value
81                            and (
82                                len(next_sim.node_mapping) > len(best_sim.node_mapping)
83                                or (
84                                    len(next_sim.edge_mapping)
85                                    > len(best_sim.edge_mapping)
86                                )
87                            )
88                        )
89                    ):
90                        best_sim = next_sim
91
92        return best_sim

 35    @dataclass(slots=True)
 36    class dfs[K, N, E, G](
 37        BaseGraphSimFunc[K, N, E, G], SimFunc[Graph[K, N, E, G], GraphSim[K]]
 38    ):
 39        """Graph similarity using depth-first search over node and edge mappings."""
 40
 41        max_iterations: int = 0
 42        upper_bound: float | None = None
 43        strictly_decreasing: bool = True
 44        timeout: float | None = None
 45        roots_func: RootsFunc[K, N, E, G] | None = None
 46
 47        def __call__(
 48            self,
 49            x: Graph[K, N, E, G],
 50            y: Graph[K, N, E, G],
 51        ) -> GraphSim[K]:
 52            x_nx = to_networkx(x)
 53            y_nx = to_networkx(y)
 54
 55            y_edges_lookup = {
 56                (e.source.key, e.target.key): e.key for e in y.edges.values()
 57            }
 58            x_edges_lookup = {
 59                (e.source.key, e.target.key): e.key for e in x.edges.values()
 60            }
 61
 62            node_pair_sims, edge_pair_sims = self.pair_similarities(x, y)
 63
 64            def node_subst_cost(y: NetworkxNode[K, N], x: NetworkxNode[K, N]) -> float:
 65                """Returns the substitution cost for a node pair as one minus similarity."""
 66                if sim := node_pair_sims.get((y["key"], x["key"])):
 67                    return 1.0 - sim
 68
 69                return float("inf")
 70
 71            def edge_subst_cost(
 72                y: NetworkxEdge[K, N, E], x: NetworkxEdge[K, N, E]
 73            ) -> float:
 74                """Returns the substitution cost for an edge pair as one minus similarity."""
 75                if sim := edge_pair_sims.get((y["key"], x["key"])):
 76                    return 1.0 - sim
 77
 78                return float("inf")
 79
 80            ged_iter = nx.optimize_edit_paths(
 81                y_nx,
 82                x_nx,
 83                node_subst_cost=node_subst_cost,
 84                edge_subst_cost=edge_subst_cost,
 85                node_del_cost=lambda y: self.node_del_cost,
 86                node_ins_cost=lambda x: self.node_ins_cost,
 87                edge_del_cost=lambda y: self.edge_del_cost,
 88                edge_ins_cost=lambda x: self.edge_ins_cost,
 89                upper_bound=self.upper_bound,
 90                strictly_decreasing=self.strictly_decreasing,
 91                timeout=self.timeout,
 92                roots=self.roots_func(x, y) if self.roots_func else None,
 93            )
 94
 95            node_edit_path: list[tuple[K, K]] = []
 96            edge_edit_path: list[tuple[tuple[K, K], tuple[K, K]]] = []
 97
 98            for idx in itertools.count():
 99                if self.max_iterations > 0 and idx >= self.max_iterations:
100                    break
101
102                try:
103                    node_edit_path, edge_edit_path, _ = next(ged_iter)
104                except StopIteration:
105                    break
106
107            node_mapping = frozendict(
108                (y_key, x_key)
109                for y_key, x_key in node_edit_path
110                if x_key is not None and y_key is not None
111            )
112            edge_mapping = frozendict(
113                (
114                    y_edges_lookup[y_key],
115                    x_edges_lookup[x_key],
116                )
117                for y_key, x_key in edge_edit_path
118                if x_key is not None
119                and y_key is not None
120                and x_key in x_edges_lookup
121                and y_key in y_edges_lookup
122            )
123
124            return self.similarity(
125                x,
126                y,
127                node_mapping,  # type: ignore[arg-type]
128                edge_mapping,
129                node_pair_sims,
130                edge_pair_sims,
131            )

16@dataclass(slots=True)
17class greedy[K, N, E, G](
18    SearchGraphSimFunc[K, N, E, G], SimFunc[Graph[K, N, E, G], GraphSim[K]]
19):
20    """Performs a Greedy search as described by [Dijkman et al. (2009)](https://doi.org/10.1007/978-3-642-03848-8_5)
21
22    Args:
23        node_sim_func: A function to compute the similarity between two nodes.
24        edge_sim_func: A function to compute the similarity between two edges.
25        node_matcher: A function that returns true if two nodes can be mapped legally.
26        edge_matcher: A function that returns true if two edges can be mapped legally.
27
28    Returns:
29        The similarity between a query and a case graph along with the mapping.
30    """
31
32    def __call__(
33        self,
34        x: Graph[K, N, E, G],
35        y: Graph[K, N, E, G],
36    ) -> GraphSim[K]:
37        node_pair_sims, edge_pair_sims = self.pair_similarities(x, y)
38
39        current_state = self.init_search_state(x, y)
40        current_sim = self.similarity(
41            x,
42            y,
43            current_state.node_mapping,
44            current_state.edge_mapping,
45            node_pair_sims,
46            edge_pair_sims,
47        )
48
49        while not self.finished(current_state):
50            # Iterate over all open pairs and find the best pair
51            next_states: list[SearchState[K]] = []
52
53            for y_key in current_state.open_y_nodes:
54                next_states.extend(self.expand_node(x, y, current_state, y_key))
55
56            for y_key in current_state.open_y_edges:
57                next_states.extend(self.expand_edge(x, y, current_state, y_key))
58
59            new_sims = [
60                self.similarity(
61                    x,
62                    y,
63                    state.node_mapping,
64                    state.edge_mapping,
65                    node_pair_sims,
66                    edge_pair_sims,
67                )
68                for state in next_states
69            ]
70
71            best_sim, best_state = max(
72                zip(new_sims, next_states, strict=True),
73                key=lambda item: item[0].value,
74            )
75
76            # Update the current state and similarity
77            current_state = best_state
78            current_sim = best_sim
79
80        return current_sim

223@dataclass(slots=True)
224class lap[K, N, E, G](
225    lap_base[K, N, E, G],
226    BaseGraphSimFunc[K, N, E, G],
227    SimFunc[Graph[K, N, E, G], GraphSim[K]],
228):
229    """Graph similarity using the Linear Assignment Problem (LAP)."""
230
231    def __call__(
232        self,
233        x: Graph[K, N, E, G],
234        y: Graph[K, N, E, G],
235    ) -> GraphSim[K]:
236        node_pair_sims, edge_pair_sims = self.pair_similarities(x, y)
237        cost_matrix = self.build_cost_matrix(x, y, node_pair_sims, edge_pair_sims)
238        row2y = {r: k for r, k in enumerate(y.nodes.keys())}
239        col2x = {c: k for c, k in enumerate(x.nodes.keys())}
240
241        row_idx, col_idx = linear_sum_assignment(cost_matrix)
242
243        node_mapping = frozendict(
244            (row2y[int(r)], col2x[int(c)])
245            for r, c in zip(row_idx, col_idx, strict=True)
246            # only consider substitutions
247            if cost_matrix[r, c] < np.inf and r in row2y and c in col2x
248        )
249
250        edge_mapping = self.induced_edge_mapping(x, y, node_mapping)
251
252        return self.similarity(
253            x,
254            y,
255            node_mapping,
256            edge_mapping,
257            node_pair_sims,
258            edge_pair_sims,
259        )

185@dataclass(slots=True)
186class vf2_networkx[K, N, E, G](VF2Base[K, N, E, G]):
187    """Graph similarity using the VF2 algorithm via NetworkX."""
188
189    def node_mappings(
190        self,
191        x: Graph[K, N, E, G],
192        y: Graph[K, N, E, G],
193    ) -> list[frozendict[K, K]]:
194        """Finds subgraph isomorphism node mappings using NetworkX."""
195        if len(y.nodes) + len(y.edges) > len(x.nodes) + len(x.edges):
196            larger_graph = to_networkx(y)
197            smaller_graph = to_networkx(x)
198            node_matcher = reverse_positional(self.node_matcher)
199            edge_matcher = reverse_positional(self.edge_matcher)
200        else:
201            larger_graph = to_networkx(x)
202            smaller_graph = to_networkx(y)
203            node_matcher = self.node_matcher
204            edge_matcher = self.edge_matcher
205
206        # `first` must be the larger graph and `second` the smaller one.
207        graph_matcher = DiGraphMatcher(
208            larger_graph,
209            smaller_graph,
210            node_match=lambda x, y: node_matcher(x["value"], y["value"]),
211            edge_match=lambda x, y: edge_matcher(x["value"], y["value"]),
212        )
213
214        mappings_iter = graph_matcher.subgraph_isomorphisms_iter()
215        node_mappings: list[frozendict[K, K]] = []
216
217        for idx in itertools.count():
218            if self.max_iterations > 0 and idx >= self.max_iterations:
219                break
220
221            try:
222                if len(y.nodes) + len(y.edges) > len(x.nodes) + len(x.edges):
223                    # y -> x (as needed)
224                    node_mappings.append(
225                        frozendict(
226                            (larger_idx, smaller_idx)
227                            for larger_idx, smaller_idx in next(mappings_iter).items()
228                        )
229                    )
230                else:
231                    # x ->  y (needs to be inverted)
232                    node_mappings.append(
233                        frozendict(
234                            (smaller_idx, larger_idx)
235                            for larger_idx, smaller_idx in next(mappings_iter).items()
236                        )
237                    )
238            except StopIteration:
239                break
240
241        return node_mappings

110@dataclass(slots=True)
111class vf2_rustworkx[K, N, E, G](VF2Base[K, N, E, G]):
112    """Graph similarity using the VF2 algorithm via rustworkx."""
113
114    id_order: bool = False
115    induced: bool = False
116    call_limit: int | None = None
117
118    def node_mappings(
119        self,
120        x: Graph[K, N, E, G],
121        y: Graph[K, N, E, G],
122    ) -> list[frozendict[K, K]]:
123        """Finds subgraph isomorphism node mappings using rustworkx."""
124        if len(y.nodes) + len(y.edges) > len(x.nodes) + len(x.edges):
125            larger_graph, larger_graph_lookup = to_rustworkx_with_lookup(y)
126            smaller_graph, smaller_graph_lookup = to_rustworkx_with_lookup(x)
127            node_matcher = reverse_positional(self.node_matcher)
128            edge_matcher = reverse_positional(self.edge_matcher)
129        else:
130            larger_graph, larger_graph_lookup = to_rustworkx_with_lookup(x)
131            smaller_graph, smaller_graph_lookup = to_rustworkx_with_lookup(y)
132            node_matcher = self.node_matcher
133            edge_matcher = self.edge_matcher
134
135        # Checks if there is a subgraph of `first` isomorphic to `second`.
136        # Returns an iterator over dictionaries of node indices from `first`
137        # to node indices in `second` representing the mapping found.
138        # As such, `first` must be the larger graph and `second` the smaller one.
139        mappings_iter = rustworkx.vf2_mapping(
140            larger_graph,
141            smaller_graph,
142            node_matcher=node_matcher,
143            edge_matcher=edge_matcher,
144            subgraph=True,
145            id_order=self.id_order,
146            induced=self.induced,
147            call_limit=self.call_limit,
148        )
149
150        node_mappings: list[frozendict[K, K]] = []
151
152        for idx in itertools.count():
153            if self.max_iterations > 0 and idx >= self.max_iterations:
154                break
155
156            try:
157                if len(y.nodes) + len(y.edges) > len(x.nodes) + len(x.edges):
158                    # y -> x (as needed)
159                    node_mappings.append(
160                        frozendict(
161                            (
162                                larger_graph_lookup[larger_idx],
163                                smaller_graph_lookup[smaller_idx],
164                            )
165                            for larger_idx, smaller_idx in next(mappings_iter).items()
166                        )
167                    )
168                else:
169                    # x ->  y (needs to be inverted)
170                    node_mappings.append(
171                        frozendict(
172                            (
173                                smaller_graph_lookup[smaller_idx],
174                                larger_graph_lookup[larger_idx],
175                            )
176                            for larger_idx, smaller_idx in next(mappings_iter).items()
177                        )
178                    )
179            except StopIteration:
180                break
181
182        return node_mappings

 99    @dataclass(slots=True, frozen=True)
100    class dtw[K, N, E, G](SimFunc[Graph[K, N, E, G], GraphSim[K]]):
101        """
102        Graph-based Dynamic Time Warping similarity function leveraging sequence alignment.
103
104        Args:
105            node_sim_func: A similarity function for nodes (SimFunc or BatchSimFunc).
106            edge_sim_func: An optional similarity function for edges (SimFunc or BatchSimFunc).
107            normalize: Whether to normalize the similarity score by the alignment length (default: True).
108
109        Examples:
110            >>> from ...model.graph import Node, Edge, Graph, SerializedGraph, from_dict
111            >>> node_sim_func = lambda n1, n2: 1.0 if n1 == n2 else 0.0
112            >>> edge_sim_func = lambda e1, e2: 1.0 if e1.value == e2.value else 0.0
113
114            >>> data_x = {
115            ...     "nodes": {
116            ...         "1": "A",
117            ...         "2": "B",
118            ...     },
119            ...     "edges": {
120            ...         "e1": {"source": "1", "target": "2", "value": "X"},
121            ...     },
122            ...     "value": None
123            ... }
124            >>> data_y = {
125            ...     "nodes": {
126            ...         "1": "A",
127            ...         "2": "C",
128            ...     },
129            ...     "edges": {
130            ...         "e1": {"source": "1", "target": "2", "value": "Y"},
131            ...     },
132            ...     "value": None
133            ... }
134            >>> graph_x = from_dict(data_x)
135            >>> graph_y = from_dict(data_y)
136
137            >>> g_dtw = dtw(node_sim_func, edge_sim_func)
138            >>> result = g_dtw(graph_x, graph_y)
139            >>> print(result)
140            GraphSim(value=0.05555555555555555,
141                node_mapping=frozendict.frozendict({'1': '2', '2': '2'}), edge_mapping=frozendict.frozendict({'e1': 'e1'}),
142                node_similarities=frozendict.frozendict({'1': 0.0, '2': 0.0}), edge_similarities=frozendict.frozendict({'e1': 0.0}))
143        """
144
145        node_sim_func: AnySimFunc[N, Float]
146        edge_sim_func: AnySimFunc[Edge[K, N, E], Float] | None = None
147        normalize: bool = True
148
149        def __call__(self, x: Graph[K, N, E, G], y: Graph[K, N, E, G]) -> GraphSim[K]:
150            sequence_x, edges_x = to_sequence(x)
151            sequence_y, edges_y = to_sequence(y)
152            node_sim_func: BatchSimFunc[Node[K, N], Float] = batchify_sim(
153                transpose_value(self.node_sim_func)
154            )
155            edge_sim_func = (
156                batchify_sim(self.edge_sim_func) if self.edge_sim_func else None
157            )
158
159            # 3) Use DTW for nodes, with alignment
160            node_result = collections_dtw(node_sim_func)(
161                sequence_x, sequence_y, return_alignment=True
162            )
163            alignment = node_result.mapping  # matched (x_node, y_node)
164
165            # 4) Build node/edge mapping & average similarity using the helper
166            similarity_data = _build_node_edge_mappings_and_similarity(
167                alignment, node_sim_func, edges_x, edges_y, edge_sim_func
168            )
169
170            # 5) Combine node & edge similarity
171            total_similarity = (
172                (similarity_data.node_similarity + similarity_data.edge_similarity)
173                / 2.0
174                if similarity_data.edge_similarity is not None
175                else similarity_data.node_similarity
176            )
177
178            # 6) Normalize by alignment length if applicable
179            if self.normalize and alignment:
180                alen = len([pair for pair in alignment if pair[0] and pair[1]])
181                if alen > 0:
182                    total_similarity /= alen
183
184            return GraphSim(
185                value=total_similarity,
186                node_mapping=frozendict(similarity_data.node_mapping),
187                edge_mapping=frozendict(similarity_data.edge_mapping),
188                node_similarities=frozendict(similarity_data.node_similarities),
189                edge_similarities=frozendict(similarity_data.edge_similarities),
190            )

196    @dataclass(slots=True)
197    class smith_waterman[K, N, E, G](SimFunc[Graph[K, N, E, G], GraphSim[K]]):
198        """
199        Graph-based Smith-Waterman similarity function leveraging local sequence alignment
200        plus ProCAKE-style weighting for local similarities.
201
202        Args:
203            node_sim_func: Base node similarity function (SimFunc or BatchSimFunc).
204            dataflow_in_sim_func: Similarity for incoming data flow (optional).
205            dataflow_out_sim_func: Similarity for outgoing data flow (optional).
206            l_t: Weight for task/node similarity. Default = 1.0
207            l_i: Weight for incoming dataflow similarity. Default = 0.0 (ignored if 0).
208            l_o: Weight for outgoing dataflow similarity. Default = 0.0 (ignored if 0).
209            edge_sim_func: If provided, an edge similarity function (SimFunc or BatchSimFunc).
210            match_score: The SW match score (default 2).
211            mismatch_penalty: The SW mismatch penalty (default -1).
212            gap_penalty: The SW gap penalty (default -1).
213            normalize: Whether to normalize the final similarity (default True).
214
215        Examples:
216            >>> from ...model.graph import Node, Edge, Graph, SerializedGraph, from_dict
217            >>> node_sim_func = lambda n1, n2: 1.0 if n1 == n2 else 0.0
218            >>> edge_sim_func = lambda e1, e2: 1.0 if e1.value == e2.value else 0.0
219            >>> def dataflow_dummy(a, b): return 0.5  # pretend data flow sim
220
221            >>> data_x = {
222            ...     "nodes": {
223            ...         "1": "A",
224            ...         "2": "B",
225            ...     },
226            ...     "edges": {
227            ...         "e1": {"source": "1", "target": "2", "value": "X"},
228            ...     },
229            ...     "value": None
230            ... }
231
232            >>> data_y = {
233            ...     "nodes": {
234            ...         "1": "A",
235            ...         "2": "C",
236            ...     },
237            ...     "edges": {
238            ...         "e1": {"source": "1", "target": "2", "value": "X"},
239            ...     },
240            ...     "value": None
241            ... }
242            >>> graph_x = from_dict(data_x)
243            >>> graph_y = from_dict(data_y)
244
245            >>> g_swa = smith_waterman(
246            ...     node_sim_func,
247            ...     edge_sim_func,
248            ...     dataflow_in_sim_func=dataflow_dummy,
249            ...     dataflow_out_sim_func=dataflow_dummy,
250            ...     l_t=1.0, l_i=0.5, l_o=0.5,
251            ...     use_procake_formula=True
252            ... )
253            >>> result = g_swa(graph_x, graph_y)
254            >>> print(result)
255            GraphSim(value=0.015,
256                node_mapping=frozendict.frozendict({'1': '1', '2': '2'}), edge_mapping=frozendict.frozendict({'e1': 'e1'}),
257                node_similarities=frozendict.frozendict({'1': 0.75, '2': 0.25}), edge_similarities=frozendict.frozendict({'e1': 1.0}))
258
259            >>> # Another example without the ProCAKE weighting:
260            >>> g_swa_naive = smith_waterman(
261            ...     node_sim_func,
262            ...     match_score=2,
263            ...     mismatch_penalty=-1,
264            ...     gap_penalty=-1,
265            ...     normalize=True,
266            ...     use_procake_formula=False
267            ... )
268            >>> result_naive = g_swa_naive(graph_x, graph_y)
269            >>> print(result_naive)
270            GraphSim(value=0.01,
271                node_mapping=frozendict.frozendict({'1': '1', '2': '2'}), edge_mapping=frozendict.frozendict({}),
272                node_similarities=frozendict.frozendict({'1': 1.0, '2': 0.0}), edge_similarities=frozendict.frozendict({}))
273        """
274
275        node_sim_func: InitVar[AnySimFunc[N, Float]]
276        edge_sim_func: InitVar[AnySimFunc[Edge[K, N, E], Float] | None] = None
277        batched_node_sim_func: BatchSimFunc[Node[K, N], Float] = field(init=False)
278        unbatched_node_sim_func: SimFunc[Node[K, N], Float] = field(init=False)
279        batched_edge_sim_func: BatchSimFunc[Edge[K, N, E], Float] | None = field(
280            init=False
281        )
282        dataflow_in_sim_func: SimFunc[Any, float] | None = None
283        dataflow_out_sim_func: SimFunc[Any, float] | None = None
284        l_t: float = 1.0
285        l_i: float = 0.0
286        l_o: float = 0.0
287
288        match_score: int = 2
289        mismatch_penalty: int = -1
290        gap_penalty: int = -1
291        normalize: bool = True
292
293        use_procake_formula: bool = False
294
295        def __post_init__(
296            self,
297            node_sim_func: AnySimFunc[N, Float],
298            edge_sim_func: AnySimFunc[Edge[K, N, E], Float] | None,
299        ) -> None:
300            self.batched_node_sim_func = batchify_sim(transpose_value(node_sim_func))
301            self.unbatched_node_sim_func = unbatchify_sim(
302                transpose_value(node_sim_func)
303            )
304            self.batched_edge_sim_func = (
305                batchify_sim(edge_sim_func) if edge_sim_func else None
306            )
307
308        def local_node_sim(self, q_node: Node[K, N], c_node: Node[K, N]) -> float:
309            """Computes weighted local node similarity using optional dataflow scores."""
310            base = unpack_float(self.unbatched_node_sim_func(q_node, c_node))
311
312            if self.use_procake_formula:
313                in_flow = 0.0
314                if self.l_i != 0.0 and self.dataflow_in_sim_func:
315                    # In real usage, pass appropriate node/edge data
316                    in_flow = self.dataflow_in_sim_func(None, None)
317
318                out_flow = 0.0
319                if self.l_o != 0.0 and self.dataflow_out_sim_func:
320                    out_flow = self.dataflow_out_sim_func(None, None)
321
322                total_weight = self.l_t + self.l_i + self.l_o
323
324                if total_weight == 0:
325                    return 0.0
326
327                return (
328                    self.l_t * base + self.l_i * in_flow + self.l_o * out_flow
329                ) / total_weight
330
331            return base
332
333        def __call__(self, x: Graph[K, N, E, G], y: Graph[K, N, E, G]) -> GraphSim[K]:
334            # 1) Convert graphs to sequences
335            sequence_x, edges_x = to_sequence(x)
336            sequence_y, edges_y = to_sequence(y)
337
338            # 3) We'll rely on the minineedle-based SW for a raw score.
339            sw_obj = collections_smith_waterman(
340                match_score=self.match_score,
341                mismatch_penalty=self.mismatch_penalty,
342                gap_penalty=self.gap_penalty,
343            )
344
345            # Use the length of each sequence as tokens
346            tokens_x = list(range(len(sequence_x)))
347            tokens_y = list(range(len(sequence_y)))
348            raw_sw_score = sw_obj(tokens_x, tokens_y)
349
350            # 4) We do not have a direct alignment from the raw minineedle call,
351            #    so let's do a naive 1:1 for the shorter length.
352            length = min(len(sequence_x), len(sequence_y))
353            alignment = [(sequence_x[i], sequence_y[i]) for i in range(length)]
354
355            # 5) Build node/edge mappings & average similarities using local_node_sim
356            similarity_data = _build_node_edge_mappings_and_similarity(
357                alignment,
358                batchify_sim(self.local_node_sim),
359                edges_x,
360                edges_y,
361                self.batched_edge_sim_func,
362            )
363
364            # 6) Combine node & edge sim, scale by raw SW score (if > 0)
365            if similarity_data.edge_similarity is not None:
366                total_similarity = (
367                    similarity_data.node_similarity + similarity_data.edge_similarity
368                ) / 2.0
369            else:
370                total_similarity = similarity_data.node_similarity
371
372            if raw_sw_score > 0:
373                total_similarity *= raw_sw_score / 100.0
374
375            # 7) Normalize if requested
376            if self.normalize and length > 0:
377                total_similarity /= length
378
379            return GraphSim(
380                value=total_similarity,
381                node_mapping=frozendict(similarity_data.node_mapping),
382                edge_mapping=frozendict(similarity_data.edge_mapping),
383                node_similarities=frozendict(similarity_data.node_similarities),
384                edge_similarities=frozendict(similarity_data.edge_similarities),
385            )