7087368175

Committed 04 Dec 2023 01:33PM UTC coverage: 43.134% (-0.5%) from 43.641%

Build # 7087368175

Build Type

push

github

Committed by

feihoo87

Commit Message

fix workflow

Run Details

7413 of 17186 relevant lines covered (43.13%)

2.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.52

/waveforms/qlisp/base.py

from __future__ import annotations

import warnings
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass, field
from enum import Flag, auto
from functools import cached_property
from typing import Any, NamedTuple, Optional, Sequence, Union

import numpy as np

from ..waveform import Waveform


class Signal(Flag):
    trace = auto()
    iq = auto()
    state = auto()

    _avg_trace = auto()
    _avg_iq = auto()
    _avg_state = auto()
    _count = auto()
    _remote = auto()

    trace_avg = trace | _avg_trace

    iq_avg = iq | _avg_iq

    population = state | _avg_state
    count = state | _count
    diag = state | _count | _avg_state

    remote_trace_avg = trace_avg | _remote
    remote_iq_avg = iq_avg | _remote
    remote_state = state | _remote
    remote_population = population | _remote
    remote_count = count | _remote


def head(st):
    if isinstance(st[0], str):
        return st[0]
    else:
        return head(st[0])


class QLispError(SyntaxError):
    pass


class Capture(NamedTuple):
    qubit: str
    cbit: tuple[str, int]
    time: float
    signal: Signal
    params: dict
    hardware: Union[ADChannel, MultADChannel] = None
    shift: float = 0


class AWGChannel(NamedTuple):
    name: str
    sampleRate: float
    size: int = -1
    amplitude: Optional[float] = None
    offset: Optional[float] = None
    delay: float = 0
    sos: Optional[np.ndarray] = None
    commandAddresses: tuple = ()


class MultAWGChannel(NamedTuple):
    I: Optional[AWGChannel] = None
    Q: Optional[AWGChannel] = None
    LO: Optional[str] = None
    lo_freq: float = -1
    lo_power: Optional[float] = None


class ADChannel(NamedTuple):
    name: str
    sampleRate: float = 1e9
    trigger: str = ''
    triggerDelay: float = 0
    triggerClockCycle: float = 8e-9
    commandAddresses: tuple = ()


class MultADChannel(NamedTuple):
    I: Optional[ADChannel] = None
    Q: Optional[ADChannel] = None
    IQ: Optional[ADChannel] = None
    Ref: Optional[ADChannel] = None
    LO: Optional[str] = None
    lo_freq: float = -1
    lo_power: Optional[float] = None


class GateConfig(NamedTuple):
    name: str
    qubits: tuple
    type: str = 'default'
    params: dict = {}


class ABCCompileConfigMixin(ABC):
    """
    Mixin for configs that can be used by compiler.
    """

    @abstractmethod
    def _getAWGChannel(self, name,
                       *qubits) -> Union[AWGChannel, MultAWGChannel]:
        """
        Get AWG channel by name and qubits.
        """
        pass

    @abstractmethod
    def _getADChannel(self, qubit) -> Union[ADChannel, MultADChannel]:
        """
        Get AD channel by qubit.
        """
        pass

    @abstractmethod
    def _getGateConfig(self, name, *qubits) -> GateConfig:
        """
        Return the gate config for the given qubits.

        Args:
            name: Name of the gate.
            qubits: Qubits to which the gate is applied.

        Returns:
            GateConfig for the given qubits.
            if the gate is not found, return None.
        """
        pass

    @abstractmethod
    def _getAllQubitLabels(self) -> list[str]:
        """
        Return all qubit labels.
        """
        pass


__config_factory = None


def set_config_factory(factory):
    global __config_factory
    __config_factory = factory


def getConfig() -> ABCCompileConfigMixin:
    if __config_factory is None:
        raise FileNotFoundError(
            'set_config_factory(factory) must be run first.')
    else:
        return __config_factory()


@dataclass
class Context():
    cfg: ABCCompileConfigMixin = field(default_factory=getConfig)
    scopes: list[dict[str, Any]] = field(default_factory=lambda: [dict()])
    qlisp: list = field(default_factory=list)
    time: dict[str,
               float] = field(default_factory=lambda: defaultdict(lambda: 0))
    addressTable: dict = field(default_factory=dict)
    waveforms: dict[str, list[Waveform]] = field(
        default_factory=lambda: defaultdict(list))
    measures: dict[str, dict[int, Capture]] = field(
        default_factory=lambda: defaultdict(dict))
    phases_ext: dict[str, dict[Union[int, str], float]] = field(
        default_factory=lambda: defaultdict(lambda: defaultdict(lambda: 0)))
    biases: dict[str,
                 float] = field(default_factory=lambda: defaultdict(lambda: 0))
    end: float = 0

    cache: dict = field(default_factory=dict)

    @property
    def channel(self):
        return self.raw_waveforms

    @property
    def phases(self):

        class D():
            __slots__ = ('ctx', )

            def __init__(self, ctx):
                self.ctx = ctx

            def __getitem__(self, qubit):
                return self.ctx.phases_ext[qubit][1] - self.ctx.phases_ext[
                    qubit][0]

            def __setitem__(self, qubit, phase):
                self.ctx.phases_ext[qubit][
                    1] = phase + self.ctx.phases_ext[qubit][0]

        return D(self)

    @property
    def params(self):
        return self.scopes[-1]

    @property
    def vars(self):
        return self.scopes[-2]

    @property
    def globals(self):
        return self.scopes[0]

    @cached_property
    def all_qubits(self):
        return self.cfg._getAllQubitLabels()

    def get_gate_config(self, name: str, qubits: tuple,
                        type: str) -> GateConfig:
        return self.cfg._getGateConfig(name, *qubits, type=type)

    def get_awg_channel(self, name: str,
                        qubits: tuple) -> AWGChannel | MultAWGChannel:
        try:
            return self.cache[(1, name, *qubits)]
        except:
            return self.cache.setdefault(
                (1, name, *qubits), self.cfg._getAWGChannel(name, *qubits))

    def get_ad_channel(self, qubit: str | int) -> ADChannel | MultADChannel:
        try:
            return self.cache[(2, qubit)]
        except:
            return self.cache.setdefault((2, qubit),
                                         self.cfg._getADChannel(qubit))

    def qubit(self, q):
        return self.addressTable[q]


@dataclass
class QLispCode():
    cfg: ABCCompileConfigMixin = field(repr=False)
    qlisp: list = field(repr=True)
    waveforms: dict[str, Waveform] = field(repr=True)
    measures: dict[tuple[str, int], list[Capture]] = field(repr=True)
    end: float = field(default=0, repr=True)
    signal: Signal = Signal.state
    shots: int = 1024
    arch: str = 'general'
    cbit_alias: dict[int, tuple[int, int]] = field(default_factory=dict)
    sub_code_count: int = 0


def set_context_factory(factory):
    warnings.warn('set_context_factory is deprecated', DeprecationWarning, 2)


def create_context(ctx: Optional[Context] = None, **kw) -> Context:
    if ctx is None:
        return Context(**kw)
    else:
        if 'cfg' not in kw:
            kw['cfg'] = ctx.cfg
        sub_ctx = Context(**kw)
        sub_ctx.time.update(ctx.time)
        # sub_ctx.phases.update(ctx.phases)
        sub_ctx.biases.update(ctx.biases)
        for k, v in ctx.phases_ext.items():
            sub_ctx.phases_ext[k].update(v)

        return sub_ctx

1	from __future__ import annotations	6✔
2
3	import warnings	6✔
4	from abc import ABC, abstractmethod	6✔
5	from collections import defaultdict	6✔
6	from dataclasses import dataclass, field	6✔
7	from enum import Flag, auto	6✔
8	from functools import cached_property	6✔
9	from typing import Any, NamedTuple, Optional, Sequence, Union	6✔
10
11	import numpy as np	6✔
12
13	from ..waveform import Waveform	6✔
14
15
16	class Signal(Flag):	6✔
17	trace = auto()	6✔
18	iq = auto()	6✔
19	state = auto()	6✔
20
21	_avg_trace = auto()	6✔
22	_avg_iq = auto()	6✔
23	_avg_state = auto()	6✔
24	_count = auto()	6✔
25	_remote = auto()	6✔
26
27	trace_avg = trace \| _avg_trace	6✔
28
29	iq_avg = iq \| _avg_iq	6✔
30
31	population = state \| _avg_state	6✔
32	count = state \| _count	6✔
33	diag = state \| _count \| _avg_state	6✔
34
35	remote_trace_avg = trace_avg \| _remote	6✔
36	remote_iq_avg = iq_avg \| _remote	6✔
37	remote_state = state \| _remote	6✔
38	remote_population = population \| _remote	6✔
39	remote_count = count \| _remote	6✔
40
41
42	def head(st):	6✔
43	if isinstance(st[0], str):	6✔
44	return st[0]	6✔
45	else:
46	return head(st[0])	6✔
47
48
49	class QLispError(SyntaxError):	6✔
50	pass	6✔
51
52
53	class Capture(NamedTuple):	6✔
54	qubit: str	6✔
55	cbit: tuple[str, int]	6✔
56	time: float	6✔
57	signal: Signal	6✔
58	params: dict	6✔
59	hardware: Union[ADChannel, MultADChannel] = None	6✔
60	shift: float = 0	6✔
61
62
63	class AWGChannel(NamedTuple):	6✔
64	name: str	6✔
65	sampleRate: float	6✔
66	size: int = -1	6✔
67	amplitude: Optional[float] = None	6✔
68	offset: Optional[float] = None	6✔
69	delay: float = 0	6✔
70	sos: Optional[np.ndarray] = None	6✔
71	commandAddresses: tuple = ()	6✔
72
73
74	class MultAWGChannel(NamedTuple):	6✔
75	I: Optional[AWGChannel] = None	6✔
76	Q: Optional[AWGChannel] = None	6✔
77	LO: Optional[str] = None	6✔
78	lo_freq: float = -1	6✔
79	lo_power: Optional[float] = None	6✔
80
81
82	class ADChannel(NamedTuple):	6✔
83	name: str	6✔
84	sampleRate: float = 1e9	6✔
85	trigger: str = ''	6✔
86	triggerDelay: float = 0	6✔
87	triggerClockCycle: float = 8e-9	6✔
88	commandAddresses: tuple = ()	6✔
89
90
91	class MultADChannel(NamedTuple):	6✔
92	I: Optional[ADChannel] = None	6✔
93	Q: Optional[ADChannel] = None	6✔
94	IQ: Optional[ADChannel] = None	6✔
95	Ref: Optional[ADChannel] = None	6✔
96	LO: Optional[str] = None	6✔
97	lo_freq: float = -1	6✔
98	lo_power: Optional[float] = None	6✔
99
100
101	class GateConfig(NamedTuple):	6✔
102	name: str	6✔
103	qubits: tuple	6✔
104	type: str = 'default'	6✔
105	params: dict = {}	6✔
106
107
108	class ABCCompileConfigMixin(ABC):	6✔
109	"""
110	Mixin for configs that can be used by compiler.
111	"""
112
113	@abstractmethod	6✔
114	def _getAWGChannel(self, name,	6✔
115	*qubits) -> Union[AWGChannel, MultAWGChannel]:
116	"""
117	Get AWG channel by name and qubits.
118	"""
119	pass	×
120
121	@abstractmethod	6✔
122	def _getADChannel(self, qubit) -> Union[ADChannel, MultADChannel]:	6✔
123	"""
124	Get AD channel by qubit.
125	"""
126	pass	×
127
128	@abstractmethod	6✔
129	def _getGateConfig(self, name, *qubits) -> GateConfig:	6✔
130	"""
131	Return the gate config for the given qubits.
132
133	Args:
134	name: Name of the gate.
135	qubits: Qubits to which the gate is applied.
136
137	Returns:
138	GateConfig for the given qubits.
139	if the gate is not found, return None.
140	"""
141	pass	×
142
143	@abstractmethod	6✔
144	def _getAllQubitLabels(self) -> list[str]:	6✔
145	"""
146	Return all qubit labels.
147	"""
148	pass	×
149
150
151	__config_factory = None	6✔
152
153
154	def set_config_factory(factory):	6✔
155	global __config_factory
156	__config_factory = factory	×
157
158
159	def getConfig() -> ABCCompileConfigMixin:	6✔
160	if __config_factory is None:	×
161	raise FileNotFoundError(	×
162	'set_config_factory(factory) must be run first.')
163	else:
164	return __config_factory()	×
165
166
167	@dataclass	6✔
168	class Context():	6✔
169	cfg: ABCCompileConfigMixin = field(default_factory=getConfig)	6✔
170	scopes: list[dict[str, Any]] = field(default_factory=lambda: [dict()])	6✔
171	qlisp: list = field(default_factory=list)	6✔
172	time: dict[str,	6✔
173	float] = field(default_factory=lambda: defaultdict(lambda: 0))
174	addressTable: dict = field(default_factory=dict)	6✔
175	waveforms: dict[str, list[Waveform]] = field(	6✔
176	default_factory=lambda: defaultdict(list))
177	measures: dict[str, dict[int, Capture]] = field(	6✔
178	default_factory=lambda: defaultdict(dict))
179	phases_ext: dict[str, dict[Union[int, str], float]] = field(	6✔
180	default_factory=lambda: defaultdict(lambda: defaultdict(lambda: 0)))
181	biases: dict[str,	6✔
182	float] = field(default_factory=lambda: defaultdict(lambda: 0))
183	end: float = 0	6✔
184
185	cache: dict = field(default_factory=dict)	6✔
186
187	@property	6✔
188	def channel(self):	6✔
189	return self.raw_waveforms	×
190
191	@property	6✔
192	def phases(self):	6✔
193
194	class D():	×
195	__slots__ = ('ctx', )	×
196
197	def __init__(self, ctx):	×
198	self.ctx = ctx	×
199
200	def __getitem__(self, qubit):	×
201	return self.ctx.phases_ext[qubit][1] - self.ctx.phases_ext[	×
202	qubit][0]
203
204	def __setitem__(self, qubit, phase):	×
205	self.ctx.phases_ext[qubit][	×
206	1] = phase + self.ctx.phases_ext[qubit][0]
207
208	return D(self)	×
209
210	@property	6✔
211	def params(self):	6✔
212	return self.scopes[-1]	6✔
213
214	@property	6✔
215	def vars(self):	6✔
216	return self.scopes[-2]	×
217
218	@property	6✔
219	def globals(self):	6✔
220	return self.scopes[0]	×
221
222	@cached_property	6✔
223	def all_qubits(self):	6✔
224	return self.cfg._getAllQubitLabels()	6✔
225
226	def get_gate_config(self, name: str, qubits: tuple,	6✔
227	type: str) -> GateConfig:
228	return self.cfg._getGateConfig(name, *qubits, type=type)	6✔
229
230	def get_awg_channel(self, name: str,	6✔
231	qubits: tuple) -> AWGChannel \| MultAWGChannel:
232	try:	6✔
233	return self.cache[(1, name, *qubits)]	6✔
234	except:	6✔
235	return self.cache.setdefault(	6✔
236	(1, name, qubits), self.cfg._getAWGChannel(name, qubits))
237
238	def get_ad_channel(self, qubit: str \| int) -> ADChannel \| MultADChannel:	6✔
239	try:	6✔
240	return self.cache[(2, qubit)]	6✔
241	except:	6✔
242	return self.cache.setdefault((2, qubit),	6✔
243	self.cfg._getADChannel(qubit))
244
245	def qubit(self, q):	6✔
246	return self.addressTable[q]	6✔
247
248
249	@dataclass	6✔
250	class QLispCode():	6✔
251	cfg: ABCCompileConfigMixin = field(repr=False)	6✔
252	qlisp: list = field(repr=True)	6✔
253	waveforms: dict[str, Waveform] = field(repr=True)	6✔
254	measures: dict[tuple[str, int], list[Capture]] = field(repr=True)	6✔
255	end: float = field(default=0, repr=True)	6✔
256	signal: Signal = Signal.state	6✔
257	shots: int = 1024	6✔
258	arch: str = 'general'	6✔
259	cbit_alias: dict[int, tuple[int, int]] = field(default_factory=dict)	6✔
260	sub_code_count: int = 0	6✔
261
262
263	def set_context_factory(factory):	6✔
264	warnings.warn('set_context_factory is deprecated', DeprecationWarning, 2)	×
265
266
267	def create_context(ctx: Optional[Context] = None, **kw) -> Context:	6✔
268	if ctx is None:	6✔
269	return Context(**kw)	6✔
270	else:
271	if 'cfg' not in kw:	×
272	kw['cfg'] = ctx.cfg	×
273	sub_ctx = Context(**kw)	×
274	sub_ctx.time.update(ctx.time)	×
275	# sub_ctx.phases.update(ctx.phases)
276	sub_ctx.biases.update(ctx.biases)	×
277	for k, v in ctx.phases_ext.items():	×
278	sub_ctx.phases_ext[k].update(v)	×
279
280	return sub_ctx	×

feihoo87 / waveforms / 7087368175

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous