app

about app

usefull functions for users to interact with QuarkServer and database

Classes:

Name	Description
Super	Super Admin Tool to interact with QuarkServer and database and so on

Super()

Bases: object

Super Admin Tool to interact with QuarkServer and database and so on

Methods:

Name	Description
init	Set path to quark.json
recipe	Create a recipe
login	Login to the QuarkServer
signup	Register a new user on the system
submit	Submit a task to a backend
getid	Get the id of a task with given idx(tid or rid)
snapshot	Get snapshot of a task with given idx(tid or rid).
rollback	Rollback the cfg with given idx(tid or rid)
result	Get data with given idx(tid or rid)
lookup	Lookup records in the database
update	Update item in the cfg
delete	Delete an item from the cfg
translate	Translate circuit to executable commands(i.e., waveforms or settings)
preview	Preview waveforms from commands
run	Execute experiment instruction sequence.
diff	Compare two snapshots or records
display	Print nested dict as a tree

Source code in quark/app/__init__.py

def __init__(self):
    self.__cache = FixedDict(maxlen=128)

init(path: str | Path = Path.cwd() / 'quark.json')

Set path to quark.json

Parameters:

Name	Type	Description	Default
path	str | Path	path to quark.json. Defaults to Path.cwd()/'quark.json'.	cwd() / 'quark.json'

Source code in quark/app/__init__.py

def init(self, path: str | Path = Path.cwd() / 'quark.json'):
    """Set path to `quark.json`

    Args:
        path (str | Path, optional): path to quark.json. Defaults to `Path.cwd()/'quark.json'`.
    """
    from quark.proxy import init
    init(path)

recipe(name: str = '', **kwds) -> Recipe

Create a recipe

Parameters:

Name	Type	Description	Default
name	str	name of the recipe. Defaults to ''.	''

Returns:

Name	Type	Description
Recipe	Recipe	a recipe object

Source code in quark/app/__init__.py

def recipe(self, name: str = '', **kwds) -> Recipe:
    """Create a recipe

    Args:
        name (str, optional): name of the recipe. Defaults to ''.

    Returns:
        Recipe: a recipe object
    """
    return Recipe(name, **kwds)

login(user: str = 'baqis', host: str = '127.0.0.1', port: int = 2088, verbose: bool = True)

Login to the QuarkServer

Parameters:

Name	Type	Description	Default
user	str	name of the user(same as signup). Defaults to 'baqis'.	'baqis'
host	str	server host. Defaults to '127.0.0.1'.	'127.0.0.1'
port	str	server port. Defaults to 2088.	2088
verbose	bool	print the login info. Defaults to True.	True

Source code in quark/app/__init__.py

def login(self, user: str = 'baqis', host: str = '127.0.0.1', port: int = 2088, verbose: bool = True):
    """Login to the QuarkServer

    Args:
        user (str, optional): name of the user(same as signup). Defaults to 'baqis'.
        host (str, optional): server host. Defaults to '127.0.0.1'.
        port (str, optional): server port. Defaults to 2088.
        verbose (bool, optional): print the login info. Defaults to True.
    """

    self._s = login(user, host, port, verbose)

    for mth in ['start', 'query', 'write', 'read', 'checkpoint', 'track', 'cancel', 'report', 'review', 'tail']:
        setattr(self, mth, getattr(self._s, mth))

signup(user: str, system: str, **kwds)

Register a new user on the system

Parameters:

Name	Type	Description	Default
user	str	name of the user	required
system	str	name of the system(i.e. the name of the cfg file)	required

Source code in quark/app/__init__.py

def signup(self, user: str, system: str, **kwds):
    """Register a new **user** on the **system**

    Args:
        user (str): name of the user
        system (str): name of the system(i.e. the name of the cfg file)
    """
    signup(user, system, **kwds)

submit(task: dict, block: bool = False, **kwds)

Submit a task to a backend

Parameters:

Name	Type	Description	Default
task	dict	description of a task generated by Recipe.	required
block	bool	block until the task is done if True	False

Kwds

Name	Type	Description
preview	list	real time display of the waveform
plot	bool	plot the result if True(1D or 2D), defaults to False.
backend	connection	connection to a backend, defaults to local machine.

Raises: TypeError: description

Source code in quark/app/__init__.py

def submit(self, task: dict, block: bool = False, **kwds):
    """Submit a task to a backend

    Args:
        task (dict): description of a task generated by `Recipe`.
        block (bool, optional): block until the task is done if True

    Keyword Arguments: Kwds
        preview (list): real time display of the waveform
        plot (bool): plot the result if True(1D or 2D), defaults to False.
        backend (connection): connection to a backend, defaults to local machine.
    Raises:
        TypeError: _description_
    """
    return submit(task, block, **kwds)

getid(idx: int = 0) -> int

Get the id of a task with given idx(tid or rid)

Parameters:

Name	Type	Description	Default
idx	int	tid or rid. Defaults to 0.	0

Returns:

Name	Type	Description
int	int	task id(tid)

Source code in quark/app/__init__.py

def getid(self, idx: int = 0) -> int:
    """Get the id of a task with given idx(**tid** or **rid**)

    Args:
        idx (int, optional): tid or rid. Defaults to 0.

    Returns:
        int: task id(tid)
    """
    try:
        if idx and self.addr[0] == '127.0.0.1':
            return int(get_tid_by_rid(idx))
        else:
            r = self.qs().getid(idx=idx)
            return int(r[1]) if idx else int(r)
    except Exception as e:
        logger.error(f'rid or tid is not found: {e}')

snapshot(idx: int = 0)

Get snapshot of a task with given idx(tid or rid). If idx is 0, current snapshot will be retrieved from QuarkServer.

Parameters:

Name	Type	Description	Default
idx	int	tid or rid. Defaults to 0.	0

Returns:

Name	Type	Description
dict		description

Source code in quark/app/__init__.py

def snapshot(self, idx: int = 0):
    """Get snapshot of a task with given idx(**tid** or **rid**). 
    If idx is 0, current snapshot will be retrieved from `QuarkServer`.

    Args:
        idx (int, optional): tid or rid. Defaults to 0.

    Returns:
        dict: _description_
    """
    if idx and self.addr[0] == '127.0.0.1':
        return get_config_by_tid(self.getid(idx))
    else:
        return self.qs().snapshot(tid=self.getid(idx) if idx else idx)

rollback(idx: int)

Rollback the cfg with given idx(tid or rid)

Parameters:

Name	Type	Description	Default
idx	int	tid or rid	required

Source code in quark/app/__init__.py

def rollback(self, idx: int):
    """Rollback the cfg with given idx(**tid** or **rid**)

    Args:
        idx (int): tid or rid
    """
    if idx:
        rollback(self.getid(idx))
    else:
        raise ValueError('rid or tid is required!')

result(idx: int, task: bool = False, clear: bool = False, **kwds)

Get data with given idx(tid or rid)

Parameters:

Name	Type	Description	Default
idx	int	tid or rid	required
task	bool	return task info if True. Defaults to False.	False
clear	bool	clear cache if True. Defaults to False.	False

Kwds

Name	Type	Description
plot	bool	plot the result in QuarkStudio after the data is loaded(1D or 2D).

Returns:

Name	Type	Description
dict		data & meta

Source code in quark/app/__init__.py

def result(self, idx: int, task: bool = False, clear: bool = False, **kwds):
    """Get data with given idx(**tid** or **rid**)

    Args:
        idx (int): tid or rid
        task (bool, optional): return task info if True. Defaults to False.
        clear (bool, optional): clear cache if True. Defaults to False.

    Keyword Arguments: Kwds
        plot (bool, optional): plot the result in QuarkStudio after the data is loaded(1D or 2D).

    Returns:
        dict: data & meta
    """
    if clear:
        self.__cache.clear()
        logger.info('Cache cleared.')

    if idx in self.__cache:
        # logger.info(f'Cache hit: {idx}')
        r = self.__cache[idx]
    else:
        if self.addr[0] == '127.0.0.1':
            r = get_data_by_tid(self.getid(idx), **kwds)
        else:
            r = self.qs().load(self.getid(idx))
            try:
                from ._db import reshape

                shape = r['meta']['other']['shape']
                r['data'] = {k: reshape(np.asarray(v), shape)
                             for k, v in r['data'].items()}
            except Exception as e:
                logger.error(f'Failed to reshape data: {e}')
        self.__cache[idx] = r

    if task:
        return r.get('task', {})
    else:
        return {'data': r['data'], 'meta': r['meta']}

lookup(start: str = '', end: str = '', name: str = '')

Lookup records in the database

Parameters:

Name	Type	Description	Default
start	str	start date. Defaults to ''.	''
end	str	end date. Defaults to ''.	''
name	str	task name. Defaults to ''.	''

Returns:

Name	Type	Description
_type_		description

Source code in quark/app/__init__.py

def lookup(self, start: str = '', end: str = '', name: str = ''):
    """Lookup records in the database

    Args:
        start (str, optional): start date. Defaults to ''.
        end (str, optional): end date. Defaults to ''.
        name (str, optional): task name. Defaults to ''.

    Returns:
        _type_: _description_
    """
    if self.addr[0] == '127.0.0.1':
        return lookup(start, end, name)
    else:
        return lookup(records=self.qs().load(0))

update(path: str, value, failed: list = [])

Update item in the cfg

Parameters:

Name	Type	Description	Default
path	str	dot-separated keys like 'usr.station.name'	required
value	_type_	value to update	required
failed	list	description. Defaults to [].	[]

Source code in quark/app/__init__.py

def update(self, path: str, value, failed: list = []):
    """Update item in the cfg

    Args:
        path (str): dot-separated keys like 'usr.station.name'
        value (_type_): value to update
        failed (list, optional): _description_. Defaults to [].
    """
    qs = self.qs()
    rs: str = qs.update(path, value)
    if rs.startswith('Failed'):
        if path.count('.') == 0:
            qs.create(path, value)
        else:
            path, _f = path.rsplit('.', 1)
            failed.append((_f, value))
            self.update(path, {}, failed)

    while failed:
        _f, v = failed.pop()
        path = f'{path}.{_f}'
        qs.update(path, v)

delete(path: str)

Delete an item from the cfg

Parameters:

Name	Type	Description	Default
path	str	dot-separated keys like 'usr.station.name'	required

Source code in quark/app/__init__.py

def delete(self, path: str):
    """Delete an item from the cfg

    Args:
        path (str): dot-separated keys like 'usr.station.name'
    """
    qs = self.qs()
    if path.count('.') > 0:
        qs.delete(path)
    else:
        qs.remove(path)

translate(circuit: list = [(('Measure', 0), 'Q0')], cfg: dict = {}, **kwds) -> tuple

Translate circuit to executable commands(i.e., waveforms or settings)

Parameters:

Name	Type	Description	Default
circuit	list	qlisp circuit. Defaults to [(('Measure', 0), 'Q0')].	[(('Measure', 0), 'Q0')]
cfg	dict	parameters of qubits in the circuit. Defaults to {} Defaults.	{}

Returns:

Name	Type	Description
tuple	tuple	context that contains cfg, translated result

Example

circuit = [('X', 'Q0'), (('Measure', 0), 'Q0')]
cfg = s.snapshot()  # get current cfg from QuarkServer
ctx, (cmds, dmap) = s.translate(circuit, cfg=cfg)

Source code in quark/app/__init__.py

def translate(self, circuit: list = [(('Measure', 0), 'Q0')], cfg: dict = {}, **kwds) -> tuple:
    """Translate circuit to executable commands(i.e., waveforms or settings)

    Args:
        circuit (list, optional): qlisp circuit. Defaults to [(('Measure', 0), 'Q0')].
        cfg (dict, optional): parameters of qubits in the circuit. Defaults to {} Defaults.

    Returns:
        tuple: context that contains cfg, translated result

    Example:
        ``` {.py3 linenums="1"}
        circuit = [('X', 'Q0'), (('Measure', 0), 'Q0')]
        cfg = s.snapshot()  # get current cfg from QuarkServer
        ctx, (cmds, dmap) = s.translate(circuit, cfg=cfg)
        ```
    """
    if not cfg:
        # for more details, see https://quarkstudio.readthedocs.io/en/latest/usage/code/cfg2/
        cfg = {
            'Q0': {
                'Measure': {
                    'duration': 4e-06,
                    'amp': 0.019,
                    'frequency': 6964370000.0,
                    'weight': 'square(2e-6)>>2e-6',
                    'phi': -2.636421695283167,
                    'threshold': 8502633802.265065,
                    'ring_up_amp': 0.024,
                    'ring_up_waist': 0.006,
                    'ring_up_time': 6e-07
                },
                'acquire': {'address': 'AD.CH13.IQ', 'TRIGD': 0, 'srate': 2e9},
                'probe': {'address': 'AWG.CH2.Waveform', 'delay': 0, 'srate': 6e9}
            },
            'station': {
                'triggercmds': ['Trigger.CH1.TRIG'],
                'lib': 'glib.gates.u3rcp',
                'arch': 'rcp',
                'align_right': False,
                'waveform_length': 1.8e-05
            }
        }
    return translate(circuit, cfg, **kwds)

preview(cmds: dict, keys: tuple[str] = ('',), calibrate: bool = True, start: float = 0, end: float = 0, srate: float = 0, unit: float = 1e-06, offset: float = 0, space: float = 0, ax=None)

Preview waveforms from commands

Parameters:

Name	Type	Description	Default
cmds	dict	command list generated by the compiler.	required
keys	tuple[str]	keywords of waveforms to to preview. Defaults to ('',).	('',)
calibrate	bool	calculate distortion if True. Defaults to True.	True
start	float	start time. Defaults to 0.	0
end	float	end time. Defaults to 0.	0
srate	float	sampling rate. Defaults to 0.	0
unit	float	unit of time. Defaults to 1e-6.	1e-06
offset	float	vertical offset between curves. Defaults to 0.	0
space	float	spacing between the curve and the y-axis. Defaults to 0.	0
ax	Axes	the Axes object (ax) in Matplotlib. Defaults to None.	None

Returns:

Name	Type	Description
dict		Sampled waveform data

Example

# cmds is generated by s.translate
wf = s.preview(cmds, keys=['Q0'], end=100e-6)

Source code in quark/app/__init__.py

def preview(self, cmds: dict, keys: tuple[str] = ('',), calibrate: bool = True,
            start: float = 0, end: float = 0, srate: float = 0,
            unit: float = 1e-6, offset: float = 0, space: float = 0, ax=None):
    """Preview waveforms from commands

    Args:
        cmds (dict): command list generated by the compiler.
        keys (tuple[str], optional): keywords of waveforms to to preview. Defaults to ('',).
        calibrate (bool, optional): calculate distortion if True. Defaults to True.
        start (float, optional): start time. Defaults to 0.
        end (float, optional): end time. Defaults to 0.
        srate (float, optional): sampling rate. Defaults to 0.
        unit (float, optional): unit of time. Defaults to 1e-6.
        offset (float, optional): vertical offset between curves. Defaults to 0.
        space (float, optional): spacing between the curve and the y-axis. Defaults to 0.
        ax (Axes, optional): the Axes object (ax) in Matplotlib. Defaults to None.

    Returns:
        dict: Sampled waveform data

    Example:
        ``` {.py3 linenums="1"}
        # cmds is generated by s.translate
        wf = s.preview(cmds, keys=['Q0'], end=100e-6)
        ```
    """
    return preview(cmds, keys, calibrate, start, end, srate, unit, offset, space, ax)

run(cmds: dict, dev: dict = {'AD': {'addr': '192.168.1.2', 'name': 'dev.VirtualDevice', 'type': 'driver'}, 'AWG': {'host': '192.168.1.3', 'port': 40001, 'type': 'remote'}, 'Trigger': {'addr': '192.168.1.4', 'name': 'dev.VirtualDevice', 'type': 'driver'}}, verbose: bool = True)

Execute experiment instruction sequence.

Parameters:

Name	Type	Description	Default
cmds	dict	Command dictionary from task.step or s.translate.	required
dev	dict	Device dictionary.	{'AD': {'addr': '192.168.1.2', 'name': 'dev.VirtualDevice', 'type': 'driver'}, 'AWG': {'host': '192.168.1.3', 'port': 40001, 'type': 'remote'}, 'Trigger': {'addr': '192.168.1.4', 'name': 'dev.VirtualDevice', 'type': 'driver'}}
verbose	bool	Whether to print execution information. Defaults to True.	True

Returns:

Name	Type	Description
dict		Dictionary containing read results.

Raises:

Type	Description
KeyError	When the target device is not in the device list.

Source code in quark/app/__init__.py

def run(self, cmds: dict, dev: dict = {
    'AD': {
        "addr": "192.168.1.2",
        "name": "dev.VirtualDevice",
        "type": "driver"
    },
    'AWG': {
        "host": "192.168.1.3",
        "port": 40001,
        "type": "remote"
    },
    'Trigger': {
        "addr": "192.168.1.4",
        "name": "dev.VirtualDevice",
        "type": "driver"
    }
}, verbose: bool = True):
    """Execute experiment instruction sequence.

    Args:
        cmds (dict): Command dictionary from `task.step` or `s.translate`.
        dev (dict): Device dictionary.
        verbose (bool): Whether to print execution information. Defaults to True.

    Returns:
        dict: Dictionary containing read results.

    Raises:
        KeyError: When the target device is not in the device list.
    """
    dhs = {}
    for alias, info in dev.items():
        if info['type'] == 'driver':
            driver = import_module(info['name']).Driver(**info)
            driver.open()
        elif info['type'] == 'remote':
            driver = connect(alias,
                             host=info['host'],
                             port=info['port'],
                             timeout=3.0
                             )
        else:
            raise ValueError('unsupported type of device')
        dhs[alias] = driver
        logger.info(f'{alias}: {driver.info()}')

    result = {}
    for step, ops in cmds.items():
        for target, params in ops.items():
            d, ch, q = target.split('.')
            try:
                ch = int(ch[2:])
            except ValueError as e:
                ch = ch[2:]
            if d in dhs:
                if verbose:
                    print(f"Execute: {step}->{target}: {params['value']}")

                if step.lower() == 'read':
                    result[target] = dhs[d].getValue(q, ch=ch)
                else:
                    dhs[d].setValue(q, params['value'], ch=ch)
            else:
                logger.error(f"Device {d}({target}) not in device list")
    return result

diff(new: int | dict | str, old: int | dict | str, fmt: str = 'dict', ignore: list[str] = ['unit', 'sid'], **kwds)

Compare two snapshots or records

Parameters:

Name	Type	Description	Default
new	int | dict | str	new snapshot or record id or dict or filepath or text	required
old	int | dict | str	old snapshot or record id or dict or filepath or text	required
fmt	str	format of the output. Defaults to 'dict'.	'dict'
ignore	list[str]	keys to be ignored. Defaults to ['unit', 'sid'].	['unit', 'sid']

Returns:

Name	Type	Description
_type_		description

Source code in quark/app/__init__.py

def diff(self, new: int | dict | str, old: int | dict | str, fmt: str = 'dict', ignore: list[str] = ['unit', 'sid'], **kwds):
    """Compare two snapshots or records

    Args:
        new (int | dict | str): new snapshot or record id or dict or filepath or text
        old (int | dict | str): old snapshot or record id or dict or filepath or text
        fmt (str, optional): format of the output. Defaults to 'dict'.
        ignore (list[str], optional): keys to be ignored. Defaults to ['unit', 'sid'].

    Returns:
        _type_: _description_
    """
    if isinstance(new, str) and isinstance(old, str):
        return 'not supported type of diff'

    return diff(new, old, fmt, ignore)

display(d: dict | np.ndarray = {}, filename: str = '', title='root', **kwds)

Print nested dict as a tree

Parameters:

Name	Type	Description	Default
d	dict	dict to be printed as tree.	{}
filename	str	hdf5 or zarr file. Defaults to ''.	''
title	str	root name. Defaults to "root".	'root'

Returns:

Name	Type	Description
_type_		description

Source code in quark/app/__init__.py

def display(self, d: dict | np.ndarray = {}, filename: str = '', title="root", **kwds):
    """Print nested dict as a tree

    Args:
        d (dict): dict to be printed as tree.
        filename (str, optional): hdf5 or zarr file. Defaults to ''.
        title (str, optional): root name. Defaults to "root".

    Returns:
        _type_: _description_
    """

    if filename and filename.endswith(('hdf5', 'zarr')):
        from ._db import get_tree_of_file
        d = get_tree_of_file(filename)
        if not d:
            return

    import rich
    rich.print(d)