assembler

Functions:

Name	Description
initialize	compiler context for current task
schedule	compile circuits to commands(saved in instruction)
assemble	assemble compiled instruction(see schedule) to corresponding devices
decode	decode target to hardware channel

initialize(tid: int, snapshot, **kwds)

compiler context for current task

Note

every task has its own context

Parameters:

Name	Type	Description	Default
tid	int	task id	required
snapshot	_type_	frozen snapshot for current task	required

Returns:

Name	Type	Description
ctx	Context	Context to be used in compilation

Source code in quark/runtime/assembler.py

def initialize(tid: int, snapshot, **kwds):
    """compiler context for current task

    Note:
        every task has its own context

    Args:
        tid (int): task id
        snapshot (_type_): frozen snapshot for current task

    Returns:
        ctx (Context): Context to be used in compilation

    """
    global ctx
    # logger.info(f'🔗 Task({tid}): initializing ...')

    if isinstance(snapshot, int):
        return os.getpid()

    ctx = create_context(tid, kwds.get('arch', 'baqis'), snapshot)
    if kwds.get('main', False):
        return ctx

schedule(sid: int, instruction: dict[str, list[tuple[str, str, Any, str]]], circuit: list, **kwds) -> tuple

compile circuits to commands(saved in instruction)

Parameters:

Name	Type	Description	Default
sid	int	step index(starts from 0)	required
instruction	dict	where commands are saved	required
circuit	list	qlisp circuit	required

Returns:

Name	Type	Description
tuple	tuple	instruction, extra arguments

Source code in quark/runtime/assembler.py

def schedule(sid: int, instruction: dict[str, list[tuple[str, str, Any, str]]], circuit: list, **kwds) -> tuple:
    """compile circuits to commands(saved in **instruction**)

    Args:
        sid (int): step index(starts from 0)
        instruction (dict): where commands are saved
        circuit (list): qlisp circuit

    Returns:
        tuple: instruction, extra arguments
    """
    logger.info(f'🔗 Step({sid}): compiling ...')

    compiled, datamap = Workflow.qcompile(circuit, **(kwds | {'ctx': ctx}))

    # merge loop body with compiled result
    for step, _cmds in compiled.items():
        if step in instruction:
            _cmds.extend(instruction[step])
            instruction[step] = _cmds  # .extend(_cmds)
        else:
            instruction[step] = _cmds

    assemble(sid, instruction)

    if sid == 0:
        kwds['clear'] = True
    logger.info(f'✅ Step({sid}): compiled!')

    return instruction, {'dataMap': datamap} | kwds

assemble(sid: int, instruction: dict[str, list[tuple[str, str, Any, str]]], **kw)

assemble compiled instruction(see schedule) to corresponding devices

Parameters:

Name	Type	Description	Default
sid	int	step index	required
instruction	dict[str, list[str, str, Any, str]]	see schedule	required

Raises:

Type	Description
TypeError	srate should be float, defaults to -1.0

Source code in quark/runtime/assembler.py

def assemble(sid: int, instruction: dict[str, list[tuple[str, str, Any, str]]], **kw):
    """assemble compiled instruction(see schedule) to corresponding devices

    Args:
        sid (int): step index
        instruction (dict[str, list[str, str, Any, str]]): see `schedule`

    Raises:
        TypeError: srate should be float, defaults to -1.0
    """

    try:
        # for s.write and s.read
        query = kw.get('ctx', ctx).query
    except AttributeError as e:
        query = ctx.query

    if sid < 0 and (atuo_clear := ctx.query('station', {}).get('auto_clear', {})):
        try:
            step = set.intersection(
                *(set(instruction), ['init', 'post'])).pop()
            instruction[step].extend([('WRITE', *cmd)
                                     for cmd in ctx.autofill(atuo_clear.get(step, []))])
        except KeyError:
            pass

    for step, operations in instruction.items():
        if not isinstance(operations, list):
            break
        scmd = {}
        for ctype, target, value, unit in operations:
            if step.lower() == 'update':
                ctx.update(target, value)
                continue

            if ctype not in ('READ', 'WRITE', 'WAIT'):
                logger.warning(f'Unknown command type: {ctype}!')
                continue

            cargs = {'sid': sid, 'target': target,
                     # 'shared': ctx.correct(query('etc.server.shared'), 0),
                     'filter': ctx.correct(query('etc.driver.filter'), [])}

            context = {}
            if 'CH' in target or ctype == 'WAIT':
                _target = target
            else:
                if not ctx.iscmd(target):
                    # logger.warning(f'Unknown target: {target}!')
                    continue
                try:
                    # logical channel to hardware channel
                    if target.endswith(('drive', 'probe', 'flux', 'acquire')):
                        try:
                            value = ctx.snapshot().cache.pop(target, value)
                        except Exception as e:
                            pass
                        context = deepcopy(query(target))
                        _target = context.pop('address', f'address: {target}')
                    else:
                        # old
                        context = query(target.split('.', 1)[0])
                        mapping = query('etc.driver.mapping')
                        _target = decode(target, context, mapping)
                except Exception as e:  # (ValueError, KeyError, AttributeError)
                    # logger.error(f'Failed to map {target}: {e}!')
                    continue

            if not (isinstance(_target, str) and _target and _target.count('.') == 2):
                logger.error(f'wrong target: {target}({_target})')
                continue

            # get sampling rate from device
            dev, channel, quantity = _target.split('.')
            srate = -1.0 if dev == 'Timer' else query(f'dev.{dev}.srate')

            # context设置, 用于calculator.calculate
            try:
                cargs['calibration'] = {
                    'srate': srate,
                    'end': context['waveform']['LEN'],
                    'offset': context.get('setting', {}).get('OFFSET', 0)
                } | context['calibration'][target.split('.')[-1]]
            except Exception as e:
                end = None
                if quantity == 'Waveform':
                    end = ctx.query('station', {}).get(
                        'waveform_length', 98e-6)
                cargs['calibration'] = {'end': end, 'srate': srate} | context

            # cmd = [ctype, value, unit, kwds]
            cmd = {'ctype': ctype, 'value': value,
                   'unit': unit, 'cargs': cargs}

            # Merge commands with the same channel
            try:
                if quantity == 'Waveform' and isinstance(cmd['value'], str):
                    cmd['value'] = Pulse.fromstr(cmd['value'])

                if _target in scmd and quantity == 'Waveform':
                    if isinstance(scmd[_target]['value'], str):
                        scmd[_target]['value'] = Pulse.fromstr(
                            scmd[_target]['value'])
                    scmd[_target]['value'] += cmd['value']

                    sh, st = scmd[_target]['cargs']['target'].split('.', 1)
                    ch, ct = cmd['cargs']['target'].split('.', 1)
                    mh = sorted(set(sh.split('_') + ch.split('_')))
                    scmd[_target]['cargs']['target'] = '_'.join(mh) + '.' + st
                else:
                    scmd[_target] = cmd
            except Exception as e:
                logger.warning(f'Channel[{_target}] mutiplexing error: {e}')
                scmd[_target] = cmd
        instruction[step] = scmd

decode(target: str, context: dict, mapping: dict = MAPPING) -> str

decode target to hardware channel

Parameters:

Name	Type	Description	Default
target	str	target to be decoded like Q0.setting.LO	required
context	dict	target location like Q0	required
mapping	dict	mapping relations. Defaults to MAPPING.	MAPPING

Raises:

Type	Description
KeyError	mapping not found
ValueError	channel not found

Returns:

Name	Type	Description
str	str	hardware channel like AD.CH1.TraceIQ

Source code in quark/runtime/assembler.py

def decode(target: str, context: dict, mapping: dict = MAPPING) -> str:
    """decode target to hardware channel

    Args:
        target (str): target to be decoded like **Q0.setting.LO**
        context (dict): target location like **Q0**
        mapping (dict, optional): mapping relations. Defaults to MAPPING.

    Raises:
        KeyError: mapping not found
        ValueError: channel not found

    Returns:
        str: hardware channel like **AD.CH1.TraceIQ**
    """
    try:
        mkey = target.split('.', 1)[-1].replace('.', '_')
        chkey, quantity = mapping[mkey].split('.', 1)
    except KeyError as e:
        raise KeyError(f'{e} not found in mapping!')

    try:
        channel = context.get('channel', {})[chkey]
    except KeyError as e:
        raise KeyError(f'{chkey} not found!')

    if channel is None:
        raise ValueError('ChannelNotFound')
    elif not isinstance(channel, str):
        raise TypeError(
            f'Wrong type of channel of {target}, string needed got {channel}')
    elif 'Marker' not in channel:
        channel = '.'.join((channel, quantity))

    return channel