Feature/mps logger (#32)

* Created new general.py file for logger and cutensornet handle. Updated copyright comment.

* Added debugging and info logging messages.

* Improved messages

* Updated Jupyter notebook with example of logger.

pytket/extensions/cutensornet/general.py

@@ -0,0 +1,42 @@
+# Copyright 2019-2023 Quantinuum
+#
+# 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.
+import logging
+from logging import Logger
+
+
+def set_logger(
+    logger_name: str,
+    level: int = logging.WARNING,
+    fmt: str = "[%(asctime)s] %(name)s (%(levelname)s) - %(message)s",
+) -> Logger:
+    """Initialises and configures a logger object.
+
+    Args:
+        logger_name: Name for the logger object.
+        level: Logger output level.
+        fmt: Logger output format.
+
+    Returns:
+        New configured logger object.
+    """
+    logger = logging.getLogger(logger_name)
+    logger.setLevel(level)
+    logger.propagate = False
+    if not logger.handlers:
+        handler = logging.StreamHandler()
+        handler.setLevel(level)
+        formatter = logging.Formatter(fmt, datefmt="%H:%M:%S")
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+    return logger

pytket/extensions/cutensornet/mps/__init__.py

@@ -1,11 +1,11 @@
-# Copyright 2019 Cambridge Quantum Computing
+# Copyright 2019-2023 Quantinuum
 #
 # 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.

pytket/extensions/cutensornet/mps/mps.py

@@ -1,18 +1,19 @@
-# Copyright 2019 Cambridge Quantum Computing
+# Copyright 2019-2023 Quantinuum
 #
 # 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 __future__ import annotations  # type: ignore
 import warnings
+import logging
 from typing import Any, Optional, Union
 from enum import Enum
 
@@ -33,6 +34,8 @@
 from pytket.circuit import Command, Op, OpType, Qubit
 from pytket.pauli import Pauli, QubitPauliString
 
+from pytket.extensions.cutensornet.general import set_logger
+
 # An alias so that `intptr_t` from CuQuantum's API (which is not available in
 # base python) has some meaningful type name.
 Handle = int
@@ -111,6 +114,7 @@ def __init__(
         chi: Optional[int] = None,
         truncation_fidelity: Optional[float] = None,
         float_precision: Optional[Union[np.float32, np.float64]] = None,
+        loglevel: int = logging.WARNING,
     ):
         """Initialise an MPS on the computational state ``|0>``.
 
@@ -139,6 +143,7 @@ def __init__(
                 choose from ``numpy`` types: ``np.float64`` or ``np.float32``.
                 Complex numbers are represented using two of such
                 ``float`` numbers. Default is ``np.float64``.
+            loglevel: Internal logger output level.
 
         Raises:
             ValueError: If less than two qubits are provided.
@@ -173,6 +178,7 @@ def __init__(
             )
 
         self._lib = libhandle
+        self._logger = set_logger("MPS", level=loglevel)
 
         #######################################
         # Initialise the MPS with a |0> state #
@@ -225,6 +231,15 @@ def is_valid(self) -> bool:
         ds_ok = set(self.canonical_form.keys()) == set(range(len(self)))
         ds_ok = ds_ok and set(self.qubit_position.values()) == set(range(len(self)))
 
+        # Debugger logging
+        self._logger.debug(
+            "Checking validity of MPS... "
+            f"chi_ok={chi_ok}, "
+            f"phys_ok={phys_ok}, "
+            f"shape_ok={shape_ok}, "
+            f"ds_ok={ds_ok}"
+        )
+
         return chi_ok and phys_ok and shape_ok and ds_ok
 
     def apply_gate(self, gate: Command) -> MPS:
@@ -258,6 +273,7 @@ def apply_gate(self, gate: Command) -> MPS:
                 "Gates can only be applied to tensors with physical"
                 + " bond dimension of 2."
             )
+        self._logger.debug(f"Applying gate {gate}")
 
         if len(positions) == 1:
             self._apply_1q_gate(positions[0], gate.op)
@@ -283,6 +299,7 @@ def apply_gate(self, gate: Command) -> MPS:
                 "Gates must act on only 1 or 2 qubits! "
                 + f"This is not satisfied by {gate}."
             )
+
         return self
 
     def canonicalise(self, l_pos: int, r_pos: int) -> None:
@@ -298,11 +315,15 @@ def canonicalise(self, l_pos: int, r_pos: int) -> None:
             r_pos: The position of the rightmost tensor that is not to be
                 canonicalised.
         """
+        self._logger.debug(f"Start canonicalisation... l_pos={l_pos}, r_pos={r_pos}")
+
         for pos in range(l_pos):
             self.canonicalise_tensor(pos, form=DirectionMPS.LEFT)
         for pos in reversed(range(r_pos + 1, len(self))):
             self.canonicalise_tensor(pos, form=DirectionMPS.RIGHT)
 
+        self._logger.debug(f"Finished canonicalisation.")
+
     def canonicalise_tensor(self, pos: int, form: DirectionMPS) -> None:
         """Canonicalises a tensor from an MPS object.
 
@@ -321,6 +342,7 @@ def canonicalise_tensor(self, pos: int, form: DirectionMPS) -> None:
         """
         if form == self.canonical_form[pos]:
             # Tensor already in canonical form, nothing needs to be done
+            self._logger.debug(f"Position {pos} already in {form}.")
             return None
 
         if self._lib._is_destroyed:
@@ -329,6 +351,7 @@ def canonicalise_tensor(self, pos: int, form: DirectionMPS) -> None:
                 "See the documentation of update_libhandle and CuTensorNetHandle.",
             )
 
+        self._logger.debug(f"Canonicalising {pos} to {form}.")
         # Glossary of bond IDs used here:
         # s -> shared virtual bond between T and Tnext
         # v -> the other virtual bond of T
@@ -360,15 +383,19 @@ def canonicalise_tensor(self, pos: int, form: DirectionMPS) -> None:
             raise ValueError("Argument form must be a value in DirectionMPS.")
 
         # Apply QR decomposition
+        self._logger.debug(f"QR decompose a {T.nbytes / 2**20} MiB tensor.")
+
         subscripts = T_bonds + "->" + Q_bonds + "," + R_bonds
         options = {"handle": self._lib.handle, "device_id": self._lib.device_id}
         Q, R = tensor.decompose(
             subscripts, T, method=tensor.QRMethod(), options=options
         )
+        self._logger.debug(f"QR decomposition finished.")
 
         # Contract R into Tnext
         subscripts = R_bonds + "," + Tnext_bonds + "->" + result_bonds
         result = cq.contract(subscripts, R, Tnext)
+        self._logger.debug(f"Contraction with {next_pos} applied.")
 
         # Update self.tensors
         self.tensors[pos] = Q
@@ -421,9 +448,12 @@ def vdot(self, other: MPS) -> complex:
 
         # Special case if only one tensor remains
         if len(self) == 1:
+            self._logger.debug("Applying trivial vdot on single tensor MPS.")
             result = cq.contract("LRp,lrp->", self.tensors[0].conj(), other.tensors[0])
 
         else:
+            self._logger.debug("Applying vdot between two MPS.")
+
             # The two MPS will be contracted from left to right, storing the
             # ``partial_result`` tensor.
             partial_result = cq.contract(
@@ -445,6 +475,7 @@ def vdot(self, other: MPS) -> complex:
                 other.tensors[-1],
             )
 
+        self._logger.debug(f"Result from vdot={result}")
         return complex(result)
 
     def sample(self) -> dict[Qubit, int]:
@@ -492,6 +523,7 @@ def measure(self, qubits: set[Qubit]) -> dict[Qubit, int]:
                 raise ValueError(f"Qubit {q} is not a qubit in the MPS.")
             position_qubit_map[self.qubit_position[q]] = q
         positions = sorted(position_qubit_map.keys())
+        self._logger.debug(f"Measuring qubits={position_qubit_map}")
 
         # Tensor for postselection to |0>
         zero_tensor = cp.zeros(2, dtype=self._complex_t)
@@ -524,6 +556,7 @@ def measure(self, qubits: set[Qubit]) -> dict[Qubit, int]:
             # Throw a coin to decide measurement outcome
             outcome = 0 if prob > random() else 1
             result[position_qubit_map[pos]] = outcome
+            self._logger.debug(f"Outcome of qubit at {pos} is {outcome}.")
 
             # Postselect the MPS for this outcome, renormalising at the same time
             postselection_tensor = cp.zeros(2, dtype=self._complex_t)
@@ -567,6 +600,7 @@ def postselect(self, qubit_outcomes: dict[Qubit, int]) -> float:
             raise ValueError(
                 "Cannot postselect all qubits. You may want to use get_amplitude()."
             )
+        self._logger.debug(f"Postselecting qubits={qubit_outcomes}")
 
         # Apply a postselection for each of the qubits
         for qubit, outcome in qubit_outcomes.items():
@@ -586,6 +620,7 @@ def postselect(self, qubit_outcomes: dict[Qubit, int]) -> float:
             self.tensors[0] = self.tensors[0] / np.sqrt(prob)
             self.canonical_form[0] = None
 
+        self._logger.debug(f"Probability of this postselection is {prob}.")
         return prob
 
     def _postselect_qubit(self, qubit: Qubit, postselection_tensor: cp.ndarray) -> None:
@@ -651,6 +686,7 @@ def expectation_value(self, pauli_string: QubitPauliString) -> float:
             if q not in self.qubit_position:
                 raise ValueError(f"Qubit {q} is not a qubit in the MPS.")
 
+        self._logger.debug(f"Calculating expectation value of {pauli_string}.")
         mps_copy = self.copy()
         pauli_optype = {Pauli.Z: OpType.Z, Pauli.X: OpType.X, Pauli.Y: OpType.Y}
 
@@ -673,6 +709,7 @@ def expectation_value(self, pauli_string: QubitPauliString) -> float:
         value = self.vdot(mps_copy)
         assert np.isclose(value.imag, 0.0, atol=self._atol)
 
+        self._logger.debug(f"Expectation value is {value.real}.")
         return value.real
 
     def get_statevector(self) -> np.ndarray:
@@ -746,7 +783,9 @@ def get_amplitude(self, state: int) -> complex:
             )
 
         assert result_tensor.shape == (1,)
-        return complex(result_tensor[0])
+        result = complex(result_tensor[0])
+        self._logger.debug(f"Amplitude of state {state} is {result}.")
+        return result
 
     def get_qubits(self) -> set[Qubit]:
         """Returns the set of qubits that this MPS is defined on."""
@@ -789,6 +828,13 @@ def get_physical_dimension(self, position: int) -> int:
         physical_dim: int = self.tensors[position].shape[2]
         return physical_dim
 
+    def get_byte_size(self) -> int:
+        """
+        Returns:
+            The number of bytes the MPS currently occupies in GPU memory.
+        """
+        return sum(t.nbytes for t in self.tensors)
+
     def get_device_id(self) -> int:
         """
         Returns:
@@ -833,6 +879,10 @@ def copy(self) -> MPS:
         new_mps._complex_t = self._complex_t
         new_mps._real_t = self._real_t
 
+        self._logger.debug(
+            "Successfully copied an MPS "
+            f"of size {new_mps.get_byte_size() / 2**20} MiB."
+        )
         return new_mps
 
     def __len__(self) -> int: