# Copyright 2019-2024 Cambridge Quantum Computing
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from collections import defaultdict
from itertools import combinations
from tempfile import NamedTemporaryFile
from typing import Optional
import networkx as nx # type: ignore
import graphviz as gv # type: ignore
from pytket.circuit import Circuit
[docs]class Graph:
[docs] def __init__(self, c: Circuit):
"""
A class for visualising a circuit as a directed acyclic graph (DAG).
Note: in order to use graph-rendering methods, such as
:py:meth:`Graph.save_DAG`, it is necessary to have the Graphviz tools installed
and on your path. See the `Graphviz website <https://graphviz.org/download/>`_
for instructions on how to install them.
:param c: Circuit
:type c: pytket.Circuit
"""
(
q_inputs,
c_inputs,
w_inputs,
q_outputs,
c_outputs,
w_outputs,
input_names,
output_names,
node_data,
edge_data,
) = c._dag_data
self.q_inputs = q_inputs
self.c_inputs = c_inputs
self.w_inputs = w_inputs
self.q_outputs = q_outputs
self.c_outputs = c_outputs
self.w_outputs = w_outputs
self.input_names = input_names
self.output_names = output_names
self.node_data = node_data
self.Gnx: Optional[nx.MultiDiGraph] = None
self.G: Optional[gv.Digraph] = None
self.Gqc: Optional[gv.Graph] = None
self.edge_data: dict[tuple[int, int], list[tuple[int, int, str]]] = defaultdict(
list
)
self.port_counts: dict = defaultdict(int)
for src_node, tgt_node, src_port, tgt_port, edge_type in edge_data:
self.edge_data[(src_node, tgt_node)].append((src_port, tgt_port, edge_type))
self.port_counts[(src_node, src_port)] += 1
[docs] def as_nx(self) -> nx.MultiDiGraph:
"""
Return a logical representation of the circuit as a DAG.
:returns: Representation of the DAG
:rtype: networkx.MultiDiGraph
"""
if self.Gnx is not None:
return self.Gnx
Gnx = nx.MultiDiGraph()
for node, desc in self.node_data.items():
Gnx.add_node(node, desc=desc)
for nodepair, portpairlist in self.edge_data.items():
src_node, tgt_node = nodepair
for src_port, tgt_port, edge_type in portpairlist:
Gnx.add_edge(
src_node,
tgt_node,
src_port=src_port,
tgt_port=tgt_port,
edge_type=edge_type,
)
# Add node IDs to edges
for edge in nx.topological_sort(nx.line_graph(Gnx)):
src_node, tgt_node, _ = edge
# List parent edges with matching port number
src_port = Gnx.edges[edge]["src_port"]
prev_edges = [
e
for e in Gnx.in_edges(src_node, keys=True)
if Gnx.edges[e]["tgt_port"] == src_port
]
if not prev_edges:
# The source must be an input node
unit_id = src_node
nx.set_edge_attributes(Gnx, {edge: {"unit_id": unit_id}})
else:
# The parent must be unique
assert len(prev_edges) == 1
prev_edge = prev_edges[0]
unit_id = Gnx.edges[prev_edge]["unit_id"]
nx.set_edge_attributes(Gnx, {edge: {"unit_id": unit_id}})
# Remove unnecessary port attributes to avoid clutter:
for node in Gnx.nodes:
if Gnx.in_degree(node) == 1:
for edge in Gnx.in_edges(node, keys=True):
nx.set_edge_attributes(Gnx, {edge: {"tgt_port": None}})
for edge in Gnx.out_edges(node, keys=True):
nx.set_edge_attributes(Gnx, {edge: {"src_port": None}})
self.Gnx = Gnx
return Gnx
[docs] def get_DAG(self) -> gv.Digraph:
"""
Return a visual representation of the DAG as a graphviz object.
:returns: Representation of the DAG
:rtype: graphviz.DiGraph
"""
if self.G is not None:
return self.G
G = gv.Digraph(
"Circuit",
strict=True,
)
G.attr(rankdir="LR", ranksep="0.3", nodesep="0.15", margin="0")
q_color = "blue"
c_color = "slategray"
b_color = "gray"
w_color = "green"
gate_color = "lightblue"
boundary_cluster_attr = {
"style": "rounded, filled",
"color": "lightgrey",
"margin": "5",
}
boundary_node_attr = {"fontname": "Courier", "fontsize": "8"}
with G.subgraph(name="cluster_q_inputs") as c:
c.attr(rank="source", **boundary_cluster_attr)
c.node_attr.update(shape="point", color=q_color)
for node in self.q_inputs:
c.node(
str((node, 0)), xlabel=self.input_names[node], **boundary_node_attr
)
with G.subgraph(name="cluster_c_inputs") as c:
c.attr(rank="source", **boundary_cluster_attr)
c.node_attr.update(shape="point", color=c_color)
for node in self.c_inputs:
c.node(
str((node, 0)), xlabel=self.input_names[node], **boundary_node_attr
)
with G.subgraph(name="cluster_w_inputs") as c:
c.attr(rank="source", **boundary_cluster_attr)
c.node_attr.update(shape="point", color=w_color)
for node in self.w_inputs:
c.node(
str((node, 0)), xlabel=self.input_names[node], **boundary_node_attr
)
with G.subgraph(name="cluster_q_outputs") as c:
c.attr(rank="sink", **boundary_cluster_attr)
c.node_attr.update(shape="point", color=q_color)
for node in self.q_outputs:
c.node(
str((node, 0)), xlabel=self.output_names[node], **boundary_node_attr
)
with G.subgraph(name="cluster_c_outputs") as c:
c.attr(rank="sink", **boundary_cluster_attr)
c.node_attr.update(shape="point", color=c_color)
for node in self.c_outputs:
c.node(
str((node, 0)), xlabel=self.output_names[node], **boundary_node_attr
)
with G.subgraph(name="cluster_w_outputs") as c:
c.attr(rank="sink", **boundary_cluster_attr)
c.node_attr.update(shape="point", color=w_color)
for node in self.w_outputs:
c.node(
str((node, 0)), xlabel=self.output_names[node], **boundary_node_attr
)
boundary_nodes = (
self.q_inputs
| self.c_inputs
| self.w_inputs
| self.q_outputs
| self.c_outputs
| self.w_outputs
)
Gnx = self.as_nx()
node_cluster_attr = {
"style": "rounded, filled",
"color": gate_color,
"fontname": "Times-Roman",
"fontsize": "10",
"margin": "5",
"lheight": "100",
}
port_node_attr = {
"shape": "point",
"weight": "2",
"fontname": "Helvetica",
"fontsize": "8",
}
for node, ndata in Gnx.nodes.items():
if node not in boundary_nodes:
with G.subgraph(name="cluster_" + str(node)) as c:
c.attr(label=ndata["desc"], **node_cluster_attr)
n_ports = Gnx.in_degree(node)
if n_ports == 1:
c.node(name=str((node, 0)), **port_node_attr)
else:
for i in range(n_ports):
c.node(name=str((node, i)), xlabel=str(i), **port_node_attr)
edge_colors = {
"Quantum": q_color,
"Boolean": b_color,
"Classical": c_color,
"WASM": w_color,
}
edge_attr = {
"weight": "2",
"arrowhead": "vee",
"arrowsize": "0.2",
"headclip": "true",
"tailclip": "true",
}
for edge, edata in Gnx.edges.items():
src_node, tgt_node, _ = edge
src_port = edata["src_port"] or 0
tgt_port = edata["tgt_port"] or 0
edge_type = edata["edge_type"]
src_nodename = str((src_node, src_port))
tgt_nodename = str((tgt_node, tgt_port))
G.edge(
src_nodename, tgt_nodename, color=edge_colors[edge_type], **edge_attr
)
self.G = G
return G
[docs] def save_DAG(self, name: str, fmt: str = "pdf") -> None:
"""
Save an image of the DAG to a file.
The actual filename will be "<name>.<fmt>". A wide range of formats is
supported. See https://graphviz.org/doc/info/output.html.
:param name: Prefix of file name
:type name: str
:param fmt: File format, e.g. "pdf", "png", ...
:type fmt: str
"""
G = self.get_DAG()
G.render(name, cleanup=True, format=fmt, quiet=True)
[docs] def view_DAG(self) -> str:
"""
View the DAG.
This method creates a temporary file, and returns its filename so that the
caller may delete it afterwards.
:returns: filename of temporary created file
"""
G = self.get_DAG()
filename = NamedTemporaryFile(delete=False).name
G.view(filename, quiet=True)
return filename
[docs] def get_qubit_graph(self) -> gv.Graph:
"""
Return a visual representation of the qubit connectivity graph as a graphviz
object.
:returns: Representation of the qubit connectivity graph of the circuit
:rtype: graphviz.Graph
"""
if self.Gqc is not None:
return self.Gqc
Gnx = self.as_nx()
Gqcnx = nx.Graph()
for node in Gnx.nodes():
qubits = []
for e in Gnx.in_edges(node, keys=True):
unit_id = Gnx.edges[e]["unit_id"]
if unit_id in self.q_inputs:
qubits.append(unit_id)
Gqcnx.add_edges_from(combinations(qubits, 2))
G = gv.Graph(
"Qubit connectivity",
node_attr={
"shape": "circle",
"color": "blue",
"fontname": "Courier",
"fontsize": "10",
},
engine="neato",
)
G.edges(
(self.input_names[src], self.input_names[tgt]) for src, tgt in Gqcnx.edges()
)
self.Gqc = G
return G
[docs] def view_qubit_graph(self) -> str:
"""
View the qubit connectivity graph.
This method creates a temporary file, and returns its filename so that the
caller may delete it afterwards.
:returns: filename of temporary created file
"""
G = self.get_qubit_graph()
filename = NamedTemporaryFile(delete=False).name
G.view(filename, quiet=True)
return filename
[docs] def save_qubit_graph(self, name: str, fmt: str = "pdf") -> None:
"""
Save an image of the qubit connectivity graph to a file.
The actual filename will be "<name>.<fmt>". A wide range of formats is
supported. See https://graphviz.org/doc/info/output.html.
:param name: Prefix of file name
:type name: str
:param fmt: File format, e.g. "pdf", "png", ...
:type fmt: str
"""
G = self.get_qubit_graph()
G.render(name, cleanup=True, format=fmt, quiet=True)
