# This code is part of Qiskit.
#
# (C) Copyright IBM 2024.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""Generic BackendV2 class that with a simulated ``run``."""

from __future__ import annotations
import warnings

from collections.abc import Iterable
import numpy as np

from qiskit import pulse
from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
from qiskit.circuit import QuantumCircuit, Instruction
from qiskit.circuit.controlflow import (
    IfElseOp,
    WhileLoopOp,
    ForLoopOp,
    SwitchCaseOp,
    BreakLoopOp,
    ContinueLoopOp,
)
from qiskit.circuit.library.standard_gates import get_standard_gate_name_mapping
from qiskit.exceptions import QiskitError
from qiskit.transpiler import CouplingMap, Target, InstructionProperties, QubitProperties
from qiskit.providers import Options
from qiskit.providers.basic_provider import BasicSimulator
from qiskit.providers.backend import BackendV2
from qiskit.providers.models import (
    PulseDefaults,
    Command,
)
from qiskit.qobj import PulseQobjInstruction, PulseLibraryItem
from qiskit.utils import optionals as _optionals

# Noise default values/ranges for duration and error of supported
# instructions. There are two possible formats:
# - (min_duration, max_duration, min_error, max_error),
#   if the defaults are ranges.
# - (duration, error), if the defaults are fixed values.
_NOISE_DEFAULTS = {
    "cx": (7.992e-08, 8.99988e-07, 1e-5, 5e-3),
    "ecr": (7.992e-08, 8.99988e-07, 1e-5, 5e-3),
    "cz": (7.992e-08, 8.99988e-07, 1e-5, 5e-3),
    "id": (2.997e-08, 5.994e-08, 9e-5, 1e-4),
    "rz": (0.0, 0.0),
    "sx": (2.997e-08, 5.994e-08, 9e-5, 1e-4),
    "x": (2.997e-08, 5.994e-08, 9e-5, 1e-4),
    "measure": (6.99966e-07, 1.500054e-06, 1e-5, 5e-3),
    "delay": (None, None),
    "reset": (None, None),
}

# Fallback values for gates with unknown noise default ranges.
_NOISE_DEFAULTS_FALLBACK = {
    "1-q": (2.997e-08, 5.994e-08, 9e-5, 1e-4),
    "multi-q": (7.992e-08, 8.99988e-07, 5e-3),
}

# Ranges to sample qubit properties from.
_QUBIT_PROPERTIES = {
    "dt": 0.222e-9,
    "t1": (100e-6, 200e-6),
    "t2": (100e-6, 200e-6),
    "frequency": (5e9, 5.5e9),
}

# The number of samples determines the pulse durations of the corresponding
# instructions. This default defines pulses with durations in multiples of
# 16 dt for consistency with the pulse granularity of real IBM devices, but
# keeps the number smaller than what would be realistic for
# manageability. If needed, more realistic durations could be added in the
# future (order of 160dt for 1q gates, 1760dt for 2q gates and measure).
_PULSE_LIBRARY = [
    PulseLibraryItem(name="pulse_1", samples=np.linspace(0, 1.0, 16, dtype=np.complex128)),  # 16dt
    PulseLibraryItem(name="pulse_2", samples=np.linspace(0, 1.0, 32, dtype=np.complex128)),  # 32dt
    PulseLibraryItem(name="pulse_3", samples=np.linspace(0, 1.0, 64, dtype=np.complex128)),  # 64dt
]


class GenericBackendV2(BackendV2):
    """Generic :class:`~.BackendV2` implementation with a configurable constructor. This class will
    return a :class:`~.BackendV2` instance that runs on a local simulator (in the spirit of fake
    backends) and contains all the necessary information to test backend-interfacing components, such
    as the transpiler. A :class:`.GenericBackendV2` instance can be constructed from as little as a
    specified ``num_qubits``, but users can additionally configure the basis gates, coupling map,
    ability to run dynamic circuits (control flow instructions), instruction calibrations and dtm.
    The remainder of the backend properties are generated by randomly sampling
    from default ranges extracted from historical IBM backend data. The seed for this random
    generation can be fixed to ensure the reproducibility of the backend output.
    This backend only supports gates in the standard library, if you need a more flexible backend,
    there is always the option to directly instantiate a :class:`.Target` object to use for
    transpilation.
    """

    def __init__(
        self,
        num_qubits: int,
        basis_gates: list[str] | None = None,
        *,
        coupling_map: list[list[int]] | CouplingMap | None = None,
        control_flow: bool = False,
        calibrate_instructions: bool | InstructionScheduleMap | None = None,
        dtm: float | None = None,
        seed: int | None = None,
    ):
        """
        Args:
            num_qubits: Number of qubits that will be used to construct the backend's target.
                Note that, while there is no limit in the size of the target that can be
                constructed, this backend runs on local noisy simulators, and these might
                present limitations in the number of qubits that can be simulated.

            basis_gates: List of basis gate names to be supported by
                the target. These must be part of the standard qiskit circuit library.
                The default set of basis gates is ``["id", "rz", "sx", "x", "cx"]``
                The ``"reset"``,  ``"delay"``, and ``"measure"`` instructions are
                always supported by default, even if not specified via ``basis_gates``.

            coupling_map: Optional coupling map
                for the backend. Multiple formats are supported:

                #. :class:`~.CouplingMap` instance
                #. List, must be given as an edge list representing the two qubit interactions
                   supported by the backend, for example:
                   ``[[0, 1], [0, 3], [1, 2], [1, 5], [2, 5], [4, 1], [5, 3]]``

                If ``coupling_map`` is specified, it must match the number of qubits
                specified in ``num_qubits``. If ``coupling_map`` is not specified,
                a fully connected coupling map will be generated with ``num_qubits``
                qubits.

            control_flow: Flag to enable control flow directives on the target
                (defaults to False).

            calibrate_instructions: Instruction calibration settings, this argument
                supports both boolean and :class:`.InstructionScheduleMap` as
                input types, and is ``None`` by default:

                #. If ``calibrate_instructions==None``, no calibrations will be added to the target.
                #. If ``calibrate_instructions==True``, all gates will be calibrated for all
                    qubits using the default pulse schedules generated internally.
                #. If ``calibrate_instructions==False``, all gates will be "calibrated" for
                    all qubits with an empty pulse schedule.
                #. If an :class:`.InstructionScheduleMap` instance is given, the calibrations
                    in this instruction schedule map will be appended to the target
                    instead of the default pulse schedules (this allows for custom calibrations).

            dtm: System time resolution of output signals in nanoseconds.
                None by default.

            seed: Optional seed for generation of default values.
        """

        super().__init__(
            provider=None,
            name=f"generic_backend_{num_qubits}q",
            description=f"This is a device with {num_qubits} qubits and generic settings.",
            backend_version="",
        )

        self._sim = None
        self._rng = np.random.default_rng(seed=seed)
        self._dtm = dtm
        self._num_qubits = num_qubits
        self._control_flow = control_flow
        self._calibrate_instructions = calibrate_instructions
        self._supported_gates = get_standard_gate_name_mapping()

        if coupling_map is None:
            self._coupling_map = CouplingMap().from_full(num_qubits)
        else:
            if isinstance(coupling_map, CouplingMap):
                self._coupling_map = coupling_map
            else:
                self._coupling_map = CouplingMap(coupling_map)

            if num_qubits != self._coupling_map.size():
                raise QiskitError(
                    f"The number of qubits (got {num_qubits}) must match "
                    f"the size of the provided coupling map (got {self._coupling_map.size()})."
                )

        self._basis_gates = (
            basis_gates if basis_gates is not None else ["cx", "id", "rz", "sx", "x"]
        )
        for name in ["reset", "delay", "measure"]:
            if name not in self._basis_gates:
                self._basis_gates.append(name)

        self._build_generic_target()
        self._build_default_channels()

    @property
    def target(self):
        return self._target

    @property
    def max_circuits(self):
        return None

    @property
    def dtm(self) -> float:
        """Return the system time resolution of output signals"""
        # converting `dtm` from nanoseconds to seconds
        return self._dtm * 1e-9 if self._dtm is not None else None

    @property
    def meas_map(self) -> list[list[int]]:
        return self._target.concurrent_measurements

    def _build_default_channels(self) -> None:
        channels_map = {
            "acquire": {(i,): [pulse.AcquireChannel(i)] for i in range(self.num_qubits)},
            "drive": {(i,): [pulse.DriveChannel(i)] for i in range(self.num_qubits)},
            "measure": {(i,): [pulse.MeasureChannel(i)] for i in range(self.num_qubits)},
            "control": {
                (edge): [pulse.ControlChannel(i)] for i, edge in enumerate(self._coupling_map)
            },
        }
        setattr(self, "channels_map", channels_map)

    def _get_noise_defaults(self, name: str, num_qubits: int) -> tuple:
        """Return noise default values/ranges for duration and error of supported
        instructions. There are two possible formats:
            - (min_duration, max_duration, min_error, max_error),
              if the defaults are ranges.
            - (duration, error), if the defaults are fixed values.
        """
        if name in _NOISE_DEFAULTS:
            return _NOISE_DEFAULTS[name]
        if num_qubits == 1:
            return _NOISE_DEFAULTS_FALLBACK["1-q"]
        return _NOISE_DEFAULTS_FALLBACK["multi-q"]

    def _get_calibration_sequence(
        self, inst: str, num_qubits: int, qargs: tuple[int]
    ) -> list[PulseQobjInstruction]:
        """Return calibration pulse sequence for given instruction (defined by name and num_qubits)
        acting on qargs.
        """

        pulse_library = _PULSE_LIBRARY
        # Note that the calibration pulses are different for
        # 1q gates vs 2q gates vs measurement instructions.
        if inst == "measure":
            sequence = [
                PulseQobjInstruction(
                    name="acquire",
                    duration=1792,
                    t0=0,
                    qubits=qargs,
                    memory_slot=qargs,
                )
            ] + [PulseQobjInstruction(name=pulse_library[1].name, ch=f"m{i}", t0=0) for i in qargs]
            return sequence
        if num_qubits == 1:
            return [
                PulseQobjInstruction(name="fc", ch=f"u{qargs[0]}", t0=0, phase="-P0"),
                PulseQobjInstruction(name=pulse_library[0].name, ch=f"d{qargs[0]}", t0=0),
            ]
        return [
            PulseQobjInstruction(name=pulse_library[1].name, ch=f"d{qargs[0]}", t0=0),
            PulseQobjInstruction(name=pulse_library[2].name, ch=f"u{qargs[0]}", t0=0),
            PulseQobjInstruction(name=pulse_library[1].name, ch=f"d{qargs[1]}", t0=0),
            PulseQobjInstruction(name="fc", ch=f"d{qargs[1]}", t0=0, phase=2.1),
        ]

    def _generate_calibration_defaults(self) -> PulseDefaults:
        """Generate pulse calibration defaults as specified with `self._calibrate_instructions`.
        If `self._calibrate_instructions` is True, the pulse schedules will be generated from
        a series of default calibration sequences. If `self._calibrate_instructions` is False,
        the pulse schedules will contain empty calibration sequences, but still be generated and
        added to the target.
        """

        # If self._calibrate_instructions==True, this method
        # will generate default pulse schedules for all gates in self._basis_gates,
        # except for `delay` and `reset`.
        calibration_buffer = self._basis_gates.copy()
        for inst in ["delay", "reset"]:
            calibration_buffer.remove(inst)

        # List of calibration commands (generated from sequences of PulseQobjInstructions)
        # corresponding to each calibrated instruction. Note that the calibration pulses
        # are different for 1q gates vs 2q gates vs measurement instructions.
        cmd_def = []
        for inst in calibration_buffer:
            num_qubits = self._supported_gates[inst].num_qubits
            qarg_set = self._coupling_map if num_qubits > 1 else list(range(self.num_qubits))
            if inst == "measure":
                cmd_def.append(
                    Command(
                        name=inst,
                        qubits=qarg_set,
                        sequence=(
                            self._get_calibration_sequence(inst, num_qubits, qarg_set)
                            if self._calibrate_instructions
                            else []
                        ),
                    )
                )
            else:
                for qarg in qarg_set:
                    qubits = [qarg] if num_qubits == 1 else qarg
                    cmd_def.append(
                        Command(
                            name=inst,
                            qubits=qubits,
                            sequence=(
                                self._get_calibration_sequence(inst, num_qubits, qubits)
                                if self._calibrate_instructions
                                else []
                            ),
                        )
                    )

        qubit_freq_est = np.random.normal(4.8, scale=0.01, size=self.num_qubits).tolist()
        meas_freq_est = np.linspace(6.4, 6.6, self.num_qubits).tolist()
        return PulseDefaults(
            qubit_freq_est=qubit_freq_est,
            meas_freq_est=meas_freq_est,
            buffer=0,
            pulse_library=_PULSE_LIBRARY,
            cmd_def=cmd_def,
        )

    def _build_generic_target(self):
        """This method generates a :class:`~.Target` instance with
        default qubit, instruction and calibration properties.
        """
        # the qubit properties are sampled from default ranges
        properties = _QUBIT_PROPERTIES
        self._target = Target(
            description=f"Generic Target with {self._num_qubits} qubits",
            num_qubits=self._num_qubits,
            dt=properties["dt"],
            qubit_properties=[
                QubitProperties(
                    t1=self._rng.uniform(properties["t1"][0], properties["t1"][1]),
                    t2=self._rng.uniform(properties["t2"][0], properties["t2"][1]),
                    frequency=self._rng.uniform(
                        properties["frequency"][0], properties["frequency"][1]
                    ),
                )
                for _ in range(self._num_qubits)
            ],
            concurrent_measurements=[list(range(self._num_qubits))],
        )

        # Generate instruction schedule map with calibrations to add to target.
        calibration_inst_map = None
        if self._calibrate_instructions is not None:
            if isinstance(self._calibrate_instructions, InstructionScheduleMap):
                calibration_inst_map = self._calibrate_instructions
            else:
                defaults = self._generate_calibration_defaults()
                calibration_inst_map = defaults.instruction_schedule_map

        # Iterate over gates, generate noise params from defaults,
        # and add instructions, noise and calibrations to target.
        for name in self._basis_gates:
            if name not in self._supported_gates:
                raise QiskitError(
                    f"Provided basis gate {name} is not an instruction "
                    f"in the standard qiskit circuit library."
                )
            gate = self._supported_gates[name]
            noise_params = self._get_noise_defaults(name, gate.num_qubits)
            self._add_noisy_instruction_to_target(gate, noise_params, calibration_inst_map)

        if self._control_flow:
            self._target.add_instruction(IfElseOp, name="if_else")
            self._target.add_instruction(WhileLoopOp, name="while_loop")
            self._target.add_instruction(ForLoopOp, name="for_loop")
            self._target.add_instruction(SwitchCaseOp, name="switch_case")
            self._target.add_instruction(BreakLoopOp, name="break")
            self._target.add_instruction(ContinueLoopOp, name="continue")

    def _add_noisy_instruction_to_target(
        self,
        instruction: Instruction,
        noise_params: tuple[float, ...] | None,
        calibration_inst_map: InstructionScheduleMap | None,
    ) -> None:
        """Add instruction properties to target for specified instruction.

        Args:
            instruction: Instance of instruction to be added to the target
            noise_params: Error and duration noise values/ranges to
                include in instruction properties.
            calibration_inst_map: Instruction schedule map with calibration defaults
        """
        qarg_set = self._coupling_map if instruction.num_qubits > 1 else range(self.num_qubits)
        props = {}
        for qarg in qarg_set:
            try:
                qargs = tuple(qarg)
            except TypeError:
                qargs = (qarg,)
            duration, error = (
                noise_params
                if len(noise_params) == 2
                else (self._rng.uniform(*noise_params[:2]), self._rng.uniform(*noise_params[2:]))
            )
            if (
                calibration_inst_map is not None
                and instruction.name not in ["reset", "delay"]
                and qarg in calibration_inst_map.qubits_with_instruction(instruction.name)
            ):
                # Do NOT call .get method. This parses Qobj immediately.
                # This operation is computationally expensive and should be bypassed.
                calibration_entry = calibration_inst_map._get_calibration_entry(
                    instruction.name, qargs
                )
            else:
                calibration_entry = None
            if duration is not None and len(noise_params) > 2:
                # Ensure exact conversion of duration from seconds to dt
                dt = _QUBIT_PROPERTIES["dt"]
                rounded_duration = round(duration / dt) * dt
                # Clamp rounded duration to be between min and max values
                duration = max(noise_params[0], min(rounded_duration, noise_params[1]))
            props.update({qargs: InstructionProperties(duration, error, calibration_entry)})
        self._target.add_instruction(instruction, props)

        # The "measure" instruction calibrations need to be added qubit by qubit, once the
        # instruction has been added to the target.
        if calibration_inst_map is not None and instruction.name == "measure":
            for qarg in calibration_inst_map.qubits_with_instruction(instruction.name):
                try:
                    qargs = tuple(qarg)
                except TypeError:
                    qargs = (qarg,)
                # Do NOT call .get method. This parses Qobj immediately.
                # This operation is computationally expensive and should be bypassed.
                calibration_entry = calibration_inst_map._get_calibration_entry(
                    instruction.name, qargs
                )
                for qubit in qargs:
                    if qubit < self.num_qubits:
                        self._target[instruction.name][(qubit,)].calibration = calibration_entry

    def run(self, run_input, **options):
        """Run on the backend using a simulator.

        This method runs circuit jobs (an individual or a list of :class:`~.QuantumCircuit`
        ) and pulse jobs (an individual or a list of :class:`~.Schedule` or
        :class:`~.ScheduleBlock`) using :class:`~.BasicSimulator` or Aer simulator and returns a
        :class:`~qiskit.providers.Job` object.

        If qiskit-aer is installed, jobs will be run using the ``AerSimulator`` with
        noise model of the backend. Otherwise, jobs will be run using the
        ``BasicSimulator`` simulator without noise.

        Noisy simulations of pulse jobs are not yet supported in :class:`~.GenericBackendV2`.

        Args:
            run_input (QuantumCircuit or Schedule or ScheduleBlock or list): An
                individual or a list of
                :class:`~qiskit.circuit.QuantumCircuit`,
                :class:`~qiskit.pulse.ScheduleBlock`, or
                :class:`~qiskit.pulse.Schedule` objects to run on the backend.
            options: Any kwarg options to pass to the backend for running the
                config. If a key is also present in the options
                attribute/object, then the expectation is that the value
                specified will be used instead of what's set in the options
                object.

        Returns:
            Job: The job object for the run

        Raises:
            QiskitError: If a pulse job is supplied and qiskit_aer is not installed.
        """
        circuits = run_input
        pulse_job = None
        if isinstance(circuits, (pulse.Schedule, pulse.ScheduleBlock)):
            pulse_job = True
        elif isinstance(circuits, QuantumCircuit):
            pulse_job = False
        elif isinstance(circuits, list):
            if circuits:
                if all(isinstance(x, (pulse.Schedule, pulse.ScheduleBlock)) for x in circuits):
                    pulse_job = True
                elif all(isinstance(x, QuantumCircuit) for x in circuits):
                    pulse_job = False
        if pulse_job is None:  # submitted job is invalid
            raise QiskitError(
                "Invalid input object %s, must be either a "
                "QuantumCircuit, Schedule, or a list of either" % circuits
            )
        if pulse_job:  # pulse job
            raise QiskitError("Pulse simulation is currently not supported for V2 backends.")
        # circuit job
        if not _optionals.HAS_AER:
            warnings.warn("Aer not found using BasicSimulator and no noise", RuntimeWarning)
        if self._sim is None:
            self._setup_sim()
        self._sim._options = self._options
        job = self._sim.run(circuits, **options)
        return job

    def _setup_sim(self) -> None:
        if _optionals.HAS_AER:
            from qiskit_aer import AerSimulator
            from qiskit_aer.noise import NoiseModel

            self._sim = AerSimulator()
            noise_model = NoiseModel.from_backend(self)
            self._sim.set_options(noise_model=noise_model)
            # Update backend default too to avoid overwriting
            # it when run() is called
            self.set_options(noise_model=noise_model)
        else:
            self._sim = BasicSimulator()

    @classmethod
    def _default_options(cls) -> Options:
        with warnings.catch_warnings():  # TODO remove catch once aer release without Provider ABC
            warnings.filterwarnings(
                "ignore",
                category=DeprecationWarning,
                message=".+abstract Provider and ProviderV1.+",
            )
            if _optionals.HAS_AER:
                from qiskit_aer import AerSimulator

                return AerSimulator._default_options()
            else:
                return BasicSimulator._default_options()

    def drive_channel(self, qubit: int):
        drive_channels_map = getattr(self, "channels_map", {}).get("drive", {})
        qubits = (qubit,)
        if qubits in drive_channels_map:
            return drive_channels_map[qubits][0]
        return None

    def measure_channel(self, qubit: int):
        measure_channels_map = getattr(self, "channels_map", {}).get("measure", {})
        qubits = (qubit,)
        if qubits in measure_channels_map:
            return measure_channels_map[qubits][0]
        return None

    def acquire_channel(self, qubit: int):
        acquire_channels_map = getattr(self, "channels_map", {}).get("acquire", {})
        qubits = (qubit,)
        if qubits in acquire_channels_map:
            return acquire_channels_map[qubits][0]
        return None

    def control_channel(self, qubits: Iterable[int]):
        control_channels_map = getattr(self, "channels_map", {}).get("control", {})
        qubits = tuple(qubits)
        if qubits in control_channels_map:
            return control_channels_map[qubits]
        return []
