6534953321

Committed 16 Oct 2023 02:19PM UTC coverage: 35.674% (-22.7%) from 58.421%

Build # 6534953321

Build Type

push

github

Committed by

feihoo87

Commit Message

fix Coveralls

Run Details

5913 of 16575 relevant lines covered (35.67%)

3.21 hits per line

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

83.24

/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, Union

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

feihoo87 / waveforms / 6534953321

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