Skip to content

qubic.pygsti.job_manager

PyGSTiJobManager

Bases: JobManager

Wrapper around JobManager class for compiling and running PyGSTi circuits

Source code in qubic/pygsti/job_manager.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class PyGSTiJobManager(JobManager):
    """
    Wrapper around JobManager class for compiling and running PyGSTi circuits
    """
    def __init__(self,                  
                 fpga_config: FPGAConfig, 
                 channel_configs: Dict[str, ChannelConfig], 
                 circuit_runner: AbstractCircuitRunner,
                 qchip: QChip,
                 gmm_manager: sd.GMMManager = None, 
                 target_platform: str = 'rfsoc'):

        if gmm_manager is None:
            raise(ValueError("PyGSTiJobManager needs a gmm_manager!"))
        super().__init__(fpga_config, channel_configs, circuit_runner, qchip,
                 gmm_manager, target_platform)

    def build_and_run_circuits(self, pygsti_circuits: List[pygsti.circuits.Circuit], 
                               qubit_map: List[str] | Dict[int, str],
                               n_total_shots: int, 
                               outputs : List[str] = ['pygsti_results'], 
                               compiler_flags: Dict[str, bool] | CompilerFlags = None, 
                               gateware_twirl = False,
                               fit_gmm: bool = False, 
                               reads_per_shot: int = 1, 
                               qchip: QChip = None, 
                               delay_before_circuit: float = 500.e-6,
                               reload_cmd: bool = True, 
                               reload_freq: bool = True, 
                               reload_env: bool = True, 
                               zero_between_reload: bool = True) -> dict:
        """
        Compile and run provided list of circuits. Output data products/analysis are controlled 
        by 'output' parameter. 

        Parameters
        ----------
        pygsti_circuits: list
            list of pygsti circuits to run
        qubit_map: List[str] | Dict[int, str]
            Maps pygsti quantum registers to physical qubits. If list
            of physical qubits, register 0 is assumed to correspond to the
            first element, etc. If dict, register -> qubit mapping is given
            directly as key value pairs.
        n_total_shots: int
            number of shots to run for each circuit
        outputs: list
            list of 's11', 'shots', 'pygsti_results', and/or 'counts'
        compiler_flags: dict | CompilerFlags
            see CompilerFlags definition for allowed keys
        gateware_twirl: bool = False
        fit_gmm: bool
        qchip: qubitconfig.qchip.QChip
            if provided, override self.qchip for compilation
        reads_per_shot : int
            number of reads (measurements) per qubit in each instruction
        reload_cmd: bool
        reload_freq: bool
        reload_env: bool
        zero_between_reload: bool

        Returns
        -------
        dict
            results with keys/types matching the provided 'outputs'

        """
        if reads_per_shot != 1:
            raise Exception(f'Multiple reads_per_shot ({reads_per_shot}) not supported')
        qubic_outputs = copy.copy(outputs)
        if 'pygsti_results' in outputs:
            qubic_outputs.remove('pygsti_results')
            qubic_outputs.append('counts')

        qubic_circuits = []
        for circuit in pygsti_circuits:
            qubic_circuits.append(transpile(circuit, qubit_map, gateware_twirl, delay_before_circuit))

        if isinstance(compiler_flags, dict):
            compiler_flags = CompilerFlags(**compiler_flags)

        compiler_flags.rc = gateware_twirl

        qubic_output_dict = super().build_and_run_circuits(qubic_circuits, n_total_shots, qubic_outputs, 
                                                                    compiler_flags=compiler_flags, fit_gmm=fit_gmm,
                                                                    reads_per_shot=reads_per_shot, qchip=qchip,
                                                                    reload_cmd=reload_cmd, reload_env=reload_env,
                                                                    reload_freq=reload_freq, 
                                                                    zero_between_reload=zero_between_reload)

        output_dict = {}
        for output_type in outputs:
            if output_type == 'pygsti_results':
                output_dict['pygsti_results'] = map_results(pygsti_circuits, qubic_output_dict['counts'], qubit_map)
            else:
                output_dict[output_type] = qubic_output_dict[output_type]

        return output_dict

build_and_run_circuits(pygsti_circuits, qubit_map, n_total_shots, outputs=['pygsti_results'], compiler_flags=None, gateware_twirl=False, fit_gmm=False, reads_per_shot=1, qchip=None, delay_before_circuit=0.0005, reload_cmd=True, reload_freq=True, reload_env=True, zero_between_reload=True)

Compile and run provided list of circuits. Output data products/analysis are controlled by 'output' parameter.

Parameters:

Name Type Description Default
pygsti_circuits List[Circuit]

list of pygsti circuits to run

 required
qubit_map List[str] | Dict[int, str]

Maps pygsti quantum registers to physical qubits. If list of physical qubits, register 0 is assumed to correspond to the first element, etc. If dict, register -> qubit mapping is given directly as key value pairs.

 required
n_total_shots int

number of shots to run for each circuit

 required
outputs List[str]

list of 's11', 'shots', 'pygsti_results', and/or 'counts'

 ['pygsti_results']
compiler_flags Dict[str, bool] | CompilerFlags

see CompilerFlags definition for allowed keys

 None
gateware_twirl
False
fit_gmm bool
False
qchip QChip

if provided, override self.qchip for compilation

 None
reads_per_shot int

number of reads (measurements) per qubit in each instruction

 1
reload_cmd bool
True
reload_freq bool
True
reload_env bool
True
zero_between_reload bool
True

Returns:

Type Description
dict

results with keys/types matching the provided 'outputs'
Source code in qubic/pygsti/job_manager.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def build_and_run_circuits(self, pygsti_circuits: List[pygsti.circuits.Circuit], 
                           qubit_map: List[str] | Dict[int, str],
                           n_total_shots: int, 
                           outputs : List[str] = ['pygsti_results'], 
                           compiler_flags: Dict[str, bool] | CompilerFlags = None, 
                           gateware_twirl = False,
                           fit_gmm: bool = False, 
                           reads_per_shot: int = 1, 
                           qchip: QChip = None, 
                           delay_before_circuit: float = 500.e-6,
                           reload_cmd: bool = True, 
                           reload_freq: bool = True, 
                           reload_env: bool = True, 
                           zero_between_reload: bool = True) -> dict:
    """
    Compile and run provided list of circuits. Output data products/analysis are controlled 
    by 'output' parameter. 

    Parameters
    ----------
    pygsti_circuits: list
        list of pygsti circuits to run
    qubit_map: List[str] | Dict[int, str]
        Maps pygsti quantum registers to physical qubits. If list
        of physical qubits, register 0 is assumed to correspond to the
        first element, etc. If dict, register -> qubit mapping is given
        directly as key value pairs.
    n_total_shots: int
        number of shots to run for each circuit
    outputs: list
        list of 's11', 'shots', 'pygsti_results', and/or 'counts'
    compiler_flags: dict | CompilerFlags
        see CompilerFlags definition for allowed keys
    gateware_twirl: bool = False
    fit_gmm: bool
    qchip: qubitconfig.qchip.QChip
        if provided, override self.qchip for compilation
    reads_per_shot : int
        number of reads (measurements) per qubit in each instruction
    reload_cmd: bool
    reload_freq: bool
    reload_env: bool
    zero_between_reload: bool

    Returns
    -------
    dict
        results with keys/types matching the provided 'outputs'

    """
    if reads_per_shot != 1:
        raise Exception(f'Multiple reads_per_shot ({reads_per_shot}) not supported')
    qubic_outputs = copy.copy(outputs)
    if 'pygsti_results' in outputs:
        qubic_outputs.remove('pygsti_results')
        qubic_outputs.append('counts')

    qubic_circuits = []
    for circuit in pygsti_circuits:
        qubic_circuits.append(transpile(circuit, qubit_map, gateware_twirl, delay_before_circuit))

    if isinstance(compiler_flags, dict):
        compiler_flags = CompilerFlags(**compiler_flags)

    compiler_flags.rc = gateware_twirl

    qubic_output_dict = super().build_and_run_circuits(qubic_circuits, n_total_shots, qubic_outputs, 
                                                                compiler_flags=compiler_flags, fit_gmm=fit_gmm,
                                                                reads_per_shot=reads_per_shot, qchip=qchip,
                                                                reload_cmd=reload_cmd, reload_env=reload_env,
                                                                reload_freq=reload_freq, 
                                                                zero_between_reload=zero_between_reload)

    output_dict = {}
    for output_type in outputs:
        if output_type == 'pygsti_results':
            output_dict['pygsti_results'] = map_results(pygsti_circuits, qubic_output_dict['counts'], qubit_map)
        else:
            output_dict[output_type] = qubic_output_dict[output_type]

    return output_dict

map_results(circuits, batchcounts, label_to_qubit)

Convert native list of CircuitCounts objects to a PyGSTi Dataset object

Parameters:

Name Type Description Default
circuits List[Circuit]
required
batchcounts List[CircuitCounts]
required
label_to_qubit list or dict

mapping from pygsti registers (indices) to physical qubits

 required
Source code in qubic/pygsti/job_manager.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def map_results(circuits: List[pygsti.circuits.Circuit], 
                batchcounts: List[CircuitCounts], 
                label_to_qubit: Dict[int, str] | List):
    """
    Convert native list of CircuitCounts objects to a PyGSTi Dataset object

    Parameters
    ----------
    circuits: List[pygsti.circuits.Circuit]
    batchcounts : List[CircuitCounts]
    label_to_qubit : list or dict
        mapping from pygsti registers (indices) to physical qubits
    """
    pygsti_dataset = pygsti.data.DataSet()
    if isinstance(label_to_qubit, dict):
        qubit_to_label = {qubit: label for label, qubit in label_to_qubit.items()}
    elif isinstance(label_to_qubit, list) or isinstance(label_to_qubit, np.ndarray):
        qubit_to_label = {qubit: label for label, qubit in enumerate(label_to_qubit)}

    for circuit, circuitcounts in zip(circuits, batchcounts):
        if list(circuitcounts.bitstring_dict.values())[0].shape[0] != 1:
            raise Exception('Multiple reads per shot not supported!')

        qubic_to_pygsti_bitstring = {}
        sorted_pygsti_labels = sorted([label for qubit, label in qubit_to_label.items() if qubit in circuitcounts.qubits])
        pygsti_labels_to_index = {label: sorted_pygsti_labels.index(label) for label in sorted_pygsti_labels}
        for pygsti_bitstring in circuitcounts.bitstring_dict.keys():
            #enumerate all bitstrings as pygsti strings, then backsolve for qubic mapping
            qb_bitstring = ''.join([pygsti_bitstring[pygsti_labels_to_index[qubit_to_label[circuitcounts.qubits[i]]]] for i in range(len(pygsti_bitstring))])
            qubic_to_pygsti_bitstring[qb_bitstring] = pygsti_bitstring

        result_dict = {qubic_to_pygsti_bitstring[bitstring]: counts for 
                       bitstring, counts in circuitcounts.bitstring_dict.items()}
        pygsti_dataset[circuit] = result_dict

    return pygsti_dataset