Skip to content

qubic.job_manager

JobManager

Class for compiling and executing circuits. Contains necessary config objects for compilation, runner for execution, and (optionally) GMMManager for state classification.

Attributes:

Name Type Description
fpga_config FPGAConfig
qchip QChip
circuit_runner CircuitRunner or CircuitRunnerClient
gmm_manager GMMManager
TODO add readout correction and heralding
Source code in qubic/job_manager.py 
 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
113
114
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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
class JobManager:
    """
    Class for compiling and executing circuits. Contains necessary
    config objects for compilation, runner for execution, and 
    (optionally) GMMManager for state classification.

    Attributes
    ----------
    fpga_config : FPGAConfig 
    qchip : qubitconfig.qchip.QChip
    circuit_runner : CircuitRunner or CircuitRunnerClient
    gmm_manager : qubic.state_disc.GMMManager


    TODO: add readout correction and heralding
    """

    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'):
        """
        Parameters
        ----------
        fpga_config: FPGAConfig
        channel_configs: Dict[str, ChannelConfig]
        circuit_runner: AbstractCircuitRunner
        qchip: QChip
        gmm_manager: sd.GMMManager
        target_platform: str
        """

        self.fpga_config = fpga_config
        self.channel_configs = channel_configs
        self.runner = circuit_runner
        self.qchip = qchip
        self.update_gmm(gmm_manager)

    def update_gmm(self, gmm_manager: sd.GMMManager):
        """
        Update the GMMManager used to classify IQ data.

        Parameters
        ----------
        gmm_manager: GMMManager
        """
        if gmm_manager is None: # instantiate empty GMMManager
            self.gmm_manager = sd.GMMManager(chanmap_or_chan_cfgs=self.channel_configs)
        elif isinstance(gmm_manager, str):
            self.gmm_manager = sd.GMMManager(load_file=gmm_manager, chanmap_or_chan_cfgs=self.channel_configs)
        else:
            assert isinstance(gmm_manager, sd.GMMManager)
            if hasattr(self,'gmm_manager'):
                self.gmm_manager.update(gmm_manager)
            else:
                self.gmm_manager=gmm_manager

    def build_and_run_circuits(self, program_list: List, 
                               n_total_shots: int, 
                               outputs : List[str] = ['s11'], 
                               compiler_flags: Dict[str, bool] | CompilerFlags = None, 
                               fit_gmm: bool = False, reads_per_shot: int = 1, 
                               qchip: QChip = None, 
                               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
        ----------
        program_list: list
            list of QubiC circuits (input to compiler layer), CompiledProgram objects,
            or raw_asm_prog dicts.
        n_total_shots: int
            number of shots to run for each circuit
        outputs: list
            list of 's11', 'shots', and/or 'counts'
        compiler_flags: dict | CompilerFlags
            see CompilerFlags definition for allowed keys
        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 qchip is None:
            qchip = self.qchip

        if not isinstance(program_list, list):
            raise TypeError("program_list of invalid type")

        if isinstance(program_list[0], list):
            if 'name' in program_list[0][0]: #this is a gate level program
                self.compiled_progs = tc.run_compile_stage(program_list, self.fpga_config, qchip, 
                                                           compiler_flags=compiler_flags)


                self.raw_asm_progs = tc.run_assemble_stage(self.compiled_progs, self.channel_configs)

            else:
                raise TypeError('{} invalid program list'.format(program_list))

        elif isinstance(program_list[0], dict): #this is raw asm
            #todo: put in check for raw asm
            self.raw_asm_progs = program_list

        elif isinstance(program_list[0], CompiledProgram):
            self.compiled_progs = program_list
            self.raw_asm_progs = tc.run_assemble_stage(self.compiled_progs, self.channel_configs)

        else:
            raise TypeError('{} invalid program list'.format(program_list))
        s11 = self.runner.run_circuit_batch(self.raw_asm_progs, n_total_shots, reads_per_shot,
                                            reload_cmd=reload_cmd, reload_freq=reload_freq, reload_env=reload_env, 
                                            zero_between_reload=zero_between_reload)

        output_dict = {}
        if fit_gmm:
            self.gmm_manager.fit(s11)
            self.gmm_manager.set_labels_maxtomin(s11, [0, 1])

        if 's11' in outputs:
            output_dict['s11'] = s11
        if 'shots' in outputs or 'counts' in outputs:
            shots = self.gmm_manager.predict(s11)
            if 'shots' in outputs:
                output_dict['shots'] = shots
            if 'counts' in outputs:
                output_dict['counts'] = ct.shots_to_batchcounts(shots)

        return output_dict

    def collect_all(self, program_list: List, num_shots_per_circuit: int, 
                    reads_per_shot: int | dict = 1, qchip: QChip = None) -> dict:
        """
        Wrapper around build_and_run_circuits with simplified args. All output data
        products (shots, counts, s11) are provided).

        Parameters
        ----------
        program_list: list
            list of QubiC circuits (input to compiler layer), CompiledProgram objects,
            or raw_asm_prog dicts.
        num_shots_per_circuit: int
            number of shots to run for each circuit
        reads_per_shot: int
        qchip: qubitconfig.qchip.QChip
            if provided, override self.qchip for compilation

        Returns
        -------
            dict
                all result types, keyed by ['s11', 'shots', 'counts']
        """
        output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['s11','shots','counts'],
                                                    reads_per_shot=reads_per_shot, qchip=qchip)
        return {k:output_dict[k] for k in ['s11','shots','counts']}

    def collect_raw_IQ(self, program_list: list, num_shots_per_circuit: int, 
                       reads_per_shot: int | dict = 1, qchip: QChip = None) -> dict:
        """
        Wrapper around build_and_run_circuits with simplified args. Returns integrated,
        unclassified IQ data.

        Parameters
        ----------
        program_list: list
            list of QubiC circuits (input to compiler layer), CompiledProgram objects,
            or raw_asm_prog dicts.
        num_shots_per_circuit: int
            number of shots to run for each circuit
        reads_per_shot: int
        qchip: qubitconfig.qchip.QChip
            if provided, override self.qchip for compilation

        Returns
        -------
        dict
            accumulated s11 shots, keyed by channel index
        """
        output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['s11'],
                                                    reads_per_shot=reads_per_shot, qchip=qchip)
        return output_dict['s11']

    def collect_classified_shots(self, program_list: list, num_shots_per_circuit: int, 
                                 reads_per_shot: int | dict, qchip: QChip = None) -> dict:
        """
        Wrapper around build_and_run_circuits with simplified args. Returns integrated,
        classified IQ data.

        Parameters
        ----------
        program_list: list
            list of QubiC circuits (input to compiler layer), CompiledProgram objects,
            or raw_asm_prog dicts.
        num_shots_per_circuit: int
            number of shots to run for each circuit
        reads_per_shot: int
        qchip: qubitconfig.qchip.QChip
            if provided, override self.qchip for compilation

        Returns
        -------
        dict
            classified shots, keyed by qubit
        """
        output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['shots'], reads_per_shot=reads_per_shot, qchip=qchip)
        return output_dict['shots']

    def collect_counts(self, program_list: list, num_shots_per_circuit: int, 
                       reads_per_shot: int | dict, qchip: QChip = None) -> List[ct.CircuitCounts]:
        """
        Wrapper around build_and_run_circuits with simplified args. Returns list of bitstring
        counts (List[CircuitCounts] object).

        Parameters
        ----------
        program_list: list
            list of QubiC circuits (input to compiler layer), CompiledProgram objects,
            or raw_asm_prog dicts.
        num_shots_per_circuit: int
            number of shots to run for each circuit
        reads_per_shot: int
        qchip: qubitconfig.qchip.QChip
            if provided, override self.qchip for compilation

        Returns
        -------
        List[CircuitCounts]
            results in the form of bitstring counts
        """
        output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['counts'],
                                                    reads_per_shot=reads_per_shot, qchip=qchip)
        return output_dict['counts']

__init__(fpga_config, channel_configs, circuit_runner, qchip, gmm_manager=None, target_platform='rfsoc')

Parameters:

Name Type Description Default
fpga_config FPGAConfig
required
channel_configs Dict[str, ChannelConfig]
required
circuit_runner AbstractCircuitRunner
required
qchip QChip
required
gmm_manager GMMManager
None
target_platform str
'rfsoc'
Source code in qubic/job_manager.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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'):
    """
    Parameters
    ----------
    fpga_config: FPGAConfig
    channel_configs: Dict[str, ChannelConfig]
    circuit_runner: AbstractCircuitRunner
    qchip: QChip
    gmm_manager: sd.GMMManager
    target_platform: str
    """

    self.fpga_config = fpga_config
    self.channel_configs = channel_configs
    self.runner = circuit_runner
    self.qchip = qchip
    self.update_gmm(gmm_manager)

build_and_run_circuits(program_list, n_total_shots, outputs=['s11'], compiler_flags=None, fit_gmm=False, reads_per_shot=1, qchip=None, 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
program_list List

list of QubiC circuits (input to compiler layer), CompiledProgram objects, or raw_asm_prog dicts.

 required
n_total_shots int

number of shots to run for each circuit

 required
outputs List[str]

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

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

see CompilerFlags definition for allowed keys

 None
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/job_manager.py 
 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
113
114
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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def build_and_run_circuits(self, program_list: List, 
                           n_total_shots: int, 
                           outputs : List[str] = ['s11'], 
                           compiler_flags: Dict[str, bool] | CompilerFlags = None, 
                           fit_gmm: bool = False, reads_per_shot: int = 1, 
                           qchip: QChip = None, 
                           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
    ----------
    program_list: list
        list of QubiC circuits (input to compiler layer), CompiledProgram objects,
        or raw_asm_prog dicts.
    n_total_shots: int
        number of shots to run for each circuit
    outputs: list
        list of 's11', 'shots', and/or 'counts'
    compiler_flags: dict | CompilerFlags
        see CompilerFlags definition for allowed keys
    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 qchip is None:
        qchip = self.qchip

    if not isinstance(program_list, list):
        raise TypeError("program_list of invalid type")

    if isinstance(program_list[0], list):
        if 'name' in program_list[0][0]: #this is a gate level program
            self.compiled_progs = tc.run_compile_stage(program_list, self.fpga_config, qchip, 
                                                       compiler_flags=compiler_flags)


            self.raw_asm_progs = tc.run_assemble_stage(self.compiled_progs, self.channel_configs)

        else:
            raise TypeError('{} invalid program list'.format(program_list))

    elif isinstance(program_list[0], dict): #this is raw asm
        #todo: put in check for raw asm
        self.raw_asm_progs = program_list

    elif isinstance(program_list[0], CompiledProgram):
        self.compiled_progs = program_list
        self.raw_asm_progs = tc.run_assemble_stage(self.compiled_progs, self.channel_configs)

    else:
        raise TypeError('{} invalid program list'.format(program_list))
    s11 = self.runner.run_circuit_batch(self.raw_asm_progs, n_total_shots, reads_per_shot,
                                        reload_cmd=reload_cmd, reload_freq=reload_freq, reload_env=reload_env, 
                                        zero_between_reload=zero_between_reload)

    output_dict = {}
    if fit_gmm:
        self.gmm_manager.fit(s11)
        self.gmm_manager.set_labels_maxtomin(s11, [0, 1])

    if 's11' in outputs:
        output_dict['s11'] = s11
    if 'shots' in outputs or 'counts' in outputs:
        shots = self.gmm_manager.predict(s11)
        if 'shots' in outputs:
            output_dict['shots'] = shots
        if 'counts' in outputs:
            output_dict['counts'] = ct.shots_to_batchcounts(shots)

    return output_dict

collect_all(program_list, num_shots_per_circuit, reads_per_shot=1, qchip=None)

Wrapper around build_and_run_circuits with simplified args. All output data products (shots, counts, s11) are provided).

Parameters:

Name Type Description Default
program_list List

list of QubiC circuits (input to compiler layer), CompiledProgram objects, or raw_asm_prog dicts.

 required
num_shots_per_circuit int

number of shots to run for each circuit

 required
reads_per_shot int | dict
1
qchip QChip

if provided, override self.qchip for compilation

 None

Returns:

Type Description
dict

all result types, keyed by ['s11', 'shots', 'counts']
Source code in qubic/job_manager.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def collect_all(self, program_list: List, num_shots_per_circuit: int, 
                reads_per_shot: int | dict = 1, qchip: QChip = None) -> dict:
    """
    Wrapper around build_and_run_circuits with simplified args. All output data
    products (shots, counts, s11) are provided).

    Parameters
    ----------
    program_list: list
        list of QubiC circuits (input to compiler layer), CompiledProgram objects,
        or raw_asm_prog dicts.
    num_shots_per_circuit: int
        number of shots to run for each circuit
    reads_per_shot: int
    qchip: qubitconfig.qchip.QChip
        if provided, override self.qchip for compilation

    Returns
    -------
        dict
            all result types, keyed by ['s11', 'shots', 'counts']
    """
    output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['s11','shots','counts'],
                                                reads_per_shot=reads_per_shot, qchip=qchip)
    return {k:output_dict[k] for k in ['s11','shots','counts']}

collect_classified_shots(program_list, num_shots_per_circuit, reads_per_shot, qchip=None)

Wrapper around build_and_run_circuits with simplified args. Returns integrated, classified IQ data.

Parameters:

Name Type Description Default
program_list list

list of QubiC circuits (input to compiler layer), CompiledProgram objects, or raw_asm_prog dicts.

 required
num_shots_per_circuit int

number of shots to run for each circuit

 required
reads_per_shot int | dict
required
qchip QChip

if provided, override self.qchip for compilation

 None

Returns:

Type Description
dict

classified shots, keyed by qubit
Source code in qubic/job_manager.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def collect_classified_shots(self, program_list: list, num_shots_per_circuit: int, 
                             reads_per_shot: int | dict, qchip: QChip = None) -> dict:
    """
    Wrapper around build_and_run_circuits with simplified args. Returns integrated,
    classified IQ data.

    Parameters
    ----------
    program_list: list
        list of QubiC circuits (input to compiler layer), CompiledProgram objects,
        or raw_asm_prog dicts.
    num_shots_per_circuit: int
        number of shots to run for each circuit
    reads_per_shot: int
    qchip: qubitconfig.qchip.QChip
        if provided, override self.qchip for compilation

    Returns
    -------
    dict
        classified shots, keyed by qubit
    """
    output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['shots'], reads_per_shot=reads_per_shot, qchip=qchip)
    return output_dict['shots']

collect_counts(program_list, num_shots_per_circuit, reads_per_shot, qchip=None)

Wrapper around build_and_run_circuits with simplified args. Returns list of bitstring counts (List[CircuitCounts] object).

Parameters:

Name Type Description Default
program_list list

list of QubiC circuits (input to compiler layer), CompiledProgram objects, or raw_asm_prog dicts.

 required
num_shots_per_circuit int

number of shots to run for each circuit

 required
reads_per_shot int | dict
required
qchip QChip

if provided, override self.qchip for compilation

 None

Returns:

Type Description
List[CircuitCounts]

results in the form of bitstring counts
Source code in qubic/job_manager.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def collect_counts(self, program_list: list, num_shots_per_circuit: int, 
                   reads_per_shot: int | dict, qchip: QChip = None) -> List[ct.CircuitCounts]:
    """
    Wrapper around build_and_run_circuits with simplified args. Returns list of bitstring
    counts (List[CircuitCounts] object).

    Parameters
    ----------
    program_list: list
        list of QubiC circuits (input to compiler layer), CompiledProgram objects,
        or raw_asm_prog dicts.
    num_shots_per_circuit: int
        number of shots to run for each circuit
    reads_per_shot: int
    qchip: qubitconfig.qchip.QChip
        if provided, override self.qchip for compilation

    Returns
    -------
    List[CircuitCounts]
        results in the form of bitstring counts
    """
    output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['counts'],
                                                reads_per_shot=reads_per_shot, qchip=qchip)
    return output_dict['counts']

collect_raw_IQ(program_list, num_shots_per_circuit, reads_per_shot=1, qchip=None)

Wrapper around build_and_run_circuits with simplified args. Returns integrated, unclassified IQ data.

Parameters:

Name Type Description Default
program_list list

list of QubiC circuits (input to compiler layer), CompiledProgram objects, or raw_asm_prog dicts.

 required
num_shots_per_circuit int

number of shots to run for each circuit

 required
reads_per_shot int | dict
1
qchip QChip

if provided, override self.qchip for compilation

 None

Returns:

Type Description
dict

accumulated s11 shots, keyed by channel index
Source code in qubic/job_manager.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def collect_raw_IQ(self, program_list: list, num_shots_per_circuit: int, 
                   reads_per_shot: int | dict = 1, qchip: QChip = None) -> dict:
    """
    Wrapper around build_and_run_circuits with simplified args. Returns integrated,
    unclassified IQ data.

    Parameters
    ----------
    program_list: list
        list of QubiC circuits (input to compiler layer), CompiledProgram objects,
        or raw_asm_prog dicts.
    num_shots_per_circuit: int
        number of shots to run for each circuit
    reads_per_shot: int
    qchip: qubitconfig.qchip.QChip
        if provided, override self.qchip for compilation

    Returns
    -------
    dict
        accumulated s11 shots, keyed by channel index
    """
    output_dict = self.build_and_run_circuits(program_list, num_shots_per_circuit, ['s11'],
                                                reads_per_shot=reads_per_shot, qchip=qchip)
    return output_dict['s11']

update_gmm(gmm_manager)

Update the GMMManager used to classify IQ data.

Parameters:

Name Type Description Default
gmm_manager GMMManager
required
Source code in qubic/job_manager.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def update_gmm(self, gmm_manager: sd.GMMManager):
    """
    Update the GMMManager used to classify IQ data.

    Parameters
    ----------
    gmm_manager: GMMManager
    """
    if gmm_manager is None: # instantiate empty GMMManager
        self.gmm_manager = sd.GMMManager(chanmap_or_chan_cfgs=self.channel_configs)
    elif isinstance(gmm_manager, str):
        self.gmm_manager = sd.GMMManager(load_file=gmm_manager, chanmap_or_chan_cfgs=self.channel_configs)
    else:
        assert isinstance(gmm_manager, sd.GMMManager)
        if hasattr(self,'gmm_manager'):
            self.gmm_manager.update(gmm_manager)
        else:
            self.gmm_manager=gmm_manager